diff --git a/notifications/manager.go b/notifications/manager.go index 3eda5d65..2ff6b087 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -45,6 +45,10 @@ const ( // htlc-confirmed subscriber before dropping the notification. htlcConfirmedSubscriberSendTimeout = 200 * time.Millisecond + // defaultMaxQueuedNotifications is the default number of notifications + // we queue per subscriber before dropping new notifications. + defaultMaxQueuedNotifications = 1024 + // current_version is the current version of the notification listener. current_version = swapserverrpc.SubscribeNotificationsRequest_V1 ) @@ -72,6 +76,10 @@ type Config struct { // MinAliveConnTime is the minimum time that the connection to the // server needs to be alive before we consider it a successful. MinAliveConnTime time.Duration + + // MaxQueuedNotifications is the maximum number of notifications that + // can wait in each subscriber's delivery queue. + MaxQueuedNotifications int } // Manager is a manager for notifications that the swap server sends to the @@ -92,6 +100,9 @@ func NewManager(cfg *Config) *Manager { if cfg.MinAliveConnTime == 0 { cfg.MinAliveConnTime = defaultMinAliveConnTime } + if cfg.MaxQueuedNotifications <= 0 { + cfg.MaxQueuedNotifications = defaultMaxQueuedNotifications + } return &Manager{ cfg: cfg, @@ -102,6 +113,113 @@ func NewManager(cfg *Config) *Manager { type subscriber struct { subCtx context.Context recvChan any + enqueue func(any) +} + +// newNotificationQueue creates a per-subscriber FIFO delivery function. +func newNotificationQueue[T any](ctx context.Context, + recvChan chan T, maxPending int) func(any) { + + type queue struct { + sync.Mutex + + pending []T + notify chan struct{} + closed bool + } + + q := &queue{ + notify: make(chan struct{}, 1), + } + + closeQueue := func() { + q.Lock() + q.closed = true + q.pending = nil + q.Unlock() + } + + go func() { + defer close(recvChan) + defer closeQueue() + + for { + select { + case <-ctx.Done(): + return + default: + } + + q.Lock() + if len(q.pending) == 0 { + q.Unlock() + + select { + case <-q.notify: + continue + + case <-ctx.Done(): + return + } + } + + ntfn := q.pending[0] + var zero T + q.pending[0] = zero + q.pending = q.pending[1:] + q.Unlock() + + select { + case recvChan <- ntfn: + case <-ctx.Done(): + return + } + } + }() + + return func(ntfn any) { + typedNtfn, ok := ntfn.(T) + if !ok { + log.Warnf("unexpected notification type %T", ntfn) + return + } + + q.Lock() + if q.closed { + q.Unlock() + return + } + if len(q.pending) >= maxPending { + q.Unlock() + log.Warnf("dropping notification for slow subscriber: "+ + "queue depth %d reached", maxPending) + return + } + + q.pending = append(q.pending, typedNtfn) + q.Unlock() + + select { + case q.notify <- struct{}{}: + default: + } + } +} + +// queueNotification queues or synchronously sends a must-deliver notification. +func queueNotification[T any](sub subscriber, recvChan chan T, ntfn T) { + if sub.enqueue != nil { + sub.enqueue(ntfn) + return + } + + log.Warnf("subscriber has no notification queue, falling back to " + + "blocking send") + + select { + case recvChan <- ntfn: + case <-sub.subCtx.Done(): + } } // SubscribeReservations subscribes to the reservation notifications. @@ -136,6 +254,9 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context, sub := subscriber{ subCtx: ctx, recvChan: notifChan, + enqueue: newNotificationQueue( + ctx, notifChan, m.cfg.MaxQueuedNotifications, + ), } m.addSubscriber(NotificationTypeStaticLoopInSweepRequest, sub) @@ -145,7 +266,6 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context, NotificationTypeStaticLoopInSweepRequest, sub, ) - close(notifChan) }) return notifChan @@ -161,12 +281,14 @@ func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context, sub := subscriber{ subCtx: ctx, recvChan: notifChan, + enqueue: newNotificationQueue( + ctx, notifChan, m.cfg.MaxQueuedNotifications, + ), } m.addSubscriber(NotificationTypeUnfinishedSwap, sub) context.AfterFunc(ctx, func() { m.removeSubscriber(NotificationTypeUnfinishedSwap, sub) - close(notifChan) }) return notifChan @@ -332,7 +454,13 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerReservationNotification) - recvChan <- reservationNtfn + select { + case recvChan <- reservationNtfn: + case <-sub.subCtx.Done(): + default: + log.Debugf("Dropping reservation " + + "notification for slow subscriber") + } } case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInSweep: // nolint: lll // We'll forward the static loop in sweep request to all @@ -345,7 +473,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerStaticLoopInSweepNotification) - recvChan <- staticLoopInSweepRequestNtfn + queueNotification(sub, recvChan, staticLoopInSweepRequestNtfn) } case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll @@ -359,7 +487,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerUnfinishedSwapNotification) - recvChan <- unfinishedSwapNtfn + queueNotification(sub, recvChan, unfinishedSwapNtfn) } case *swapserverrpc.SubscribeNotificationsResponse_HtlcConfirmed: @@ -403,7 +531,7 @@ func (m *Manager) removeSubscriber(notifType NotificationType, sub subscriber) { subs := m.subscribers[notifType] newSubs := make([]subscriber, 0, len(subs)) for _, s := range subs { - if s != sub { + if s.recvChan != sub.recvChan { newSubs = append(newSubs, s) } } diff --git a/notifications/manager_test.go b/notifications/manager_test.go index 44300d7a..1768b004 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -20,7 +20,7 @@ import ( var ( testReservationId = []byte{0x01, 0x02} - testReservationId2 = []byte{0x01, 0x02} + testReservationId2 = []byte{0x03, 0x04} ) // mockNotificationsClient implements the NotificationsClient interface for testing. @@ -190,6 +190,276 @@ func getTestNotification(resId []byte) *swapserverrpc.SubscribeNotificationsResp } } +// unfinishedSwapNotification builds an unfinished swap notification. +func unfinishedSwapNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_UnfinishedSwap{ + UnfinishedSwap: &swapserverrpc. + ServerUnfinishedSwapNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +// staticLoopInSweepNotification builds a static loop-in sweep notification. +func staticLoopInSweepNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_StaticLoopInSweep{ + StaticLoopInSweep: &swapserverrpc. + ServerStaticLoopInSweepNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +// TestManager_SlowReservationSubscriberDoesNotBlock tests that a reservation +// subscriber with a full notification channel does not block delivery to other +// subscribers. Reservation notifications are best-effort, so slow subscribers +// drop new notifications instead of queueing them. +func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + slowCtx, slowCancel := context.WithCancel(t.Context()) + defer slowCancel() + slowChan := mgr.SubscribeReservations(slowCtx) + + fastCtx, fastCancel := context.WithCancel(t.Context()) + defer fastCancel() + fastChan := mgr.SubscribeReservations(fastCtx) + + firstNotif := getTestNotification(testReservationId) + mgr.handleNotification(firstNotif) + + received := <-fastChan + require.Equal(t, testReservationId, received.ReservationId) + + secondNotif := getTestNotification(testReservationId2) + done := make(chan struct{}) + go func() { + mgr.handleNotification(secondNotif) + close(done) + }() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + select { + case received = <-fastChan: + require.Equal(t, testReservationId2, received.ReservationId) + + case <-time.After(time.Second): + t.Fatal("fast subscriber did not receive notification") + } + + require.Len(t, slowChan, 1) + + select { + case received = <-slowChan: + require.Equal(t, testReservationId, received.ReservationId) + + case <-time.After(time.Second): + t.Fatal("slow subscriber did not receive first notification") + } + + select { + case received = <-slowChan: + t.Fatalf("slow subscriber received dropped notification %x", + received.ReservationId) + + case <-time.After(50 * time.Millisecond): + } +} + +// TestManager_UnfinishedSwapNotificationWaitsForSubscriber verifies that +// unfinished swap recovery notifications are not dropped when the local +// subscriber is briefly behind. +func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) { + t.Parallel() + + assertQueuedSwapHashNotifications( + t, + func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc. + ServerUnfinishedSwapNotification { + + return mgr.SubscribeUnfinishedSwaps(ctx) + }, + unfinishedSwapNotification, + func(ntfn *swapserverrpc.ServerUnfinishedSwapNotification) []byte { + return ntfn.SwapHash + }, + lntypes.Hash{0x02, 0x03}, lntypes.Hash{0x04, 0x05}, + "did not receive first unfinished swap notification", + "second unfinished swap notification was dropped", + ) +} + +// TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber verifies +// that a full static-loop-in sweep subscriber channel does not block the global +// notification receive loop. +func TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber( + t *testing.T) { + + t.Parallel() + + assertQueuedSwapHashNotifications( + t, + func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc. + ServerStaticLoopInSweepNotification { + + return mgr.SubscribeStaticLoopInSweepRequests(ctx) + }, + staticLoopInSweepNotification, + func(ntfn *swapserverrpc.ServerStaticLoopInSweepNotification) []byte { + return ntfn.SwapHash + }, + lntypes.Hash{0x12, 0x13}, lntypes.Hash{0x14, 0x15}, + "did not receive first sweep notification", + "second sweep notification was not queued", + ) +} + +// TestManager_QueuedNotificationChannelClosesOnCancel verifies that queued +// subscribers own their channel shutdown even when delivery is blocked. +func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + subChan := mgr.SubscribeUnfinishedSwaps(subCtx) + + swapHashA := lntypes.Hash{0x21, 0x22} + mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + + require.Eventually(t, func() bool { + return len(subChan) == 1 + }, time.Second, 10*time.Millisecond) + + swapHashB := lntypes.Hash{0x23, 0x24} + done := make(chan struct{}) + go func() { + mgr.handleNotification(unfinishedSwapNotification(swapHashB)) + close(done) + }() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + subCancel() + + select { + case received, ok := <-subChan: + require.True(t, ok) + require.Equal(t, swapHashA[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("first unfinished swap notification was not delivered") + } + + require.Eventually(t, func() bool { + select { + case _, ok := <-subChan: + return !ok + default: + return false + } + }, time.Second, 10*time.Millisecond) +} + +// TestNotificationQueueDropsAtCapacity checks the queue's explicit drop policy +// once a subscriber reaches its configured backlog limit. +func TestNotificationQueueDropsAtCapacity(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + recvChan := make(chan int, 1) + enqueue := newNotificationQueue(ctx, recvChan, 0) + + enqueue(1) + + select { + case ntfn := <-recvChan: + t.Fatalf("received dropped notification %d", ntfn) + + case <-time.After(50 * time.Millisecond): + } +} + +// assertQueuedSwapHashNotifications checks queued delivery for swap hashes. +func assertQueuedSwapHashNotifications[T any](t *testing.T, + subscribe func(*Manager, context.Context) <-chan T, + notification func(lntypes.Hash) *swapserverrpc. + SubscribeNotificationsResponse, + swapHash func(T) []byte, swapHashA, swapHashB lntypes.Hash, + firstFailureMsg, secondFailureMsg string) { + + t.Helper() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := subscribe(mgr, subCtx) + + mgr.handleNotification(notification(swapHashA)) + + done := make(chan struct{}) + go func() { + mgr.handleNotification(notification(swapHashB)) + close(done) + }() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + select { + case received := <-subChan: + require.Equal(t, swapHashA[:], swapHash(received)) + + case <-time.After(time.Second): + t.Fatal(firstFailureMsg) + } + + select { + case received := <-subChan: + require.Equal(t, swapHashB[:], swapHash(received)) + + case <-time.After(time.Second): + t.Fatal(secondFailureMsg) + } +} + // TestManager_Backoff verifies that repeated failures in // subscribeNotifications cause the Manager to space out subscription attempts // via a predictable incremental backoff. diff --git a/staticaddr/deposit/actions.go b/staticaddr/deposit/actions.go index 362417e7..437c3ac2 100644 --- a/staticaddr/deposit/actions.go +++ b/staticaddr/deposit/actions.go @@ -161,14 +161,25 @@ func (f *FSM) WaitForExpirySweepAction(ctx context.Context, // FinalizeDepositAction is the final action after a withdrawal. It signals to // the manager that the deposit has been swept and the FSM can be removed. -func (f *FSM) FinalizeDepositAction(ctx context.Context, +func (f *FSM) FinalizeDepositAction(_ context.Context, _ fsm.EventContext) fsm.EventType { - select { - case <-ctx.Done(): - return fsm.OnError + outpoint := f.deposit.OutPoint - case f.finalizedDepositChan <- f.deposit.OutPoint: - return fsm.NoOp - } + // The finalization notification only tells the manager to remove the + // deposit from its active set. Send it asynchronously so a busy manager + // loop can't stall withdrawal confirmation while deposit locks are held. + go func() { + select { + case <-f.quitChan: + // The deposit is already in a final state. If shutdown wins + // this race, startup recovery will skip it instead of + // re-adding it to the active set. + return + + case f.finalizedDepositChan <- outpoint: + } + }() + + return fsm.NoOp } diff --git a/staticaddr/deposit/actions_test.go b/staticaddr/deposit/actions_test.go new file mode 100644 index 00000000..8c021121 --- /dev/null +++ b/staticaddr/deposit/actions_test.go @@ -0,0 +1,135 @@ +package deposit + +import ( + "context" + "testing" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/loop/fsm" + "github.com/stretchr/testify/require" +) + +// TestFinalizeDepositActionDoesNotBlock ensures the final cleanup notification +// does not block the withdrawal completion path while the manager loop is busy. +func TestFinalizeDepositActionDoesNotBlock(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: 1, + } + + depositFSM := &FSM{ + deposit: &Deposit{ + OutPoint: outpoint, + }, + quitChan: make(chan struct{}), + finalizedDepositChan: make(chan wire.OutPoint), + } + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- depositFSM.FinalizeDepositAction(ctx, nil) + }() + + select { + case result := <-resultChan: + require.Equal(t, fsm.NoOp, result) + + case <-time.After(100 * time.Millisecond): + t.Fatal("FinalizeDepositAction blocked on manager cleanup") + } + + select { + case gotOutpoint := <-depositFSM.finalizedDepositChan: + require.Equal(t, outpoint, gotOutpoint) + + case <-time.After(time.Second): + t.Fatal("finalization cleanup notification was not delivered") + } +} + +// TestFinalizeDepositActionIgnoresRequestCancellation ensures the cleanup +// notification is tied to the FSM lifetime, not the caller's request context. +func TestFinalizeDepositActionIgnoresRequestCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + quitChan := make(chan struct{}) + defer close(quitChan) + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{2}, + Index: 2, + } + + depositFSM := &FSM{ + deposit: &Deposit{ + OutPoint: outpoint, + }, + quitChan: quitChan, + finalizedDepositChan: make(chan wire.OutPoint), + } + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- depositFSM.FinalizeDepositAction(ctx, nil) + }() + + select { + case result := <-resultChan: + require.Equal(t, fsm.NoOp, result) + + case <-time.After(100 * time.Millisecond): + t.Fatal("FinalizeDepositAction blocked on manager cleanup") + } + + cancel() + + select { + case gotOutpoint := <-depositFSM.finalizedDepositChan: + require.Equal(t, outpoint, gotOutpoint) + + case <-time.After(time.Second): + t.Fatal("finalization cleanup notification was dropped after " + + "request cancellation") + } +} + +// TestFinalizeDepositActionIgnoresCanceledContext ensures the final cleanup +// notification is still queued even if the caller's context is already done. +func TestFinalizeDepositActionIgnoresCanceledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + quitChan := make(chan struct{}) + defer close(quitChan) + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{3}, + Index: 3, + } + + depositFSM := &FSM{ + deposit: &Deposit{ + OutPoint: outpoint, + }, + quitChan: quitChan, + finalizedDepositChan: make(chan wire.OutPoint), + } + + result := depositFSM.FinalizeDepositAction(ctx, nil) + require.Equal(t, fsm.NoOp, result) + + select { + case gotOutpoint := <-depositFSM.finalizedDepositChan: + require.Equal(t, outpoint, gotOutpoint) + + case <-time.After(time.Second): + t.Fatal("finalization cleanup notification was dropped for " + + "an already-canceled request context") + } +} diff --git a/staticaddr/loopin/actions.go b/staticaddr/loopin/actions.go index 70a27811..ccf86b93 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -36,6 +36,8 @@ const ( defaultConfTarget = 3 DefaultPaymentTimeoutSeconds = 60 + + defaultInvoiceCleanupTimeout = 5 * time.Second ) var ( @@ -57,6 +59,24 @@ var ( func (f *FSM) InitHtlcAction(ctx context.Context, _ fsm.EventContext) fsm.EventType { + var event fsm.EventType + invoiceNeedsCleanup := false + defer func() { + // If we created the private invoice but failed before persisting the + // swap, cancel it so retries do not accumulate orphan invoices. + if !invoiceNeedsCleanup || event != fsm.OnError { + return + } + + f.cancelSwapInvoice() + }() + + returnError := func(err error) fsm.EventType { + event = f.HandleError(err) + + return event + } + // Lock the deposits and transition them to the LoopingIn state. err := f.cfg.DepositManager.TransitionDeposits( ctx, f.loopIn.Deposits, deposit.OnLoopInInitiated, @@ -65,7 +85,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { err = fmt.Errorf("unable to loop-in deposits: %w", err) - return f.HandleError(err) + return returnError(err) } // Calculate the swap invoice amount. The server needs to pay us the @@ -88,7 +108,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("unable to create random swap preimage: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.SwapPreimage = swapPreimage f.loopIn.SwapHash = swapPreimage.Hash() @@ -100,7 +120,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { err = fmt.Errorf("unable to derive client htlc key: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.ClientPubkey = keyDesc.PubKey f.loopIn.HtlcKeyLocator = keyDesc.KeyLocator @@ -119,10 +139,14 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { err = fmt.Errorf("unable to create swap invoice: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.SwapInvoice = swapInvoice + // From here until CreateLoopIn succeeds, any error path would otherwise + // leave behind a live invoice with no persisted swap to recover it. + invoiceNeedsCleanup = true + f.loopIn.ProtocolVersion = version.AddressProtocolVersion( version.CurrentRPCProtocolVersion(), ) @@ -149,7 +173,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("unable to initiate the loop-in with the "+ "server: %w", err) - return f.HandleError(err) + return returnError(err) } // Pushing empty sigs signals the server that we abandoned the swap @@ -171,7 +195,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, pushEmptySigs() err = fmt.Errorf("unable to parse server pubkey: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.ServerPubkey = serverPubkey @@ -185,7 +209,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("server response parameters are outside "+ "our allowed range: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.HtlcCltvExpiry = loopInResp.HtlcExpiry @@ -194,7 +218,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, pushEmptySigs() err = fmt.Errorf("unable to convert server nonces: %w", err) - return f.HandleError(err) + return returnError(err) } f.htlcServerNoncesHighFee, err = toNonces( loopInResp.HighFeeHtlcInfo.Nonces, @@ -202,7 +226,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { pushEmptySigs() - return f.HandleError(err) + return returnError(err) } f.htlcServerNoncesExtremelyHighFee, err = toNonces( loopInResp.ExtremeFeeHtlcInfo.Nonces, @@ -210,7 +234,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { pushEmptySigs() - return f.HandleError(err) + return returnError(err) } // We need to defend against the server setting high fees for the htlc @@ -232,7 +256,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, log.Errorf("server htlc tx fee is higher than the configured "+ "allowed maximum: %v > %v", fee, maxHtlcTxFee) - return f.HandleError(ErrFeeTooHigh) + return returnError(ErrFeeTooHigh) } f.loopIn.HtlcTxFeeRate = feeRate @@ -246,7 +270,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, "configured allowed maximum: %v > %v", fee, maxHtlcTxBackupFee) - return f.HandleError(ErrFeeTooHigh) + return returnError(ErrFeeTooHigh) } f.loopIn.HtlcTxHighFeeRate = highFeeRate @@ -262,7 +286,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, "configured allowed maximum: %v > %v", fee, maxHtlcTxBackupFee) - return f.HandleError(ErrFeeTooHigh) + return returnError(ErrFeeTooHigh) } f.loopIn.HtlcTxExtremelyHighFeeRate = extremelyHighFeeRate @@ -276,7 +300,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("unable to derive htlc timeout sweep "+ "address: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.HtlcTimeoutSweepAddress = sweepAddress @@ -286,10 +310,34 @@ func (f *FSM) InitHtlcAction(ctx context.Context, pushEmptySigs() err = fmt.Errorf("unable to store loop-in in db: %w", err) - return f.HandleError(err) + return returnError(err) } - return OnHtlcInitiated + // Once the swap is stored, restart/recovery code owns invoice lifecycle. + invoiceNeedsCleanup = false + + event = OnHtlcInitiated + + return event +} + +// cancelSwapInvoice best-effort cancels the current swap invoice using a +// detached timeout-limited context. +func (f *FSM) cancelSwapInvoice() { + if f.loopIn.SwapInvoice == "" { + return + } + + cleanupCtx, cancel := context.WithTimeout( + context.Background(), defaultInvoiceCleanupTimeout, + ) + 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) + } } // SignHtlcTxAction is called if the htlc was initialized and the server @@ -557,11 +605,9 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // Cancel the lndclient invoice subscription. cancelInvoiceSubscription() - err = f.cfg.InvoicesClient.CancelInvoice(ctx, f.loopIn.SwapHash) - if err != nil { - f.Warnf("unable to cancel invoice "+ - "for swap hash: %v", err) - } + // Reuse the same helper as InitHtlcAction so timeout cleanup + // follows the same detached-context path as early-init cleanup. + f.cancelSwapInvoice() } for { @@ -609,10 +655,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // re-enable them for loop-ins and withdrawals. cancelInvoice() - event := f.UnlockDepositsAction(ctx, nil) - if event != fsm.OnError { - f.Errorf("unable to unlock deposits after " + - "payment deadline") + err = f.unlockDeposits(ctx) + if err != nil { + f.Errorf("unable to unlock deposits after "+ + "payment deadline: %v", err) } case currentHeight := <-blockChan: @@ -824,18 +870,27 @@ func (f *FSM) PaymentReceivedAction(ctx context.Context, func (f *FSM) UnlockDepositsAction(ctx context.Context, _ fsm.EventContext) fsm.EventType { - err := f.cfg.DepositManager.TransitionDeposits( - ctx, f.loopIn.Deposits, fsm.OnError, deposit.Deposited, - ) - if err != nil { - err = fmt.Errorf("unable to unlock deposits: %w", err) + f.cancelSwapInvoice() + err := f.unlockDeposits(ctx) + if err != nil { return f.HandleError(err) } return fsm.OnError } +func (f *FSM) unlockDeposits(ctx context.Context) error { + err := f.cfg.DepositManager.TransitionDeposits( + ctx, f.loopIn.Deposits, fsm.OnError, deposit.Deposited, + ) + if err != nil { + return fmt.Errorf("unable to unlock deposits: %w", err) + } + + return nil +} + // createAndPublishHtlcTimeoutSweepTx creates and publishes the htlc timeout // sweep transaction. func (f *FSM) createAndPublishHtlcTimeoutSweepTx(ctx context.Context) error { diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index b058e328..d75adf6c 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -270,6 +270,192 @@ func testValidateLoopInContract(_ int32, _ int32) error { return nil } +// 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) + defer cancel() + + mockLnd := test.NewMockLnd() + + loopIn := &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{ + Value: 200_000, + }}, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + PaymentTimeoutSeconds: DefaultPaymentTimeoutSeconds, + ProtocolVersion: version.ProtocolVersion_V0, + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + DepositManager: &noopDepositManager{}, + WalletKit: mockLnd.WalletKit, + LndClient: mockLnd.Client, + InvoicesClient: mockLnd.LndServices.Invoices, + Server: &initHtlcTestServer{ + loopInErr: errors.New("server rejected swap"), + }, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + // The init step should fail and synchronously trigger deferred invoice + // cleanup. + event := f.InitHtlcAction(ctx, nil) + require.Equal(t, fsm.OnError, event) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, loopIn.SwapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } +} + +// 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) + defer cancel() + + mockLnd := test.NewMockLnd() + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + loopIn := &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{ + Value: 200_000, + }}, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + PaymentTimeoutSeconds: DefaultPaymentTimeoutSeconds, + ProtocolVersion: version.ProtocolVersion_V0, + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + DepositManager: &noopDepositManager{}, + WalletKit: mockLnd.WalletKit, + LndClient: mockLnd.Client, + InvoicesClient: mockLnd.LndServices.Invoices, + Server: &initHtlcTestServer{ + loopInResp: &swapserverrpc.ServerStaticAddressLoopInResponse{ + HtlcServerPubKey: serverKey.PubKey(). + SerializeCompressed(), + HtlcExpiry: mockLnd.Height + + DefaultLoopInOnChainCltvDelta, + StandardHtlcInfo: &swapserverrpc.ServerHtlcSigningInfo{ + FeeRate: 1_000_000, + }, + HighFeeHtlcInfo: &swapserverrpc.ServerHtlcSigningInfo{}, + ExtremeFeeHtlcInfo: &swapserverrpc. + ServerHtlcSigningInfo{}, + }, + }, + ValidateLoopInContract: func(int32, int32) error { + return nil + }, + MaxStaticAddrHtlcFeePercentage: 0, + MaxStaticAddrHtlcBackupFeePercentage: 1, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + // The fee guard runs before persistence, so the deferred cleanup must + // cancel the invoice on this error path as well. + event := f.InitHtlcAction(ctx, nil) + require.Equal(t, fsm.OnError, event) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, loopIn.SwapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } +} + +// 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) + defer cancel() + + mockLnd := test.NewMockLnd() + dep := &deposit.Deposit{ + Value: 200_000, + } + swapHash := lntypes.Hash{0x44, 0x55} + depositMgr := &recordingDepositManager{} + + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + DepositManager: depositMgr, + InvoicesClient: mockLnd.LndServices.Invoices, + }, + loopIn: &StaticAddressLoopIn{ + SwapHash: swapHash, + SwapInvoice: "lnbc1test", + Deposits: []*deposit.Deposit{dep}, + }, + } + + event := f.UnlockDepositsAction(ctx, nil) + require.Equal(t, fsm.OnError, event) + require.NoError(t, f.LastActionError) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } + + require.Len(t, depositMgr.transitions, 1) + require.Equal(t, []*deposit.Deposit{dep}, depositMgr.transitions[0].deposits) + require.Equal(t, fsm.OnError, depositMgr.transitions[0].event) + require.Equal(t, deposit.Deposited, depositMgr.transitions[0].state) +} + +// TestUnlockDepositsActionReportsTransitionError ensures the unlock path +// preserves the real deposit transition failure for callers that need to log it. +func TestUnlockDepositsActionReportsTransitionError(t *testing.T) { + depositMgr := &recordingDepositManager{ + err: errors.New("transition failed"), + } + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + DepositManager: depositMgr, + }, + loopIn: &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{Value: 200_000}}, + }, + } + + event := f.UnlockDepositsAction(t.Context(), nil) + require.Equal(t, fsm.OnError, event) + require.ErrorContains( + t, f.LastActionError, "unable to unlock deposits", + ) + require.ErrorContains(t, f.LastActionError, "transition failed") +} + // mockAddressManager is a minimal AddressManager implementation used by the // test FSM setup. type mockAddressManager struct { @@ -327,3 +513,56 @@ func (n *noopDepositManager) GetActiveDepositsInState(fsm.StateType) ( return nil, nil } + +type depositTransition struct { + deposits []*deposit.Deposit + event fsm.EventType + state fsm.StateType +} + +type recordingDepositManager struct { + noopDepositManager + + err error + transitions []depositTransition +} + +// TransitionDeposits records the transition and returns the configured error. +func (r *recordingDepositManager) TransitionDeposits(_ context.Context, + deposits []*deposit.Deposit, event fsm.EventType, + state fsm.StateType) error { + + r.transitions = append(r.transitions, depositTransition{ + deposits: deposits, + event: event, + state: state, + }) + + return r.err +} + +// initHtlcTestServer lets InitHtlcAction tests inject a deterministic server +// response without standing up the full gRPC client. +type initHtlcTestServer struct { + swapserverrpc.StaticAddressServerClient + + loopInResp *swapserverrpc.ServerStaticAddressLoopInResponse + loopInErr error +} + +// ServerStaticAddressLoopIn returns the canned response configured by the test. +func (s *initHtlcTestServer) ServerStaticAddressLoopIn(context.Context, + *swapserverrpc.ServerStaticAddressLoopInRequest, ...grpc.CallOption, +) (*swapserverrpc.ServerStaticAddressLoopInResponse, error) { + + return s.loopInResp, s.loopInErr +} + +// PushStaticAddressHtlcSigs accepts the abandonment signal used by error-path +// tests without adding additional assertions. +func (s *initHtlcTestServer) PushStaticAddressHtlcSigs(context.Context, + *swapserverrpc.PushStaticAddressHtlcSigsRequest, ...grpc.CallOption, +) (*swapserverrpc.PushStaticAddressHtlcSigsResponse, error) { + + return &swapserverrpc.PushStaticAddressHtlcSigsResponse{}, nil +} diff --git a/staticaddr/loopin/sql_store.go b/staticaddr/loopin/sql_store.go index d06c5c18..d8c253be 100644 --- a/staticaddr/loopin/sql_store.go +++ b/staticaddr/loopin/sql_store.go @@ -56,8 +56,8 @@ type Querier interface { swapHash []byte) (sqlc.GetStaticAddressLoopInSwapRow, error) // GetStaticAddressLoopInSwapsByStates retrieves all swaps with the - // given states. The states string is an input for the IN primitive in - // sqlite, hence the format needs to be '{State1,State2,...}'. + // given states. The states string is comma-separated so the query can + // match complete state names by wrapping it with comma sentinels. GetStaticAddressLoopInSwapsByStates(ctx context.Context, states sql.NullString) ([]sqlc.GetStaticAddressLoopInSwapsByStatesRow, error) @@ -203,7 +203,7 @@ func (s *SqlStore) GetStaticAddressLoopInSwapsByStates(ctx context.Context, } func toJointStringStates(states []fsm.StateType) string { - return "{" + strings.Join(toStrings(states), ",") + "}" + return strings.Join(toStrings(states), ",") } func toStrings(states []fsm.StateType) []string { diff --git a/staticaddr/loopin/sql_store_test.go b/staticaddr/loopin/sql_store_test.go index 1e30081d..81fd10a7 100644 --- a/staticaddr/loopin/sql_store_test.go +++ b/staticaddr/loopin/sql_store_test.go @@ -41,8 +41,10 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { } loopingDepositID := newID() + timeoutDepositID := newID() loopedInDepositID := newID() - d1, d2 := &deposit.Deposit{ + failedDepositID := newID() + d1, d2, d3, d4 := &deposit.Deposit{ ID: loopingDepositID, OutPoint: wire.OutPoint{ Hash: chainhash.Hash{0x1a, 0x2b, 0x3c, 0x4d}, @@ -54,7 +56,7 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { }, }, &deposit.Deposit{ - ID: loopedInDepositID, + ID: timeoutDepositID, OutPoint: wire.OutPoint{ Hash: chainhash.Hash{0x2a, 0x2b, 0x3c, 0x4e}, Index: 1, @@ -63,29 +65,67 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { TimeOutSweepPkScript: []byte{ 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4d, }, + }, + &deposit.Deposit{ + ID: loopedInDepositID, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x3a, 0x2b, 0x3c, 0x4e}, + Index: 2, + }, + Value: btcutil.Amount(300_000), + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4f, + }, + }, + &deposit.Deposit{ + ID: failedDepositID, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x4a, 0x2b, 0x3c, 0x4e}, + Index: 3, + }, + Value: btcutil.Amount(400_000), + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x50, + }, } err := depositStore.CreateDeposit(ctxb, d1) require.NoError(t, err) err = depositStore.CreateDeposit(ctxb, d2) require.NoError(t, err) + err = depositStore.CreateDeposit(ctxb, d3) + require.NoError(t, err) + err = depositStore.CreateDeposit(ctxb, d4) + require.NoError(t, err) // Add two updates per deposit, expect the last to be retrieved. d1.SetState(deposit.Deposited) d2.SetState(deposit.Deposited) + d3.SetState(deposit.Deposited) + d4.SetState(deposit.Deposited) err = depositStore.UpdateDeposit(ctxb, d1) require.NoError(t, err) err = depositStore.UpdateDeposit(ctxb, d2) require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d3) + require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d4) + require.NoError(t, err) d1.SetState(deposit.LoopingIn) - d2.SetState(deposit.LoopedIn) + d2.SetState(deposit.HtlcTimeoutSwept) + d3.SetState(deposit.LoopedIn) + d4.SetState(deposit.Deposited) err = depositStore.UpdateDeposit(ctxb, d1) require.NoError(t, err) err = depositStore.UpdateDeposit(ctxb, d2) require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d3) + require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d4) + require.NoError(t, err) _, clientPubKey := test.CreateKey(1) _, serverPubKey := test.CreateKey(2) @@ -108,13 +148,30 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { err = swapStore.CreateLoopIn(ctxb, &swapPending) require.NoError(t, err) + // Create htlc-timeout-swept swap. HtlcTimeoutSwept is the first final + // state, so this exercises the state-list query boundary. + swapHashTimeoutSwept := lntypes.Hash{0x4, 0x2, 0x3, 0x5} + swapTimeoutSwept := StaticAddressLoopIn{ + SwapHash: swapHashTimeoutSwept, + SwapPreimage: lntypes.Preimage{0x4, 0x2, 0x3, 0x5}, + DepositOutpoints: []string{d2.OutPoint.String()}, + Deposits: []*deposit.Deposit{d2}, + ClientPubkey: clientPubKey, + ServerPubkey: serverPubKey, + HtlcTimeoutSweepAddress: addr, + } + swapTimeoutSwept.SetState(HtlcTimeoutSwept) + + err = swapStore.CreateLoopIn(ctxb, &swapTimeoutSwept) + require.NoError(t, err) + // Create succeeded swap. swapHashSucceeded := lntypes.Hash{0x2, 0x2, 0x3, 0x5} swapSucceeded := StaticAddressLoopIn{ SwapHash: swapHashSucceeded, SwapPreimage: lntypes.Preimage{0x2, 0x2, 0x3, 0x5}, - DepositOutpoints: []string{d2.OutPoint.String()}, - Deposits: []*deposit.Deposit{d2}, + DepositOutpoints: []string{d3.OutPoint.String()}, + Deposits: []*deposit.Deposit{d3}, ClientPubkey: clientPubKey, ServerPubkey: serverPubKey, HtlcTimeoutSweepAddress: addr, @@ -124,6 +181,23 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { err = swapStore.CreateLoopIn(ctxb, &swapSucceeded) require.NoError(t, err) + // Create failed swap. Failed is the last final state, so this + // exercises the state-list query boundary. + swapHashFailed := lntypes.Hash{0x3, 0x2, 0x3, 0x5} + swapFailed := StaticAddressLoopIn{ + SwapHash: swapHashFailed, + SwapPreimage: lntypes.Preimage{0x3, 0x2, 0x3, 0x5}, + DepositOutpoints: []string{d4.OutPoint.String()}, + Deposits: []*deposit.Deposit{d4}, + ClientPubkey: clientPubKey, + ServerPubkey: serverPubKey, + HtlcTimeoutSweepAddress: addr, + } + swapFailed.SetState(Failed) + + err = swapStore.CreateLoopIn(ctxb, &swapFailed) + require.NoError(t, err) + pendingSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(ctxb, PendingStates) require.NoError(t, err) @@ -142,16 +216,33 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { finalizedSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(ctxb, FinalStates) require.NoError(t, err) - require.Len(t, finalizedSwaps, 1) - require.Equal(t, swapHashSucceeded, finalizedSwaps[0].SwapHash) - require.Equal(t, []string{d2.OutPoint.String()}, finalizedSwaps[0].DepositOutpoints) - require.Equal(t, Succeeded, finalizedSwaps[0].GetState()) + require.Len(t, finalizedSwaps, 3) + finalizedByState := make(map[string]*StaticAddressLoopIn) + for _, swap := range finalizedSwaps { + finalizedByState[string(swap.GetState())] = swap + } - finalizedDeposits := finalizedSwaps[0].Deposits + timeoutSweptSwap := finalizedByState[string(HtlcTimeoutSwept)] + require.NotNil(t, timeoutSweptSwap) + require.Equal(t, swapHashTimeoutSwept, timeoutSweptSwap.SwapHash) + require.Equal(t, HtlcTimeoutSwept, timeoutSweptSwap.GetState()) + + succeededSwap := finalizedByState[string(Succeeded)] + require.NotNil(t, succeededSwap) + require.Equal(t, swapHashSucceeded, succeededSwap.SwapHash) + require.Equal(t, []string{d3.OutPoint.String()}, succeededSwap.DepositOutpoints) + require.Equal(t, Succeeded, succeededSwap.GetState()) + + failedSwap := finalizedByState[string(Failed)] + require.NotNil(t, failedSwap) + require.Equal(t, swapHashFailed, failedSwap.SwapHash) + require.Equal(t, Failed, failedSwap.GetState()) + + finalizedDeposits := succeededSwap.Deposits require.Len(t, finalizedDeposits, 1) - require.Equal(t, d2.ID, finalizedDeposits[0].ID) - require.Equal(t, d2.OutPoint, finalizedDeposits[0].OutPoint) - require.Equal(t, d2.Value, finalizedDeposits[0].Value) + require.Equal(t, d3.ID, finalizedDeposits[0].ID) + require.Equal(t, d3.OutPoint, finalizedDeposits[0].OutPoint) + require.Equal(t, d3.Value, finalizedDeposits[0].Value) require.Equal(t, deposit.LoopedIn, finalizedDeposits[0].GetState()) }