diff --git a/staticaddr/deposit/deposit.go b/staticaddr/deposit/deposit.go index 19c5c2b0..b784b511 100644 --- a/staticaddr/deposit/deposit.go +++ b/staticaddr/deposit/deposit.go @@ -52,7 +52,8 @@ type Deposit struct { Value btcutil.Amount // ConfirmationHeight is the absolute height at which the deposit was - // first confirmed. + // first confirmed. A value of zero means the deposit is still + // unconfirmed. ConfirmationHeight int64 // TimeOutSweepPkScript is the pk script that is used to sweep the @@ -91,6 +92,10 @@ func (d *Deposit) IsExpired(currentHeight, expiry uint32) bool { d.Lock() defer d.Unlock() + if d.ConfirmationHeight <= 0 { + return false + } + return currentHeight >= uint32(d.ConfirmationHeight)+expiry } diff --git a/staticaddr/deposit/deposit_test.go b/staticaddr/deposit/deposit_test.go new file mode 100644 index 00000000..3215e116 --- /dev/null +++ b/staticaddr/deposit/deposit_test.go @@ -0,0 +1,17 @@ +package deposit + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// TestDepositIsExpiredUnconfirmed verifies that unconfirmed deposits do not +// expire because their CSV timeout has not started yet. +func TestDepositIsExpiredUnconfirmed(t *testing.T) { + t.Parallel() + + d := &Deposit{} + + require.False(t, d.IsExpired(1_000, 144)) +} diff --git a/staticaddr/deposit/fsm.go b/staticaddr/deposit/fsm.go index 7dd8715d..c5bb85c3 100644 --- a/staticaddr/deposit/fsm.go +++ b/staticaddr/deposit/fsm.go @@ -42,8 +42,8 @@ var ( // States. var ( - // Deposited signals that funds at a static address have reached the - // confirmation height. + // Deposited signals that funds at a static address have been detected + // and are available to the client. Deposited = fsm.StateType("Deposited") // Withdrawing signals that the withdrawal transaction has been @@ -93,8 +93,8 @@ var ( // Events. var ( // OnStart is sent to the fsm once the deposit outpoint has been - // sufficiently confirmed. It transitions the fsm into the Deposited - // state from where we can trigger a withdrawal, a loopin or an expiry. + // detected. It transitions the fsm into the Deposited state from where + // we can trigger a withdrawal, a loopin or an expiry. OnStart = fsm.EventType("OnStart") // OnWithdrawInitiated is sent to the fsm when a withdrawal has been diff --git a/staticaddr/deposit/manager.go b/staticaddr/deposit/manager.go index 103b51c5..4d892f7e 100644 --- a/staticaddr/deposit/manager.go +++ b/staticaddr/deposit/manager.go @@ -6,6 +6,7 @@ import ( "fmt" "sort" "sync" + "sync/atomic" "time" "github.com/btcsuite/btcd/txscript" @@ -17,9 +18,8 @@ import ( ) const ( - // MinConfs is the minimum number of confirmations we require for a - // deposit to be considered available for loop-ins, coop-spends and - // timeouts. + // MinConfs is the legacy minimum confirmation target deposits had to + // reach before they were considered ready to be used for swaps. MinConfs = 6 // MaxConfs is unset since we don't require a max number of @@ -86,6 +86,9 @@ type Manager struct { // been finalized. The manager will adjust its internal state and flush // finalized deposits from its memory. finalizedDepositChan chan wire.OutPoint + + // currentHeight stores the currently best known block height. + currentHeight atomic.Uint32 } // NewManager creates a new deposit manager. @@ -107,6 +110,19 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { return err } + var startupHeight uint32 + select { + case height := <-newBlockChan: + startupHeight = uint32(height) + m.currentHeight.Store(startupHeight) + + case err = <-newBlockErrChan: + return err + + case <-ctx.Done(): + return ctx.Err() + } + // Recover previous deposits and static address parameters from the DB. err = m.recoverDeposits(ctx) if err != nil { @@ -132,7 +148,15 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { for { select { case height := <-newBlockChan: - err := m.notifyActiveDeposits(ctx, uint32(height)) + m.currentHeight.Store(uint32(height)) + + err := m.reconcileDeposits(ctx) + if err != nil { + log.Errorf("unable to reconcile deposits: %v", err) + continue + } + + err = m.notifyActiveDeposits(ctx, uint32(height)) if err != nil { return err } @@ -226,8 +250,10 @@ func (m *Manager) recoverDeposits(ctx context.Context) error { return nil } -// pollDeposits polls new deposits to our static address and notifies the -// manager's event loop about them. +// pollDeposits periodically polls for new deposits to our static address. This +// complements the block-driven reconciliation in the main event loop: while new +// blocks trigger reconcileDeposits to promptly detect confirmations, the ticker +// here catches deposits that appear in the mempool between blocks. func (m *Manager) pollDeposits(ctx context.Context) { log.Debugf("Waiting for new static address deposits...") @@ -261,12 +287,19 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error { log.Tracef("Reconciling new deposits...") utxos, err := m.cfg.AddressManager.ListUnspent( - ctx, MinConfs, MaxConfs, + ctx, 0, MaxConfs, ) if err != nil { return fmt.Errorf("unable to list new deposits: %w", err) } + currentHeight := m.currentHeight.Load() + err = m.updateDepositConfirmations(ctx, utxos, currentHeight) + if err != nil { + return fmt.Errorf("unable to update deposit "+ + "confirmations: %w", err) + } + newDeposits := m.filterNewDeposits(utxos) if len(newDeposits) == 0 { log.Tracef("No new deposits...") @@ -274,7 +307,7 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error { } for _, utxo := range newDeposits { - deposit, err := m.createNewDeposit(ctx, utxo) + deposit, err := m.createNewDeposit(ctx, utxo, currentHeight) if err != nil { return fmt.Errorf("unable to retain new deposit: %w", err) @@ -294,9 +327,11 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error { // createNewDeposit transforms the wallet utxo into a deposit struct and stores // it in our database and manager memory. func (m *Manager) createNewDeposit(ctx context.Context, - utxo *lnwallet.Utxo) (*Deposit, error) { + utxo *lnwallet.Utxo, currentHeight uint32) (*Deposit, error) { - blockHeight, err := m.getBlockHeight(ctx, utxo) + confirmationHeight, err := confirmationHeightForUtxo( + currentHeight, utxo, + ) if err != nil { return nil, err } @@ -324,7 +359,7 @@ func (m *Manager) createNewDeposit(ctx context.Context, state: Deposited, OutPoint: utxo.OutPoint, Value: utxo.Value, - ConfirmationHeight: int64(blockHeight), + ConfirmationHeight: confirmationHeight, TimeOutSweepPkScript: timeoutSweepPkScript, } @@ -340,37 +375,77 @@ func (m *Manager) createNewDeposit(ctx context.Context, return deposit, nil } -// getBlockHeight retrieves the block height of a given utxo. -func (m *Manager) getBlockHeight(ctx context.Context, - utxo *lnwallet.Utxo) (uint32, error) { +// confirmationHeightForUtxo derives the first confirmation height of a wallet +// UTXO from the manager's current block height. Unconfirmed UTXOs return 0. +func confirmationHeightForUtxo(currentHeight uint32, + utxo *lnwallet.Utxo) (int64, error) { - addressParams, err := m.cfg.AddressManager.GetStaticAddressParameters( - ctx, - ) - if err != nil { - return 0, fmt.Errorf("couldn't get confirmation height for "+ - "deposit, %w", err) + if utxo.Confirmations <= 0 { + return 0, nil } - notifChan, errChan, err := - m.cfg.ChainNotifier.RegisterConfirmationsNtfn( - ctx, &utxo.OutPoint.Hash, addressParams.PkScript, - MinConfs, addressParams.InitiationHeight, - ) - if err != nil { - return 0, err + if currentHeight == 0 { + return 0, errors.New("current block height unavailable") } - select { - case tx := <-notifChan: - return tx.BlockHeight, nil - - case err := <-errChan: - return 0, err - - case <-ctx.Done(): - return 0, ctx.Err() + firstConfirmationHeight := int64(currentHeight) - utxo.Confirmations + 1 + if firstConfirmationHeight <= 0 { + return 0, fmt.Errorf("invalid confirmation height %d for %v "+ + "with current height %d and %d confirmations", + firstConfirmationHeight, utxo.OutPoint, currentHeight, + utxo.Confirmations) } + + return firstConfirmationHeight, nil +} + +// updateDepositConfirmations syncs first confirmation heights for deposits that +// are visible in lnd's wallet view. +func (m *Manager) updateDepositConfirmations(ctx context.Context, + utxos []*lnwallet.Utxo, currentHeight uint32) error { + + for _, utxo := range utxos { + m.mu.Lock() + deposit, ok := m.deposits[utxo.OutPoint] + m.mu.Unlock() + if !ok { + continue + } + + err := func() error { + deposit.Lock() + defer deposit.Unlock() + + previousConfirmationHeight := deposit.ConfirmationHeight + + confirmationHeight, err := confirmationHeightForUtxo( + currentHeight, utxo, + ) + if err != nil { + return err + } + + if deposit.ConfirmationHeight == confirmationHeight { + return nil + } + + deposit.ConfirmationHeight = confirmationHeight + + err = m.cfg.Store.UpdateDeposit(ctx, deposit) + if err != nil { + deposit.ConfirmationHeight = previousConfirmationHeight + + return err + } + + return nil + }() + if err != nil { + return err + } + } + + return nil } // filterNewDeposits filters the given utxos for new deposits that we haven't diff --git a/staticaddr/deposit/manager_height_test.go b/staticaddr/deposit/manager_height_test.go new file mode 100644 index 00000000..78ad27bb --- /dev/null +++ b/staticaddr/deposit/manager_height_test.go @@ -0,0 +1,39 @@ +package deposit + +import ( + "testing" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/lnwallet" + "github.com/stretchr/testify/require" +) + +// TestConfirmationHeightForUtxo verifies confirmation heights are derived from +// the current block height and wallet confirmation count. +func TestConfirmationHeightForUtxo(t *testing.T) { + t.Run("unconfirmed", func(t *testing.T) { + height, err := confirmationHeightForUtxo(0, &lnwallet.Utxo{}) + require.NoError(t, err) + require.Zero(t, height) + }) + + t.Run("confirmed", func(t *testing.T) { + height, err := confirmationHeightForUtxo(101, &lnwallet.Utxo{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: 2, + }, + Confirmations: 6, + }) + require.NoError(t, err) + require.EqualValues(t, 96, height) + }) + + t.Run("invalid current height", func(t *testing.T) { + _, err := confirmationHeightForUtxo(2, &lnwallet.Utxo{ + Confirmations: 6, + }) + require.ErrorContains(t, err, "invalid confirmation height") + }) +} diff --git a/staticaddr/deposit/manager_reconcile_test.go b/staticaddr/deposit/manager_reconcile_test.go index 972feefd..b71f1d6c 100644 --- a/staticaddr/deposit/manager_reconcile_test.go +++ b/staticaddr/deposit/manager_reconcile_test.go @@ -1,14 +1,237 @@ package deposit import ( + "context" + "errors" + "sync" + "sync/atomic" "testing" "time" + "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/loop/staticaddr/script" + "github.com/lightninglabs/loop/staticaddr/version" + "github.com/lightninglabs/loop/test" + "github.com/lightningnetwork/lnd/lnwallet" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) +// TestReconcileDepositsSerialized verifies reconciliation is serialized across +// concurrent callers. +func TestReconcileDepositsSerialized(t *testing.T) { + ctx := context.Background() + mockLnd := test.NewMockLnd() + utxo := &lnwallet.Utxo{ + AddressType: lnwallet.TaprootPubkey, + Value: btcutil.Amount(100_000), + Confirmations: 0, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: 1, + }, + } + + mockAddressManager := new(mockAddressManager) + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{utxo}, nil) + mockAddressManager.On( + "GetStaticAddressParameters", mock.Anything, + ).Return((*script.Parameters)(nil), errors.New("fsm init failed")) + + mockStore := new(mockStore) + var createCalls atomic.Int32 + createEntered := make(chan struct{}) + releaseCreate := make(chan struct{}) + mockStore.On( + "CreateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(mock.Arguments) { + if createCalls.Add(1) == 1 { + close(createEntered) + } + + <-releaseCreate + }) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: mockStore, + WalletKit: mockLnd.WalletKit, + Signer: mockLnd.Signer, + }) + + var wg sync.WaitGroup + wg.Add(2) + + errs := make(chan error, 2) + go func() { + defer wg.Done() + errs <- manager.reconcileDeposits(ctx) + }() + + <-createEntered + + go func() { + defer wg.Done() + errs <- manager.reconcileDeposits(ctx) + }() + + time.Sleep(100 * time.Millisecond) + close(releaseCreate) + wg.Wait() + close(errs) + + var gotErrs []error + for err := range errs { + gotErrs = append(gotErrs, err) + } + + require.EqualValues(t, 1, createCalls.Load()) + require.Len(t, manager.deposits, 1) + require.Empty(t, manager.activeDeposits) + require.Len(t, gotErrs, 2) + + var errCount int + for _, err := range gotErrs { + if err == nil { + continue + } + + errCount++ + require.ErrorContains(t, err, "unable to start new deposit FSM") + } + require.Equal(t, 1, errCount) +} + +// TestReconcileConfirmedDepositUsesCurrentHeight verifies confirmation heights +// are derived from the manager's current block height. +func TestReconcileConfirmedDepositUsesCurrentHeight(t *testing.T) { + ctx := context.Background() + mockLnd := test.NewMockLnd() + utxo := &lnwallet.Utxo{ + AddressType: lnwallet.TaprootPubkey, + Value: btcutil.Amount(100_000), + Confirmations: 3, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{8}, + Index: 1, + }, + } + + mockAddressManager := new(mockAddressManager) + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{utxo}, nil) + mockAddressManager.On( + "GetStaticAddressParameters", mock.Anything, + ).Return((*script.Parameters)(nil), errors.New("fsm init failed")) + + mockStore := new(mockStore) + mockStore.On( + "CreateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(args mock.Arguments) { + createdDeposit := args.Get(1).(*Deposit) + require.EqualValues(t, 98, createdDeposit.ConfirmationHeight) + }) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: mockStore, + WalletKit: mockLnd.WalletKit, + Signer: mockLnd.Signer, + }) + manager.currentHeight.Store(100) + + err := manager.reconcileDeposits(ctx) + require.ErrorContains(t, err, "unable to start new deposit FSM") +} + +// TestUpdateDepositConfirmationsResetsReorgedDeposit verifies that a deposit +// which remains wallet-visible but loses confirmations has its confirmation +// height reset. This can happen if a confirmed transaction is reorged back into +// the mempool. +func TestUpdateDepositConfirmationsResetsReorgedDeposit(t *testing.T) { + ctx := context.Background() + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{7}, + Index: 2, + } + + deposit := &Deposit{ + OutPoint: outpoint, + ConfirmationHeight: 99, + } + deposit.SetState(Deposited) + + utxo := &lnwallet.Utxo{ + OutPoint: outpoint, + Confirmations: 0, + } + + mockStore := new(mockStore) + mockStore.On( + "UpdateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(args mock.Arguments) { + updatedDeposit := args.Get(1).(*Deposit) + require.Zero(t, updatedDeposit.ConfirmationHeight) + }) + + manager := NewManager(&ManagerConfig{ + Store: mockStore, + }) + manager.deposits[outpoint] = deposit + + err := manager.updateDepositConfirmations(ctx, []*lnwallet.Utxo{utxo}, 0) + require.NoError(t, err) + require.Zero(t, deposit.ConfirmationHeight) + mockStore.AssertExpectations(t) +} + +// TestUpdateDepositConfirmationsRecomputesPositiveHeight verifies that a +// deposit which is confirmed again at a different height after a reorg does +// not retain its stale, positive confirmation height. +func TestUpdateDepositConfirmationsRecomputesPositiveHeight(t *testing.T) { + ctx := context.Background() + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{10}, + Index: 3, + } + + deposit := &Deposit{ + OutPoint: outpoint, + ConfirmationHeight: 100, + } + deposit.SetState(Deposited) + + utxo := &lnwallet.Utxo{ + OutPoint: outpoint, + Confirmations: 2, + } + + mockStore := new(mockStore) + mockStore.On( + "UpdateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(args mock.Arguments) { + updatedDeposit := args.Get(1).(*Deposit) + require.EqualValues(t, 109, updatedDeposit.ConfirmationHeight) + }) + + manager := NewManager(&ManagerConfig{ + Store: mockStore, + }) + manager.deposits[outpoint] = deposit + + err := manager.updateDepositConfirmations( + ctx, []*lnwallet.Utxo{utxo}, 110, + ) + require.NoError(t, err) + require.EqualValues(t, 109, deposit.ConfirmationHeight) + mockStore.AssertExpectations(t) +} + // TestAllOutpointsActiveDepositsRejectsDuplicateOutpoints verifies that a // duplicated selection is rejected before the manager tries to lock the same // deposit twice. @@ -126,3 +349,87 @@ func TestLockDepositsAllowsReversedConcurrentRequests(t *testing.T) { } } } + +// TestReconcileReplacementDepositCreatesNewDeposit ensures that a replacement +// UTXO is retained as a new deposit while an in-flight deposit remains tied to +// the outpoint selected by a loop-in. +func TestReconcileReplacementDepositCreatesNewDeposit(t *testing.T) { + ctx := context.Background() + mockLnd := test.NewMockLnd() + oldOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{4}, + Index: 8, + } + newOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{5}, + Index: 9, + } + + depositID, err := GetRandomDepositID() + require.NoError(t, err) + + deposit := &Deposit{ + ID: depositID, + OutPoint: oldOutpoint, + Value: btcutil.Amount(100_000), + } + deposit.SetState(LoopingIn) + + utxo := &lnwallet.Utxo{ + OutPoint: newOutpoint, + Value: deposit.Value, + Confirmations: 0, + } + + mockAddressManager := new(mockAddressManager) + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{utxo}, nil) + mockAddressManager.On( + "GetStaticAddressParameters", mock.Anything, + ).Return(&script.Parameters{ + ProtocolVersion: version.ProtocolVersion_V0, + }, nil) + mockAddressManager.On( + "GetStaticAddress", mock.Anything, + ).Return((*script.StaticAddress)(nil), nil) + + mockStore := new(mockStore) + var createdDeposit *Deposit + mockStore.On( + "CreateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(args mock.Arguments) { + createdDeposit = args.Get(1).(*Deposit) + }) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: mockStore, + WalletKit: mockLnd.WalletKit, + Signer: mockLnd.Signer, + }) + manager.deposits[oldOutpoint] = deposit + fsm := &FSM{} + manager.activeDeposits[oldOutpoint] = fsm + + require.NoError(t, manager.reconcileDeposits(ctx)) + + require.Same(t, deposit, manager.deposits[oldOutpoint]) + require.Equal(t, oldOutpoint, deposit.OutPoint) + require.Equal(t, LoopingIn, deposit.GetState()) + + replacement, ok := manager.deposits[newOutpoint] + require.True(t, ok) + require.Same(t, createdDeposit, replacement) + require.NotEqual(t, depositID, replacement.ID) + require.Equal(t, newOutpoint, replacement.OutPoint) + require.Equal(t, Deposited, replacement.GetState()) + require.Zero(t, replacement.ConfirmationHeight) + + require.Same(t, fsm, manager.activeDeposits[oldOutpoint]) + require.NotSame(t, fsm, manager.activeDeposits[newOutpoint]) + + mockStore.AssertNotCalled( + t, "UpdateDeposit", mock.Anything, mock.Anything, + ) +} diff --git a/staticaddr/deposit/manager_test.go b/staticaddr/deposit/manager_test.go index 1e798ea3..880e286c 100644 --- a/staticaddr/deposit/manager_test.go +++ b/staticaddr/deposit/manager_test.go @@ -133,6 +133,9 @@ func (m *mockAddressManager) ListUnspent(ctx context.Context, minConfs, maxConfs int32) ([]*lnwallet.Utxo, error) { args := m.Called(ctx, minConfs, maxConfs) + if listUnspent, ok := args.Get(0).(func() []*lnwallet.Utxo); ok { + return listUnspent(), args.Error(1) + } return args.Get(0).([]*lnwallet.Utxo), args.Error(1) @@ -237,6 +240,10 @@ func TestManager(t *testing.T) { runErrChan <- testContext.manager.Run(ctx, initChan) }() + // Send an initial block so the manager can proceed past its startup + // block wait. + testContext.blockChan <- int32(defaultDepositConfirmations) + // Ensure that the manager has been initialized. select { case <-initChan: @@ -366,6 +373,7 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext { "UpdateDeposit", mock.Anything, mock.Anything, ).Return(nil) + var manager *Manager mockAddressManager.On( "GetStaticAddressParameters", mock.Anything, ).Return(&script.Parameters{ @@ -374,7 +382,19 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext { mockAddressManager.On( "ListUnspent", mock.Anything, mock.Anything, mock.Anything, - ).Return([]*lnwallet.Utxo{utxo}, nil) + ).Return(func() []*lnwallet.Utxo { + currentUtxo := *utxo + currentHeight := manager.currentHeight.Load() + if currentHeight < defaultDepositConfirmations { + currentUtxo.Confirmations = 0 + } else { + currentUtxo.Confirmations = int64( + currentHeight - defaultDepositConfirmations + 1, + ) + } + + return []*lnwallet.Utxo{¤tUtxo} + }, nil) // Define the expected return values for the mocks. mockChainNotifier.On( @@ -394,7 +414,7 @@ func newManagerTestContext(t *testing.T) *ManagerTestContext { Signer: mockLnd.Signer, } - manager := NewManager(cfg) + manager = NewManager(cfg) testContext := &ManagerTestContext{ manager: manager,