notifications: queue blocking fanout

Required notification fanout should not block the manager lock, but subscribers still need ordered delivery once brief backpressure clears. Sending must-deliver notifications directly can couple manager progress to subscriber receive timing, while queueing optional reservation notifications would contradict their best-effort delivery semantics.

Add bounded per-subscriber queues for must-deliver notifications, let those queues own channel shutdown instead of relying on recover for closed-channel sends, keep reservation fanout best-effort, and cover queued delivery, queue cleanup, and capacity drops in manager tests.

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
@@ -351,10 +473,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerStaticLoopInSweepNotification)
 
-			select {
-			case recvChan <- staticLoopInSweepRequestNtfn:
-			case <-sub.subCtx.Done():
-			}
+			queueNotification(sub, recvChan, staticLoopInSweepRequestNtfn)
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
@@ -368,10 +487,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerUnfinishedSwapNotification)
 
-			select {
-			case recvChan <- unfinishedSwapNtfn:
-			case <-sub.subCtx.Done():
-			}
+			queueNotification(sub, recvChan, unfinishedSwapNtfn)
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_HtlcConfirmed:
@@ -415,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)
 		}
 	}

notifications/manager_test.go

@@ -205,9 +205,26 @@ func unfinishedSwapNotification(
 	}
 }
 
-// TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full
-// notification channel does not block delivery to other subscribers.
-func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) {
+// 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{})
@@ -251,6 +268,22 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) {
 	}
 
 	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
@@ -259,45 +292,171 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) {
 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())
-	defer subCancel()
-
 	subChan := mgr.SubscribeUnfinishedSwaps(subCtx)
 
-	swapHashA := lntypes.Hash{0x02, 0x03}
-	swapHashB := lntypes.Hash{0x04, 0x05}
-
+	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 := <-subChan:
+	case received, ok := <-subChan:
+		require.True(t, ok)
 		require.Equal(t, swapHashA[:], received.SwapHash)
 
 	case <-time.After(time.Second):
-		t.Fatal("did not receive first unfinished swap notification")
+		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 <-done:
+	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("second unfinished swap notification did not unblock")
+		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 unfinished swap notification was dropped")
+		t.Fatal(secondFailureMsg)
 	}
 }