diff --git a/staticaddr/loopin/actions.go b/staticaddr/loopin/actions.go index 47ffbbe7..ae831d0c 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -1,7 +1,6 @@ package loopin import ( - "bytes" "context" "crypto/rand" "errors" @@ -40,6 +39,8 @@ const ( DefaultPaymentTimeoutSeconds = 60 defaultInvoiceCleanupTimeout = 5 * time.Second + + monitorRetryDelay = time.Second ) var ( @@ -70,7 +71,10 @@ func (f *FSM) InitHtlcAction(ctx context.Context, return } - f.cancelSwapInvoice() + if err := f.cancelSwapInvoice(); err != nil { + f.Warnf("unable to clean up invoice for swap %v: %v", + f.loopIn.SwapHash, err) + } }() returnError := func(err error) fsm.EventType { @@ -342,11 +346,12 @@ func (f *FSM) InitHtlcAction(ctx context.Context, return event } -// cancelSwapInvoice best-effort cancels the current swap invoice using a -// detached timeout-limited context. -func (f *FSM) cancelSwapInvoice() { +// cancelSwapInvoice cancels the current swap invoice using a detached, +// timeout-limited context. Callers that must not proceed while the invoice may +// still be payable can use the returned error to retry. +func (f *FSM) cancelSwapInvoice() error { if f.loopIn.SwapHash == (lntypes.Hash{}) { - return + return nil } cleanupCtx, cancel := context.WithTimeout( @@ -355,10 +360,7 @@ func (f *FSM) cancelSwapInvoice() { defer cancel() err := f.cfg.InvoicesClient.CancelInvoice(cleanupCtx, f.loopIn.SwapHash) - if err != nil { - f.Warnf("unable to cancel invoice for swap %v: %v", - f.loopIn.SwapHash, err) - } + return err } // handleInvoiceUpdate applies the monitor state's invoice-update semantics and @@ -384,9 +386,13 @@ func (f *FSM) handleInvoiceUpdate(update lndclient.InvoiceUpdate) ( 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 + // An unknown state is not evidence that the invoice can no longer + // settle. Keep monitoring rather than leaving the deposits available + // for reuse. + f.Warnf("unexpected invoice state %v for swap hash %v", + update.State, f.loopIn.SwapHash) + + return fsm.NoOp, false } } @@ -556,7 +562,10 @@ func (f *FSM) SignHtlcTxAction(ctx context.Context, if outpointUnavailable { err = errors.New("original deposit outpoint no longer available") f.Warnf("%v, canceling swap invoice", err) - f.cancelSwapInvoice() + if cancelErr := f.cancelSwapInvoice(); cancelErr != nil { + f.Warnf("unable to cancel invoice for swap %v: %v", + f.loopIn.SwapHash, cancelErr) + } return f.HandleError(err) } @@ -782,6 +791,25 @@ func (f *FSM) cleanUpSessions(ctx context.Context, func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, _ fsm.EventContext) fsm.EventType { + retryMonitor := func(err error) fsm.EventType { + f.Errorf("monitoring failed: %v, retrying", err) + + invoice, lookupErr := f.cfg.LndClient.LookupInvoice( + ctx, f.loopIn.SwapHash, + ) + if lookupErr == nil && invoice.State == invoices.ContractSettled { + return OnPaymentReceived + } + + select { + case <-time.After(monitorRetryDelay): + return OnRecover + + case <-ctx.Done(): + return fsm.NoOp + } + } + // Subscribe to the state of the swap invoice. If upon restart recovery, // we land here and observe that the invoice is already canceled, it can // only be the case where a user-provided payment timeout was hit, the @@ -804,18 +832,20 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, err = fmt.Errorf("unable to subscribe to swap "+ "invoice: %w", err) - return f.HandleError(err) + return retryMonitor(err) } htlc, err := f.loopIn.getHtlc(f.cfg.ChainParams) if err != nil { err = fmt.Errorf("unable to get htlc: %w", err) - return f.HandleError(err) + return retryMonitor(err) } // Subscribe to htlc tx confirmation. reorgChan := make(chan struct{}, 1) + // registerHtlcConf registers for the HTLC transaction confirmation using + // the current reorg channel. registerHtlcConf := func() (chan *chainntnfs.TxConfirmation, chan error, error) { @@ -835,7 +865,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, err = fmt.Errorf("unable to monitor htlc tx confirmation: %w", err) - return f.HandleError(err) + return retryMonitor(err) } // Subscribe to new blocks. @@ -848,28 +878,16 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, err = fmt.Errorf("unable to subscribe to new blocks: %w", err) - return f.HandleError(err) + return retryMonitor(err) } - var ( - riskAcceptedChan <-chan *swapserverrpc. - ServerStaticLoopInRiskAcceptedNotification - riskRejectedChan <-chan *swapserverrpc. - ServerStaticLoopInRiskRejectedNotification - cancelRiskNotificationSubscriptions = func() {} + // The watcher keeps notification normalization and timestamp restoration + // outside of the swap-state handling below. + riskWatcher := newConfirmationRiskWatcher( + f.cfg, f.loopIn.SwapHash, f.Warnf, ) - if f.cfg.NotificationManager != nil { - notificationCtx, cancel := context.WithCancel(ctx) - cancelRiskNotificationSubscriptions = cancel - riskAcceptedChan = f.cfg.NotificationManager. - SubscribeStaticLoopInRiskAccepted( - notificationCtx, f.loopIn.SwapHash, - ) - riskRejectedChan = f.cfg.NotificationManager. - SubscribeStaticLoopInRiskRejected( - notificationCtx, f.loopIn.SwapHash, - ) - } + riskUpdateChan, cancelRiskNotificationSubscriptions := + riskWatcher.subscribe(ctx) defer cancelRiskNotificationSubscriptions() // Look up the current invoice state after registering subscriptions so @@ -880,10 +898,24 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, return fsm.NoOp } - err = fmt.Errorf("unable to look up invoice by swap hash: %w", - err) + // A failed lookup leaves the invoice state unknown. The active + // subscription can still provide an authoritative update, so keep + // monitoring and, most importantly, keep the deposits locked. + f.Warnf("unable to look up invoice by swap hash: %v", err) + invoice = &lndclient.Invoice{} + } - return f.HandleError(err) + // A settled invoice always takes precedence over a recovered risk + // rejection or an elapsed payment deadline. + if invoice.State == invoices.ContractSettled { + return OnPaymentReceived + } + + invoiceCanceledForNonPayment := invoice.State == invoices.ContractCanceled + if invoiceCanceledForNonPayment { + // If the invoice was canceled previously we end our + // subscription to invoice updates. + cancelInvoiceSubscription() } // Create the swap payment timeout timer after the server confirms @@ -894,12 +926,15 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, deadlineTimer *time.Timer deadlineStarted bool ) + // Stop the payment deadline timer when leaving the monitor action. defer func() { if deadlineTimer != nil { deadlineTimer.Stop() } }() + // depositsInState reports whether all selected deposits are currently + // in the requested state. depositsInState := func(state fsm.StateType) bool { if len(f.loopIn.Deposits) == 0 { return false @@ -910,10 +945,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, return false } - d.Lock() - inState := d.IsInStateNoLock(state) - d.Unlock() - if !inState { + if !d.IsInState(state) { return false } } @@ -921,11 +953,8 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, return true } - invoiceCanceledForNonPayment := invoice.State == invoices.ContractCanceled - depositsLockedForHtlcTimeout := depositsInState( - deposit.SweepHtlcTimeout, - ) - + // startPaymentDeadline arms the server payment timeout from the decision + // time when one is available. startPaymentDeadline := func(reason string, startedAt time.Time) { if deadlineStarted || invoice.State == invoices.ContractCanceled { return @@ -945,6 +974,12 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, deadlineStarted = true } + depositsLockedForHtlcTimeout := depositsInState( + deposit.SweepHtlcTimeout, + ) + + // transitionDepositsToHtlcTimeout locks deposits into timeout sweeping once + // the HTLC is confirmed and the invoice cannot be paid. transitionDepositsToHtlcTimeout := func(reason string) { if depositsLockedForHtlcTimeout || depositsInState(deposit.SweepHtlcTimeout) { @@ -969,111 +1004,64 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, depositsLockedForHtlcTimeout = true } + // startLegacyFallback starts the payment deadline once the old local + // minimum-confirmation rule has been satisfied. startLegacyFallback := func(reason string, currentHeight int32) { - if deadlineStarted || invoice.State == invoices.ContractCanceled { + if deadlineStarted || invoice.State == invoices.ContractCanceled || + f.loopIn.ConfirmationRiskDecision != + ConfirmationRiskDecisionNone { + return } if f.shouldStartLegacyConfirmationFallback(ctx, currentHeight) { - startPaymentDeadline(reason, time.Time{}) + decisionTime, ok := riskWatcher.durableDecisionTime( + ctx, ConfirmationRiskDecisionAccepted, + ) + if !ok { + return + } + + f.loopIn.ConfirmationRiskDecision = + ConfirmationRiskDecisionAccepted + f.loopIn.ConfirmationRiskDecisionTime = decisionTime + startPaymentDeadline(reason, decisionTime) } } - if invoice.State == invoices.ContractCanceled { - // If the invoice was canceled previously we end our - // subscription to invoice updates. + // cancelInvoice only marks the invoice canceled after lnd acknowledges + // the request or the lookup/subscription already observed that state. + // Failures recover the monitor state without releasing deposits. + cancelInvoice := func(reason string) (fsm.EventType, bool) { + if invoice.State != invoices.ContractCanceled { + f.Errorf("%s, canceling invoice", reason) + if err := f.cancelSwapInvoice(); err != nil { + return retryMonitor(err), false + } + } + cancelInvoiceSubscription() - } - - cancelInvoice := func() { - f.Errorf("timeout waiting for invoice to be " + - "paid, canceling invoice") - - // Cancel the lndclient invoice subscription. - cancelInvoiceSubscription() - - // Reuse the same helper as InitHtlcAction so timeout cleanup - // follows the same detached-context path as early-init cleanup. - f.cancelSwapInvoice() invoice.State = invoices.ContractCanceled invoiceCanceledForNonPayment = true + + return fsm.NoOp, true } - riskDecisionTime := func(decision ConfirmationRiskDecision) time.Time { - now := time.Now() - if f.cfg.Store == nil { - return now - } + // handleRiskRejected records a server rejection and only exits through + // the generic error path once the invoice can no longer settle. + handleRiskRejected := func(reason string, + decisionTime time.Time) fsm.EventType { - storedLoopIn, err := f.cfg.Store.GetLoopInByHash( - ctx, f.loopIn.SwapHash, - ) - if err != nil { - f.Warnf("unable to reload persisted risk decision for "+ - "swap %v: %v", f.loopIn.SwapHash, err) - - return now - } - - if storedLoopIn == nil { - return now - } - - hasPersistedDecision := - storedLoopIn.ConfirmationRiskDecision == decision && - !storedLoopIn.ConfirmationRiskDecisionTime.IsZero() - - if !hasPersistedDecision { - err = f.cfg.Store.RecordStaticAddressRiskDecision( - ctx, f.loopIn.SwapHash, decision, - ) - if err != nil { - f.Warnf("unable to persist replayed risk "+ - "decision for swap %v: %v", - f.loopIn.SwapHash, err) - - return now - } - - storedLoopIn, err = f.cfg.Store.GetLoopInByHash( - ctx, f.loopIn.SwapHash, - ) - if err != nil { - f.Warnf("unable to reload persisted risk "+ - "decision for swap %v: %v", - f.loopIn.SwapHash, err) - - return now - } - if storedLoopIn == nil || - storedLoopIn.ConfirmationRiskDecision != decision || - storedLoopIn.ConfirmationRiskDecisionTime.IsZero() { - - return now - } - } - - f.loopIn.ConfirmationRiskDecision = - storedLoopIn.ConfirmationRiskDecision - f.loopIn.ConfirmationRiskDecisionTime = - storedLoopIn.ConfirmationRiskDecisionTime - - return storedLoopIn.ConfirmationRiskDecisionTime - } - - handleRiskRejected := func(reason string) fsm.EventType { - cancelInvoiceSubscription() - f.cancelSwapInvoice() - invoice.State = invoices.ContractCanceled - invoiceCanceledForNonPayment = true - decisionTime := riskDecisionTime( - ConfirmationRiskDecisionRejected, - ) f.loopIn.ConfirmationRiskDecision = ConfirmationRiskDecisionRejected f.loopIn.ConfirmationRiskDecisionTime = decisionTime - riskAcceptedChan = nil - riskRejectedChan = nil + + event, canceled := cancelInvoice( + "server rejected confirmation risk wait after " + reason, + ) + if !canceled { + return event + } return f.HandleError(fmt.Errorf( "server rejected confirmation risk wait after %s", reason, @@ -1088,7 +1076,13 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, ) case ConfirmationRiskDecisionRejected: - return handleRiskRejected("recovered risk rejection") + decisionTime := riskWatcher.decisionTime( + ctx, ConfirmationRiskDecisionRejected, + ) + + return handleRiskRejected( + "recovered risk rejection", decisionTime, + ) } info, err := f.cfg.LndClient.GetInfo(ctx) @@ -1136,7 +1130,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, err = fmt.Errorf("unable to re-register for "+ "htlc tx confirmation: %w", err) - return f.HandleError(err) + return retryMonitor(err) } case <-reorgChan: @@ -1154,7 +1148,7 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, err = fmt.Errorf("unable to monitor htlc tx "+ "confirmation: %v", err) - return f.HandleError(err) + return retryMonitor(err) } case <-deadlineChan: @@ -1163,7 +1157,12 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // If the server didn't pay the invoice on time, we cancel // it and keep monitoring the htlc tx. Confirmed HTLC // deposits remain locked for timeout sweeping. - cancelInvoice() + event, canceled := cancelInvoice( + "timeout waiting for invoice to be paid", + ) + if !canceled { + return event + } if htlcConfirmed { transitionDepositsToHtlcTimeout("payment deadline") continue @@ -1175,45 +1174,31 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, "payment deadline: %v", err) } - case riskAccepted, ok := <-riskAcceptedChan: + case riskUpdate, ok := <-riskUpdateChan: if !ok { - riskAcceptedChan = nil + riskUpdateChan = nil continue } - if !bytes.Equal( - riskAccepted.SwapHash, f.loopIn.SwapHash[:], - ) { - - continue - } - - startedAt := riskDecisionTime( - ConfirmationRiskDecisionAccepted, - ) - f.loopIn.ConfirmationRiskDecision = - ConfirmationRiskDecisionAccepted - f.loopIn.ConfirmationRiskDecisionTime = startedAt - startPaymentDeadline( - "risk accepted notification", - f.loopIn.ConfirmationRiskDecisionTime, + decisionTime := riskWatcher.decisionTime( + ctx, riskUpdate.decision, ) + f.loopIn.ConfirmationRiskDecision = riskUpdate.decision + f.loopIn.ConfirmationRiskDecisionTime = decisionTime - case riskRejected, ok := <-riskRejectedChan: - if !ok { - riskRejectedChan = nil - continue + switch riskUpdate.decision { + case ConfirmationRiskDecisionAccepted: + startPaymentDeadline( + riskUpdate.reason, + f.loopIn.ConfirmationRiskDecisionTime, + ) + + case ConfirmationRiskDecisionRejected: + return handleRiskRejected( + riskUpdate.reason, decisionTime, + ) } - if !bytes.Equal( - riskRejected.SwapHash, f.loopIn.SwapHash[:], - ) { - - continue - } - - return handleRiskRejected("risk rejection") - case currentHeight := <-blockChan: startLegacyFallback( "legacy confirmation fallback", currentHeight, @@ -1236,10 +1221,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, f.Infof("htlc timed out at block height %v", currentHeight) - - // If the timeout path opened up we consider the swap - // failed and cancel the invoice. - cancelInvoice() + event, canceled := cancelInvoice("htlc timed out") + if !canceled { + return event + } if !htlcConfirmed { f.Infof("swap timed out, htlc not confirmed") @@ -1247,13 +1232,13 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // If the htlc hasn't confirmed but the timeout // path opened up, and we didn't receive the // swap payment, we consider the swap attempt to - // be failed. Now that the HTLC can no longer - // confirm, the deposits can be made available - // again. + // be failed. Now that the invoice is canceled and + // the HTLC can no longer confirm, its deposits can be + // made available again. err = f.unlockDeposits(ctx) if err != nil { - f.Errorf("unable to unlock deposits "+ - "after htlc timeout: %v", err) + f.Errorf("unable to unlock deposits after "+ + "htlc timeout: %v", err) } return OnSwapTimedOut @@ -1272,10 +1257,16 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, f.Errorf("block subscription error: %v", err) - return f.HandleError(err) + return retryMonitor(err) case update, ok := <-invoiceUpdateChan: if !ok { + if !invoiceCanceledForNonPayment { + return retryMonitor(errors.New( + "invoice update subscription closed", + )) + } + invoiceUpdateChan = nil continue } @@ -1284,13 +1275,30 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, return event } + invoice.State = update.State + if update.State == invoices.ContractCanceled { + invoiceCanceledForNonPayment = true + } + case err, ok := <-invoiceErrChan: if !ok { + if !invoiceCanceledForNonPayment { + return retryMonitor(errors.New( + "invoice error subscription closed", + )) + } + invoiceErrChan = nil continue } - f.Errorf("invoice subscription error: %v", err) + if ctx.Err() != nil { + return fsm.NoOp + } + + return retryMonitor(fmt.Errorf( + "invoice subscription error: %w", err, + )) case <-ctx.Done(): return fsm.NoOp @@ -1425,7 +1433,10 @@ func (f *FSM) PaymentReceivedAction(ctx context.Context, func (f *FSM) UnlockDepositsAction(ctx context.Context, _ fsm.EventContext) fsm.EventType { - f.cancelSwapInvoice() + if err := f.cancelSwapInvoice(); err != nil { + f.Warnf("unable to cancel invoice for swap %v: %v", + f.loopIn.SwapHash, err) + } err := f.unlockDeposits(ctx) if err != nil { diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index 030aed30..24f92b2c 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -25,6 +25,8 @@ import ( "google.golang.org/grpc" ) +const testTimeout = 5 * time.Second + // TestHandleInvoiceUpdate verifies that invoice state updates map to the // monitor events expected by the static address loop-in FSM. func TestHandleInvoiceUpdate(t *testing.T) { @@ -60,11 +62,9 @@ func TestHandleInvoiceUpdate(t *testing.T) { event: fsm.NoOp, }, { - name: "unexpected", - state: invoices.ContractState(99), - event: fsm.OnError, - done: true, - errString: "unexpected invoice state", + name: "unexpected", + state: invoices.ContractState(99), + event: fsm.NoOp, }, } @@ -103,12 +103,304 @@ func TestHandleInvoiceUpdate(t *testing.T) { } } +// TestMonitorInvoiceSettledWinsOverRecoveredRiskRejection verifies that an +// authoritative settled state takes precedence over a persisted server risk +// rejection during recovery. +func TestMonitorInvoiceSettledWinsOverRecoveredRiskRejection(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) + defer cancel() + + mockLnd := test.NewMockLnd() + swapHash := lntypes.Hash{1, 2, 5} + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractSettled, + }) + + f, depositMgr := newInvoiceMonitorTestFSM( + t, ctx, mockLnd, swapHash, ConfirmationRiskDecisionRejected, + mockLnd.LndServices.Invoices, + ) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + select { + case event := <-resultChan: + require.Equal(t, OnPaymentReceived, event) + + case <-ctx.Done(): + t.Fatalf("monitor action did not exit: %v", ctx.Err()) + } + + require.Empty(t, depositMgr.transitions) + select { + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("settled invoice was canceled: %v", hash) + + default: + } +} + +// TestMonitorInvoiceCancelErrorKeepsMonitoring verifies that cancellation +// failures neither unlock deposits nor stop invoice monitoring. Recovery +// rechecks the authoritative invoice state, and a later settlement wins. +func TestMonitorInvoiceCancelErrorKeepsMonitoring(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) + defer cancel() + + mockLnd := test.NewMockLnd() + swapHash := lntypes.Hash{1, 2, 6} + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + cancelCalls := make(chan lntypes.Hash, 2) + releaseCancel := make(chan struct{}) + invoicesClient := &failingCancelInvoices{ + InvoicesClient: mockLnd.LndServices.Invoices, + cancelCalls: cancelCalls, + release: releaseCancel, + err: errors.New("invoice backend unavailable"), + } + f, depositMgr := newInvoiceMonitorTestFSM( + t, ctx, mockLnd, swapHash, ConfirmationRiskDecisionRejected, + invoicesClient, + ) + f.ActionEntryFunc = nil + + resultChan := make(chan error, 1) + go func() { + resultChan <- f.SendEvent(ctx, OnRecover, nil) + }() + waitForMonitorSubscriptions(t, ctx, mockLnd) + + select { + case hash := <-cancelCalls: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("cancellation attempt not received: %v", ctx.Err()) + } + + select { + case transition := <-depositMgr.transitionChan: + t.Fatalf("deposits unlocked after cancellation error: %v", + transition) + + default: + } + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractSettled, + }) + close(releaseCancel) + + select { + case err := <-resultChan: + require.NoError(t, err) + + case <-ctx.Done(): + t.Fatalf("monitor action did not exit: %v", ctx.Err()) + } + + require.Equal(t, []fsm.StateType{deposit.LoopedIn}, depositMgr.states) +} + +// TestMonitorInvoiceUnknownStateKeepsMonitoring verifies that a failed lookup +// and an unknown subscription state do not release deposits. A subsequent +// authoritative settlement still completes the swap. +func TestMonitorInvoiceUnknownStateKeepsMonitoring(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) + defer cancel() + + mockLnd := test.NewMockLnd() + swapHash := lntypes.Hash{1, 2, 7} + f, depositMgr := newInvoiceMonitorTestFSM( + t, ctx, mockLnd, swapHash, ConfirmationRiskDecisionNone, + mockLnd.LndServices.Invoices, + ) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + var invoiceSub *test.SingleInvoiceSubscription + select { + case invoiceSub = <-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()) + } + + invoiceSub.Update <- lndclient.InvoiceUpdate{ + Invoice: lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractState(99), + }, + } + + select { + case event := <-resultChan: + t.Fatalf("unknown invoice state ended monitor with %v", event) + + case transition := <-depositMgr.transitionChan: + t.Fatalf("unknown invoice state unlocked deposits: %v", transition) + + case <-time.After(100 * time.Millisecond): + } + + invoiceSub.Update <- lndclient.InvoiceUpdate{ + Invoice: lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractSettled, + }, + } + + select { + case event := <-resultChan: + require.Equal(t, OnPaymentReceived, event) + + case <-ctx.Done(): + t.Fatalf("monitor action did not exit: %v", ctx.Err()) + } + + require.Empty(t, depositMgr.transitions) +} + +// TestMonitorInvoiceSetupFailureRecoversSettledInvoice verifies that a +// transient subscription failure cannot route an already-settled swap through +// deposit cleanup before the authoritative invoice lookup runs. +func TestMonitorInvoiceSetupFailureRecoversSettledInvoice(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) + defer cancel() + + mockLnd := test.NewMockLnd() + swapHash := lntypes.Hash{1, 2, 8} + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractSettled, + }) + invoicesClient := &flakySubscribeInvoices{ + InvoicesClient: mockLnd.LndServices.Invoices, + err: errors.New("invoice backend unavailable"), + } + f, depositMgr := newInvoiceMonitorTestFSM( + t, ctx, mockLnd, swapHash, ConfirmationRiskDecisionRejected, + invoicesClient, + ) + f.ActionEntryFunc = nil + + resultChan := make(chan error, 1) + go func() { + resultChan <- f.SendEvent(ctx, OnRecover, nil) + }() + + select { + case err := <-resultChan: + require.NoError(t, err) + + case <-ctx.Done(): + t.Fatalf("monitor did not recover: %v", ctx.Err()) + } + + require.Equal(t, 1, invoicesClient.subscribeCalls) + require.Equal(t, []fsm.StateType{deposit.LoopedIn}, depositMgr.states) + select { + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("settled invoice was canceled: %v", hash) + + default: + } +} + +// TestMonitorInvoiceStreamErrorRecoversSettlement verifies that a dead invoice +// stream checks the latest invoice state before entering recovery. +func TestMonitorInvoiceStreamErrorRecoversSettlement(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) + defer cancel() + + mockLnd := test.NewMockLnd() + swapHash := lntypes.Hash{1, 2, 9} + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + f, depositMgr := newInvoiceMonitorTestFSM( + t, ctx, mockLnd, swapHash, ConfirmationRiskDecisionNone, + mockLnd.LndServices.Invoices, + ) + lookupStarted := make(chan struct{}) + releaseLookup := make(chan struct{}) + f.cfg.LndClient = &firstLookupBarrier{ + LightningClient: mockLnd.Client, + lookupStarted: lookupStarted, + release: releaseLookup, + } + f.ActionEntryFunc = nil + + resultChan := make(chan error, 1) + go func() { + resultChan <- f.SendEvent(ctx, OnRecover, nil) + }() + + var invoiceSub *test.SingleInvoiceSubscription + select { + case invoiceSub = <-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()) + } + select { + case <-lookupStarted: + case <-ctx.Done(): + t.Fatalf("initial invoice lookup not received: %v", ctx.Err()) + } + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractSettled, + }) + close(releaseLookup) + select { + case invoiceSub.Err <- errors.New("invoice stream failed"): + case <-ctx.Done(): + t.Fatalf("invoice stream error was not consumed: %v", ctx.Err()) + } + + select { + case err := <-resultChan: + require.NoError(t, err) + + case <-ctx.Done(): + t.Fatalf("monitor did not recover: %v", ctx.Err()) + } + + require.Equal(t, []fsm.StateType{deposit.LoopedIn}, depositMgr.states) +} + // 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 // test would time out waiting for the second one. func TestMonitorInvoiceAndHtlcTxReRegistersOnConfErr(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -211,7 +503,7 @@ func TestMonitorInvoiceAndHtlcTxReRegistersOnConfErr(t *testing.T) { // 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) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() runCtx, stop := context.WithCancel(ctx) @@ -325,7 +617,7 @@ func TestSweepHtlcTimeoutActionNoOpOnShutdown(t *testing.T) { // 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) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -362,7 +654,7 @@ func TestMonitorHtlcTimeoutSweepActionNoOpOnShutdown(t *testing.T) { require.Equal(t, fsm.NoOp, event) require.Nil(t, f.LastActionError) - case <-time.After(5 * time.Second): + case <-time.After(testTimeout): t.Fatal("timeout sweep monitor did not return") } } @@ -370,7 +662,7 @@ func TestMonitorHtlcTimeoutSweepActionNoOpOnShutdown(t *testing.T) { // TestMonitorInvoiceAndHtlcTxShutdownDoesNotUnlock verifies that daemon // shutdown exits the monitor action without treating the swap as failed. func TestMonitorInvoiceAndHtlcTxShutdownDoesNotUnlock(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() runCtx, stop := context.WithCancel(ctx) @@ -669,7 +961,7 @@ func testValidateLoopInContract(_ int32, _ int32) error { // payment timeout starts on risk acceptance and keeps confirmed HTLC deposits // locked for timeout sweeping. func TestMonitorInvoiceAndHtlcTxLocksConfirmedHtlcAtDeadline(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -806,11 +1098,149 @@ func TestMonitorInvoiceAndHtlcTxLocksConfirmedHtlcAtDeadline(t *testing.T) { } } +// TestMonitorInvoiceAndHtlcTxIgnoresWrongHashRiskNotifications verifies that +// risk notifications for another swap do not start the payment deadline or +// persist a decision through the monitor action. +func TestMonitorInvoiceAndHtlcTxIgnoresWrongHashRiskNotifications( + t *testing.T) { + + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) + defer cancel() + + mockLnd := test.NewMockLnd() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + swapHash := lntypes.Hash{4, 5, 8} + otherHash := lntypes.Hash{8, 5, 4} + depositOutpoint := wire.OutPoint{ + Hash: chainhash.Hash{9}, + Index: 0, + } + + 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, + DepositOutpoints: []string{ + depositOutpoint.String(), + }, + Deposits: []*deposit.Deposit{{ + OutPoint: depositOutpoint, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + mockLnd.SetInvoice(&lndclient.Invoice{ + Hash: swapHash, + State: invoices.ContractOpen, + }) + + notificationMgr := &mockNotificationManager{ + riskAccepted: make( + chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification, 2, + ), + riskRejected: make( + chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification, 1, + ), + } + store := &recordingRiskStore{ + mockStore: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: {}, + }, + }, + decisions: make(chan ConfirmationRiskDecision, 1), + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: &noopDepositManager{}, + InvoicesClient: mockLnd.LndServices.Invoices, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + NotificationManager: notificationMgr, + Store: store, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- f.MonitorInvoiceAndHtlcTxAction(ctx, nil) + }() + + waitForMonitorSubscriptions(t, ctx, mockLnd) + + notificationMgr.riskAccepted <- &swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification{ + SwapHash: otherHash[:], + } + notificationMgr.riskRejected <- &swapserverrpc. + ServerStaticLoopInRiskRejectedNotification{ + SwapHash: otherHash[:], + } + + select { + case decision := <-store.decisions: + t.Fatalf("persisted wrong-hash risk decision: %v", decision) + + case hash := <-mockLnd.FailInvoiceChannel: + t.Fatalf("canceled invoice for wrong-hash risk decision: %v", hash) + + case event := <-resultChan: + t.Fatalf("monitor action exited after wrong-hash risk decision: %v", + event) + + case <-time.After(200 * time.Millisecond): + } + + notificationMgr.riskAccepted <- &swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification{ + SwapHash: swapHash[:], + } + + select { + case decision := <-store.decisions: + require.Equal(t, ConfirmationRiskDecisionAccepted, decision) + + case <-ctx.Done(): + t.Fatalf("risk decision was not persisted: %v", ctx.Err()) + } + + cancel() + select { + case event := <-resultChan: + require.Equal(t, fsm.NoOp, event) + + case <-time.After(time.Second): + t.Fatal("monitor action did not exit") + } +} + // TestMonitorInvoiceAndHtlcTxUsesPersistedAcceptedRiskTime verifies that live // risk notifications use the durable receipt time, not the local channel // receive time, when reconstructing the payment deadline. func TestMonitorInvoiceAndHtlcTxUsesPersistedAcceptedRiskTime(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -925,7 +1355,7 @@ func TestMonitorInvoiceAndHtlcTxUsesPersistedAcceptedRiskTime(t *testing.T) { // TestMonitorInvoiceAndHtlcTxPersistsReplayedRiskAccepted verifies that a risk // notification replayed after the swap row exists is written back to the store. func TestMonitorInvoiceAndHtlcTxPersistsReplayedRiskAccepted(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1038,7 +1468,7 @@ func TestMonitorInvoiceAndHtlcTxPersistsReplayedRiskAccepted(t *testing.T) { // confirmation risk rejection is persisted and exits through the generic error // path so the FSM unlocks deposits. func TestMonitorInvoiceAndHtlcTxPersistsRiskRejected(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1161,7 +1591,7 @@ func TestMonitorInvoiceAndHtlcTxPersistsRiskRejected(t *testing.T) { // persisted risk acceptance restarts the payment deadline with elapsed time // preserved after restart. func TestMonitorInvoiceAndHtlcTxRecoversAcceptedRiskDecision(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1262,7 +1692,7 @@ func TestMonitorInvoiceAndHtlcTxRecoversAcceptedRiskDecision(t *testing.T) { // persisted risk rejection still cancels after restart and exits through the // generic error path so the FSM unlocks deposits. func TestMonitorInvoiceAndHtlcTxRecoversRejectedRiskDecision(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1360,7 +1790,7 @@ func TestMonitorInvoiceAndHtlcTxRecoversRejectedRiskDecision(t *testing.T) { func TestMonitorInvoiceAndHtlcTxDoesNotCancelWhenOriginalOutpointVanishes( t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1452,7 +1882,7 @@ func TestMonitorInvoiceAndHtlcTxDoesNotCancelWhenOriginalOutpointVanishes( func TestMonitorInvoiceAndHtlcTxDoesNotCancelAcceptedInvoiceForMissingOutpoint( t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1537,7 +1967,7 @@ func TestMonitorInvoiceAndHtlcTxDoesNotCancelAcceptedInvoiceForMissingOutpoint( // monitor action preserves the legacy payment deadline fallback when no risk // decision has been observed locally. func TestMonitorInvoiceAndHtlcTxStartsDeadlineAtLegacyMinConfs(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1649,7 +2079,7 @@ func TestMonitorInvoiceAndHtlcTxStartsDeadlineAtLegacyMinConfs(t *testing.T) { func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackWithNotificationManager( t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1761,7 +2191,7 @@ func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackWithNotificationManager( func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackAtCurrentHeight( t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -1805,6 +2235,14 @@ func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackAtCurrentHeight( Hash: swapHash, State: invoices.ContractOpen, }) + store := &recordingRiskStore{ + mockStore: &mockStore{ + loopIns: map[lntypes.Hash]*StaticAddressLoopIn{ + swapHash: {}, + }, + }, + decisions: make(chan ConfirmationRiskDecision, 1), + } cfg := &Config{ AddressManager: &mockAddressManager{ @@ -1823,6 +2261,7 @@ func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackAtCurrentHeight( InvoicesClient: mockLnd.LndServices.Invoices, LndClient: mockLnd.Client, ChainParams: mockLnd.ChainParams, + Store: store, } f, err := NewFSM(ctx, loopIn, cfg, false) @@ -1835,6 +2274,19 @@ func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackAtCurrentHeight( waitForMonitorSubscriptions(t, ctx, mockLnd) + select { + case decision := <-store.decisions: + require.Equal(t, ConfirmationRiskDecisionAccepted, decision) + + case <-ctx.Done(): + t.Fatalf("legacy fallback decision was not persisted: %v", + ctx.Err()) + } + require.Equal(t, ConfirmationRiskDecisionAccepted, + store.loopIns[swapHash].ConfirmationRiskDecision) + require.False(t, + store.loopIns[swapHash].ConfirmationRiskDecisionTime.IsZero()) + select { case hash := <-mockLnd.FailInvoiceChannel: t.Fatalf("invoice canceled before payment deadline: %v", hash) @@ -1864,7 +2316,7 @@ func TestMonitorInvoiceAndHtlcTxStartsLegacyFallbackAtCurrentHeight( // recovered monitor state does not rely on stale selected-deposit snapshots when // deciding whether the legacy payment deadline fallback has opened. func TestMonitorInvoiceAndHtlcTxRefreshesDepositsForLegacyFallback(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -2033,7 +2485,7 @@ func TestLegacyConfirmationFallbackStopsOnFreshnessFailure(t *testing.T) { func TestMonitorInvoiceAndHtlcTxUnlocksOnHtlcTimeoutWithoutDeadline( t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -2139,6 +2591,140 @@ func waitForMonitorSubscriptions(t *testing.T, ctx context.Context, } } +// newInvoiceMonitorTestFSM creates the minimal monitor-state setup shared by +// invoice precedence and cancellation tests. +func newInvoiceMonitorTestFSM(t *testing.T, ctx context.Context, + mockLnd *test.LndMockServices, swapHash lntypes.Hash, + decision ConfirmationRiskDecision, + invoicesClient lndclient.InvoicesClient) (*FSM, *recordingDepositManager) { + + t.Helper() + + clientKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + loopIn := &StaticAddressLoopIn{ + SwapHash: swapHash, + HtlcCltvExpiry: mockLnd.Height + 1_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + ProtocolVersion: version.ProtocolVersion_V0, + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + PaymentTimeoutSeconds: 3_600, + ConfirmationRiskDecision: decision, + ConfirmationRiskDecisionTime: time.Now(), + Deposits: []*deposit.Deposit{{ + Value: 200_000, + }}, + } + loopIn.SetState(MonitorInvoiceAndHtlcTx) + + depositMgr := &recordingDepositManager{ + transitionChan: make(chan depositTransition, 1), + } + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ClientPubkey: clientKey.PubKey(), + ServerPubkey: serverKey.PubKey(), + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + ChainNotifier: mockLnd.ChainNotifier, + DepositManager: depositMgr, + InvoicesClient: invoicesClient, + LndClient: mockLnd.Client, + ChainParams: mockLnd.ChainParams, + } + + f, err := NewFSM(ctx, loopIn, cfg, true) + require.NoError(t, err) + + return f, depositMgr +} + +// failingCancelInvoices records cancellation attempts and returns a configured +// error after its release channel is closed. +type failingCancelInvoices struct { + lndclient.InvoicesClient + + cancelCalls chan lntypes.Hash + release chan struct{} + err error +} + +// flakySubscribeInvoices counts subscription attempts and returns a configured +// subscription error. +type flakySubscribeInvoices struct { + lndclient.InvoicesClient + + subscribeCalls int + err error +} + +// firstLookupBarrier blocks the first invoice lookup until its release channel +// is closed. +type firstLookupBarrier struct { + lndclient.LightningClient + + lookupStarted chan struct{} + release chan struct{} + firstLookup bool +} + +func (f *firstLookupBarrier) LookupInvoice(ctx context.Context, + hash lntypes.Hash) (*lndclient.Invoice, error) { + + invoice, err := f.LightningClient.LookupInvoice(ctx, hash) + if f.firstLookup { + return invoice, err + } + + f.firstLookup = true + close(f.lookupStarted) + select { + case <-f.release: + return invoice, err + + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func (f *flakySubscribeInvoices) SubscribeSingleInvoice(ctx context.Context, + hash lntypes.Hash) (<-chan lndclient.InvoiceUpdate, <-chan error, error) { + + f.subscribeCalls++ + if f.subscribeCalls == 1 { + return nil, nil, f.err + } + + return f.InvoicesClient.SubscribeSingleInvoice(ctx, hash) +} + +func (f *failingCancelInvoices) CancelInvoice(ctx context.Context, + hash lntypes.Hash) error { + + select { + case f.cancelCalls <- hash: + case <-ctx.Done(): + return ctx.Err() + } + + if f.release != nil { + select { + case <-f.release: + case <-ctx.Done(): + return ctx.Err() + } + } + + return f.err +} + // TestOriginalDepositOutpointUnavailableRequiresMissingTxOut verifies that a // present txout does not trigger the RBF cancellation path. func TestOriginalDepositOutpointUnavailableRequiresMissingTxOut(t *testing.T) { @@ -2172,7 +2758,7 @@ func TestOriginalDepositOutpointUnavailableRequiresMissingTxOut(t *testing.T) { // pending loop-in is canceled before HTLC signing if GetTxOuts reports that // one of the originally selected outpoints is gone. func TestSignHtlcTxActionCancelsWhenOriginalOutpointUnavailable(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -2223,7 +2809,7 @@ func TestSignHtlcTxActionCancelsWhenOriginalOutpointUnavailable(t *testing.T) { // failures are treated as errors, but do not cancel the invoice. The invoice is // only canceled when GetTxOuts omits an original outpoint. func TestSignHtlcTxActionDoesNotCancelOnTxOutLookupError(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -2271,7 +2857,7 @@ func TestSignHtlcTxActionDoesNotCancelOnTxOutLookupError(t *testing.T) { // TestInitHtlcActionCancelsInvoiceOnServerError verifies that an invoice // created before a server-side rejection is canceled immediately. func TestInitHtlcActionCancelsInvoiceOnServerError(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -2321,7 +2907,7 @@ func TestInitHtlcActionCancelsInvoiceOnServerError(t *testing.T) { // TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure verifies that the early // fee guard also cancels the pre-created invoice before returning an error. func TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() @@ -2389,7 +2975,7 @@ func TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure(t *testing.T) { // TestUnlockDepositsActionCancelsInvoice verifies that stored swaps that enter // the generic error unlock path also clean up their swap invoice. func TestUnlockDepositsActionCancelsInvoice(t *testing.T) { - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + ctx, cancel := context.WithTimeout(t.Context(), testTimeout) defer cancel() mockLnd := test.NewMockLnd() diff --git a/staticaddr/loopin/risk_watcher.go b/staticaddr/loopin/risk_watcher.go index 4da95a20..6df9a745 100644 --- a/staticaddr/loopin/risk_watcher.go +++ b/staticaddr/loopin/risk_watcher.go @@ -126,14 +126,16 @@ func (w *confirmationRiskWatcher) subscribe(ctx context.Context) ( return riskUpdates, cancel } -// decisionTime returns the durable decision timestamp, recording the decision -// first if the notification was replayed before it could be persisted. -func (w *confirmationRiskWatcher) decisionTime(ctx context.Context, - decision ConfirmationRiskDecision) time.Time { +// durableDecisionTime returns the durable decision timestamp, recording the +// decision first if the notification was replayed before it could be persisted. +// The bool is false when a configured store could not durably record or reload +// the decision. +func (w *confirmationRiskWatcher) durableDecisionTime(ctx context.Context, + decision ConfirmationRiskDecision) (time.Time, bool) { now := time.Now() if w.store == nil { - return now + return now, true } storedLoopIn, err := w.store.GetLoopInByHash(ctx, w.swapHash) @@ -141,11 +143,11 @@ func (w *confirmationRiskWatcher) decisionTime(ctx context.Context, w.warnf("unable to reload persisted risk decision for swap %v: %v", w.swapHash, err) - return now + return time.Time{}, false } if storedLoopIn == nil { - return now + return time.Time{}, false } hasPersistedDecision := @@ -160,7 +162,7 @@ func (w *confirmationRiskWatcher) decisionTime(ctx context.Context, w.warnf("unable to persist replayed risk decision for "+ "swap %v: %v", w.swapHash, err) - return now + return time.Time{}, false } storedLoopIn, err = w.store.GetLoopInByHash(ctx, w.swapHash) @@ -168,15 +170,27 @@ func (w *confirmationRiskWatcher) decisionTime(ctx context.Context, w.warnf("unable to reload persisted risk decision for "+ "swap %v: %v", w.swapHash, err) - return now + return time.Time{}, false } if storedLoopIn == nil || storedLoopIn.ConfirmationRiskDecision != decision || storedLoopIn.ConfirmationRiskDecisionTime.IsZero() { - return now + return time.Time{}, false } } - return storedLoopIn.ConfirmationRiskDecisionTime + return storedLoopIn.ConfirmationRiskDecisionTime, true +} + +// decisionTime retains the best-effort behavior used for server notifications. +func (w *confirmationRiskWatcher) decisionTime(ctx context.Context, + decision ConfirmationRiskDecision) time.Time { + + decisionTime, ok := w.durableDecisionTime(ctx, decision) + if !ok { + return time.Now() + } + + return decisionTime }