diff --git a/notifications/manager.go b/notifications/manager.go index ed408195..eb47c1ed 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -88,6 +88,13 @@ type Config struct { // MaxQueuedNotifications is the maximum number of notifications that // can wait in each subscriber's delivery queue. MaxQueuedNotifications int + + // PersistStaticLoopInRiskDecision durably records static loop-in + // confirmation-risk decisions. If this fails, the notification is still + // cached and forwarded so a later subscriber can process it after the swap + // row exists. + PersistStaticLoopInRiskDecision func(context.Context, lntypes.Hash, + bool) error } // Manager is a manager for notifications that the swap server sends to the @@ -99,13 +106,26 @@ type Manager struct { hasL402 bool + // subscribers holds active notification subscribers by notification + // type. It is guarded by the Manager mutex. subscribers map[NotificationType][]subscriber + // staticLoopInRiskAccepted caches accepted risk decisions by swap hash + // so a later matching subscriber can receive a previously delivered + // server decision. staticLoopInRiskAccepted map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskAcceptedNotification + // staticLoopInRiskRejected caches rejected risk decisions by swap hash + // so a later matching subscriber can receive a previously delivered + // server decision. staticLoopInRiskRejected map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskRejectedNotification + + // staticLoopInRiskPersisted records whether the cached risk decision for + // a swap hash was durably persisted. Unpersisted decisions remain cached + // after subscriber cancellation so they can be replayed. + staticLoopInRiskPersisted map[lntypes.Hash]bool } // NewManager creates a new notification manager. @@ -129,14 +149,15 @@ func NewManager(cfg *Config) *Manager { map[lntypes.Hash]*swapserverrpc. ServerStaticLoopInRiskRejectedNotification, ), + staticLoopInRiskPersisted: make(map[lntypes.Hash]bool), } } type subscriber struct { subCtx context.Context recvChan any - enqueue func(any) swapHash *lntypes.Hash + enqueue func(any) } // newNotificationQueue creates a per-subscriber FIFO delivery function. @@ -245,6 +266,19 @@ func queueNotification[T any](sub subscriber, recvChan chan T, ntfn T) { } } +// dropNotification sends a best-effort notification to a subscriber. +func dropNotification[T any](sub subscriber, recvChan chan T, ntfn T, + description string) { + + select { + case recvChan <- ntfn: + case <-sub.subCtx.Done(): + default: + log.Debugf("Dropping %s notification for slow subscriber", + description) + } +} + // SubscribeReservations subscribes to the reservation notifications. func (m *Manager) SubscribeReservations(ctx context.Context, ) <-chan *swapserverrpc.ServerReservationNotification { @@ -294,16 +328,11 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context, return notifChan } -// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted -// notifications. -func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, - swapHash lntypes.Hash, -) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification { - - notifChan := make( - chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification, 1, - ) +func subscribeStaticLoopInRiskDecision[T any](m *Manager, ctx context.Context, + swapHash lntypes.Hash, notifType NotificationType, + notifications map[lntypes.Hash]T) <-chan T { + notifChan := make(chan T, 1) sub := subscriber{ subCtx: ctx, recvChan: notifChan, @@ -311,19 +340,25 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, } m.Lock() - m.subscribers[NotificationTypeStaticLoopInRiskAccepted] = append( - m.subscribers[NotificationTypeStaticLoopInRiskAccepted], sub, - ) - if ntfn, ok := m.staticLoopInRiskAccepted[swapHash]; ok { + m.subscribers[notifType] = append(m.subscribers[notifType], sub) + if ntfn, ok := notifications[swapHash]; ok { notifChan <- ntfn - delete(m.staticLoopInRiskAccepted, swapHash) + if m.staticLoopInRiskPersisted[swapHash] { + delete(notifications, swapHash) + delete(m.staticLoopInRiskPersisted, swapHash) + } } m.Unlock() context.AfterFunc(ctx, func() { - m.removeSubscriber(NotificationTypeStaticLoopInRiskAccepted, sub) + m.removeSubscriber(notifType, sub) m.Lock() - delete(m.staticLoopInRiskAccepted, swapHash) + if _, ok := notifications[swapHash]; ok && + m.staticLoopInRiskPersisted[swapHash] { + + delete(notifications, swapHash) + delete(m.staticLoopInRiskPersisted, swapHash) + } m.Unlock() close(notifChan) }) @@ -331,41 +366,28 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, return notifChan } +// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted +// notifications. +func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context, + swapHash lntypes.Hash, +) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification { + + return subscribeStaticLoopInRiskDecision( + m, ctx, swapHash, NotificationTypeStaticLoopInRiskAccepted, + m.staticLoopInRiskAccepted, + ) +} + // SubscribeStaticLoopInRiskRejected subscribes to static loop in risk rejected // notifications. func (m *Manager) SubscribeStaticLoopInRiskRejected(ctx context.Context, swapHash lntypes.Hash, ) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification { - notifChan := make( - chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification, 1, + return subscribeStaticLoopInRiskDecision( + m, ctx, swapHash, NotificationTypeStaticLoopInRiskRejected, + m.staticLoopInRiskRejected, ) - - sub := subscriber{ - subCtx: ctx, - recvChan: notifChan, - swapHash: &swapHash, - } - - m.Lock() - m.subscribers[NotificationTypeStaticLoopInRiskRejected] = append( - m.subscribers[NotificationTypeStaticLoopInRiskRejected], sub, - ) - if ntfn, ok := m.staticLoopInRiskRejected[swapHash]; ok { - notifChan <- ntfn - delete(m.staticLoopInRiskRejected, swapHash) - } - m.Unlock() - - context.AfterFunc(ctx, func() { - m.removeSubscriber(NotificationTypeStaticLoopInRiskRejected, sub) - m.Lock() - delete(m.staticLoopInRiskRejected, swapHash) - m.Unlock() - close(notifChan) - }) - - return notifChan } // SubscribeUnfinishedSwaps subscribes to the unfinished swap notifications. @@ -525,7 +547,7 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error { notification, err := notifStream.Recv() if err == nil && notification != nil { log.Tracef("Received notification: %v", notification) - m.handleNotification(notification) + m.handleNotification(ctx, notification) continue } @@ -535,9 +557,73 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error { } } +// staticLoopInRiskDecisionName returns the log label for a risk decision. +func staticLoopInRiskDecisionName(accepted bool) string { + if accepted { + return "accepted" + } + + return "rejected" +} + +// handleStaticLoopInRiskDecision persists, caches, and forwards a risk +// decision notification to the matching subscriber. +func (m *Manager) handleStaticLoopInRiskDecision(ctx context.Context, + swapHashBytes []byte, accepted bool, notifType NotificationType, + cacheDecision func(lntypes.Hash, bool), + notifySubscriber func(subscriber)) { + + decision := staticLoopInRiskDecisionName(accepted) + persisted := m.cfg.PersistStaticLoopInRiskDecision == nil + + var ( + swapHash lntypes.Hash + hasSwapHash bool + ) + if swapHashBytes != nil { + hash, err := lntypes.MakeHash(swapHashBytes) + if err != nil { + log.Warnf("Received invalid static loop in risk "+ + "%s notification: %v", decision, err) + } else { + swapHash = hash + hasSwapHash = true + } + } + + if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil { + err := m.cfg.PersistStaticLoopInRiskDecision( + ctx, swapHash, accepted, + ) + if err != nil { + log.Errorf("Unable to persist static loop in risk "+ + "%s notification: %v", decision, err) + } else { + persisted = true + } + } + + m.Lock() + defer m.Unlock() + + if hasSwapHash { + cacheDecision(swapHash, persisted) + } + + for _, sub := range m.subscribers[notifType] { + if !hasSwapHash || sub.swapHash == nil || + *sub.swapHash != swapHash { + + continue + } + + notifySubscriber(sub) + } +} + // handleNotification handles an incoming notification from the server, // forwarding it to the appropriate subscribers. -func (m *Manager) handleNotification(ntfn *swapserverrpc. +func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc. SubscribeNotificationsResponse) { switch ntfn.Notification.(type) { @@ -577,89 +663,57 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. // We'll forward the static loop in risk accepted notification to the // subscriber for the matching swap. riskAcceptedNtfn := ntfn.GetStaticLoopInRiskAccepted() - m.Lock() - defer m.Unlock() - - var ( - swapHash lntypes.Hash - hasSwapHash bool - ) + var swapHashBytes []byte if riskAcceptedNtfn != nil { - hash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash) - if err != nil { - log.Warnf("Received invalid static loop in risk "+ - "accepted notification: %v", err) - } else { - swapHash = hash - hasSwapHash = true - m.staticLoopInRiskAccepted[hash] = + swapHashBytes = riskAcceptedNtfn.SwapHash + } + + m.handleStaticLoopInRiskDecision( + ctx, swapHashBytes, true, + NotificationTypeStaticLoopInRiskAccepted, + func(swapHash lntypes.Hash, persisted bool) { + m.staticLoopInRiskAccepted[swapHash] = riskAcceptedNtfn - delete(m.staticLoopInRiskRejected, hash) - } - } - - for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskAccepted] { // nolint: lll - if !hasSwapHash || sub.swapHash == nil || - *sub.swapHash != swapHash { - - continue - } - - recvChan := sub.recvChan.(chan *swapserverrpc. - ServerStaticLoopInRiskAcceptedNotification) - - select { - case recvChan <- riskAcceptedNtfn: - case <-sub.subCtx.Done(): - default: - log.Debugf("Dropping static loop in risk " + - "accepted notification for slow subscriber") - } - } + m.staticLoopInRiskPersisted[swapHash] = persisted + delete(m.staticLoopInRiskRejected, swapHash) + }, + func(sub subscriber) { + recvChan := sub.recvChan.(chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification) + dropNotification( + sub, recvChan, riskAcceptedNtfn, + "static loop in risk accepted", + ) + }, + ) case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInRiskRejected: // nolint: lll // We'll forward the static loop in risk rejected notification to the // subscriber for the matching swap. riskRejectedNtfn := ntfn.GetStaticLoopInRiskRejected() - m.Lock() - defer m.Unlock() - - var ( - swapHash lntypes.Hash - hasSwapHash bool - ) + var swapHashBytes []byte if riskRejectedNtfn != nil { - hash, err := lntypes.MakeHash(riskRejectedNtfn.SwapHash) - if err != nil { - log.Warnf("Received invalid static loop in risk "+ - "rejected notification: %v", err) - } else { - swapHash = hash - hasSwapHash = true - m.staticLoopInRiskRejected[hash] = + swapHashBytes = riskRejectedNtfn.SwapHash + } + + m.handleStaticLoopInRiskDecision( + ctx, swapHashBytes, false, + NotificationTypeStaticLoopInRiskRejected, + func(swapHash lntypes.Hash, persisted bool) { + m.staticLoopInRiskRejected[swapHash] = riskRejectedNtfn - delete(m.staticLoopInRiskAccepted, hash) - } - } - - for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskRejected] { // nolint: lll - if !hasSwapHash || sub.swapHash == nil || - *sub.swapHash != swapHash { - - continue - } - - recvChan := sub.recvChan.(chan *swapserverrpc. - ServerStaticLoopInRiskRejectedNotification) - - select { - case recvChan <- riskRejectedNtfn: - case <-sub.subCtx.Done(): - default: - log.Debugf("Dropping static loop in risk " + - "rejected notification for slow subscriber") - } - } + m.staticLoopInRiskPersisted[swapHash] = persisted + delete(m.staticLoopInRiskAccepted, swapHash) + }, + func(sub subscriber) { + recvChan := sub.recvChan.(chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification) + dropNotification( + sub, recvChan, riskRejectedNtfn, + "static loop in risk rejected", + ) + }, + ) case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll // We'll forward the unfinished swap notification to all diff --git a/notifications/manager_test.go b/notifications/manager_test.go index b165c494..59ecab78 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -220,6 +220,7 @@ func staticLoopInSweepNotification( } } +// staticLoopInRiskAcceptedNotification builds a risk accepted notification. func staticLoopInRiskAcceptedNotification( swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { @@ -271,7 +272,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[ subChanA := subscribe(mgr, subCtx, swapHashA) subChanB := subscribe(mgr, subCtx, swapHashB) - mgr.handleNotification(notification(swapHashA)) + mgr.handleNotification(t.Context(), notification(swapHashA)) select { case received := <-subChanA: @@ -290,7 +291,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[ default: } - mgr.handleNotification(notification(swapHashB)) + mgr.handleNotification(t.Context(), notification(swapHashB)) select { case received := <-subChanB: @@ -320,7 +321,7 @@ func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) { fastChan := mgr.SubscribeReservations(fastCtx) firstNotif := getTestNotification(testReservationId) - mgr.handleNotification(firstNotif) + mgr.handleNotification(t.Context(), firstNotif) received := <-fastChan require.Equal(t, testReservationId, received.ReservationId) @@ -328,7 +329,7 @@ func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) { secondNotif := getTestNotification(testReservationId2) done := make(chan struct{}) go func() { - mgr.handleNotification(secondNotif) + mgr.handleNotification(t.Context(), secondNotif) close(done) }() @@ -427,7 +428,7 @@ func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) { subChan := mgr.SubscribeUnfinishedSwaps(subCtx) swapHashA := lntypes.Hash{0x21, 0x22} - mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashA)) require.Eventually(t, func() bool { return len(subChan) == 1 @@ -436,7 +437,7 @@ func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) { swapHashB := lntypes.Hash{0x23, 0x24} done := make(chan struct{}) go func() { - mgr.handleNotification(unfinishedSwapNotification(swapHashB)) + mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashB)) close(done) }() @@ -508,11 +509,11 @@ func assertQueuedSwapHashNotifications[T any](t *testing.T, subChan := subscribe(mgr, subCtx) - mgr.handleNotification(notification(swapHashA)) + mgr.handleNotification(t.Context(), notification(swapHashA)) done := make(chan struct{}) go func() { - mgr.handleNotification(notification(swapHashB)) + mgr.handleNotification(t.Context(), notification(swapHashB)) close(done) }() @@ -557,6 +558,7 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) { subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash) mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ @@ -577,6 +579,169 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) { } } +// TestManager_StaticLoopInRiskDecisionPersists verifies that risk decisions are +// handed to the durable callback before they are treated as delivered. +func TestManager_StaticLoopInRiskDecisionPersists(t *testing.T) { + t.Parallel() + + type persistedDecision struct { + swapHash lntypes.Hash + accepted bool + } + + persisted := make(chan persistedDecision, 2) + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func(_ context.Context, + swapHash lntypes.Hash, accepted bool) error { + + persisted <- persistedDecision{ + swapHash: swapHash, + accepted: accepted, + } + + return nil + }, + }) + + acceptedHash := lntypes.Hash{0x16, 0x17} + rejectedHash := lntypes.Hash{0x18, 0x19} + + mgr.handleNotification( + t.Context(), staticLoopInRiskAcceptedNotification(acceptedHash), + ) + mgr.handleNotification( + t.Context(), staticLoopInRiskRejectedNotification(rejectedHash), + ) + + select { + case decision := <-persisted: + require.Equal(t, acceptedHash, decision.swapHash) + require.True(t, decision.accepted) + + case <-time.After(time.Second): + t.Fatal("accepted risk decision was not persisted") + } + + select { + case decision := <-persisted: + require.Equal(t, rejectedHash, decision.swapHash) + require.False(t, decision.accepted) + + case <-time.After(time.Second): + t.Fatal("rejected risk decision was not persisted") + } +} + +// TestManager_StaticLoopInRiskDecisionReplayOnPersistFailure verifies that an +// early risk notification is still cached if the swap row does not exist yet. +func TestManager_StaticLoopInRiskDecisionReplayOnPersistFailure(t *testing.T) { + t.Parallel() + + swapHash := lntypes.Hash{0x1a, 0x1b} + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func(_ context.Context, + _ lntypes.Hash, _ bool) error { + + return errors.New("swap not stored yet") + }, + }) + + mgr.handleNotification( + t.Context(), staticLoopInRiskAcceptedNotification(swapHash), + ) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash) + + select { + case received := <-subChan: + require.Equal(t, swapHash[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("did not replay risk notification after persist failure") + } +} + +// TestManager_StaticLoopInRiskDecisionReplaysAfterSubscriberCancel verifies that +// a non-persisted risk decision remains replayable if the subscriber is canceled +// before the FSM has a chance to process it. +func TestManager_StaticLoopInRiskDecisionReplaysAfterSubscriberCancel( + t *testing.T) { + + t.Parallel() + + assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel( + t, + (*Manager).SubscribeStaticLoopInRiskAccepted, + staticLoopInRiskAcceptedNotification, + ) + assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel( + t, + (*Manager).SubscribeStaticLoopInRiskRejected, + staticLoopInRiskRejectedNotification, + ) +} + +func assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel[ + T staticLoopInRiskNotification](t *testing.T, + subscribe func(*Manager, context.Context, lntypes.Hash) <-chan T, + notification func(lntypes.Hash) *swapserverrpc. + SubscribeNotificationsResponse) { + + t.Helper() + + swapHash := lntypes.Hash{0x2a, 0x2b} + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func(_ context.Context, + _ lntypes.Hash, _ bool) error { + + return errors.New("swap not stored yet") + }, + }) + + subCtx, subCancel := context.WithCancel(t.Context()) + subChan := subscribe(mgr, subCtx, swapHash) + + mgr.handleNotification(t.Context(), notification(swapHash)) + + require.Eventually(t, func() bool { + return len(subChan) == 1 + }, time.Second, 10*time.Millisecond) + + subCancel() + + select { + case <-subChan: + + case <-time.After(time.Second): + t.Fatal("risk decision notification was not delivered before " + + "cancel") + } + + select { + case _, ok := <-subChan: + require.False(t, ok) + + case <-time.After(time.Second): + t.Fatal("risk decision subscription did not close after cancel") + } + + replayCtx, replayCancel := context.WithCancel(t.Context()) + defer replayCancel() + + replayChan := subscribe(mgr, replayCtx, swapHash) + select { + case received := <-replayChan: + require.Equal(t, swapHash[:], received.GetSwapHash()) + + case <-time.After(time.Second): + t.Fatal("cached risk decision was lost after subscriber " + + "cancellation") + } +} + // TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped verifies that a // notification for one swap does not occupy another swap's subscriber channel. func TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped(t *testing.T) { @@ -603,6 +768,7 @@ func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) { swapHash := lntypes.Hash{0x06, 0x07} mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskAccepted{ @@ -643,6 +809,7 @@ func TestManager_StaticLoopInRiskRejectedNotification(t *testing.T) { subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash) mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskRejected{ @@ -689,6 +856,7 @@ func TestManager_StaticLoopInRiskRejectedNotificationReplay(t *testing.T) { swapHash := lntypes.Hash{0x0a, 0x0b} mgr.handleNotification( + t.Context(), &swapserverrpc.SubscribeNotificationsResponse{ Notification: &swapserverrpc. SubscribeNotificationsResponse_StaticLoopInRiskRejected{