diff --git a/release_notes.md b/release_notes.md index 38afc603e..5813dd193 100644 --- a/release_notes.md +++ b/release_notes.md @@ -20,4 +20,13 @@ This file tracks release notes for the loop client. #### Bug Fixes +* [Static Address Withdrawals: Add retry and reorg handling for withdrawal + confirmations](https://github.com/lightninglabs/loop/issues/1087). + Previously, if `RegisterSpendNtfn` or `RegisterConfirmationsNtfn` failed, + the withdrawal monitoring would silently stop. Additionally, reorgs were not + handled, and state was not persisted before monitoring setup, causing + recovery failures on restart. This fix adds automatic retry on next block + for registration failures, proper reorg detection via `WithReOrgChan`, and + reorders state persistence to occur before monitoring setup. + #### Maintenance diff --git a/staticaddr/withdraw/manager.go b/staticaddr/withdraw/manager.go index 99fddd267..ea5677b0b 100644 --- a/staticaddr/withdraw/manager.go +++ b/staticaddr/withdraw/manager.go @@ -22,6 +22,7 @@ import ( "github.com/lightninglabs/loop/staticaddr/deposit" "github.com/lightninglabs/loop/staticaddr/staticutil" staticaddressrpc "github.com/lightninglabs/loop/swapserverrpc" + "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/funding" "github.com/lightningnetwork/lnd/input" "github.com/lightningnetwork/lnd/lnrpc" @@ -238,6 +239,18 @@ func (m *Manager) recoverWithdrawals(ctx context.Context) error { return err } + // If there are no withdrawing deposits, we have nothing to recover. + if len(withdrawingDeposits) == 0 { + return nil + } + + // Get static address pkScript for handleWithdrawal. We fetch this once + // before grouping deposits and spawning goroutines. + addrParams, err := m.cfg.AddressManager.GetStaticAddressParameters(ctx) + if err != nil { + return fmt.Errorf("failed to get address params: %w", err) + } + // Group the deposits by their finalized withdrawal transaction. depositsByWithdrawalTx := make(map[chainhash.Hash][]*deposit.Deposit) hash2tx := make(map[chainhash.Hash]*wire.MsgTx) @@ -279,13 +292,12 @@ func (m *Manager) recoverWithdrawals(ctx context.Context) error { return err } - err = m.handleWithdrawal( + // Spawn goroutine to monitor for spend/confirmation. + // Best-effort: tx is published, caller continues. + m.handleWithdrawal( ctx, deposits, tx.TxHash(), - tx.TxOut[0].PkScript, + tx.TxOut[0].PkScript, addrParams.PkScript, ) - if err != nil { - return err - } m.mu.Lock() m.finalizedWithdrawalTxns[tx.TxHash()] = tx @@ -443,27 +455,9 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, "pkscript: %w", err) } - // If this is the first time this cluster of deposits is withdrawn, we - // start a goroutine that listens for the spent of the first input of - // the withdrawal transaction. - // Since we ensure above that the same ensemble of deposits is - // republished in case of a fee bump, it suffices if only one spent - // notifier is run. - if allDeposited { - // Persist info about the finalized withdrawal. - err = m.cfg.Store.CreateWithdrawal(ctx, deposits) - if err != nil { - log.Errorf("Error persisting "+ - "withdrawal: %v", err) - } - - err = m.handleWithdrawal( - ctx, deposits, finalizedTx.TxHash(), withdrawalPkScript, - ) - if err != nil { - return "", "", err - } - } + // Persist state before starting the spend monitor. If monitoring setup + // fails, recoverWithdrawals will find deposits in Withdrawing state on + // restart and retry. // If a previous withdrawal existed across the selected deposits, and // it isn't the same as the new withdrawal, we remove it from the @@ -500,7 +494,7 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, ctx, deposits, deposit.OnWithdrawInitiated, deposit.Withdrawing, ) if err != nil { - return "", "", fmt.Errorf("failed to transition deposits %w", + return "", "", fmt.Errorf("failed to transition deposits: %w", err) } @@ -509,7 +503,35 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, err = m.cfg.DepositManager.UpdateDeposit(ctx, d) if err != nil { return "", "", fmt.Errorf("failed to update "+ - "deposit %w", err) + "deposit: %w", err) + } + } + + // allDeposited is true when deposits were in Deposited state (vs + // Withdrawing), meaning this is the first withdrawal attempt. Start the + // spend monitor goroutine. Fee bumps reuse the existing monitor since + // we ensure the same deposit ensemble is used for republishing. + if allDeposited { + // Persist info about the finalized withdrawal. + err = m.cfg.Store.CreateWithdrawal(ctx, deposits) + if err != nil { + log.Errorf("Error persisting withdrawal: %v", err) + } + + // Get static address pkScript for spend registration. + addrParams, err := m.cfg.AddressManager.GetStaticAddressParameters(ctx) + if err == nil { + // Spawn goroutine to monitor for spend/confirmation. + // Best-effort: tx is published, caller continues. + m.handleWithdrawal( + ctx, deposits, finalizedTx.TxHash(), + withdrawalPkScript, addrParams.PkScript, + ) + } else { + log.Errorf("Failed to get address params: %v", err) + // Return success - tx published and state persisted. + // handleWithdrawal will be picked up on restart via + // recovery. } } @@ -654,97 +676,228 @@ func (m *Manager) publishFinalizedWithdrawalTx(ctx context.Context, } // handleWithdrawal starts a goroutine that listens for the spent of the first -// input of the withdrawal transaction. +// input of the withdrawal transaction. It handles retry on registration +// failure and reorg detection. func (m *Manager) handleWithdrawal(ctx context.Context, deposits []*deposit.Deposit, txHash chainhash.Hash, - withdrawalPkscript []byte) error { - - addrParams, err := m.cfg.AddressManager.GetStaticAddressParameters(ctx) - if err != nil { - log.Errorf("error retrieving address params: %v", err) - - return fmt.Errorf("withdrawal failed") - } + withdrawalPkscript []byte, staticAddrPkScript []byte) { d := deposits[0] - spentChan, errChan, err := m.cfg.ChainNotifier.RegisterSpendNtfn( - ctx, &d.OutPoint, addrParams.PkScript, - int32(d.ConfirmationHeight), - ) - if err != nil { - return fmt.Errorf("unable to register spend ntfn: %w", err) - } go func() { - select { - case spentTx := <-spentChan: - spendingHeight := uint32(spentTx.SpendingHeight) - // If the transaction received one confirmation, we - // ensure re-org safety by waiting for some more - // confirmations. - confChan, confErrChan, err := - m.cfg.ChainNotifier.RegisterConfirmationsNtfn( - ctx, spentTx.SpenderTxHash, - withdrawalPkscript, MinConfs, - int32(m.initiationHeight.Load()), + // Register for block notifications (for retry-on-next-block). + blockChan, blockErrChan, err := + m.cfg.ChainNotifier.RegisterBlockEpochNtfn(ctx) + if err != nil { + log.Errorf("handleWithdrawal: failed to register "+ + "block epoch: %v", err) + return + } + + // Loop for spend registration retry and reorg recovery. + for { + // Create sub-context and fresh reorg channel per + // iteration to ensure clean subscription cleanup. + // When using WithReOrgChan, lndclient's goroutine + // keeps running until context is canceled. + iterCtx, iterCancel := context.WithCancel(ctx) + reorgChan := make(chan struct{}, 1) + + // Register spend notification with reorg channel. + spendChan, spendErrChan, regErr := + m.cfg.ChainNotifier.RegisterSpendNtfn( + iterCtx, &d.OutPoint, staticAddrPkScript, + int32(d.ConfirmationHeight), + lndclient.WithReOrgChan(reorgChan), ) - if err != nil { - // TODO(#1087): Retry registration on - // next block instead of giving up. - log.Errorf("Error registering confirmation "+ - "notification: %v", err) + if regErr != nil { + iterCancel() + log.Errorf("RegisterSpendNtfn failed for %v, "+ + "retrying on next block: %v", + d.OutPoint, regErr) + + select { + case _, ok := <-blockChan: + if !ok { + return + } + continue + case err := <-blockErrChan: + log.Errorf("Block subscription "+ + "error: %v", err) + return + case <-ctx.Done(): + return + } + } - return + if !m.handleWithdrawalEvents( + iterCtx, deposits, txHash, + withdrawalPkscript, staticAddrPkScript, + spendChan, spendErrChan, reorgChan, + blockChan, blockErrChan, + ) { + // Cancel subscription before re-registering. + iterCancel() + continue } + iterCancel() + return + } + }() +} + +// handleWithdrawalEvents handles spend/confirmation events. +// Returns true if complete, false if should re-register. +func (m *Manager) handleWithdrawalEvents(ctx context.Context, + deposits []*deposit.Deposit, txHash chainhash.Hash, + withdrawalPkscript []byte, staticAddrPkScript []byte, + spendChan <-chan *chainntnfs.SpendDetail, + spendErrChan <-chan error, + reorgChan <-chan struct{}, + blockChan <-chan int32, + blockErrChan <-chan error) bool { + + select { + case spentTx, ok := <-spendChan: + if !ok { + return false + } + spendingHeight := uint32(spentTx.SpendingHeight) + + // Wait for confirmations with reorg handling. + return m.waitForConfirmations( + ctx, deposits, txHash, withdrawalPkscript, + staticAddrPkScript, spentTx, spendingHeight, + blockChan, blockErrChan, + ) + + case err := <-spendErrChan: + log.Errorf("Spend notification error for %v, "+ + "re-registering: %v", txHash, err) + return false // Re-register + + case <-reorgChan: + log.Warnf("Reorg detected before spend for withdrawal %v", + txHash) + return false // Re-register + + case err := <-blockErrChan: + log.Errorf("Block subscription error: %v", err) + return true // Exit + + case <-ctx.Done(): + return true // Exit + } +} + +// waitForConfirmations waits for withdrawal confirmation with reorg handling. +func (m *Manager) waitForConfirmations(ctx context.Context, + deposits []*deposit.Deposit, txHash chainhash.Hash, + withdrawalPkscript []byte, staticAddrPkScript []byte, + spentTx *chainntnfs.SpendDetail, spendingHeight uint32, + blockChan <-chan int32, blockErrChan <-chan error) bool { + + for { + // Create sub-context and fresh reorg channel per iteration + // to ensure clean subscription cleanup. + confCtx, confCancel := context.WithCancel(ctx) + confReorgChan := make(chan struct{}, 1) + + confChan, confErrChan, err := + m.cfg.ChainNotifier.RegisterConfirmationsNtfn( + confCtx, spentTx.SpenderTxHash, + withdrawalPkscript, MinConfs, + int32(m.initiationHeight.Load()), + lndclient.WithReOrgChan(confReorgChan), + ) + if err != nil { + confCancel() + log.Errorf("RegisterConfirmationsNtfn failed, "+ + "retrying on next block: %v", err) select { - case tx := <-confChan: - err = m.cfg.DepositManager.TransitionDeposits( - ctx, deposits, deposit.OnWithdrawn, - deposit.Withdrawn, - ) - if err != nil { - log.Errorf("Error transitioning "+ - "deposits: %v", err) + case _, ok := <-blockChan: + if !ok { + return true } + continue + case err := <-blockErrChan: + log.Errorf("Block subscription error: %v", err) + return true + case <-ctx.Done(): + return true + } + } + + select { + case tx, ok := <-confChan: + if !ok { + // Channel closed, retry registration. + confCancel() + continue + } + // Withdrawal confirmed - transition state. + err = m.cfg.DepositManager.TransitionDeposits( + ctx, deposits, deposit.OnWithdrawn, + deposit.Withdrawn, + ) + if err != nil { + log.Errorf("Error transitioning deposits: %v", + err) + } - // Remove the withdrawal tx from the active - // withdrawals to stop republishing it on block - // arrivals. + // Remove the withdrawal tx from the active withdrawals + // to stop republishing it on block arrivals. Use + // SpenderTxHash (not the original txHash) to handle + // RBF: when fee-bumped, the confirmed tx has a + // different hash than the original. + if spentTx.SpenderTxHash != nil { m.mu.Lock() - delete(m.finalizedWithdrawalTxns, txHash) + delete(m.finalizedWithdrawalTxns, *spentTx.SpenderTxHash) m.mu.Unlock() + } - // Persist info about the finalized withdrawal. - err = m.cfg.Store.UpdateWithdrawal( - ctx, deposits, tx.Tx, spendingHeight, - addrParams.PkScript, - ) - if err != nil { - log.Errorf("Error persisting "+ - "withdrawal: %v", err) - } - - case err := <-confErrChan: - // TODO(#1087): Handle reorgs by retrying - // confirmation registration on next block. - log.Errorf("Error waiting for confirmation: %v", + // Persist info about the finalized withdrawal. + err = m.cfg.Store.UpdateWithdrawal( + ctx, deposits, tx.Tx, spendingHeight, + staticAddrPkScript, + ) + if err != nil { + log.Errorf("Error persisting withdrawal: %v", err) - - case <-ctx.Done(): - log.Errorf("Withdrawal tx confirmation wait " + - "canceled") } - case err := <-errChan: - log.Errorf("Error waiting for spending: %v", err) + confCancel() + return true + + case err := <-confErrChan: + // Cancel subscription before re-registering. + confCancel() + log.Errorf("Confirmation error, re-registering: %v", + err) + continue + + case <-confReorgChan: + // A reorg after spend detection means the spend tx + // might have been reorged out. Return false to go back + // to the outer loop and re-register for spend + // notification, not just confirmations. + confCancel() + log.Warnf("Reorg detected for withdrawal %v, "+ + "re-watching for spend", txHash) + return false + + case err := <-blockErrChan: + confCancel() + log.Errorf("Block subscription error: %v", err) + return true case <-ctx.Done(): - log.Errorf("Withdrawal tx confirmation wait canceled") + confCancel() + return true } - }() - - return nil + } } func toOutpoints(deposits []*deposit.Deposit) []wire.OutPoint { diff --git a/staticaddr/withdraw/manager_test.go b/staticaddr/withdraw/manager_test.go index 4ffd4e1e8..bb81f99db 100644 --- a/staticaddr/withdraw/manager_test.go +++ b/staticaddr/withdraw/manager_test.go @@ -2,24 +2,37 @@ package withdraw import ( "context" + "errors" + "sync" "testing" + "time" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/lndclient" "github.com/lightninglabs/loop/staticaddr/deposit" "github.com/lightninglabs/loop/swapserverrpc" "github.com/lightninglabs/loop/test" + "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/funding" "github.com/lightningnetwork/lnd/input" "github.com/lightningnetwork/lnd/lnrpc" + "github.com/lightningnetwork/lnd/lnrpc/chainrpc" "github.com/lightningnetwork/lnd/lnwallet" "github.com/lightningnetwork/lnd/lnwallet/chainfee" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) +func init() { + // Initialize logger for tests to avoid nil pointer panics. + UseLogger(btclog.Disabled) +} + // TestNewManagerHeightValidation ensures the constructor rejects zero heights. func TestNewManagerHeightValidation(t *testing.T) { t.Parallel() @@ -606,3 +619,538 @@ func TestCalculateWithdrawalTxValues(t *testing.T) { }) } } + +// --------------------------------------------------------------------------- +// Mock types for handleWithdrawal tests +// --------------------------------------------------------------------------- + +// mockChainNotifier is a mock implementation of lndclient.ChainNotifierClient. +type mockChainNotifier struct { + mock.Mock + + mu sync.Mutex + + // Track call counts for conditional behavior. + spendNtfnCalls int + confNtfnCalls int +} + +func (m *mockChainNotifier) RegisterSpendNtfn(ctx context.Context, + outpoint *wire.OutPoint, pkScript []byte, heightHint int32, + opts ...lndclient.NotifierOption) (chan *chainntnfs.SpendDetail, + chan error, error) { + + m.mu.Lock() + m.spendNtfnCalls++ + callNum := m.spendNtfnCalls + m.mu.Unlock() + + args := m.Called(ctx, outpoint, pkScript, heightHint, callNum) + + spendChan := args.Get(0) + errChan := args.Get(1) + + var sc chan *chainntnfs.SpendDetail + var ec chan error + + if spendChan != nil { + sc = spendChan.(chan *chainntnfs.SpendDetail) + } + if errChan != nil { + ec = errChan.(chan error) + } + + return sc, ec, args.Error(2) +} + +func (m *mockChainNotifier) RegisterConfirmationsNtfn(ctx context.Context, + txid *chainhash.Hash, pkScript []byte, numConfs, heightHint int32, + opts ...lndclient.NotifierOption) (chan *chainntnfs.TxConfirmation, + chan error, error) { + + m.mu.Lock() + m.confNtfnCalls++ + callNum := m.confNtfnCalls + m.mu.Unlock() + + args := m.Called(ctx, txid, pkScript, numConfs, heightHint, callNum) + + confChan := args.Get(0) + errChan := args.Get(1) + + var cc chan *chainntnfs.TxConfirmation + var ec chan error + + if confChan != nil { + cc = confChan.(chan *chainntnfs.TxConfirmation) + } + if errChan != nil { + ec = errChan.(chan error) + } + + return cc, ec, args.Error(2) +} + +func (m *mockChainNotifier) RegisterBlockEpochNtfn(ctx context.Context) ( + chan int32, chan error, error) { + + args := m.Called(ctx) + + blockChan := args.Get(0) + errChan := args.Get(1) + + var bc chan int32 + var ec chan error + + if blockChan != nil { + bc = blockChan.(chan int32) + } + if errChan != nil { + ec = errChan.(chan error) + } + + return bc, ec, args.Error(2) +} + +func (m *mockChainNotifier) RawClientWithMacAuth(ctx context.Context) ( + context.Context, time.Duration, chainrpc.ChainNotifierClient) { + + return ctx, 0, nil +} + +// --------------------------------------------------------------------------- +// handleWithdrawal tests +// --------------------------------------------------------------------------- + +// testDeposit creates a test deposit with the given parameters. +func testDeposit(hashByte byte, value btcutil.Amount, + confHeight uint32) *deposit.Deposit { + + hash := chainhash.Hash{} + hash[0] = hashByte + return &deposit.Deposit{ + OutPoint: wire.OutPoint{ + Hash: hash, + Index: 0, + }, + Value: value, + ConfirmationHeight: int64(confHeight), + } +} + +// TestHandleWithdrawal_HappyPath tests the successful withdrawal flow up to +// confirmation registration. Full flow including storage persistence requires +// integration tests since Store is a concrete type (*SqlStore). +// +// Tests: spend detected -> confirmation registration succeeds. +func TestHandleWithdrawal_HappyPath(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Create test data. + deposits := []*deposit.Deposit{testDeposit(1, 100000, 100)} + txHash := chainhash.Hash{0x01} + spenderTxHash := chainhash.Hash{0x02} + withdrawalPkScript := []byte{0x51, 0x20} + staticAddrPkScript := []byte{0x51, 0x21} + + // Create channels for notifications. + blockChan := make(chan int32, 1) + blockErrChan := make(chan error, 1) + spendChan := make(chan *chainntnfs.SpendDetail, 1) + spendErrChan := make(chan error, 1) + confChan := make(chan *chainntnfs.TxConfirmation, 1) + confErrChan := make(chan error, 1) + + // Set up mocks. + mockNotifier := &mockChainNotifier{} + + mockNotifier.On("RegisterBlockEpochNtfn", mock.Anything). + Return(blockChan, blockErrChan, nil) + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 1). + Return(spendChan, spendErrChan, nil) + mockNotifier.On("RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1). + Return(confChan, confErrChan, nil) + + // Create manager with mocks. + // Note: Store is nil, so we can't test the full flow including + // UpdateWithdrawal. That requires integration tests. + m := &Manager{ + cfg: &ManagerConfig{ + ChainNotifier: mockNotifier, + }, + finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx), + } + m.initiationHeight.Store(100) + + // Call handleWithdrawal (spawns goroutine). + m.handleWithdrawal(ctx, deposits, txHash, withdrawalPkScript, + staticAddrPkScript) + + // Send spend notification. + spendChan <- &chainntnfs.SpendDetail{ + SpenderTxHash: &spenderTxHash, + SpendingHeight: 105, + } + + // Give time for spend to be processed and conf registration to occur. + time.Sleep(100 * time.Millisecond) + + // Verify spend and confirmation registration happened. + mockNotifier.AssertCalled(t, "RegisterSpendNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, 1) + mockNotifier.AssertCalled(t, "RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1) + + // Cancel to clean up goroutine (we're not testing full confirmation + // flow since Store is nil). + cancel() +} + +// TestHandleWithdrawal_SpendRegistrationRetry tests that spend registration +// is retried on next block when it fails. +func TestHandleWithdrawal_SpendRegistrationRetry(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Create test data. + deposits := []*deposit.Deposit{testDeposit(1, 100000, 100)} + txHash := chainhash.Hash{0x01} + spenderTxHash := chainhash.Hash{0x02} + withdrawalPkScript := []byte{0x51, 0x20} + staticAddrPkScript := []byte{0x51, 0x21} + + // Create channels. + blockChan := make(chan int32, 2) + blockErrChan := make(chan error, 1) + spendChan := make(chan *chainntnfs.SpendDetail, 1) + spendErrChan := make(chan error, 1) + confChan := make(chan *chainntnfs.TxConfirmation, 1) + confErrChan := make(chan error, 1) + + // Set up mocks. + mockNotifier := &mockChainNotifier{} + + mockNotifier.On("RegisterBlockEpochNtfn", mock.Anything). + Return(blockChan, blockErrChan, nil) + + // First call fails, second succeeds. + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 1). + Return(nil, nil, errors.New("temporary failure")) + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 2). + Return(spendChan, spendErrChan, nil) + + mockNotifier.On("RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1). + Return(confChan, confErrChan, nil) + + // Create manager. + m := &Manager{ + cfg: &ManagerConfig{ + ChainNotifier: mockNotifier, + }, + finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx), + } + m.initiationHeight.Store(100) + + // Call handleWithdrawal. + m.handleWithdrawal(ctx, deposits, txHash, withdrawalPkScript, + staticAddrPkScript) + + // Send block to trigger retry. + blockChan <- 101 + + // Give time for retry. + time.Sleep(50 * time.Millisecond) + + // Send spend notification (on successful registration). + spendChan <- &chainntnfs.SpendDetail{ + SpenderTxHash: &spenderTxHash, + SpendingHeight: 105, + } + + // Give time for confirmation registration. + time.Sleep(100 * time.Millisecond) + + // Verify both spend registrations were attempted. + mockNotifier.AssertNumberOfCalls(t, "RegisterSpendNtfn", 2) + // Verify we reached confirmation registration (retry worked). + mockNotifier.AssertCalled(t, "RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1) + + cancel() +} + +// TestHandleWithdrawal_ConfirmationRegistrationRetry tests that confirmation +// registration is retried on next block when it fails. +func TestHandleWithdrawal_ConfirmationRegistrationRetry(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Create test data. + deposits := []*deposit.Deposit{testDeposit(1, 100000, 100)} + txHash := chainhash.Hash{0x01} + spenderTxHash := chainhash.Hash{0x02} + withdrawalPkScript := []byte{0x51, 0x20} + staticAddrPkScript := []byte{0x51, 0x21} + + // Create channels. + blockChan := make(chan int32, 2) + blockErrChan := make(chan error, 1) + spendChan := make(chan *chainntnfs.SpendDetail, 1) + spendErrChan := make(chan error, 1) + confChan := make(chan *chainntnfs.TxConfirmation, 1) + confErrChan := make(chan error, 1) + + // Set up mocks. + mockNotifier := &mockChainNotifier{} + + mockNotifier.On("RegisterBlockEpochNtfn", mock.Anything). + Return(blockChan, blockErrChan, nil) + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 1). + Return(spendChan, spendErrChan, nil) + + // First confirmation registration fails, second succeeds. + mockNotifier.On("RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1). + Return(nil, nil, errors.New("temporary failure")) + mockNotifier.On("RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 2). + Return(confChan, confErrChan, nil) + + // Create manager. + m := &Manager{ + cfg: &ManagerConfig{ + ChainNotifier: mockNotifier, + }, + finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx), + } + m.initiationHeight.Store(100) + + // Call handleWithdrawal. + m.handleWithdrawal(ctx, deposits, txHash, withdrawalPkScript, + staticAddrPkScript) + + // Send spend notification. + spendChan <- &chainntnfs.SpendDetail{ + SpenderTxHash: &spenderTxHash, + SpendingHeight: 105, + } + + // Give time for first conf registration to fail. + time.Sleep(50 * time.Millisecond) + + // Send block to trigger retry. + blockChan <- 106 + + // Give time for retry. + time.Sleep(100 * time.Millisecond) + + // Verify both confirmation registrations were attempted. + mockNotifier.AssertNumberOfCalls(t, "RegisterConfirmationsNtfn", 2) + + cancel() +} + +// TestHandleWithdrawal_ContextCanceled tests that the goroutine exits cleanly +// when the context is canceled. +func TestHandleWithdrawal_ContextCanceled(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + + // Create test data. + deposits := []*deposit.Deposit{testDeposit(1, 100000, 100)} + txHash := chainhash.Hash{0x01} + withdrawalPkScript := []byte{0x51, 0x20} + staticAddrPkScript := []byte{0x51, 0x21} + + // Create channels. + blockChan := make(chan int32, 1) + blockErrChan := make(chan error, 1) + spendChan := make(chan *chainntnfs.SpendDetail, 1) + spendErrChan := make(chan error, 1) + + // Set up mocks. + mockNotifier := &mockChainNotifier{} + + mockNotifier.On("RegisterBlockEpochNtfn", mock.Anything). + Return(blockChan, blockErrChan, nil) + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 1). + Return(spendChan, spendErrChan, nil) + + // Create manager. + m := &Manager{ + cfg: &ManagerConfig{ + ChainNotifier: mockNotifier, + }, + finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx), + } + m.initiationHeight.Store(100) + + // Call handleWithdrawal. + m.handleWithdrawal(ctx, deposits, txHash, withdrawalPkScript, + staticAddrPkScript) + + // Give time for goroutine to start. + time.Sleep(50 * time.Millisecond) + + // Cancel context. + cancel() + + // Give time for goroutine to exit. + time.Sleep(100 * time.Millisecond) + + // If we get here without hanging, the test passes. + // The goroutine should have exited cleanly. + mockNotifier.AssertExpectations(t) +} + +// TestHandleWithdrawal_BlockChannelClosed tests that the goroutine exits +// cleanly when the block channel is closed. +func TestHandleWithdrawal_BlockChannelClosed(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Create test data. + deposits := []*deposit.Deposit{testDeposit(1, 100000, 100)} + txHash := chainhash.Hash{0x01} + withdrawalPkScript := []byte{0x51, 0x20} + staticAddrPkScript := []byte{0x51, 0x21} + + // Create channels. + blockChan := make(chan int32) + blockErrChan := make(chan error, 1) + + // Set up mocks - spend registration fails to force retry path. + mockNotifier := &mockChainNotifier{} + + mockNotifier.On("RegisterBlockEpochNtfn", mock.Anything). + Return(blockChan, blockErrChan, nil) + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 1). + Return(nil, nil, errors.New("temporary failure")) + + // Create manager. + m := &Manager{ + cfg: &ManagerConfig{ + ChainNotifier: mockNotifier, + }, + finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx), + } + m.initiationHeight.Store(100) + + // Call handleWithdrawal. + m.handleWithdrawal(ctx, deposits, txHash, withdrawalPkScript, + staticAddrPkScript) + + // Give time for goroutine to hit retry wait. + time.Sleep(50 * time.Millisecond) + + // Close block channel. + close(blockChan) + + // Give time for goroutine to exit. + time.Sleep(100 * time.Millisecond) + + // If we get here without hanging, the test passes. + mockNotifier.AssertExpectations(t) +} + +// TestHandleWithdrawal_SpendErrorReregisters tests that an error on the spend +// error channel causes re-registration. +func TestHandleWithdrawal_SpendErrorReregisters(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Create test data. + deposits := []*deposit.Deposit{testDeposit(1, 100000, 100)} + txHash := chainhash.Hash{0x01} + spenderTxHash := chainhash.Hash{0x02} + withdrawalPkScript := []byte{0x51, 0x20} + staticAddrPkScript := []byte{0x51, 0x21} + + // Create channels. + blockChan := make(chan int32, 1) + blockErrChan := make(chan error, 1) + spendChan1 := make(chan *chainntnfs.SpendDetail, 1) + spendErrChan1 := make(chan error, 1) + spendChan2 := make(chan *chainntnfs.SpendDetail, 1) + spendErrChan2 := make(chan error, 1) + confChan := make(chan *chainntnfs.TxConfirmation, 1) + confErrChan := make(chan error, 1) + + // Set up mocks. + mockNotifier := &mockChainNotifier{} + + mockNotifier.On("RegisterBlockEpochNtfn", mock.Anything). + Return(blockChan, blockErrChan, nil) + + // First spend registration succeeds but will send error. + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 1). + Return(spendChan1, spendErrChan1, nil) + // Second spend registration succeeds normally. + mockNotifier.On("RegisterSpendNtfn", mock.Anything, mock.Anything, + mock.Anything, mock.Anything, 2). + Return(spendChan2, spendErrChan2, nil) + + mockNotifier.On("RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1). + Return(confChan, confErrChan, nil) + + // Create manager. + m := &Manager{ + cfg: &ManagerConfig{ + ChainNotifier: mockNotifier, + }, + finalizedWithdrawalTxns: make(map[chainhash.Hash]*wire.MsgTx), + } + m.initiationHeight.Store(100) + + // Call handleWithdrawal. + m.handleWithdrawal(ctx, deposits, txHash, withdrawalPkScript, + staticAddrPkScript) + + // Give time for goroutine to register. + time.Sleep(50 * time.Millisecond) + + // Send error on first spend channel to trigger re-registration. + spendErrChan1 <- errors.New("spend notification error") + + // Give time for re-registration. + time.Sleep(50 * time.Millisecond) + + // Send spend on second channel. + spendChan2 <- &chainntnfs.SpendDetail{ + SpenderTxHash: &spenderTxHash, + SpendingHeight: 105, + } + + // Give time for confirmation registration. + time.Sleep(100 * time.Millisecond) + + // Verify both spend registrations were called. + mockNotifier.AssertNumberOfCalls(t, "RegisterSpendNtfn", 2) + // Verify we reached confirmation registration. + mockNotifier.AssertCalled(t, "RegisterConfirmationsNtfn", mock.Anything, + mock.Anything, mock.Anything, mock.Anything, mock.Anything, 1) + + cancel() +}