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)