diff --git a/.golangci.yml b/.golangci.yml index 5b870ccb..329dcde2 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -3,7 +3,7 @@ run: go: "1.26" # timeout for analysis - timeout: 4m + timeout: 6m linters: default: all diff --git a/go.mod b/go.mod index 1e312f90..c853da92 100644 --- a/go.mod +++ b/go.mod @@ -238,4 +238,10 @@ replace gonum.org/v1/plot => github.com/gonum/plot v0.10.1 // checking later if the domain reappears and replace can be removed. replace dario.cat/mergo => github.com/darccio/mergo v1.0.1 +// The go.augendre.info vanity import endpoints are currently unavailable, +// but these tags still declare the original module paths. +replace go.augendre.info/arangolint => github.com/Crocmagnon/arangolint v0.4.0 + +replace go.augendre.info/fatcontext => github.com/Crocmagnon/fatcontext v0.9.0 + go 1.25.10 diff --git a/loopd/daemon.go b/loopd/daemon.go index 91267811..dd29d1a5 100644 --- a/loopd/daemon.go +++ b/loopd/daemon.go @@ -692,6 +692,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error { Server: staticAddressClient, QuoteGetter: swapClient.Server, LndClient: d.lnd.Client, + TxOutChecker: loopin.NewLndTxOutChecker(d.lnd.Client), InvoicesClient: d.lnd.Invoices, NodePubkey: d.lnd.NodePubkey, AddressManager: staticAddressManager, diff --git a/loopd/swapclient_server.go b/loopd/swapclient_server.go index 86d44a52..7bf5ab53 100644 --- a/loopd/swapclient_server.go +++ b/loopd/swapclient_server.go @@ -1896,7 +1896,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(), ), @@ -1987,7 +1987,8 @@ func (s *swapClientServer) ListStaticAddressSwaps(ctx context.Context, protoDeposits = make([]*looprpc.Deposit, 0, len(ds)) for _, d := range ds { state := toClientDepositState(d.GetState()) - blocksUntilExpiry := d.ConfirmationHeight + + confirmationHeight := d.GetConfirmationHeight() + blocksUntilExpiry := confirmationHeight + int64(addrParams.Expiry) - int64(lndInfo.BlockHeight) @@ -1996,7 +1997,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, } @@ -2304,7 +2305,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/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 4cb64bc9..19c5c2b0 100644 --- a/staticaddr/deposit/deposit.go +++ b/staticaddr/deposit/deposit.go @@ -29,6 +29,13 @@ func (r *ID) FromByteSlice(b []byte) error { // Deposit bundles an utxo at a static address together with manager-relevant // data. +// +// 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 @@ -69,6 +76,12 @@ func (d *Deposit) IsInFinalState() bool { d.Lock() defer d.Unlock() + return d.isInFinalStateNoLock() +} + +// isInFinalStateNoLock returns true if the deposit is final without acquiring +// the deposit lock. +func (d *Deposit) isInFinalStateNoLock() bool { return d.state == Expired || d.state == Withdrawn || d.state == LoopedIn || d.state == HtlcTimeoutSwept || d.state == ChannelPublished @@ -88,6 +101,10 @@ func (d *Deposit) GetState() fsm.StateType { return d.state } +func (d *Deposit) getStateNoLock() fsm.StateType { + return d.state +} + func (d *Deposit) SetState(state fsm.StateType) { d.Lock() defer d.Unlock() @@ -95,7 +112,7 @@ func (d *Deposit) SetState(state fsm.StateType) { d.state = state } -func (d *Deposit) SetStateNoLock(state fsm.StateType) { +func (d *Deposit) setStateNoLock(state fsm.StateType) { d.state = state } @@ -106,10 +123,24 @@ func (d *Deposit) IsInState(state fsm.StateType) bool { return d.state == state } -func (d *Deposit) IsInStateNoLock(state fsm.StateType) bool { +func (d *Deposit) isInStateNoLock(state fsm.StateType) bool { return d.state == state } +// 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 +} + // GetRandomDepositID generates a random deposit ID. func GetRandomDepositID() (ID, error) { var id ID diff --git a/staticaddr/deposit/fsm.go b/staticaddr/deposit/fsm.go index 6dadd127..7dd8715d 100644 --- a/staticaddr/deposit/fsm.go +++ b/staticaddr/deposit/fsm.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" @@ -160,6 +161,12 @@ type FSM struct { blockNtfnChan chan uint32 + // stopChan requests shutdown of the block notification loop. + stopChan chan struct{} + + // stopOnce ensures Stop is idempotent. + stopOnce sync.Once + // quitChan stops after the FSM stops consuming blockNtfnChan. quitChan chan struct{} @@ -191,6 +198,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, params: params, address: address, blockNtfnChan: make(chan uint32), + stopChan: make(chan struct{}), quitChan: make(chan struct{}), finalizedDepositChan: finalizedDepositChan, } @@ -226,6 +234,9 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, ctx, currentHeight, ) + case <-fsm.stopChan: + return + case <-ctx.Done(): return } @@ -235,12 +246,27 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, return depoFsm, nil } +// Stop requests shutdown of the FSM's block notification loop. +func (f *FSM) Stop() { + if f == nil || f.stopChan == nil { + return + } + + f.stopOnce.Do(func() { + close(f.stopChan) + }) +} + // handleBlockNotification inspects the current block height and sends the // OnExpiry event to publish the expiry sweep transaction if the deposit timed // out, or it republishes the expiry sweep transaction if it was not yet swept. func (f *FSM) handleBlockNotification(ctx context.Context, currentHeight uint32) { + if f.deposit.IsInFinalState() { + return + } + // If the deposit is expired but not yet sufficiently confirmed, we // republish the expiry sweep transaction. if f.deposit.IsExpired(currentHeight, f.params.Expiry) { @@ -353,6 +379,11 @@ func (f *FSM) DepositStatesV0() fsm.States { // still pending, we publish the expiry sweep. OnExpiry: PublishExpirySweep, + // If the server publishes the HTLC without + // paying us, we need to keep the deposit locked + // until the HTLC timeout path can be swept. + OnSweepingHtlcTimeout: SweepHtlcTimeout, + OnLoopInInitiated: LoopingIn, OnRecover: LoopingIn, @@ -362,7 +393,7 @@ func (f *FSM) DepositStatesV0() fsm.States { }, LoopedIn: fsm.State{ Transitions: fsm.Transitions{ - OnExpiry: Expired, + OnExpiry: LoopedIn, }, Action: f.FinalizeDepositAction, }, @@ -381,7 +412,7 @@ func (f *FSM) DepositStatesV0() fsm.States { }, Withdrawn: fsm.State{ Transitions: fsm.Transitions{ - OnExpiry: Expired, + OnExpiry: Withdrawn, OnWithdrawn: Withdrawn, }, Action: f.FinalizeDepositAction, @@ -421,17 +452,14 @@ func (f *FSM) updateDeposit(ctx context.Context, return } - type checkStateFunc func(state fsm.StateType) bool - type setStateFunc func(state fsm.StateType) - checkFunc := checkStateFunc(f.deposit.IsInState) - setFunc := setStateFunc(f.deposit.SetState) - if _, ok := lockedEvents[notification.Event]; ok { - checkFunc = f.deposit.IsInStateNoLock - setFunc = f.deposit.SetStateNoLock + _, alreadyLocked := lockedEvents[notification.Event] + if !alreadyLocked { + f.deposit.Lock() + defer f.deposit.Unlock() } - setFunc(notification.NextState) - if isUpdateSkipped(notification, checkFunc) { + f.deposit.setStateNoLock(notification.NextState) + if isUpdateSkipped(notification, f.deposit.isInStateNoLock) { return } diff --git a/staticaddr/deposit/fsm_test.go b/staticaddr/deposit/fsm_test.go new file mode 100644 index 00000000..f174ad1d --- /dev/null +++ b/staticaddr/deposit/fsm_test.go @@ -0,0 +1,159 @@ +package deposit + +import ( + "testing" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/loop/fsm" + "github.com/lightninglabs/loop/staticaddr/script" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// TestHandleBlockNotificationIgnoresFinalStates verifies that a block-driven +// expiry notification cannot mutate deposits that already reached a final +// state but have not yet been removed from the manager's active set. +func TestHandleBlockNotificationIgnoresFinalStates(t *testing.T) { + t.Parallel() + + finalStates := []fsm.StateType{ + Expired, + Withdrawn, + LoopedIn, + HtlcTimeoutSwept, + ChannelPublished, + } + + for i, state := range finalStates { + t.Run(string(state), func(t *testing.T) { + t.Parallel() + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{byte(i + 1)}, + Index: uint32(i), + } + deposit := &Deposit{ + OutPoint: outpoint, + ConfirmationHeight: 1, + } + deposit.SetState(state) + + depositFSM := &FSM{ + cfg: &ManagerConfig{ + Store: new(mockStore), + }, + deposit: deposit, + params: &script.Parameters{Expiry: 1}, + quitChan: make(chan struct{}), + finalizedDepositChan: make(chan wire.OutPoint, 1), + } + depositFSM.StateMachine = fsm.NewStateMachineWithState( + depositFSM.DepositStatesV0(), state, + DefaultObserverSize, + ) + depositFSM.ActionEntryFunc = depositFSM.updateDeposit + + depositFSM.handleBlockNotification(t.Context(), 3) + + require.Never(t, func() bool { + return deposit.GetState() != state + }, 100*time.Millisecond, 10*time.Millisecond) + + select { + case finalized := <-depositFSM.finalizedDepositChan: + t.Fatalf("unexpected finalization for %v", finalized) + + default: + } + }) + } +} + +// TestFinalStatesIgnoreQueuedExpiry verifies that a queued OnExpiry event cannot +// overwrite a deposit that already reached a final state. +func TestFinalStatesIgnoreQueuedExpiry(t *testing.T) { + t.Parallel() + + finalStates := []fsm.StateType{ + Expired, + Withdrawn, + LoopedIn, + HtlcTimeoutSwept, + ChannelPublished, + } + + for i, state := range finalStates { + t.Run(string(state), func(t *testing.T) { + t.Parallel() + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{byte(i + 1)}, + Index: uint32(i), + } + deposit := &Deposit{ + OutPoint: outpoint, + } + deposit.SetState(state) + + depositFSM := &FSM{ + cfg: &ManagerConfig{ + Store: new(mockStore), + }, + deposit: deposit, + quitChan: make(chan struct{}), + finalizedDepositChan: make(chan wire.OutPoint, 1), + } + depositFSM.StateMachine = fsm.NewStateMachineWithState( + depositFSM.DepositStatesV0(), state, + DefaultObserverSize, + ) + depositFSM.ActionEntryFunc = depositFSM.updateDeposit + + err := depositFSM.SendEvent(t.Context(), OnExpiry, nil) + require.NoError(t, err) + require.Equal(t, state, deposit.GetState()) + }) + } +} + +// TestLoopingInTransitionsToSweepHtlcTimeout verifies that a deposit selected +// by a loop-in can be moved into the timeout sweep state if the server confirms +// the HTLC without paying the invoice. +func TestLoopingInTransitionsToSweepHtlcTimeout(t *testing.T) { + t.Parallel() + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{9}, + Index: 9, + } + deposit := &Deposit{ + OutPoint: outpoint, + } + deposit.SetState(LoopingIn) + + store := new(mockStore) + store.On( + "UpdateDeposit", mock.Anything, mock.Anything, + ).Return(nil).Once() + + depositFSM := &FSM{ + cfg: &ManagerConfig{ + Store: store, + }, + deposit: deposit, + params: &script.Parameters{Expiry: 1}, + } + depositFSM.StateMachine = fsm.NewStateMachineWithState( + depositFSM.DepositStatesV0(), LoopingIn, DefaultObserverSize, + ) + depositFSM.ActionEntryFunc = depositFSM.updateDeposit + + err := depositFSM.SendEvent( + t.Context(), OnSweepingHtlcTimeout, nil, + ) + require.NoError(t, err) + require.Equal(t, SweepHtlcTimeout, deposit.GetState()) + store.AssertExpectations(t) +} diff --git a/staticaddr/deposit/manager.go b/staticaddr/deposit/manager.go index af882030..103b51c5 100644 --- a/staticaddr/deposit/manager.go +++ b/staticaddr/deposit/manager.go @@ -58,12 +58,21 @@ type ManagerConfig struct { } // Manager manages the address state machines. +// +// 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. Multiple deposits must be locked with lockDeposits, which +// canonicalizes lock order by outpoint. type Manager struct { cfg *ManagerConfig // mu guards access to the activeDeposits map. mu sync.Mutex + // reconcileMu serializes deposit reconciliation so new deposits are + // discovered and retained exactly once per outpoint. + reconcileMu sync.Mutex + // activeDeposits contains all the active static address outputs. activeDeposits map[wire.OutPoint]*FSM @@ -123,32 +132,15 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { for { select { case height := <-newBlockChan: - // Inform all active deposits about a new block arrival. - m.mu.Lock() - activeDeposits := make([]*FSM, 0, len(m.activeDeposits)) - for _, fsm := range m.activeDeposits { - activeDeposits = append(activeDeposits, fsm) - } - m.mu.Unlock() - - for _, fsm := range activeDeposits { - select { - case fsm.blockNtfnChan <- uint32(height): - - case <-fsm.quitChan: - continue - - case <-ctx.Done(): - return ctx.Err() - } + err := m.notifyActiveDeposits(ctx, uint32(height)) + if err != nil { + return err } case outpoint := <-m.finalizedDepositChan: // If deposits notify us about their finalization, flush // the finalized deposit from memory. - m.mu.Lock() - delete(m.activeDeposits, outpoint) - m.mu.Unlock() + m.removeActiveDeposit(outpoint) case err = <-newBlockErrChan: return err @@ -159,6 +151,33 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error { } } +// notifyActiveDeposits informs all active deposit FSMs about a new block +// height. +func (m *Manager) notifyActiveDeposits(ctx context.Context, + height uint32) error { + + m.mu.Lock() + activeDeposits := make([]*FSM, 0, len(m.activeDeposits)) + for _, fsm := range m.activeDeposits { + activeDeposits = append(activeDeposits, fsm) + } + m.mu.Unlock() + + for _, fsm := range activeDeposits { + select { + case fsm.blockNtfnChan <- height: + + case <-fsm.quitChan: + continue + + case <-ctx.Done(): + return ctx.Err() + } + } + + return nil +} + // recoverDeposits recovers static address parameters, previous deposits and // state machines from the database and starts the deposit notifier. func (m *Manager) recoverDeposits(ctx context.Context) error { @@ -236,6 +255,9 @@ func (m *Manager) pollDeposits(ctx context.Context) { // far. It picks the newly identified deposits and starts a state machine per // deposit to track its progress. func (m *Manager) reconcileDeposits(ctx context.Context) error { + m.reconcileMu.Lock() + defer m.reconcileMu.Unlock() + log.Tracef("Reconciling new deposits...") utxos, err := m.cfg.AddressManager.ListUnspent( @@ -412,12 +434,12 @@ func (m *Manager) GetActiveDepositsInState(stateFilter fsm.StateType) ( deposits = append(deposits, fsm.deposit) } - lockDeposits(deposits) - defer unlockDeposits(deposits) + lockedDeposits := lockDeposits(deposits) + defer unlockDeposits(lockedDeposits) filteredDeposits := make([]*Deposit, 0, len(deposits)) for _, d := range deposits { - if !d.IsInStateNoLock(stateFilter) { + if !d.isInStateNoLock(stateFilter) { continue } @@ -425,8 +447,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 @@ -439,6 +461,10 @@ func (m *Manager) GetActiveDepositsInState(stateFilter fsm.StateType) ( func (m *Manager) AllOutpointsActiveDeposits(outpoints []wire.OutPoint, targetState fsm.StateType) ([]*Deposit, bool) { + if CheckDuplicates(outpoints) != nil { + return nil, false + } + m.mu.Lock() defer m.mu.Unlock() @@ -453,10 +479,10 @@ func (m *Manager) AllOutpointsActiveDeposits(outpoints []wire.OutPoint, return deposits, true } - lockDeposits(deposits) - defer unlockDeposits(deposits) + lockedDeposits := lockDeposits(deposits) + defer unlockDeposits(lockedDeposits) for _, d := range deposits { - if !d.IsInStateNoLock(targetState) { + if !d.isInStateNoLock(targetState) { return nil, false } } @@ -495,8 +521,15 @@ func (m *Manager) TransitionDeposits(ctx context.Context, deposits []*Deposit, outpoints := make([]wire.OutPoint, len(deposits)) for i, d := range deposits { + if d == nil { + return fmt.Errorf("nil deposit at index %d", i) + } + outpoints[i] = d.OutPoint } + if err := CheckDuplicates(outpoints); err != nil { + return fmt.Errorf("duplicate deposit outpoint: %w", err) + } m.mu.Lock() stateMachines, _ := m.toActiveDeposits(&outpoints) @@ -506,8 +539,16 @@ func (m *Manager) TransitionDeposits(ctx context.Context, deposits []*Deposit, return fmt.Errorf("deposits not found in active deposits") } - lockDeposits(deposits) - defer unlockDeposits(deposits) + lockedDeposits := lockDeposits(deposits) + defer unlockDeposits(lockedDeposits) + for _, deposit := range deposits { + if deposit.isInFinalStateNoLock() { + return fmt.Errorf("deposit %v is no longer active in "+ + "state %v", deposit.OutPoint, + deposit.getStateNoLock()) + } + } + for _, sm := range stateMachines { err := sm.SendEvent(ctx, event, nil) if err != nil { @@ -525,15 +566,41 @@ func (m *Manager) TransitionDeposits(ctx context.Context, deposits []*Deposit, return nil } -func lockDeposits(deposits []*Deposit) { - for _, d := range deposits { +// lockDeposits locks deposits in canonical outpoint order and returns the +// ordered slice that must be passed to unlockDeposits. +func lockDeposits(deposits []*Deposit) []*Deposit { + lockedDeposits := append([]*Deposit(nil), deposits...) + sort.Slice(lockedDeposits, func(i, j int) bool { + return lockedDeposits[i].OutPoint.String() < + lockedDeposits[j].OutPoint.String() + }) + + for _, d := range lockedDeposits { d.Lock() } + + return lockedDeposits +} + +// unlockDeposits unlocks deposits in reverse lock order. +func unlockDeposits(deposits []*Deposit) { + for i := len(deposits) - 1; i >= 0; i-- { + d := deposits[i] + d.Unlock() + } } -func unlockDeposits(deposits []*Deposit) { - for _, d := range deposits { - d.Unlock() +// removeActiveDeposit removes and stops the FSM for an active outpoint. +func (m *Manager) removeActiveDeposit(outpoint wire.OutPoint) { + m.mu.Lock() + fsm, ok := m.activeDeposits[outpoint] + if ok { + delete(m.activeDeposits, outpoint) + } + m.mu.Unlock() + + if ok { + fsm.Stop() } } @@ -544,6 +611,9 @@ func (m *Manager) GetAllDeposits(ctx context.Context) ([]*Deposit, error) { // UpdateDeposit overrides all fields of the deposit with given ID in the store. func (m *Manager) UpdateDeposit(ctx context.Context, d *Deposit) error { + d.Lock() + defer d.Unlock() + return m.cfg.Store.UpdateDeposit(ctx, d) } diff --git a/staticaddr/deposit/manager_reconcile_test.go b/staticaddr/deposit/manager_reconcile_test.go new file mode 100644 index 00000000..972feefd --- /dev/null +++ b/staticaddr/deposit/manager_reconcile_test.go @@ -0,0 +1,128 @@ +package deposit + +import ( + "testing" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/stretchr/testify/require" +) + +// TestAllOutpointsActiveDepositsRejectsDuplicateOutpoints verifies that a +// duplicated selection is rejected before the manager tries to lock the same +// deposit twice. +func TestAllOutpointsActiveDepositsRejectsDuplicateOutpoints(t *testing.T) { + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{12}, + Index: 6, + } + + deposit := &Deposit{ + OutPoint: outpoint, + } + deposit.SetState(Deposited) + + manager := NewManager(&ManagerConfig{}) + manager.deposits[outpoint] = deposit + manager.activeDeposits[outpoint] = &FSM{ + deposit: deposit, + } + + deposits, ok := manager.AllOutpointsActiveDeposits( + []wire.OutPoint{outpoint, outpoint}, Deposited, + ) + require.False(t, ok) + require.Nil(t, deposits) +} + +// TestTransitionDepositsRejectsDuplicateOutpoints verifies that transition +// callers cannot deadlock the manager by passing the same deposit twice. +func TestTransitionDepositsRejectsDuplicateOutpoints(t *testing.T) { + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{13}, + Index: 6, + } + + deposit := &Deposit{ + OutPoint: outpoint, + } + deposit.SetState(Deposited) + + manager := NewManager(&ManagerConfig{}) + err := manager.TransitionDeposits( + t.Context(), []*Deposit{deposit, deposit}, OnLoopInInitiated, + LoopingIn, + ) + require.ErrorContains(t, err, "duplicate deposit outpoint") + require.Equal(t, Deposited, deposit.GetState()) +} + +// TestLockDepositsCanonicalizesOutpoints verifies that lockDeposits takes a +// canonical copy of the caller's slice so overlapping multi-deposit operations +// cannot lock deposits in conflicting request orders. +func TestLockDepositsCanonicalizesOutpoints(t *testing.T) { + depositA := &Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: 0, + }, + } + depositB := &Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{2}, + Index: 0, + }, + } + + deposits := []*Deposit{depositB, depositA} + lockedDeposits := lockDeposits(deposits) + defer unlockDeposits(lockedDeposits) + + require.Equal(t, []*Deposit{depositA, depositB}, lockedDeposits) + require.Equal(t, []*Deposit{depositB, depositA}, deposits) +} + +// TestLockDepositsAllowsReversedConcurrentRequests exercises the reviewer +// case where overlapping callers request the same deposits in opposite orders. +func TestLockDepositsAllowsReversedConcurrentRequests(t *testing.T) { + depositA := &Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{3}, + Index: 0, + }, + } + depositB := &Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{4}, + Index: 0, + }, + } + + start := make(chan struct{}) + done := make(chan struct{}, 2) + lockAndUnlock := func(deposits []*Deposit) { + <-start + + for range 100 { + lockedDeposits := lockDeposits(deposits) + unlockDeposits(lockedDeposits) + } + + done <- struct{}{} + } + + go lockAndUnlock([]*Deposit{depositA, depositB}) + go lockAndUnlock([]*Deposit{depositB, depositA}) + + close(start) + + for range 2 { + select { + case <-done: + + case <-time.After(time.Second): + t.Fatal("reversed deposit lock requests deadlocked") + } + } +} diff --git a/staticaddr/deposit/outpoint.go b/staticaddr/deposit/outpoint.go new file mode 100644 index 00000000..64a3a701 --- /dev/null +++ b/staticaddr/deposit/outpoint.go @@ -0,0 +1,21 @@ +package deposit + +import ( + "fmt" + + "github.com/btcsuite/btcd/wire" +) + +// CheckDuplicates returns an error if the outpoint list contains duplicates. +func CheckDuplicates(outpoints []wire.OutPoint) error { + seen := make(map[wire.OutPoint]struct{}, len(outpoints)) + for _, outpoint := range outpoints { + if _, ok := seen[outpoint]; ok { + return fmt.Errorf("duplicate outpoint %v", outpoint) + } + + seen[outpoint] = struct{}{} + } + + return nil +} diff --git a/staticaddr/deposit/outpoint_test.go b/staticaddr/deposit/outpoint_test.go new file mode 100644 index 00000000..fff86458 --- /dev/null +++ b/staticaddr/deposit/outpoint_test.go @@ -0,0 +1,40 @@ +package deposit + +import ( + "testing" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/stretchr/testify/require" +) + +func TestCheckDuplicates(t *testing.T) { + duplicate := wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: 2, + } + + outpoints := []wire.OutPoint{{ + Hash: chainhash.Hash{3}, + Index: 4, + }, duplicate, { + Hash: chainhash.Hash{5}, + Index: 6, + }, duplicate} + + err := CheckDuplicates(outpoints) + require.ErrorContains(t, err, "duplicate outpoint") + require.ErrorContains(t, err, duplicate.String()) +} + +func TestCheckDuplicatesNoDuplicate(t *testing.T) { + outpoints := []wire.OutPoint{{ + Hash: chainhash.Hash{1}, + Index: 2, + }, { + Hash: chainhash.Hash{3}, + Index: 4, + }} + + require.NoError(t, CheckDuplicates(outpoints)) +} diff --git a/staticaddr/deposit/sql_store.go b/staticaddr/deposit/sql_store.go index d9f4249f..a49550e5 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, } @@ -69,11 +69,15 @@ func (s *SqlStore) CreateDeposit(ctx context.Context, deposit *Deposit) error { } // UpdateDeposit updates the deposit in the database. +// +// Callers that pass a live deposit must hold the deposit lock while calling +// this method. The deposit FSM already does this for state transitions, and +// Manager.UpdateDeposit wraps external callers with the same lock. func (s *SqlStore) UpdateDeposit(ctx context.Context, deposit *Deposit) error { insertUpdateArgs := sqlc.InsertDepositUpdateParams{ DepositID: deposit.ID[:], UpdateTimestamp: s.clock.Now().UTC(), - UpdateState: string(deposit.state), + UpdateState: string(deposit.getStateNoLock()), } var ( @@ -83,7 +87,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.go b/staticaddr/loopin/actions.go index cf21fc5f..bfb3d333 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -12,6 +12,7 @@ import ( "github.com/btcsuite/btcd/btcec/v2/schnorr/musig2" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/txscript" + "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcwallet/chain" "github.com/lightninglabs/lndclient" "github.com/lightninglabs/loop" @@ -359,6 +360,35 @@ func (f *FSM) cancelSwapInvoice() { } } +// handleInvoiceUpdate applies the monitor state's invoice-update semantics and +// reports whether the update produced a terminal event. +func (f *FSM) handleInvoiceUpdate(update lndclient.InvoiceUpdate) ( + fsm.EventType, bool) { + + switch update.State { + case invoices.ContractOpen: + return fsm.NoOp, false + + case invoices.ContractAccepted: + return fsm.NoOp, false + + case invoices.ContractSettled: + f.Debugf("received off-chain payment update %v", update.State) + return OnPaymentReceived, true + + case invoices.ContractCanceled: + // If the invoice was canceled we only log here since we still need + // to monitor until the htlc timed out. + log.Warnf("invoice for swap hash %v canceled", f.loopIn.SwapHash) + return fsm.NoOp, false + + default: + err := fmt.Errorf("unexpected invoice state %v for swap hash %v "+ + "canceled", update.State, f.loopIn.SwapHash) + return f.HandleError(err), true + } +} + // SignHtlcTxAction is called if the htlc was initialized and the server // provided the necessary information to construct the htlc tx. We sign the htlc // tx and send the signatures to the server. @@ -384,6 +414,11 @@ func (f *FSM) SignHtlcTxAction(ctx context.Context, return f.HandleError(err) } + err = f.checkDepositsAvailable(ctx) + if err != nil { + return f.HandleError(err) + } + // Create a musig2 session for each deposit and different htlc tx fee // rates. createSession := staticutil.CreateMusig2Sessions @@ -497,6 +532,68 @@ func (f *FSM) SignHtlcTxAction(ctx context.Context, return OnHtlcTxSigned } +// checkDepositsAvailable verifies that all loop-in deposits are still available +// before the client signs the HTLC transaction. +func (f *FSM) checkDepositsAvailable(ctx context.Context) error { + outpoints, err := f.validateSigningDepositOutpoints() + if err != nil { + return err + } + + if f.cfg.TxOutChecker == nil { + return nil + } + + txOuts, err := f.cfg.TxOutChecker.GetTxOuts(ctx, outpoints) + if err != nil { + return fmt.Errorf("unable to check deposits: %w", err) + } + + for _, outpoint := range outpoints { + if txOuts[outpoint] == nil { + return fmt.Errorf("deposit %v is no longer available", + outpoint) + } + } + + return nil +} + +// validateSigningDepositOutpoints verifies that the current deposit rows match +// the server-side outpoint snapshot before signing the HTLC transaction. +func (f *FSM) validateSigningDepositOutpoints() ([]wire.OutPoint, error) { + currentOutpoints := f.loopIn.Outpoints() + if len(f.loopIn.DepositOutpoints) == 0 { + return currentOutpoints, nil + } + + if len(f.loopIn.DepositOutpoints) != len(currentOutpoints) { + return nil, fmt.Errorf("deposit outpoint snapshot has %d "+ + "outpoints, current deposits have %d", + len(f.loopIn.DepositOutpoints), len(currentOutpoints)) + } + + snapshotOutpoints := make( + []wire.OutPoint, len(f.loopIn.DepositOutpoints), + ) + for i, snapshot := range f.loopIn.DepositOutpoints { + outpoint, err := wire.NewOutPointFromString(snapshot) + if err != nil { + return nil, fmt.Errorf("unable to parse deposit "+ + "outpoint snapshot %q: %w", snapshot, err) + } + + snapshotOutpoints[i] = *outpoint + if *outpoint != currentOutpoints[i] { + return nil, fmt.Errorf("deposit outpoint snapshot "+ + "mismatch at index %d: snapshot %v, "+ + "current %v", i, outpoint, currentOutpoints[i]) + } + } + + return snapshotOutpoints, nil +} + // cleanUpSessions releases allocated memory of the musig2 sessions. func (f *FSM) cleanUpSessions(ctx context.Context, sessions []*input.MuSig2SessionInfo) { @@ -536,6 +633,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, subscribeCtx, f.loopIn.SwapHash, ) if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to subscribe to swap "+ "invoice: %w", err) @@ -563,6 +664,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, htlcConfChan, htlcErrConfChan, err := registerHtlcConf() if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to monitor htlc tx confirmation: %w", err) @@ -573,15 +678,23 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, registerBlocks := f.cfg.ChainNotifier.RegisterBlockEpochNtfn blockChan, blockChanErr, err := registerBlocks(ctx) if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to subscribe to new blocks: %w", err) return f.HandleError(err) } - htlcConfirmed := false - + // Look up the current invoice state after registering subscriptions so + // recovery can resume the payment deadline from the latest known state. invoice, err := f.cfg.LndClient.LookupInvoice(ctx, f.loopIn.SwapHash) if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to look up invoice by swap hash: %w", err) @@ -596,8 +709,8 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, if invoice.State != invoices.ContractCanceled { // If the invoice is still live we set the timeout to the // remaining payment time. If too much time has elapsed, e.g. - // after a restart, we set the timeout to 0 to cancel the - // invoice and unlock the deposits immediately. + // after a restart, we cancel the invoice immediately and keep + // monitoring the HTLC until it can no longer confirm. remainingTimeSeconds := f.loopIn.RemainingPaymentTimeSeconds() // If the invoice isn't cancelled yet and the payment timeout @@ -629,6 +742,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, f.cancelSwapInvoice() } + htlcConfirmed := false for { select { case <-htlcConfChan: @@ -637,6 +751,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, htlcConfirmed = true case err = <-htlcErrConfChan: + if ctx.Err() != nil { + return fsm.NoOp + } + f.Errorf("htlc tx conf chan error, re-registering: "+ "%v", err) @@ -647,6 +765,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // Re-register for htlc confirmation. htlcConfChan, htlcErrConfChan, err = registerHtlcConf() if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to re-register for "+ "htlc tx confirmation: %w", err) @@ -661,6 +783,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, htlcConfChan, htlcErrConfChan, err = registerHtlcConf() if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to monitor htlc tx "+ "confirmation: %v", err) @@ -732,40 +858,34 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, return OnSweepHtlcTimeout case err = <-blockChanErr: + if ctx.Err() != nil { + return fsm.NoOp + } + f.Errorf("block subscription error: %v", err) return f.HandleError(err) - case update := <-invoiceUpdateChan: - switch update.State { - case invoices.ContractOpen: - case invoices.ContractAccepted: - case invoices.ContractSettled: - f.Debugf("received off-chain payment update "+ - "%v", update.State) - - return OnPaymentReceived - - case invoices.ContractCanceled: - // If the invoice was canceled we only log here - // since we still need to monitor until the htlc - // timed out. - log.Warnf("invoice for swap hash %v canceled", - f.loopIn.SwapHash) - - default: - err = fmt.Errorf("unexpected invoice state %v "+ - "for swap hash %v canceled", - update.State, f.loopIn.SwapHash) - - return f.HandleError(err) + case update, ok := <-invoiceUpdateChan: + if !ok { + invoiceUpdateChan = nil + continue + } + + if event, done := f.handleInvoiceUpdate(update); done { + return event + } + + case err, ok := <-invoiceErrChan: + if !ok { + invoiceErrChan = nil + continue } - case err = <-invoiceErrChan: f.Errorf("invoice subscription error: %v", err) case <-ctx.Done(): - return f.HandleError(ctx.Err()) + return fsm.NoOp } } } @@ -791,10 +911,10 @@ func (f *FSM) SweepHtlcTimeoutAction(ctx context.Context, select { // The context is cancelled when the server is shutting - // down. In that case we give up broadcasting attempts - // and return an error. + // down. Keep the current state so recovery resumes + // broadcasting attempts after restart. case <-ctx.Done(): - return f.HandleError(ctx.Err()) + return fsm.NoOp case <-time.After(htlcTimeoutSweepRetryDelay): } @@ -828,6 +948,10 @@ func (f *FSM) MonitorHtlcTimeoutSweepAction(ctx context.Context, ) if err != nil { + if ctx.Err() != nil { + return fsm.NoOp + } + err = fmt.Errorf("unable to register to the htlc timeout "+ "sweep tx: %w", err) @@ -837,6 +961,10 @@ func (f *FSM) MonitorHtlcTimeoutSweepAction(ctx context.Context, for { select { case err := <-errChan: + if ctx.Err() != nil { + return fsm.NoOp + } + return f.HandleError(err) case conf := <-htlcTimeoutTxidChan: @@ -860,7 +988,7 @@ func (f *FSM) MonitorHtlcTimeoutSweepAction(ctx context.Context, return OnHtlcTimeoutSwept case <-ctx.Done(): - return f.HandleError(ctx.Err()) + return fsm.NoOp } } } diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index d75adf6c..d463bf37 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -3,6 +3,7 @@ package loopin import ( "context" "errors" + "fmt" "testing" "time" @@ -24,6 +25,84 @@ import ( "google.golang.org/grpc" ) +// TestHandleInvoiceUpdate verifies that invoice state updates map to the +// monitor events expected by the static address loop-in FSM. +func TestHandleInvoiceUpdate(t *testing.T) { + t.Parallel() + + swapHash := lntypes.Hash{1, 2, 3} + tests := []struct { + name string + state invoices.ContractState + event fsm.EventType + done bool + errString string + }{ + { + name: "open", + state: invoices.ContractOpen, + event: fsm.NoOp, + }, + { + name: "accepted", + state: invoices.ContractAccepted, + event: fsm.NoOp, + }, + { + name: "settled", + state: invoices.ContractSettled, + event: OnPaymentReceived, + done: true, + }, + { + name: "canceled", + state: invoices.ContractCanceled, + event: fsm.NoOp, + }, + { + name: "unexpected", + state: invoices.ContractState(99), + event: fsm.OnError, + done: true, + errString: "unexpected invoice state", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + loopIn: &StaticAddressLoopIn{ + SwapHash: swapHash, + }, + } + + event, done := f.handleInvoiceUpdate( + lndclient.InvoiceUpdate{ + Invoice: lndclient.Invoice{ + State: test.state, + }, + }, + ) + require.Equal(t, test.event, event) + require.Equal(t, test.done, done) + + if test.errString == "" { + require.Nil(t, f.LastActionError) + } else { + require.ErrorContains( + t, f.LastActionError, test.errString, + ) + require.ErrorContains( + t, f.LastActionError, fmt.Sprint(swapHash), + ) + } + }) + } +} + // TestMonitorInvoiceAndHtlcTxReRegistersOnConfErr ensures that an error from // the HTLC confirmation subscription triggers a re-registration. Without the // regression fix, only the initial registration would be performed and the @@ -128,6 +207,166 @@ func TestMonitorInvoiceAndHtlcTxReRegistersOnConfErr(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxNoOpOnShutdown ensures that a shutdown while the +// client is monitoring an HTLC-signed loop-in keeps the swap resumable instead +// of entering the generic unlock path. +func TestMonitorInvoiceAndHtlcTxNoOpOnShutdown(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + runCtx, stop := context.WithCancel(ctx) + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{4, 5, 6} + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: 2_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 3_600, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.Invoices[swapHash] = &lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + } + + depositMgr := &recordingDepositManager{} + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + } + + f, err := NewFSM(runCtx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(runCtx, nil) + }() + + select { + case <-mockLnd.SingleInvoiceSubcribeChannel: + case <-ctx.Done(): + t.Fatalf("invoice subscription not registered: %v", ctx.Err()) + } + + select { + case <-mockLnd.RegisterConfChannel: + case <-ctx.Done(): + t.Fatalf("htlc conf registration not received: %v", ctx.Err()) + } + + stop() + + select { + case event := <-resultChan: + require.Equal(t, fsm.NoOp, event) + + case <-ctx.Done(): + t.Fatalf("monitor action did not exit: %v", ctx.Err()) + } + + require.Nil(t, f.LastActionError) + require.Empty(t, depositMgr.transitions) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("invoice canceled on shutdown: %v", hash) + + default: + } +} + +// TestSweepHtlcTimeoutActionNoOpOnShutdown ensures that a shutdown during +// timeout sweep publication keeps the FSM in the same state so it can resume +// after restart. +func TestSweepHtlcTimeoutActionNoOpOnShutdown(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + mockLnd := test.NewMockLnd() + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + LndClient: mockLnd.Client, + WalletKit: mockLnd.WalletKit, + }, + loopIn: &StaticAddressLoopIn{}, + } + + event := f.SweepHtlcTimeoutAction(ctx, nil) + require.Equal(t, fsm.NoOp, event) + require.Nil(t, f.LastActionError) +} + +// TestMonitorHtlcTimeoutSweepActionNoOpOnShutdown ensures that a shutdown +// while waiting for the timeout sweep confirmation keeps the FSM resumable. +func TestMonitorHtlcTimeoutSweepActionNoOpOnShutdown(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + sweepAddr, err := mockLnd.WalletKit.NextAddr(ctx, "", 0, false) + require.NoError(t, err) + + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + ChainNotifier: mockLnd.ChainNotifier, + }, + loopIn: &StaticAddressLoopIn{ + HtlcTimeoutSweepAddress: sweepAddr, + InitiationHeight: uint32(mockLnd.Height), + }, + } + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorHtlcTimeoutSweepAction(ctx, nil) + }() + + select { + case <-mockLnd.RegisterConfChannel: + case <-ctx.Done(): + t.Fatalf("timeout sweep conf registration not received: %v", + ctx.Err()) + } + + cancel() + + select { + case event := <-resultChan: + require.Equal(t, fsm.NoOp, event) + require.Nil(t, f.LastActionError) + + case <-time.After(5 * time.Second): + t.Fatal("timeout sweep monitor did not return") + } +} + // TestInitHtlcActionPreservesRouteHints asserts that static-address loop-in // propagates explicit route hints into the encoded swap invoice sent to the // server. @@ -191,6 +430,72 @@ func TestInitHtlcActionPreservesRouteHints(t *testing.T) { test.RequireRouteHintsEqual(t, loopIn.RouteHints, routeHints) } +func TestSignHtlcTxActionChecksDepositAvailability(t *testing.T) { + dep := &deposit.Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x77}, + Index: 2, + }, + Value: 200_000, + } + checker := &recordingTxOutChecker{} + + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + TxOutChecker: checker, + }, + loopIn: &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{dep}, + DepositOutpoints: []string{dep.OutPoint.String()}, + }, + } + + event := f.SignHtlcTxAction(t.Context(), nil) + require.Equal(t, fsm.OnError, event) + require.ErrorContains( + t, f.LastActionError, "deposit "+ + dep.OutPoint.String()+" is no longer available", + ) + require.Equal(t, [][]wire.OutPoint{{dep.OutPoint}}, checker.outpoints) +} + +func TestCheckDepositsAvailableRejectsDivergentDepositOutpoints( + t *testing.T) { + + currentOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{0x89}, + Index: 1, + } + snapshotOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{0x88}, + Index: 0, + } + checker := &recordingTxOutChecker{} + + f := &FSM{ + cfg: &Config{ + TxOutChecker: checker, + }, + loopIn: &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{ + OutPoint: currentOutpoint, + Value: 200_000, + }}, + DepositOutpoints: []string{snapshotOutpoint.String()}, + }, + } + + err := f.checkDepositsAvailable(t.Context()) + require.ErrorContains(t, err, "deposit outpoint snapshot mismatch") + require.Empty(t, checker.outpoints) +} + // mockStaticAddressServer captures static-address loop-in requests in tests. type mockStaticAddressServer struct { swapserverrpc.StaticAddressServerClient @@ -541,6 +846,26 @@ func (r *recordingDepositManager) TransitionDeposits(_ context.Context, return r.err } +type recordingTxOutChecker struct { + outpoints [][]wire.OutPoint + txOuts map[wire.OutPoint]*wire.TxOut + err error +} + +// GetTxOuts records the request and returns the configured available outputs. +func (r *recordingTxOutChecker) GetTxOuts(_ context.Context, + outpoints []wire.OutPoint) (map[wire.OutPoint]*wire.TxOut, error) { + + r.outpoints = append( + r.outpoints, append([]wire.OutPoint(nil), outpoints...), + ) + if r.err != nil { + return nil, r.err + } + + return r.txOuts, nil +} + // initHtlcTestServer lets InitHtlcAction tests inject a deterministic server // response without standing up the full gRPC client. type initHtlcTestServer struct { diff --git a/staticaddr/loopin/autoloop_dp.go b/staticaddr/loopin/autoloop_dp.go index c605c43e..44dd12f4 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 @@ -246,7 +246,7 @@ func filterAutoloopCandidateDeposits(maxAmount btcutil.Amount, continue } - residualLife := candidateDeposit.ConfirmationHeight + + residualLife := confirmationHeight + int64(csvExpiry) - int64(blockHeight) eligibleDeposits = append( diff --git a/staticaddr/loopin/interface.go b/staticaddr/loopin/interface.go index c4bbb2b7..52139020 100644 --- a/staticaddr/loopin/interface.go +++ b/staticaddr/loopin/interface.go @@ -4,6 +4,7 @@ import ( "context" "github.com/btcsuite/btcd/btcutil" + "github.com/btcsuite/btcd/wire" "github.com/lightninglabs/loop" "github.com/lightninglabs/loop/fsm" "github.com/lightninglabs/loop/staticaddr/deposit" @@ -105,6 +106,15 @@ type QuoteGetter interface { numDeposits uint32, fast bool) (*loop.LoopInQuote, error) } +// TxOutChecker checks whether outpoints are still available in the chain +// backend's UTXO view. +type TxOutChecker interface { + // GetTxOuts returns entries for the requested outpoints that are + // available and unspent. Missing entries are unavailable or spent. + GetTxOuts(ctx context.Context, outpoints []wire.OutPoint) ( + map[wire.OutPoint]*wire.TxOut, error) +} + type NotificationManager interface { // SubscribeStaticLoopInSweepRequests subscribes to the static loop in // sweep requests. These are sent by the server to the client to request diff --git a/staticaddr/loopin/loopin.go b/staticaddr/loopin/loopin.go index 0be2ffe4..9b0fc03b 100644 --- a/staticaddr/loopin/loopin.go +++ b/staticaddr/loopin/loopin.go @@ -93,8 +93,6 @@ type StaticAddressLoopIn struct { // The outpoints in the format txid:vout that are part of the loop-in // swap. - // TODO(hieblmi): Replace this with a getter method that fetches the - // outpoints from the deposits. DepositOutpoints []string // SelectedAmount is the amount that the user selected for the swap. If @@ -469,12 +467,28 @@ func (l *StaticAddressLoopIn) TotalDepositAmount() btcutil.Amount { // RemainingPaymentTimeSeconds returns the remaining time in seconds until the // payment timeout is reached. The remaining time is calculated from the -// initiation time of the swap. If more than the swaps configured payment +// initiation time of the swap. If more than the swap's configured payment // timeout has passed, the remaining time will be negative. func (l *StaticAddressLoopIn) RemainingPaymentTimeSeconds() int64 { - elapsedSinceInitiation := time.Since(l.InitiationTime).Seconds() + deadline := l.InitiationTime.Add(l.PaymentTimeoutDuration()) - return int64(l.PaymentTimeoutSeconds) - int64(elapsedSinceInitiation) + return int64(time.Until(deadline).Seconds()) +} + +// PaymentTimeoutDuration returns the configured payment timeout duration, +// falling back to the default if the swap predates the persisted timeout field. +func (l *StaticAddressLoopIn) PaymentTimeoutDuration() time.Duration { + return time.Duration(l.paymentTimeoutSeconds()) * time.Second +} + +// paymentTimeoutSeconds returns the configured timeout in seconds. +func (l *StaticAddressLoopIn) paymentTimeoutSeconds() int64 { + timeoutSeconds := int64(l.PaymentTimeoutSeconds) + if timeoutSeconds == 0 { + timeoutSeconds = int64(DefaultPaymentTimeoutSeconds) + } + + return timeoutSeconds } // Outpoints returns the wire outpoints of the deposits. diff --git a/staticaddr/loopin/loopin_test.go b/staticaddr/loopin/loopin_test.go index d4be020f..8b0892e6 100644 --- a/staticaddr/loopin/loopin_test.go +++ b/staticaddr/loopin/loopin_test.go @@ -147,6 +147,42 @@ func TestCreateHtlcSweepTxSweepValue(t *testing.T) { "the HTLC output") } +// TestPaymentTimeoutDuration verifies that zero timeout values fall back to the +// default payment timeout duration. +func TestPaymentTimeoutDuration(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + paymentTimeoutSeconds uint32 + expected time.Duration + }{ + { + name: "default", + expected: time.Duration(DefaultPaymentTimeoutSeconds) * time.Second, + }, + { + name: "configured", + paymentTimeoutSeconds: 42, + expected: 42 * time.Second, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + loopIn := &StaticAddressLoopIn{ + PaymentTimeoutSeconds: test.paymentTimeoutSeconds, + } + + require.Equal( + t, test.expected, loopIn.PaymentTimeoutDuration(), + ) + }) + } +} + // newStaticAddress creates a StaticAddress for testing. func newStaticAddress(clientKey, serverKey *btcec.PublicKey, csvExpiry int64) (*script.StaticAddress, error) { diff --git a/staticaddr/loopin/manager.go b/staticaddr/loopin/manager.go index a76cbba3..51969578 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -56,6 +56,10 @@ type Config struct { // LndClient is used to add invoices and select hop hints. LndClient lndclient.LightningClient + // TxOutChecker checks that selected deposits are still available before + // the client gives the server HTLC signatures. + TxOutChecker TxOutChecker + // InvoicesClient is used to subscribe to invoice settlements and // cancel invoices. InvoicesClient lndclient.InvoicesClient @@ -533,17 +537,15 @@ func (m *Manager) recoverLoopIns(ctx context.Context) error { for _, loopIn := range pendingLoopIns { log.Debugf("Recovering loopIn %x", loopIn.SwapHash[:]) - // Retrieve all deposits regardless of deposit state. If any of - // the deposits is not active in the in-mem map of the deposits - // manager we log it, but continue to recover the loop-in. - var allActive bool - loopIn.Deposits, allActive = - m.cfg.DepositManager.AllStringOutpointsActiveDeposits( - loopIn.DepositOutpoints, fsm.EmptyState, - ) - + // Retrieve all deposits regardless of deposit state. If all + // deposits are active in the in-mem map of the deposits manager, + // use those active instances. Otherwise, keep the store's + // swap_hash/deposit-id reconstruction and continue recovery. + activeDeposits, allActive := m.activeDepositsForLoopIn(loopIn) if !allActive { log.Errorf("one or more deposits are not active") + } else { + loopIn.Deposits = activeDeposits } loopIn.AddressParams, err = @@ -754,8 +756,12 @@ func (m *Manager) initiateLoopIn(ctx context.Context, } swap := &StaticAddressLoopIn{ - SelectedAmount: req.SelectedAmount, - DepositOutpoints: selectedOutpoints, + SelectedAmount: req.SelectedAmount, + // Copy into a nil slice so the swap owns a stable snapshot + // instead of aliasing the caller's selectedOutpoints slice. + DepositOutpoints: append( + []string(nil), selectedOutpoints..., + ), Deposits: selectedDeposits, Label: req.Label, Initiator: req.Initiator, @@ -814,35 +820,29 @@ func (m *Manager) startLoopInFsm(ctx context.Context, func (m *Manager) GetAllSwaps(ctx context.Context) ([]*StaticAddressLoopIn, error) { - swaps, err := m.cfg.Store.GetStaticAddressLoopInSwapsByStates( + return m.cfg.Store.GetStaticAddressLoopInSwapsByStates( ctx, AllStates, ) - if err != nil { - return nil, err - } +} - allDeposits, err := m.cfg.DepositManager.GetAllDeposits(ctx) - if err != nil { - return nil, err - } +// activeDepositsForLoopIn returns the active deposit instances for a loop-in +// using the current deposit outpoints reconstructed by the store. The stored +// deposit outpoint snapshots remain the original swap inputs and are not the +// source of truth for current deposit rows. +func (m *Manager) activeDepositsForLoopIn(loopIn *StaticAddressLoopIn) ( + []*deposit.Deposit, bool) { - var depositLookup = make(map[string]*deposit.Deposit) - for i, d := range allDeposits { - depositLookup[d.OutPoint.String()] = allDeposits[i] - } - - for i, s := range swaps { - var deposits []*deposit.Deposit - for _, outpoint := range s.DepositOutpoints { - if d, ok := depositLookup[outpoint]; ok { - deposits = append(deposits, d) - } + outpoints := loopIn.DepositOutpoints + if len(loopIn.Deposits) > 0 { + outpoints = make([]string, 0, len(loopIn.Deposits)) + for _, d := range loopIn.Deposits { + outpoints = append(outpoints, d.OutPoint.String()) } - - swaps[i].Deposits = deposits } - return swaps, nil + return m.cfg.DepositManager.AllStringOutpointsActiveDeposits( + outpoints, fsm.EmptyState, + ) } // SelectDeposits sorts the deposits by amount in descending order, then by @@ -857,8 +857,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 "+ @@ -874,9 +875,9 @@ func SelectDeposits(targetAmount btcutil.Amount, // blocks-until-expiry in ascending order. sort.Slice(deposits, func(i, j int) bool { if deposits[i].Value == deposits[j].Value { - iExp := uint32(deposits[i].ConfirmationHeight) + + iExp := uint32(deposits[i].GetConfirmationHeight()) + csvExpiry - blockHeight - jExp := uint32(deposits[j].ConfirmationHeight) + + jExp := uint32(deposits[j].GetConfirmationHeight()) + csvExpiry - blockHeight return iExp < jExp diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 85178a05..ed31df8a 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -298,6 +298,65 @@ func TestHandleLoopInSweepReqRejectsInvalidServerNonce(t *testing.T) { require.ErrorContains(t, err, depOutpoint) } +// TestActiveDepositsForLoopInUsesCurrentDepositOutpoints verifies that +// recovery checks the current deposit outpoints reconstructed by the store +// rather than the original outpoint snapshot persisted on the swap. +func TestActiveDepositsForLoopInUsesCurrentDepositOutpoints(t *testing.T) { + oldOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{0xaa}, + Index: 0, + } + currentDeposit := makeDeposit(0xbb, 1, 10_000, 42) + + manager := &Manager{ + cfg: &Config{ + DepositManager: &mockDepositManager{ + byOutpoint: map[string]*deposit.Deposit{ + currentDeposit.OutPoint.String(): currentDeposit, + }, + }, + }, + } + + deposits, allActive := manager.activeDepositsForLoopIn( + &StaticAddressLoopIn{ + DepositOutpoints: []string{oldOutpoint.String()}, + Deposits: []*deposit.Deposit{currentDeposit}, + }, + ) + require.True(t, allActive) + require.Equal(t, []*deposit.Deposit{currentDeposit}, deposits) +} + +// TestGetAllSwapsPreservesStoreDeposits verifies that list responses keep the +// store's swap_hash/deposit-id reconstruction even when DepositOutpoints is an +// original input snapshot and the deposit's current outpoint has changed. +func TestGetAllSwapsPreservesStoreDeposits(t *testing.T) { + oldOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{0xcc}, + Index: 0, + } + currentDeposit := makeDeposit(0xdd, 1, 10_000, 42) + swap := &StaticAddressLoopIn{ + DepositOutpoints: []string{oldOutpoint.String()}, + Deposits: []*deposit.Deposit{currentDeposit}, + } + + manager := &Manager{ + cfg: &Config{ + Store: &mockStore{ + swaps: []*StaticAddressLoopIn{swap}, + }, + }, + } + + swaps, err := manager.GetAllSwaps(t.Context()) + require.NoError(t, err) + require.Len(t, swaps, 1) + require.Equal(t, []string{oldOutpoint.String()}, swaps[0].DepositOutpoints) + require.Equal(t, []*deposit.Deposit{currentDeposit}, swaps[0].Deposits) +} + // mockDepositManager implements DepositManager for tests. type mockDepositManager struct { // activeDeposits is the set returned by GetActiveDepositsInState. @@ -316,7 +375,7 @@ func (m *mockDepositManager) GetAllDeposits(_ context.Context) ( func (m *mockDepositManager) AllStringOutpointsActiveDeposits(outpoints []string, state fsm.StateType) ([]*deposit.Deposit, bool) { - if state != deposit.Deposited { + if state != deposit.Deposited && state != fsm.EmptyState { return nil, false } @@ -408,6 +467,7 @@ func (m *mockQuoteGetter) GetLoopInQuote(_ context.Context, // mockStore implements StaticAddressLoopInStore for tests. type mockStore struct { + swaps []*StaticAddressLoopIn loopIns map[lntypes.Hash]*StaticAddressLoopIn mapIDs map[lntypes.Hash][]deposit.ID } @@ -427,7 +487,7 @@ func (s *mockStore) UpdateLoopIn(_ context.Context, func (s *mockStore) GetStaticAddressLoopInSwapsByStates(_ context.Context, _ []fsm.StateType) ([]*StaticAddressLoopIn, error) { - return nil, nil + return s.swaps, nil } func (s *mockStore) IsStored(_ context.Context, _ lntypes.Hash) (bool, error) { return false, nil diff --git a/staticaddr/loopin/sql_store.go b/staticaddr/loopin/sql_store.go index d8c253be..cbaf78fa 100644 --- a/staticaddr/loopin/sql_store.go +++ b/staticaddr/loopin/sql_store.go @@ -507,9 +507,12 @@ func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params, } } - depositOutpoints := strings.Split( - swap.DepositOutpoints, OutpointSeparator, - ) + var depositOutpoints []string + if swap.DepositOutpoints != "" { + depositOutpoints = strings.Split( + swap.DepositOutpoints, OutpointSeparator, + ) + } timeoutAddressString := swap.HtlcTimeoutSweepAddress var timeoutAddress btcutil.Address @@ -555,6 +558,7 @@ func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params, depositList = append(depositList, deposit) } + depositList = orderDepositsBySnapshot(depositList, depositOutpoints) loopIn := &StaticAddressLoopIn{ SwapHash: swapHash, @@ -596,3 +600,37 @@ func toStaticAddressLoopIn(_ context.Context, network *chaincfg.Params, return loopIn, nil } + +// orderDepositsBySnapshot returns deposits ordered by the stored outpoint +// snapshot when the current deposit rows still match that snapshot. If any +// snapshot outpoint no longer maps to a current deposit row, recovery keeps the +// store reconstruction untouched so callers can handle the divergence. +func orderDepositsBySnapshot(deposits []*deposit.Deposit, + depositOutpoints []string) []*deposit.Deposit { + + if len(deposits) != len(depositOutpoints) { + return deposits + } + + byOutpoint := make(map[string]*deposit.Deposit, len(deposits)) + for _, d := range deposits { + outpoint := d.OutPoint.String() + if _, ok := byOutpoint[outpoint]; ok { + return deposits + } + + byOutpoint[outpoint] = d + } + + orderedDeposits := make([]*deposit.Deposit, len(depositOutpoints)) + for i, outpoint := range depositOutpoints { + d, ok := byOutpoint[outpoint] + if !ok { + return deposits + } + + orderedDeposits[i] = d + } + + return orderedDeposits +} diff --git a/staticaddr/loopin/sql_store_test.go b/staticaddr/loopin/sql_store_test.go index 81fd10a7..fd4534b7 100644 --- a/staticaddr/loopin/sql_store_test.go +++ b/staticaddr/loopin/sql_store_test.go @@ -377,3 +377,160 @@ func TestCreateLoopIn(t *testing.T) { time.Microsecond, ) } + +// TestGetLoopInByHashOrdersDepositsBySnapshot ensures recovered deposits are +// ordered by the stored swap input snapshot, which is the signing order shared +// with the server. +func TestGetLoopInByHashOrdersDepositsBySnapshot(t *testing.T) { + ctx := context.Background() + testDb := loopdb.NewTestDB(t) + testClock := clock.NewTestClock(time.Now()) + defer testDb.Close() + + depositStore := deposit.NewSqlStore(testDb.BaseDB) + swapStore := NewSqlStore( + loopdb.NewTypedStore[Querier](testDb), testClock, + &chaincfg.RegressionNetParams, + ) + + newID := func() deposit.ID { + did, err := deposit.GetRandomDepositID() + require.NoError(t, err) + + return did + } + + d1 := &deposit.Deposit{ + ID: newID(), + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x11}, + Index: 0, + }, + Value: 100_000, + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x41, + }, + } + d2 := &deposit.Deposit{ + ID: newID(), + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x22}, + Index: 1, + }, + Value: 200_000, + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4d, + }, + } + + require.NoError(t, depositStore.CreateDeposit(ctx, d1)) + require.NoError(t, depositStore.CreateDeposit(ctx, d2)) + + d1.SetState(deposit.LoopingIn) + d2.SetState(deposit.LoopingIn) + require.NoError(t, depositStore.UpdateDeposit(ctx, d1)) + require.NoError(t, depositStore.UpdateDeposit(ctx, d2)) + + _, clientPubKey := test.CreateKey(1) + _, serverPubKey := test.CreateKey(2) + addr, err := btcutil.DecodeAddress(P2wkhAddr, nil) + require.NoError(t, err) + + swapHash := lntypes.Hash{0x1, 0x2, 0x3, 0x4} + swap := StaticAddressLoopIn{ + SwapHash: swapHash, + SwapPreimage: lntypes.Preimage{0x1, 0x2, 0x3, 0x4}, + DepositOutpoints: []string{ + d2.OutPoint.String(), d1.OutPoint.String(), + }, + Deposits: []*deposit.Deposit{d2, d1}, + ClientPubkey: clientPubKey, + ServerPubkey: serverPubKey, + HtlcTimeoutSweepAddress: addr, + } + swap.SetState(SignHtlcTx) + + require.NoError(t, swapStore.CreateLoopIn(ctx, &swap)) + + storedSwap, err := swapStore.GetLoopInByHash(ctx, swapHash) + require.NoError(t, err) + require.Equal(t, []string{ + d2.OutPoint.String(), d1.OutPoint.String(), + }, storedSwap.DepositOutpoints) + require.Len(t, storedSwap.Deposits, 2) + require.Equal(t, d2.ID, storedSwap.Deposits[0].ID) + require.Equal(t, d1.ID, storedSwap.Deposits[1].ID) +} + +// TestGetLoopInByHashPreservesStoredDepositOutpoints ensures recovered loop-ins +// keep the original outpoint snapshot stored when the swap was created. +func TestGetLoopInByHashPreservesStoredDepositOutpoints(t *testing.T) { + ctxb := context.Background() + testDb := loopdb.NewTestDB(t) + testClock := clock.NewTestClock(time.Now()) + defer testDb.Close() + + depositStore := deposit.NewSqlStore(testDb.BaseDB) + swapStore := NewSqlStore( + loopdb.NewTypedStore[Querier](testDb), testClock, + &chaincfg.RegressionNetParams, + ) + + depositID, err := deposit.GetRandomDepositID() + require.NoError(t, err) + + oldOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{0x1a, 0x2b, 0x3c, 0x4d}, + Index: 0, + } + currentOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{0x5a, 0x6b, 0x7c, 0x8d}, + Index: 1, + } + + d := &deposit.Deposit{ + ID: depositID, + OutPoint: oldOutpoint, + Value: btcutil.Amount(100_000), + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x41, + }, + } + require.NoError(t, depositStore.CreateDeposit(ctxb, d)) + + d.SetState(deposit.LoopingIn) + require.NoError(t, depositStore.UpdateDeposit(ctxb, d)) + + _, clientPubKey := test.CreateKey(1) + _, serverPubKey := test.CreateKey(2) + addr, err := btcutil.DecodeAddress(P2wkhAddr, nil) + require.NoError(t, err) + + swapHash := lntypes.Hash{0x1, 0x2, 0x3, 0x4} + swap := StaticAddressLoopIn{ + SwapHash: swapHash, + SwapPreimage: lntypes.Preimage{0x1, 0x2, 0x3, 0x4}, + DepositOutpoints: []string{oldOutpoint.String()}, + Deposits: []*deposit.Deposit{d}, + ClientPubkey: clientPubKey, + ServerPubkey: serverPubKey, + HtlcTimeoutSweepAddress: addr, + } + swap.SetState(SignHtlcTx) + + require.NoError(t, swapStore.CreateLoopIn(ctxb, &swap)) + + d.OutPoint = currentOutpoint + d.ConfirmationHeight = 42 + require.NoError(t, depositStore.UpdateDeposit(ctxb, d)) + + storedSwap, err := swapStore.GetLoopInByHash(ctxb, swapHash) + require.NoError(t, err) + require.Equal( + t, []string{oldOutpoint.String()}, + storedSwap.DepositOutpoints, + ) + require.Len(t, storedSwap.Deposits, 1) + require.Equal(t, currentOutpoint, storedSwap.Deposits[0].OutPoint) + require.Equal(t, int64(42), storedSwap.Deposits[0].ConfirmationHeight) +} diff --git a/staticaddr/loopin/txout_checker.go b/staticaddr/loopin/txout_checker.go new file mode 100644 index 00000000..da14be7c --- /dev/null +++ b/staticaddr/loopin/txout_checker.go @@ -0,0 +1,79 @@ +package loopin + +import ( + "context" + + "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/lndclient" +) + +// lndTxOutChecker checks outpoint availability using lnd's wallet transaction +// view. It omits outputs already spent by a wallet-known transaction. +type lndTxOutChecker struct { + client lndclient.LightningClient +} + +// NewLndTxOutChecker creates a TxOutChecker backed by lnd. +func NewLndTxOutChecker(client lndclient.LightningClient) TxOutChecker { + return &lndTxOutChecker{ + client: client, + } +} + +// GetTxOuts returns all requested tx outputs that lnd's transaction view still +// reports as unspent. +func (c *lndTxOutChecker) GetTxOuts(ctx context.Context, + outpoints []wire.OutPoint) (map[wire.OutPoint]*wire.TxOut, error) { + + outpointByString := make(map[string]wire.OutPoint, len(outpoints)) + outpointsByHash := make(map[string][]wire.OutPoint, len(outpoints)) + for _, outpoint := range outpoints { + outpointByString[outpoint.String()] = outpoint + outpointsByHash[outpoint.Hash.String()] = append( + outpointsByHash[outpoint.Hash.String()], outpoint, + ) + } + + // We need lnd's wallet transaction view rather than only the funding + // transaction: a matching previous outpoint tells us the deposit has + // already been spent by a wallet-known transaction. Use endHeight=-1 so + // lnd includes unconfirmed transactions and mempool spends. + txs, err := c.client.ListTransactions(ctx, 0, -1) + if err != nil { + return nil, err + } + + txOuts := make(map[wire.OutPoint]*wire.TxOut, len(outpoints)) + spent := make(map[wire.OutPoint]struct{}, len(outpoints)) + for _, tx := range txs { + for _, prevOutpoint := range tx.PreviousOutpoints { + outpoint, ok := outpointByString[prevOutpoint.GetOutpoint()] + if ok { + spent[outpoint] = struct{}{} + } + } + + if tx.Tx == nil { + continue + } + + txHash := tx.TxHash + if txHash == "" { + txHash = tx.Tx.TxHash().String() + } + + for _, outpoint := range outpointsByHash[txHash] { + if int(outpoint.Index) >= len(tx.Tx.TxOut) { + continue + } + + txOuts[outpoint] = tx.Tx.TxOut[outpoint.Index] + } + } + + for outpoint := range spent { + delete(txOuts, outpoint) + } + + return txOuts, nil +} diff --git a/staticaddr/loopin/txout_checker_test.go b/staticaddr/loopin/txout_checker_test.go new file mode 100644 index 00000000..491d9ce9 --- /dev/null +++ b/staticaddr/loopin/txout_checker_test.go @@ -0,0 +1,112 @@ +package loopin + +import ( + "context" + "errors" + "testing" + + "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/lndclient" + "github.com/lightningnetwork/lnd/lnrpc" + "github.com/stretchr/testify/require" +) + +func TestLndTxOutChecker(t *testing.T) { + fundingTx := wire.NewMsgTx(2) + fundingTx.AddTxOut(wire.NewTxOut(1000, []byte{0x01})) + fundingTx.AddTxOut(wire.NewTxOut(2000, []byte{0x02})) + + outpoint := wire.OutPoint{ + Hash: fundingTx.TxHash(), + Index: 1, + } + + t.Run("returns live tx outputs", func(t *testing.T) { + otherOutpoint := wire.OutPoint{ + Hash: fundingTx.TxHash(), + Index: 0, + } + client := &mockTxListLightningClient{ + txs: []lndclient.Transaction{{ + Tx: fundingTx, + }}, + } + + checker := NewLndTxOutChecker(client) + txOuts, err := checker.GetTxOuts( + t.Context(), []wire.OutPoint{outpoint, otherOutpoint}, + ) + require.NoError(t, err) + require.Equal(t, fundingTx.TxOut[outpoint.Index], txOuts[outpoint]) + require.Equal( + t, fundingTx.TxOut[otherOutpoint.Index], + txOuts[otherOutpoint], + ) + require.Equal(t, []txListCall{{ + startHeight: 0, + endHeight: -1, + }}, client.calls) + }) + + t.Run("returns nil for known spend", func(t *testing.T) { + client := &mockTxListLightningClient{ + txs: []lndclient.Transaction{{ + Tx: fundingTx, + }, { + PreviousOutpoints: []*lnrpc.PreviousOutPoint{{ + Outpoint: outpoint.String(), + }}, + }}, + } + + checker := NewLndTxOutChecker(client) + txOuts, err := checker.GetTxOuts( + t.Context(), []wire.OutPoint{outpoint}, + ) + require.NoError(t, err) + require.Nil(t, txOuts[outpoint]) + require.Equal(t, []txListCall{{ + startHeight: 0, + endHeight: -1, + }}, client.calls) + }) + + t.Run("returns error", func(t *testing.T) { + expectedErr := errors.New("list transactions failed") + client := &mockTxListLightningClient{ + err: expectedErr, + } + + checker := NewLndTxOutChecker(client) + txOuts, err := checker.GetTxOuts( + t.Context(), []wire.OutPoint{outpoint}, + ) + require.ErrorIs(t, err, expectedErr) + require.Nil(t, txOuts) + }) +} + +type txListCall struct { + startHeight int32 + endHeight int32 +} + +type mockTxListLightningClient struct { + lndclient.LightningClient + + txs []lndclient.Transaction + err error + calls []txListCall +} + +func (m *mockTxListLightningClient) ListTransactions(_ context.Context, + startHeight, endHeight int32, _ ...lndclient.ListTransactionsOption) ( + []lndclient.Transaction, error) { + + m.calls = append(m.calls, txListCall{ + startHeight: startHeight, + endHeight: endHeight, + }) + + return m.txs, m.err +} diff --git a/staticaddr/openchannel/manager.go b/staticaddr/openchannel/manager.go index 17f6c6a1..d1fda2f3 100644 --- a/staticaddr/openchannel/manager.go +++ b/staticaddr/openchannel/manager.go @@ -284,13 +284,8 @@ func (m *Manager) OpenChannel(ctx context.Context, // Check for duplicate outpoints which would lead to fee // miscalculation and an invalid PSBT with the same input // listed twice. - seen := make(map[wire.OutPoint]struct{}, len(outpoints)) - for _, op := range outpoints { - if _, ok := seen[op]; ok { - return nil, fmt.Errorf("duplicate outpoint "+ - "%v in request", op) - } - seen[op] = struct{}{} + if err := deposit.CheckDuplicates(outpoints); err != nil { + return nil, fmt.Errorf("%w in request", err) } deposits, allActive = @@ -620,7 +615,7 @@ func (m *Manager) openChannelPsbt(ctx context.Context, "address: %w", err) } - //nolint:ll + //nolint:lll signedTx, unsignedPsbt, err := m.cfg.WithdrawalManager.CreateFinalizedWithdrawalTx( ctx, deposits, channelFundingAddress, feeRate, fundingAmount, req.CommitmentType, diff --git a/staticaddr/staticutil/utils.go b/staticaddr/staticutil/utils.go index a8a5e404..a2509333 100644 --- a/staticaddr/staticutil/utils.go +++ b/staticaddr/staticutil/utils.go @@ -24,20 +24,21 @@ import ( func ToPrevOuts(deposits []*deposit.Deposit, pkScript []byte) (map[wire.OutPoint]*wire.TxOut, error) { + outpoints := make([]wire.OutPoint, len(deposits)) + for i, d := range deposits { + outpoints[i] = d.OutPoint + } + if err := deposit.CheckDuplicates(outpoints); err != nil { + return nil, err + } + prevOuts := make(map[wire.OutPoint]*wire.TxOut, len(deposits)) - for _, d := range deposits { - outpoint := wire.OutPoint{ - Hash: d.Hash, - Index: d.Index, - } + for i, d := range deposits { + outpoint := outpoints[i] txOut := &wire.TxOut{ Value: int64(d.Value), PkScript: pkScript, } - if _, ok := prevOuts[outpoint]; ok { - return nil, fmt.Errorf("duplicate outpoint %v", - outpoint) - } prevOuts[outpoint] = txOut } diff --git a/staticaddr/withdraw/manager.go b/staticaddr/withdraw/manager.go index 71967623..22acb79e 100644 --- a/staticaddr/withdraw/manager.go +++ b/staticaddr/withdraw/manager.go @@ -669,7 +669,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) diff --git a/tools/go.mod b/tools/go.mod index 819df12f..1f2d73ff 100644 --- a/tools/go.mod +++ b/tools/go.mod @@ -222,3 +222,9 @@ require ( go.augendre.info/fatcontext v0.9.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect ) + +// The go.augendre.info vanity import endpoints are currently unavailable, +// but these tags still declare the original module paths. +replace go.augendre.info/arangolint => github.com/Crocmagnon/arangolint v0.4.0 + +replace go.augendre.info/fatcontext => github.com/Crocmagnon/fatcontext v0.9.0 diff --git a/tools/go.sum b/tools/go.sum index 243eb20e..b04c04cd 100644 --- a/tools/go.sum +++ b/tools/go.sum @@ -61,6 +61,10 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03 github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= +github.com/Crocmagnon/arangolint v0.4.0 h1:hqwqcPrdvYouEmwLxT9lZGY5/Zwn25m5WvLa+JDxYhc= +github.com/Crocmagnon/arangolint v0.4.0/go.mod h1:l+f/b4plABuFISuKnTGD4RioXiCCgghv2xqst/xOvAA= +github.com/Crocmagnon/fatcontext v0.9.0 h1:uCKrygUTja+hDpqURwbOCpIOqdeywN196fUA+/9r9lU= +github.com/Crocmagnon/fatcontext v0.9.0/go.mod h1:L94brOAT1OOUNue6ph/2HnwxoNlds9aXDF2FcUntbNw= github.com/Djarvur/go-err113 v0.1.1 h1:eHfopDqXRwAi+YmCUas75ZE0+hoBHJ2GQNLYRSxao4g= github.com/Djarvur/go-err113 v0.1.1/go.mod h1:IaWJdYFLg76t2ihfflPZnM1LIQszWOsFDh2hhhAVF6k= github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0= @@ -622,10 +626,6 @@ go-simpler.org/musttag v0.14.0 h1:XGySZATqQYSEV3/YTy+iX+aofbZZllJaqwFWs+RTtSo= go-simpler.org/musttag v0.14.0/go.mod h1:uP8EymctQjJ4Z1kUnjX0u2l60WfUdQxCwSNKzE1JEOE= go-simpler.org/sloglint v0.11.1 h1:xRbPepLT/MHPTCA6TS/wNfZrDzkGvCCqUv4Bdwc3H7s= go-simpler.org/sloglint v0.11.1/go.mod h1:2PowwiCOK8mjiF+0KGifVOT8ZsCNiFzvfyJeJOIt8MQ= -go.augendre.info/arangolint v0.4.0 h1:xSCZjRoS93nXazBSg5d0OGCi9APPLNMmmLrC995tR50= -go.augendre.info/arangolint v0.4.0/go.mod h1:l+f/b4plABuFISuKnTGD4RioXiCCgghv2xqst/xOvAA= -go.augendre.info/fatcontext v0.9.0 h1:Gt5jGD4Zcj8CDMVzjOJITlSb9cEch54hjRRlN3qDojE= -go.augendre.info/fatcontext v0.9.0/go.mod h1:L94brOAT1OOUNue6ph/2HnwxoNlds9aXDF2FcUntbNw= go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU= go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8= go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=