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.

Author: hieblmi

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
-			}
-		}
-
-		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)
-			}
+			swapHashBytes = riskAcceptedNtfn.SwapHash
 		}
 
-		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

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{})
-
-	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)
+	assertQueuedSwapHashNotifications(
+		t,
+		func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc.
+			ServerUnfinishedSwapNotification {
 
-	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)
 	}
 }