From 5506d28f794fed43e1ce2746304b5e1bcf48dc1f Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 26 Jun 2026 12:52:51 +0200 Subject: [PATCH] staticaddr: sync active deposits with wallet Treat lnd wallet view as the source of spendable static-address outputs while keeping historical deposit records in the DB. Reconcile active FSMs against the current wallet view and reactivate known deposits when their outpoints are visible again. Refresh deposits before selection, withdrawal, loop-in, and channel-open paths, and filter list and summary responses through the live active set so stale Deposited records are not exposed as available funds. --- loopd/swapclient_server.go | 43 ++- loopd/swapclient_server_staticaddr_test.go | 103 +++++ staticaddr/deposit/actions.go | 2 +- staticaddr/deposit/deposit.go | 17 + staticaddr/deposit/manager.go | 180 +++++++-- staticaddr/deposit/manager_reconcile_test.go | 373 ++++++++++++++++++- staticaddr/deposit/sql_store.go | 4 +- staticaddr/loopin/actions_test.go | 5 + staticaddr/loopin/autoloop.go | 5 + staticaddr/loopin/autoloop_dp.go | 7 +- staticaddr/loopin/interface.go | 3 + staticaddr/loopin/manager.go | 22 +- staticaddr/loopin/manager_test.go | 4 + staticaddr/openchannel/interface.go | 3 + staticaddr/openchannel/manager.go | 21 +- staticaddr/openchannel/manager_test.go | 4 + staticaddr/withdraw/interface.go | 10 + staticaddr/withdraw/manager.go | 14 +- 18 files changed, 757 insertions(+), 63 deletions(-) diff --git a/loopd/swapclient_server.go b/loopd/swapclient_server.go index 9e84316c..d3843a3d 100644 --- a/loopd/swapclient_server.go +++ b/loopd/swapclient_server.go @@ -921,6 +921,12 @@ func (s *swapClientServer) GetLoopInQuote(ctx context.Context, // number of deposits to quote for. numDeposits := 0 if autoSelectDeposits { + err = s.depositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, fmt.Errorf("unable to refresh deposits: %w", + err) + } + deposits, err := s.depositManager.GetActiveDepositsInState( deposit.Deposited, ) @@ -955,6 +961,12 @@ func (s *swapClientServer) GetLoopInQuote(ctx context.Context, numDeposits = len(selectedDeposits) } else if len(req.DepositOutpoints) > 0 { + err = s.depositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, fmt.Errorf("unable to refresh deposits: %w", + err) + } + // If deposits are selected, we need to retrieve them to // calculate the total value which we request a quote for. depositList, err := s.ListStaticAddressDeposits( @@ -985,9 +997,11 @@ func (s *swapClientServer) GetLoopInQuote(ctx context.Context, // server can probe the selected value and calculate the per // input fee. for _, deposit := range depositList.FilteredDeposits { - // For a manual quote we require the current state to be - // Deposited so a stale client-side outpoint selection - // fails early instead of making it to swap initiation. + // ListStaticAddressDeposits only returns deposits that are visible + // in the manager's live view. For a manual quote we additionally + // require the current state to be Deposited so stale client-side + // outpoint selection fails early instead of making it to swap + // initiation. if deposit.State != looprpc.DepositState_DEPOSITED { return nil, fmt.Errorf("deposit %s is not "+ "currently available", deposit.Outpoint) @@ -1768,6 +1782,12 @@ func (s *swapClientServer) WithdrawDeposits(ctx context.Context, return nil, fmt.Errorf("must select either all or some utxos") case isAllSelected: + err = s.depositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, fmt.Errorf("unable to refresh deposits: %w", + err) + } + deposits, err := s.depositManager.GetActiveDepositsInState( deposit.Deposited, ) @@ -1808,7 +1828,7 @@ func withdrawAllDepositOutpoints(deposits []*deposit.Deposit) ([]wire.OutPoint, outpoints := make([]wire.OutPoint, 0, len(deposits)) for _, d := range deposits { - if d.ConfirmationHeight <= 0 { + if d.GetConfirmationHeight() <= 0 { return nil, fmt.Errorf("can't withdraw all deposits while " + "some deposits are unconfirmed") } @@ -1834,7 +1854,7 @@ func (s *swapClientServer) ListStaticAddressDeposits(ctx context.Context, "outpoints") } - allDeposits, err := s.depositManager.GetAllDeposits(ctx) + allDeposits, err := s.depositManager.GetVisibleDeposits(ctx) if err != nil { return nil, err } @@ -1900,7 +1920,7 @@ func (s *swapClientServer) ListStaticAddressWithdrawals(ctx context.Context, Id: d.ID[:], Outpoint: d.OutPoint.String(), Value: int64(d.Value), - ConfirmationHeight: d.ConfirmationHeight, + ConfirmationHeight: d.GetConfirmationHeight(), State: toClientDepositState( d.GetState(), ), @@ -1991,8 +2011,9 @@ func (s *swapClientServer) ListStaticAddressSwaps(ctx context.Context, protoDeposits = make([]*looprpc.Deposit, 0, len(ds)) for _, d := range ds { state := toClientDepositState(d.GetState()) + confirmationHeight := d.GetConfirmationHeight() blocksUntilExpiry := depositBlocksUntilExpiry( - d.ConfirmationHeight, addrParams.Expiry, + confirmationHeight, addrParams.Expiry, int64(lndInfo.BlockHeight), ) @@ -2001,7 +2022,7 @@ func (s *swapClientServer) ListStaticAddressSwaps(ctx context.Context, State: state, Outpoint: d.OutPoint.String(), Value: int64(d.Value), - ConfirmationHeight: d.ConfirmationHeight, + ConfirmationHeight: confirmationHeight, SwapHash: d.SwapHash[:], BlocksUntilExpiry: blocksUntilExpiry, } @@ -2070,7 +2091,7 @@ func (s *swapClientServer) GetStaticAddressSummary(ctx context.Context, _ *looprpc.StaticAddressSummaryRequest) ( *looprpc.StaticAddressSummaryResponse, error) { - allDeposits, err := s.depositManager.GetAllDeposits(ctx) + allDeposits, err := s.depositManager.GetVisibleDeposits(ctx) if err != nil { return nil, err } @@ -2091,7 +2112,7 @@ func (s *swapClientServer) GetStaticAddressSummary(ctx context.Context, value := int64(d.Value) switch d.GetState() { case deposit.Deposited: - if d.ConfirmationHeight <= 0 { + if d.GetConfirmationHeight() <= 0 { valueUnconfirmed += value } else { valueDeposited += value @@ -2316,7 +2337,7 @@ func filter(deposits []*deposit.Deposit, f filterFunc) []*looprpc.Deposit { ), Outpoint: outpoint, Value: int64(d.Value), - ConfirmationHeight: d.ConfirmationHeight, + ConfirmationHeight: d.GetConfirmationHeight(), SwapHash: swapHash, } diff --git a/loopd/swapclient_server_staticaddr_test.go b/loopd/swapclient_server_staticaddr_test.go index 3d229a60..bb4cc01c 100644 --- a/loopd/swapclient_server_staticaddr_test.go +++ b/loopd/swapclient_server_staticaddr_test.go @@ -4,6 +4,7 @@ import ( "context" "testing" + "github.com/btcsuite/btcd/btcec/v2" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" @@ -13,6 +14,7 @@ import ( "github.com/lightninglabs/loop/staticaddr/deposit" "github.com/lightninglabs/loop/staticaddr/script" mock_lnd "github.com/lightninglabs/loop/test" + "github.com/lightningnetwork/lnd/lnwallet" "github.com/stretchr/testify/require" ) @@ -60,6 +62,33 @@ func (s *staticAddrDepositStore) AllDeposits(context.Context) ( return s.allDeposits, nil } +type staticAddrTestAddressManager struct{} + +func (s *staticAddrTestAddressManager) GetStaticAddressParameters( + context.Context) (*script.Parameters, error) { + + return nil, nil +} + +func (s *staticAddrTestAddressManager) GetStaticAddress( + context.Context) (*script.StaticAddress, error) { + + return nil, nil +} + +func (s *staticAddrTestAddressManager) ListUnspent(context.Context, + int32, int32) ([]*lnwallet.Utxo, error) { + + return nil, nil +} + +func (s *staticAddrTestAddressManager) GetTaprootAddress( + *btcec.PublicKey, *btcec.PublicKey, int64) (*btcutil.AddressTaproot, + error) { + + return nil, nil +} + // newTestDepositManager creates a deposit manager backed by seeded deposits. func newTestDepositManager( deposits ...*deposit.Deposit) *deposit.Manager { @@ -70,6 +99,7 @@ func newTestDepositManager( } return deposit.NewManager(&deposit.ManagerConfig{ + AddressManager: &staticAddrTestAddressManager{}, Store: &staticAddrDepositStore{ allDeposits: deposits, byOutpoint: byOutpoint, @@ -106,6 +136,79 @@ func newTestStaticAddressContext(t *testing.T) (*address.Manager, return addrMgr, mock } +// TestListStaticAddressDepositsReturnsVisibleDeposits verifies normal deposit +// listings include visible deposit records. +func TestListStaticAddressDepositsReturnsVisibleDeposits(t *testing.T) { + t.Parallel() + + available := &deposit.Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{2}, + Index: 2, + }, + } + available.SetState(deposit.Deposited) + + addrMgr, lnd := newTestStaticAddressContext(t) + server := &swapClientServer{ + depositManager: newTestDepositManager(available), + staticAddressManager: addrMgr, + lnd: &lnd.LndServices, + } + + resp, err := server.ListStaticAddressDeposits( + context.Background(), &looprpc.ListStaticAddressDepositsRequest{}, + ) + require.NoError(t, err) + require.Len(t, resp.FilteredDeposits, 1) + require.Equal( + t, available.OutPoint.String(), + resp.FilteredDeposits[0].Outpoint, + ) +} + +// TestGetStaticAddressSummaryTotalsDeposits verifies visible deposits are +// included in static address summary totals. +func TestGetStaticAddressSummaryTotalsDeposits(t *testing.T) { + t.Parallel() + + unconfirmed := &deposit.Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{4}, + Index: 4, + }, + Value: btcutil.Amount(2_000), + ConfirmationHeight: 0, + } + unconfirmed.SetState(deposit.Deposited) + + confirmed := &deposit.Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{5}, + Index: 5, + }, + Value: btcutil.Amount(3_000), + ConfirmationHeight: 123, + } + confirmed.SetState(deposit.Deposited) + + addrMgr, _ := newTestStaticAddressContext(t) + server := &swapClientServer{ + depositManager: newTestDepositManager( + unconfirmed, confirmed, + ), + staticAddressManager: addrMgr, + } + + resp, err := server.GetStaticAddressSummary( + context.Background(), &looprpc.StaticAddressSummaryRequest{}, + ) + require.NoError(t, err) + require.EqualValues(t, 2, resp.TotalNumDeposits) + require.EqualValues(t, 2_000, resp.ValueUnconfirmedSatoshis) + require.EqualValues(t, 3_000, resp.ValueDepositedSatoshis) +} + // TestGetLoopInQuoteRejectsUnavailableSelectedDeposit verifies manual quote // requests fail for selected deposits that are no longer available. func TestGetLoopInQuoteRejectsUnavailableSelectedDeposit(t *testing.T) { diff --git a/staticaddr/deposit/actions.go b/staticaddr/deposit/actions.go index 437c3ac2..77560eb2 100644 --- a/staticaddr/deposit/actions.go +++ b/staticaddr/deposit/actions.go @@ -139,7 +139,7 @@ func (f *FSM) WaitForExpirySweepAction(ctx context.Context, spendChan, errSpendChan, err := f.cfg.ChainNotifier.RegisterConfirmationsNtfn( //nolint:lll ctx, txID, f.deposit.TimeOutSweepPkScript, DefaultConfTarget, - int32(f.deposit.ConfirmationHeight), + int32(f.deposit.GetConfirmationHeight()), ) if err != nil { return f.HandleError(err) diff --git a/staticaddr/deposit/deposit.go b/staticaddr/deposit/deposit.go index d9625660..165a3754 100644 --- a/staticaddr/deposit/deposit.go +++ b/staticaddr/deposit/deposit.go @@ -33,6 +33,9 @@ func (r *ID) FromByteSlice(b []byte) error { // Lock order: if both Manager.mu and a Deposit lock are needed, acquire // Manager.mu before Deposit.Lock. Never acquire Manager.mu while holding a // Deposit lock. +// +// The state and ConfirmationHeight fields are mutable and protected by the +// deposit lock. type Deposit struct { sync.Mutex @@ -94,6 +97,20 @@ func (d *Deposit) IsExpired(currentHeight, expiry uint32) bool { return currentHeight >= uint32(d.ConfirmationHeight)+expiry } +// GetConfirmationHeight returns the deposit confirmation height. +func (d *Deposit) GetConfirmationHeight() int64 { + d.Lock() + defer d.Unlock() + + return d.ConfirmationHeight +} + +// GetConfirmationHeightNoLock returns the deposit confirmation height without +// acquiring the deposit lock. +func (d *Deposit) GetConfirmationHeightNoLock() int64 { + return d.ConfirmationHeight +} + func (d *Deposit) GetState() fsm.StateType { d.Lock() defer d.Unlock() diff --git a/staticaddr/deposit/manager.go b/staticaddr/deposit/manager.go index 55f0d7da..ec51e72e 100644 --- a/staticaddr/deposit/manager.go +++ b/staticaddr/deposit/manager.go @@ -285,6 +285,15 @@ func (m *Manager) pollDeposits(ctx context.Context) { }() } +// EnsureDepositsFresh reconciles the cached active deposit set with lnd's +// current wallet view. Spending paths call this before selecting deposits so +// stale persisted records are not treated as live funds. This can happen when +// an unconfirmed funding transaction is replaced, a confirmed deposit is +// reorged out, or the output was spent outside the active manager path. +func (m *Manager) EnsureDepositsFresh(ctx context.Context) error { + return m.reconcileDeposits(ctx) +} + // reconcileDeposits fetches all spends to our static addresses from our lnd // wallet and matches it against the deposits in our memory that we've seen so // far. It picks the newly identified deposits and starts a state machine per @@ -306,6 +315,11 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error { "confirmations: %w", err) } + err = m.syncActiveDeposits(ctx, utxos) + if err != nil { + return fmt.Errorf("unable to sync active deposits: %w", err) + } + newDeposits := m.filterNewDeposits(utxos) if len(newDeposits) == 0 { log.Tracef("No new deposits...") @@ -359,34 +373,38 @@ func (m *Manager) listUnspentWithBestHeight(ctx context.Context) ( "confirmed deposits") } - const maxAttempts = 3 - for range maxAttempts { - _, beforeHeight, err := m.cfg.ChainKit.GetBestBlock(ctx) - if err != nil { - return nil, 0, fmt.Errorf("unable to get best block "+ - "before listing deposits: %w", err) - } + _, beforeHeight, err := m.cfg.ChainKit.GetBestBlock(ctx) + if err != nil { + return nil, 0, fmt.Errorf("unable to get best block "+ + "before listing deposits: %w", err) + } + + utxos, err = m.cfg.AddressManager.ListUnspent(ctx, 0, MaxConfs) + if err != nil { + return nil, 0, fmt.Errorf("unable to list new deposits: %w", + err) + } + + _, afterHeight, err := m.cfg.ChainKit.GetBestBlock(ctx) + if err != nil { + return nil, 0, fmt.Errorf("unable to get best block "+ + "after listing deposits: %w", err) + } + + if beforeHeight != afterHeight { + log.Debugf("Best block changed from %d to %d while listing "+ + "deposits, relisting deposits at latest height", + beforeHeight, afterHeight) utxos, err = m.cfg.AddressManager.ListUnspent(ctx, 0, MaxConfs) if err != nil { return nil, 0, fmt.Errorf("unable to list new deposits: %w", err) } - - _, afterHeight, err := m.cfg.ChainKit.GetBestBlock(ctx) - if err != nil { - return nil, 0, fmt.Errorf("unable to get best block "+ - "after listing deposits: %w", err) - } - - if beforeHeight == afterHeight { - m.currentHeight.Store(uint32(afterHeight)) - return utxos, afterHeight, nil - } } - return nil, 0, errors.New("unable to get stable best block while " + - "listing deposits") + m.currentHeight.Store(uint32(afterHeight)) + return utxos, afterHeight, nil } // createNewDeposit transforms the wallet utxo into a deposit struct and stores @@ -506,6 +524,92 @@ func (m *Manager) updateDepositConfirmations(ctx context.Context, return nil } +// syncActiveDeposits reconciles the live active set with lnd's current wallet +// view. Known Deposited records that are visible but inactive become active +// again, and active Deposited records that are no longer wallet-visible are +// removed from the live set. The DB record is left untouched as historical +// evidence that the outpoint was once detected. +func (m *Manager) syncActiveDeposits(ctx context.Context, + utxos []*lnwallet.Utxo) error { + + currentUtxos := make(map[wire.OutPoint]struct{}, len(utxos)) + for _, utxo := range utxos { + currentUtxos[utxo.OutPoint] = struct{}{} + } + + type deactivatedDeposit struct { + outpoint wire.OutPoint + fsm *FSM + } + + m.mu.Lock() + toActivate := make([]*Deposit, 0, len(utxos)) + for _, utxo := range utxos { + deposit, ok := m.deposits[utxo.OutPoint] + if !ok { + continue + } + + if _, active := m.activeDeposits[utxo.OutPoint]; active { + continue + } + + if !deposit.IsInState(Deposited) { + continue + } + + toActivate = append(toActivate, deposit) + } + + toDeactivate := make( + []deactivatedDeposit, 0, len(m.activeDeposits), + ) + for outpoint, fsm := range m.activeDeposits { + if _, ok := currentUtxos[outpoint]; ok { + continue + } + + if fsm == nil || fsm.deposit == nil { + continue + } + + if !fsm.deposit.IsInState(Deposited) { + continue + } + + delete(m.activeDeposits, outpoint) + toDeactivate = append(toDeactivate, deactivatedDeposit{ + outpoint: outpoint, + fsm: fsm, + }) + } + m.mu.Unlock() + + for _, deactivated := range toDeactivate { + deactivated.fsm.Stop() + + log.Infof("Removed vanished deposit %v from active set", + deactivated.outpoint) + } + + for _, deposit := range toActivate { + if !deposit.IsInState(Deposited) { + continue + } + + err := m.startDepositFsm(ctx, deposit) + if err != nil { + m.removeActiveDeposit(deposit.OutPoint) + + return err + } + + log.Infof("Reactivated visible deposit %v", deposit.OutPoint) + } + + return nil +} + // filterNewDeposits filters the given utxos for new deposits that we haven't // seen before. func (m *Manager) filterNewDeposits(utxos []*lnwallet.Utxo) []*lnwallet.Utxo { @@ -580,8 +684,8 @@ func (m *Manager) GetActiveDepositsInState(stateFilter fsm.StateType) ( } sort.Slice(filteredDeposits, func(i, j int) bool { - return filteredDeposits[i].ConfirmationHeight < - filteredDeposits[j].ConfirmationHeight + return filteredDeposits[i].GetConfirmationHeightNoLock() < + filteredDeposits[j].GetConfirmationHeightNoLock() }) return filteredDeposits, nil @@ -741,11 +845,41 @@ func (m *Manager) removeActiveDeposit(outpoint wire.OutPoint) { } } -// GetAllDeposits returns all active deposits. +// GetAllDeposits returns all known deposits from the database. func (m *Manager) GetAllDeposits(ctx context.Context) ([]*Deposit, error) { return m.cfg.Store.AllDeposits(ctx) } +// GetVisibleDeposits returns deposits that should be exposed through normal +// user-facing views. The database can contain historical Deposited rows whose +// outpoints are no longer present in lnd's current wallet view, for example +// after replacement or reorg. Once the manager has recovered its live cache, +// plain Deposited records are only visible while their outpoint is in the +// active set. +func (m *Manager) GetVisibleDeposits(ctx context.Context) ([]*Deposit, error) { + deposits, err := m.cfg.Store.AllDeposits(ctx) + if err != nil { + return nil, err + } + + m.mu.Lock() + defer m.mu.Unlock() + + liveCacheReady := len(m.deposits) > 0 + filtered := make([]*Deposit, 0, len(deposits)) + for _, d := range deposits { + if liveCacheReady && d.IsInState(Deposited) { + if _, ok := m.activeDeposits[d.OutPoint]; !ok { + continue + } + } + + filtered = append(filtered, d) + } + + return filtered, nil +} + // UpdateDeposit overrides all fields of the deposit with given ID in the store. func (m *Manager) UpdateDeposit(ctx context.Context, d *Deposit) error { return m.cfg.Store.UpdateDeposit(ctx, d) diff --git a/staticaddr/deposit/manager_reconcile_test.go b/staticaddr/deposit/manager_reconcile_test.go index 6249a72d..169c2f6b 100644 --- a/staticaddr/deposit/manager_reconcile_test.go +++ b/staticaddr/deposit/manager_reconcile_test.go @@ -3,6 +3,7 @@ package deposit import ( "context" "errors" + "strings" "sync" "sync/atomic" "testing" @@ -11,6 +12,7 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/loop/fsm" "github.com/lightninglabs/loop/staticaddr/script" "github.com/lightninglabs/loop/staticaddr/version" "github.com/lightninglabs/loop/test" @@ -108,9 +110,18 @@ func TestReconcileDepositsSerialized(t *testing.T) { } errCount++ - require.ErrorContains(t, err, "unable to start new deposit FSM") + errMsg := err.Error() + require.True( + t, + strings.Contains( + errMsg, "unable to start new deposit FSM", + ) || strings.Contains( + errMsg, "unable to sync active deposits", + ), + "unexpected error: %v", err, + ) } - require.Equal(t, 1, errCount) + require.Equal(t, 2, errCount) } // TestReconcileConfirmedDepositUsesBestBlockHeight verifies confirmation @@ -159,6 +170,75 @@ func TestReconcileConfirmedDepositUsesBestBlockHeight(t *testing.T) { require.ErrorContains(t, err, "unable to start new deposit FSM") } +// TestReconcileConfirmedDepositRelistsOnBlockChange verifies that a block +// arriving while listing deposits does not fail reconciliation. Instead we +// relist deposits and use the latest height. +func TestReconcileConfirmedDepositRelistsOnBlockChange(t *testing.T) { + ctx := context.Background() + mockLnd := test.NewMockLnd() + initialUtxo := &lnwallet.Utxo{ + AddressType: lnwallet.TaprootPubkey, + Value: btcutil.Amount(100_000), + Confirmations: 1, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{18}, + Index: 1, + }, + } + latestUtxo := &lnwallet.Utxo{ + AddressType: lnwallet.TaprootPubkey, + Value: btcutil.Amount(100_000), + Confirmations: 2, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{18}, + Index: 1, + }, + } + + mockAddressManager := new(mockAddressManager) + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{initialUtxo}, nil).Once() + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{initialUtxo}, nil).Once() + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{latestUtxo}, nil).Once() + mockAddressManager.On( + "GetStaticAddressParameters", mock.Anything, + ).Return((*script.Parameters)(nil), errors.New("fsm init failed")) + + mockChainKit := new(MockChainKit) + mockChainKit.On( + "GetBestBlock", mock.Anything, + ).Return(chainhash.Hash{}, int32(100), nil).Once() + mockChainKit.On( + "GetBestBlock", mock.Anything, + ).Return(chainhash.Hash{}, int32(101), nil).Once() + + 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, 100, createdDeposit.ConfirmationHeight) + }) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + ChainKit: mockChainKit, + Store: mockStore, + WalletKit: mockLnd.WalletKit, + Signer: mockLnd.Signer, + }) + + err := manager.reconcileDeposits(ctx) + require.ErrorContains(t, err, "unable to start new deposit FSM") + mockAddressManager.AssertExpectations(t) + mockChainKit.AssertExpectations(t) +} + // 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 @@ -200,6 +280,102 @@ func TestUpdateDepositConfirmationsResetsReorgedDeposit(t *testing.T) { mockStore.AssertExpectations(t) } +// TestReconcileDepositsDeactivatesVanishedUnconfirmedDeposit verifies that a +// missing wallet outpoint is removed from the live active set without mutating +// its historical DB state. +func TestReconcileDepositsDeactivatesVanishedUnconfirmedDeposit(t *testing.T) { + ctx := context.Background() + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{2}, + Index: 7, + } + + deposit := &Deposit{ + OutPoint: outpoint, + } + deposit.SetState(Deposited) + + mockAddressManager := new(mockAddressManager) + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{}, nil) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: new(mockStore), + }) + manager.deposits[outpoint] = deposit + fsm := &FSM{ + deposit: deposit, + stopChan: make(chan struct{}), + quitChan: make(chan struct{}), + } + go func() { + <-fsm.stopChan + close(fsm.quitChan) + }() + manager.activeDeposits[outpoint] = fsm + + require.NoError(t, manager.reconcileDeposits(ctx)) + require.Equal(t, Deposited, deposit.GetState()) + require.Empty(t, manager.activeDeposits) + select { + case <-fsm.quitChan: + + case <-time.After(time.Second): + t.Fatal("fsm did not stop after deposit vanished") + } +} + +// TestReconcileDepositsDeactivatesVanishedConfirmedDeposit verifies that a +// previously confirmed deposit is also removed from the live active set if it +// vanishes from the wallet view. +func TestReconcileDepositsDeactivatesVanishedConfirmedDeposit(t *testing.T) { + ctx := context.Background() + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{9}, + Index: 4, + } + + deposit := &Deposit{ + OutPoint: outpoint, + ConfirmationHeight: 123, + } + deposit.SetState(Deposited) + + mockAddressManager := new(mockAddressManager) + mockAddressManager.On( + "ListUnspent", mock.Anything, int32(0), int32(MaxConfs), + ).Return([]*lnwallet.Utxo{}, nil) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: new(mockStore), + }) + manager.deposits[outpoint] = deposit + fsm := &FSM{ + deposit: deposit, + stopChan: make(chan struct{}), + quitChan: make(chan struct{}), + } + go func() { + <-fsm.stopChan + close(fsm.quitChan) + }() + manager.activeDeposits[outpoint] = fsm + + require.NoError(t, manager.reconcileDeposits(ctx)) + require.Equal(t, Deposited, deposit.GetState()) + require.EqualValues(t, 123, deposit.ConfirmationHeight) + require.Empty(t, manager.activeDeposits) + select { + case <-fsm.quitChan: + + case <-time.After(time.Second): + t.Fatal("fsm did not stop after confirmed deposit vanished") + } +} + // TestAllOutpointsActiveDepositsRejectsDuplicateOutpoints verifies that a // duplicated selection is rejected before the manager tries to lock the same // deposit twice. @@ -249,6 +425,199 @@ func TestTransitionDepositsRejectsDuplicateOutpoints(t *testing.T) { require.Equal(t, Deposited, deposit.GetState()) } +// TestReconcileDepositsReactivatesReappearedDeposit verifies that the same +// outpoint can become active again if lnd reports it after a prior wallet-view +// miss. +func TestReconcileDepositsReactivatesReappearedDeposit(t *testing.T) { + ctx := context.Background() + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{3}, + Index: 5, + } + + deposit := &Deposit{ + OutPoint: outpoint, + Value: btcutil.Amount(100_000), + ConfirmationHeight: 77, + } + deposit.SetState(Deposited) + + utxo := &lnwallet.Utxo{ + OutPoint: outpoint, + 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 updateStates []fsm.StateType + mockStore.On( + "UpdateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(args mock.Arguments) { + updatedDeposit := args.Get(1).(*Deposit) + updateStates = append(updateStates, updatedDeposit.state) + if updatedDeposit.IsInStateNoLock(Deposited) { + require.Zero(t, updatedDeposit.ConfirmationHeight) + } + }) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: mockStore, + }) + manager.deposits[outpoint] = deposit + + // Reconciliation should reactivate the existing record instead of + // creating a second deposit entry for the same outpoint. + require.NoError(t, manager.reconcileDeposits(ctx)) + require.Equal(t, Deposited, deposit.GetState()) + require.Zero(t, deposit.ConfirmationHeight) + require.Len(t, manager.activeDeposits, 1) + require.Equal(t, []fsm.StateType{Deposited}, updateStates) +} + +// TestReconcileDepositsKeepsInactiveOnFSMStartFailure verifies that a failed +// reactivation does not leave memory saying a deposit is active without an FSM. +func TestReconcileDepositsKeepsInactiveOnFSMStartFailure(t *testing.T) { + ctx := context.Background() + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{11}, + Index: 5, + } + + deposit := &Deposit{ + OutPoint: outpoint, + Value: btcutil.Amount(100_000), + ConfirmationHeight: 77, + } + deposit.SetState(Deposited) + + utxo := &lnwallet.Utxo{ + OutPoint: outpoint, + 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)(nil), errors.New("fsm init failed")) + + var ( + updateStates []fsm.StateType + updateHeights []int64 + ) + mockStore := new(mockStore) + mockStore.On( + "UpdateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Run(func(args mock.Arguments) { + updatedDeposit := args.Get(1).(*Deposit) + updateStates = append(updateStates, updatedDeposit.state) + updateHeights = append( + updateHeights, updatedDeposit.ConfirmationHeight, + ) + }) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: mockStore, + }) + manager.deposits[outpoint] = deposit + + err := manager.reconcileDeposits(ctx) + require.ErrorContains(t, err, "unable to sync active deposits") + require.Equal(t, Deposited, deposit.GetState()) + require.Zero(t, deposit.ConfirmationHeight) + require.Empty(t, manager.activeDeposits) + require.Equal(t, []fsm.StateType{Deposited}, updateStates) + require.EqualValues(t, []int64{0}, updateHeights) +} + +// TestReconcileDepositsDeactivatesBeforeActivationFailure verifies that a +// failed reactivation of one visible deposit does not leave another vanished +// deposit in the live active set. +func TestReconcileDepositsDeactivatesBeforeActivationFailure(t *testing.T) { + ctx := context.Background() + visibleOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{21}, + Index: 5, + } + vanishedOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{22}, + Index: 6, + } + + visibleDeposit := &Deposit{ + OutPoint: visibleOutpoint, + Value: btcutil.Amount(100_000), + } + visibleDeposit.SetState(Deposited) + + vanishedDeposit := &Deposit{ + OutPoint: vanishedOutpoint, + Value: btcutil.Amount(100_000), + } + vanishedDeposit.SetState(Deposited) + + utxo := &lnwallet.Utxo{ + OutPoint: visibleOutpoint, + Value: visibleDeposit.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)(nil), errors.New("fsm init failed")) + + manager := NewManager(&ManagerConfig{ + AddressManager: mockAddressManager, + Store: new(mockStore), + }) + manager.deposits[visibleOutpoint] = visibleDeposit + manager.deposits[vanishedOutpoint] = vanishedDeposit + + vanishedFsm := &FSM{ + deposit: vanishedDeposit, + stopChan: make(chan struct{}), + quitChan: make(chan struct{}), + } + go func() { + <-vanishedFsm.stopChan + close(vanishedFsm.quitChan) + }() + manager.activeDeposits[vanishedOutpoint] = vanishedFsm + + err := manager.reconcileDeposits(ctx) + require.ErrorContains(t, err, "unable to sync active deposits") + require.Empty(t, manager.activeDeposits) + + select { + case <-vanishedFsm.quitChan: + + case <-time.After(time.Second): + t.Fatal("vanished deposit fsm did not stop") + } +} + // 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. diff --git a/staticaddr/deposit/sql_store.go b/staticaddr/deposit/sql_store.go index d9f4249f..dbaaa70f 100644 --- a/staticaddr/deposit/sql_store.go +++ b/staticaddr/deposit/sql_store.go @@ -47,7 +47,7 @@ func (s *SqlStore) CreateDeposit(ctx context.Context, deposit *Deposit) error { TxHash: deposit.Hash[:], OutIndex: int32(deposit.Index), Amount: int64(deposit.Value), - ConfirmationHeight: deposit.ConfirmationHeight, + ConfirmationHeight: deposit.GetConfirmationHeight(), TimeoutSweepPkScript: deposit.TimeOutSweepPkScript, } @@ -83,7 +83,7 @@ func (s *SqlStore) UpdateDeposit(ctx context.Context, deposit *Deposit) error { Valid: true, } confirmationHeight = sql.NullInt64{ - Int64: deposit.ConfirmationHeight, + Int64: deposit.GetConfirmationHeightNoLock(), } ) diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index d75adf6c..ee935de8 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -479,6 +479,11 @@ func (m *mockAddressManager) GetStaticAddress(_ context.Context) ( // noopDepositManager is a stub DepositManager used to satisfy FSM config. type noopDepositManager struct{} +// EnsureDepositsFresh implements DepositManager with a no-op. +func (n *noopDepositManager) EnsureDepositsFresh(context.Context) error { + return nil +} + // GetAllDeposits implements DepositManager with a no-op. func (n *noopDepositManager) GetAllDeposits(_ context.Context) ( []*deposit.Deposit, error) { diff --git a/staticaddr/loopin/autoloop.go b/staticaddr/loopin/autoloop.go index 337d73e1..0bfddaa4 100644 --- a/staticaddr/loopin/autoloop.go +++ b/staticaddr/loopin/autoloop.go @@ -30,6 +30,11 @@ func (m *Manager) PrepareAutoloopLoopIn(ctx context.Context, return nil, 0, false, ErrNoAutoloopCandidate } + err := m.cfg.DepositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, 0, false, err + } + allDeposits, err := m.cfg.DepositManager.GetActiveDepositsInState( deposit.Deposited, ) diff --git a/staticaddr/loopin/autoloop_dp.go b/staticaddr/loopin/autoloop_dp.go index d8abde50..86e67fd1 100644 --- a/staticaddr/loopin/autoloop_dp.go +++ b/staticaddr/loopin/autoloop_dp.go @@ -234,9 +234,9 @@ func filterAutoloopCandidateDeposits(maxAmount btcutil.Amount, continue } + confirmationHeight := candidateDeposit.GetConfirmationHeight() swappable := IsSwappable( - uint32(candidateDeposit.ConfirmationHeight), - blockHeight, csvExpiry, + uint32(confirmationHeight), blockHeight, csvExpiry, ) if !swappable { continue @@ -247,8 +247,7 @@ func filterAutoloopCandidateDeposits(maxAmount btcutil.Amount, } residualLife := int64(blocksUntilDepositExpiry( - uint32(candidateDeposit.ConfirmationHeight), - blockHeight, csvExpiry, + uint32(confirmationHeight), blockHeight, csvExpiry, )) eligibleDeposits = append( diff --git a/staticaddr/loopin/interface.go b/staticaddr/loopin/interface.go index c4bbb2b7..f9b5f3ce 100644 --- a/staticaddr/loopin/interface.go +++ b/staticaddr/loopin/interface.go @@ -44,6 +44,9 @@ type AddressManager interface { // DepositManager handles the interaction of loop-ins with deposits. type DepositManager interface { + // EnsureDepositsFresh reconciles active deposits with the wallet view. + EnsureDepositsFresh(ctx context.Context) error + // GetAllDeposits returns all known deposits from the database store. GetAllDeposits(ctx context.Context) ([]*deposit.Deposit, error) diff --git a/staticaddr/loopin/manager.go b/staticaddr/loopin/manager.go index a78adfe4..50512f98 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -628,6 +628,11 @@ func (m *Manager) initiateLoopIn(ctx context.Context, selectedDeposits []*deposit.Deposit ) + err = m.cfg.DepositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, fmt.Errorf("unable to refresh deposits: %w", err) + } + // Determine which deposits to use for the loop-in swap. If none are // selected by the client, we will coin-select them based on the amount. switch { @@ -858,8 +863,9 @@ func SelectDeposits(targetAmount btcutil.Amount, // Filter out deposits that are too close to expiry to be swapped. var deposits []*deposit.Deposit for _, d := range unfilteredDeposits { + confirmationHeight := d.GetConfirmationHeight() if !IsSwappable( - uint32(d.ConfirmationHeight), blockHeight, csvExpiry, + uint32(confirmationHeight), blockHeight, csvExpiry, ) { log.Debugf("Skipping deposit %s as it expires before "+ @@ -875,20 +881,22 @@ func SelectDeposits(targetAmount btcutil.Amount, // prefers deposits the server can accept immediately. Within each group // we prefer larger deposits, then earlier expiries. sort.Slice(deposits, func(i, j int) bool { - iConfirmed := deposits[i].ConfirmationHeight > 0 - jConfirmed := deposits[j].ConfirmationHeight > 0 + iConfirmationHeight := deposits[i].GetConfirmationHeight() + jConfirmationHeight := deposits[j].GetConfirmationHeight() + iConfirmed := iConfirmationHeight > 0 + jConfirmed := jConfirmationHeight > 0 if iConfirmed != jConfirmed { return iConfirmed } if deposits[i].Value == deposits[j].Value { iExp := blocksUntilDepositExpiry( - uint32(deposits[i].ConfirmationHeight), - blockHeight, csvExpiry, + uint32(iConfirmationHeight), blockHeight, + csvExpiry, ) jExp := blocksUntilDepositExpiry( - uint32(deposits[j].ConfirmationHeight), - blockHeight, csvExpiry, + uint32(jConfirmationHeight), blockHeight, + csvExpiry, ) return iExp < jExp diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 58549747..cb7eb7e7 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -334,6 +334,10 @@ type mockDepositManager struct { byOutpoint map[string]*deposit.Deposit } +func (m *mockDepositManager) EnsureDepositsFresh(context.Context) error { + return nil +} + func (m *mockDepositManager) GetAllDeposits(_ context.Context) ( []*deposit.Deposit, error) { diff --git a/staticaddr/openchannel/interface.go b/staticaddr/openchannel/interface.go index 73010a2b..ee6ed31a 100644 --- a/staticaddr/openchannel/interface.go +++ b/staticaddr/openchannel/interface.go @@ -12,6 +12,9 @@ import ( ) type DepositManager interface { + // EnsureDepositsFresh reconciles active deposits with the wallet view. + EnsureDepositsFresh(ctx context.Context) error + // AllOutpointsActiveDeposits returns all deposits that are in the // given state. If the state filter is fsm.StateTypeNone, all deposits // are returned. diff --git a/staticaddr/openchannel/manager.go b/staticaddr/openchannel/manager.go index ba8e636e..14f76528 100644 --- a/staticaddr/openchannel/manager.go +++ b/staticaddr/openchannel/manager.go @@ -267,11 +267,6 @@ func (m *Manager) OpenChannel(ctx context.Context, ).FeePerKWeight() } - // There are three ways in which we select deposits to open a channel - // with. 1.) The user manually selects the deposits. 2.) The user only - // selects a local channel amount in which case we coin-select deposits - // to cover for it. 3.) The user selects the fundmax flag, in which case - // we select all deposits to fund the channel. if len(req.Outpoints) > 0 { // Ensure that the deposits are in a state in which they are // available for a channel open. @@ -293,6 +288,12 @@ func (m *Manager) OpenChannel(ctx context.Context, seen[op] = struct{}{} } + err = m.cfg.DepositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, fmt.Errorf("unable to refresh deposits: %w", + err) + } + deposits, allActive = m.cfg.DepositManager.AllOutpointsActiveDeposits( outpoints, deposit.Deposited, @@ -301,6 +302,12 @@ func (m *Manager) OpenChannel(ctx context.Context, return nil, ErrOpeningChannelUnavailableDeposits } } else { + err = m.cfg.DepositManager.EnsureDepositsFresh(ctx) + if err != nil { + return nil, fmt.Errorf("unable to refresh deposits: %w", + err) + } + // We have to select the deposits that are used to fund the // channel. deposits, err = m.cfg.DepositManager.GetActiveDepositsInState( @@ -331,7 +338,7 @@ func (m *Manager) OpenChannel(ctx context.Context, for _, d := range deposits { // Deposited now includes mempool outputs for static loop-ins, but // channel opens still require the deposit input to be confirmed. - if d.ConfirmationHeight <= 0 { + if d.GetConfirmationHeight() <= 0 { return nil, ErrOpeningChannelUnavailableDeposits } } @@ -417,7 +424,7 @@ func (m *Manager) OpenChannel(ctx context.Context, func filterConfirmedDeposits(deposits []*deposit.Deposit) []*deposit.Deposit { confirmed := make([]*deposit.Deposit, 0, len(deposits)) for _, d := range deposits { - if d.ConfirmationHeight <= 0 { + if d.GetConfirmationHeight() <= 0 { continue } diff --git a/staticaddr/openchannel/manager_test.go b/staticaddr/openchannel/manager_test.go index 7250498c..d46c8c00 100644 --- a/staticaddr/openchannel/manager_test.go +++ b/staticaddr/openchannel/manager_test.go @@ -36,6 +36,10 @@ type mockDepositManager struct { calls []transitionCall } +func (m *mockDepositManager) EnsureDepositsFresh(context.Context) error { + return nil +} + func (m *mockDepositManager) AllOutpointsActiveDeposits([]wire.OutPoint, fsm.StateType) ([]*deposit.Deposit, bool) { diff --git a/staticaddr/withdraw/interface.go b/staticaddr/withdraw/interface.go index ff878488..0f32697a 100644 --- a/staticaddr/withdraw/interface.go +++ b/staticaddr/withdraw/interface.go @@ -21,14 +21,24 @@ type AddressManager interface { } type DepositManager interface { + // EnsureDepositsFresh reconciles active deposits with the wallet view. + EnsureDepositsFresh(ctx context.Context) error + + // GetActiveDepositsInState returns all active deposits in the given + // state. GetActiveDepositsInState(stateFilter fsm.StateType) ([]*deposit.Deposit, error) + // AllOutpointsActiveDeposits returns all active deposits referenced by + // the outpoints if every deposit is active and in the given state. AllOutpointsActiveDeposits(outpoints []wire.OutPoint, stateFilter fsm.StateType) ([]*deposit.Deposit, bool) + // TransitionDeposits transitions the deposits with the given event and + // waits until they reach the expected final state. TransitionDeposits(ctx context.Context, deposits []*deposit.Deposit, event fsm.EventType, expectedFinalState fsm.StateType) error + // UpdateDeposit persists the current deposit fields. UpdateDeposit(ctx context.Context, d *deposit.Deposit) error } diff --git a/staticaddr/withdraw/manager.go b/staticaddr/withdraw/manager.go index 7f085c6c..3a7927a4 100644 --- a/staticaddr/withdraw/manager.go +++ b/staticaddr/withdraw/manager.go @@ -320,6 +320,11 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, allWithdrawing bool ) + err := m.cfg.DepositManager.EnsureDepositsFresh(ctx) + if err != nil { + return "", "", fmt.Errorf("unable to refresh deposits: %w", err) + } + // Ensure that the deposits are in a state in which they can be // withdrawn. deposits, allDeposited = m.cfg.DepositManager.AllOutpointsActiveDeposits( @@ -384,16 +389,13 @@ func (m *Manager) WithdrawDeposits(ctx context.Context, for _, d := range deposits { // Deposited now includes mempool outputs for static loop-ins, but // withdrawals still require the deposit input to be confirmed. - if d.ConfirmationHeight <= 0 { + if d.GetConfirmationHeight() <= 0 { return "", "", fmt.Errorf("can't withdraw, " + "unconfirmed deposits can't be withdrawn") } } - var ( - withdrawalAddress btcutil.Address - err error - ) + var withdrawalAddress btcutil.Address // Check if the user provided an address to withdraw to. If not, we'll // generate a new address for them. @@ -678,7 +680,7 @@ func (m *Manager) handleWithdrawal(ctx context.Context, d := deposits[0] spentChan, errChan, err := m.cfg.ChainNotifier.RegisterSpendNtfn( ctx, &d.OutPoint, addrParams.PkScript, - int32(d.ConfirmationHeight), + int32(d.GetConfirmationHeight()), ) if err != nil { return fmt.Errorf("unable to register spend ntfn: %w", err)