From 8ae163c15d4fa311ee8347146b5aa76527011d1b Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 15 May 2026 15:24:56 +0200 Subject: [PATCH] notifications: deduplicate risk fanout Risk accepted and rejected notifications had nearly identical parsing, persistence, cache update, and subscriber fanout code. The duplication made it easy for the two decision paths to drift. Introduce a shared risk-decision handler that validates the swap hash, persists the decision, updates the matching cache, clears the opposite cache, and fans out only to the subscriber for that swap. Keep risk-decision delivery best-effort for slow subscribers, while queued delivery remains limited to notification types that must not be dropped. Fold the queue tests through a common helper so both queued notification paths keep the same behavior. --- notifications/manager.go | 209 ++++++++++++++++++---------------- notifications/manager_test.go | 101 ++++++++-------- 2 files changed, 159 insertions(+), 151 deletions(-) diff --git a/notifications/manager.go b/notifications/manager.go index 3fe101f8..95007b4e 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -215,6 +215,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 { @@ -503,6 +516,66 @@ 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), notifySubscriber func(subscriber)) { + + decision := staticLoopInRiskDecisionName(accepted) + + 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) + } + } + + m.Lock() + defer m.Unlock() + + if hasSwapHash { + cacheDecision(swapHash) + } + + 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(ctx context.Context, ntfn *swapserverrpc. @@ -545,115 +618,55 @@ func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc. // We'll forward the static loop in risk accepted notification to the // subscriber for the matching swap. riskAcceptedNtfn := ntfn.GetStaticLoopInRiskAccepted() - 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 - } + swapHashBytes = riskAcceptedNtfn.SwapHash } - if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil { - err := m.cfg.PersistStaticLoopInRiskDecision( - ctx, swapHash, true, - ) - if err != nil { - log.Errorf("Unable to persist static loop in "+ - "risk accepted notification: %v", err) - } - } - - m.Lock() - defer m.Unlock() - - if hasSwapHash { - m.staticLoopInRiskAccepted[swapHash] = - riskAcceptedNtfn - delete(m.staticLoopInRiskRejected, swapHash) - } - - 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.handleStaticLoopInRiskDecision( + ctx, swapHashBytes, true, + NotificationTypeStaticLoopInRiskAccepted, + func(swapHash lntypes.Hash) { + m.staticLoopInRiskAccepted[swapHash] = + riskAcceptedNtfn + 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() - 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 - } + swapHashBytes = riskRejectedNtfn.SwapHash } - if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil { - err := m.cfg.PersistStaticLoopInRiskDecision( - ctx, swapHash, false, - ) - if err != nil { - log.Errorf("Unable to persist static loop in "+ - "risk rejected notification: %v", err) - } - } - - m.Lock() - defer m.Unlock() - - if hasSwapHash { - m.staticLoopInRiskRejected[swapHash] = - riskRejectedNtfn - delete(m.staticLoopInRiskAccepted, swapHash) - } - - 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.handleStaticLoopInRiskDecision( + ctx, swapHashBytes, false, + NotificationTypeStaticLoopInRiskRejected, + func(swapHash lntypes.Hash) { + m.staticLoopInRiskRejected[swapHash] = + riskRejectedNtfn + 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 ae5630a1..42168746 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -357,48 +357,21 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) { t.Parallel() - mgr := NewManager(&Config{}) + assertQueuedSwapHashNotifications( + t, + func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc. + ServerUnfinishedSwapNotification { - subCtx, subCancel := context.WithCancel(t.Context()) - defer subCancel() - - subChan := mgr.SubscribeUnfinishedSwaps(subCtx) - - swapHashA := lntypes.Hash{0x02, 0x03} - swapHashB := lntypes.Hash{0x04, 0x05} - - mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashA)) - - done := make(chan struct{}) - go func() { - mgr.handleNotification(t.Context(), unfinishedSwapNotification(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[:], received.SwapHash) - - case <-time.After(time.Second): - t.Fatal("did not receive first unfinished swap notification") - } - - select { - case received := <-subChan: - require.Equal(t, swapHashB[:], received.SwapHash) - - case <-time.After(time.Second): - t.Fatal("second unfinished swap notification was dropped") - } + 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 @@ -409,23 +382,45 @@ func TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber( 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", + ) +} + +// 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 := mgr.SubscribeStaticLoopInSweepRequests(subCtx) + subChan := subscribe(mgr, subCtx) - swapHashA := lntypes.Hash{0x12, 0x13} - swapHashB := lntypes.Hash{0x14, 0x15} - - mgr.handleNotification(t.Context(), staticLoopInSweepNotification(swapHashA)) + mgr.handleNotification(t.Context(), notification(swapHashA)) done := make(chan struct{}) go func() { - mgr.handleNotification( - t.Context(), staticLoopInSweepNotification(swapHashB), - ) + mgr.handleNotification(t.Context(), notification(swapHashB)) close(done) }() @@ -440,18 +435,18 @@ func TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber( select { case received := <-subChan: - require.Equal(t, swapHashA[:], received.SwapHash) + require.Equal(t, swapHashA[:], swapHash(received)) case <-time.After(time.Second): - t.Fatal("did not receive first sweep notification") + t.Fatal(firstFailureMsg) } select { case received := <-subChan: - require.Equal(t, swapHashB[:], received.SwapHash) + require.Equal(t, swapHashB[:], swapHash(received)) case <-time.After(time.Second): - t.Fatal("second sweep notification was not queued") + t.Fatal(secondFailureMsg) } }