Merge pull request #967 from starius/neutrino-fixes

staticaddr: neutrino fixes

Author: starius

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/btcsuite/btcd/btcutil v1.1.5
 	github.com/btcsuite/btcd/btcutil/psbt v1.1.10
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0
-	github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c // indirect
+	github.com/btcsuite/btclog/v2 v2.0.1-0.20250110154127-3ae4bf1cb318
 	github.com/btcsuite/btcwallet v0.16.13
 	github.com/btcsuite/btcwallet/wtxmgr v1.5.6
 	github.com/davecgh/go-spew v1.1.1
@@ -36,15 +36,14 @@ require (
 	github.com/urfave/cli v1.22.14
 	go.etcd.io/bbolt v1.3.11
 	golang.org/x/net v0.38.0
+	golang.org/x/sync v0.12.0
 	google.golang.org/grpc v1.64.1
 	google.golang.org/protobuf v1.34.2
 	gopkg.in/macaroon-bakery.v2 v2.3.0
 	gopkg.in/macaroon.v2 v2.1.0
 	modernc.org/sqlite v1.34.5
 )
 
-require github.com/btcsuite/btclog/v2 v2.0.1-0.20250110154127-3ae4bf1cb318
-
 require (
 	dario.cat/mergo v1.0.1 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
@@ -56,6 +55,7 @@ require (
 	github.com/aead/chacha20 v0.0.0-20180709150244-8b13a72661da // indirect
 	github.com/aead/siphash v1.0.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c // indirect
 	github.com/btcsuite/btcwallet/wallet/txauthor v1.3.5 // indirect
 	github.com/btcsuite/btcwallet/wallet/txrules v1.2.2 // indirect
 	github.com/btcsuite/btcwallet/wallet/txsizes v1.2.5 // indirect
@@ -185,7 +185,6 @@ require (
 	golang.org/x/crypto v0.36.0 // indirect
 	golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 // indirect
 	golang.org/x/mod v0.21.0 // indirect
-	golang.org/x/sync v0.12.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/term v0.30.0 // indirect
 	golang.org/x/text v0.23.0 // indirect

loopd/daemon.go

@@ -922,10 +922,15 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 			}
 		}()
 
+		// We need a higher timeout here, because withdrawalManager
+		// publishes transactions and each PublishTransaction call can
+		// wait for getting inv messages from a peer (neutrino).
+		const withdrawalManagerTimeout = time.Minute
+
 		// Wait for the static address withdrawal manager to be ready
 		// before starting the grpc server.
 		timeOutCtx, cancel := context.WithTimeout(
-			d.mainCtx, initManagerTimeout,
+			d.mainCtx, withdrawalManagerTimeout,
 		)
 		select {
 		case <-timeOutCtx.Done():

staticaddr/deposit/fsm.go

@@ -399,7 +399,7 @@ func (f *FSM) updateDeposit(ctx context.Context,
 
 	err := f.cfg.Store.UpdateDeposit(ctx, f.deposit)
 	if err != nil {
-		f.Errorf("unable to update deposit: %w", err)
+		f.Errorf("unable to update deposit: %v", err)
 	}
 }
 

staticaddr/withdraw/interface.go

@@ -8,7 +8,6 @@ import (
 	"github.com/lightninglabs/loop/staticaddr/address"
 	"github.com/lightninglabs/loop/staticaddr/deposit"
 	"github.com/lightninglabs/loop/staticaddr/script"
-	"github.com/lightningnetwork/lnd/lnwallet"
 )
 
 // Store is the database interface that is used to store and retrieve
@@ -32,10 +31,6 @@ type AddressManager interface {
 	// GetStaticAddress returns the deposit address for the given
 	// client and server public keys.
 	GetStaticAddress(ctx context.Context) (*script.StaticAddress, error)
-
-	// ListUnspent returns a list of utxos at the static address.
-	ListUnspent(ctx context.Context, minConfs,
-		maxConfs int32) ([]*lnwallet.Utxo, error)
 }
 
 type DepositManager interface {

staticaddr/withdraw/manager.go

@@ -26,6 +26,7 @@ import (
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
+	"golang.org/x/sync/errgroup"
 )
 
 var (
@@ -226,44 +227,65 @@ func (m *Manager) recoverWithdrawals(ctx context.Context) error {
 	}
 
 	// Group the deposits by their finalized withdrawal transaction.
-	depositsByWithdrawalTx := make(map[*wire.MsgTx][]*deposit.Deposit)
+	depositsByWithdrawalTx := make(map[chainhash.Hash][]*deposit.Deposit)
+	hash2tx := make(map[chainhash.Hash]*wire.MsgTx)
 	for _, d := range activeDeposits {
 		withdrawalTx := d.FinalizedWithdrawalTx
 		if withdrawalTx == nil {
 			continue
 		}
+		txid := withdrawalTx.TxHash()
+		hash2tx[txid] = withdrawalTx
 
-		depositsByWithdrawalTx[withdrawalTx] = append(
-			depositsByWithdrawalTx[withdrawalTx], d,
+		depositsByWithdrawalTx[txid] = append(
+			depositsByWithdrawalTx[txid], d,
 		)
 	}
 
+	// Publishing a transaction can take a while in neutrino mode, so
+	// do it in parallel.
+	eg := &errgroup.Group{}
+
 	// We can now reinstate each cluster of deposits for a withdrawal.
-	for finalizedWithdrawalTx, deposits := range depositsByWithdrawalTx {
-		tx := finalizedWithdrawalTx
-		err = m.cfg.DepositManager.TransitionDeposits(
-			ctx, deposits, deposit.OnWithdrawInitiated,
-			deposit.Withdrawing,
-		)
-		if err != nil {
-			return err
-		}
+	for txid, deposits := range depositsByWithdrawalTx {
+		eg.Go(func() error {
+			err := m.cfg.DepositManager.TransitionDeposits(
+				ctx, deposits, deposit.OnWithdrawInitiated,
+				deposit.Withdrawing,
+			)
+			if err != nil {
+				return err
+			}
 
-		_, err = m.publishFinalizedWithdrawalTx(ctx, tx)
-		if err != nil {
-			return err
-		}
+			tx, ok := hash2tx[txid]
+			if !ok {
+				return fmt.Errorf("can't find tx %v", txid)
+			}
 
-		err = m.handleWithdrawal(
-			ctx, deposits, tx.TxHash(), tx.TxOut[0].PkScript,
-		)
-		if err != nil {
-			return err
-		}
+			_, err = m.publishFinalizedWithdrawalTx(ctx, tx)
+			if err != nil {
+				return err
+			}
 
-		m.mu.Lock()
-		m.finalizedWithdrawalTxns[tx.TxHash()] = tx
-		m.mu.Unlock()
+			err = m.handleWithdrawal(
+				ctx, deposits, tx.TxHash(),
+				tx.TxOut[0].PkScript,
+			)
+			if err != nil {
+				return err
+			}
+
+			m.mu.Lock()
+			m.finalizedWithdrawalTxns[tx.TxHash()] = tx
+			m.mu.Unlock()
+
+			return nil
+		})
+	}
+
+	// Wait for all goroutines to report back.
+	if err := eg.Wait(); err != nil {
+		return fmt.Errorf("error recovering withdrawals: %w", err)
 	}
 
 	return nil
@@ -558,6 +580,9 @@ func (m *Manager) publishFinalizedWithdrawalTx(ctx context.Context,
 			"withdrawal tx is nil")
 	}
 
+	log.Debugf("Publishing deposit withdrawal with txid: %v ...",
+		tx.TxHash())
+
 	txLabel := fmt.Sprintf("deposit-withdrawal-%v", tx.TxHash())
 
 	// Publish the withdrawal sweep transaction.
@@ -577,7 +602,7 @@ func (m *Manager) publishFinalizedWithdrawalTx(ctx context.Context,
 			return false, nil
 		}
 	} else {
-		log.Debugf("published deposit withdrawal with txid: %v",
+		log.Debugf("Published deposit withdrawal with txid: %v",
 			tx.TxHash())
 	}