diff --git a/go.mod b/go.mod index 5cd43d6c5..a609a47be 100644 --- a/go.mod +++ b/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,6 +36,7 @@ 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 @@ -43,8 +44,6 @@ require ( 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 diff --git a/loopd/daemon.go b/loopd/daemon.go index 9feca79e3..7212933ad 100644 --- a/loopd/daemon.go +++ b/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(): diff --git a/staticaddr/deposit/fsm.go b/staticaddr/deposit/fsm.go index 2d1751ffc..1fad4dc20 100644 --- a/staticaddr/deposit/fsm.go +++ b/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) } } diff --git a/staticaddr/withdraw/interface.go b/staticaddr/withdraw/interface.go index dd6587e82..065a6f8d5 100644 --- a/staticaddr/withdraw/interface.go +++ b/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 { diff --git a/staticaddr/withdraw/manager.go b/staticaddr/withdraw/manager.go index 0cd4adade..408d0338e 100644 --- a/staticaddr/withdraw/manager.go +++ b/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()) }