From d0c613e5c9c05a3a7039d024d1905bd12ff4d8cd Mon Sep 17 00:00:00 2001
From: Slyghtning <hieblmi@protonmail.com>
Date: Fri, 19 Jun 2026 14:48:02 +0200
Subject: [PATCH] notifications: drop best-effort messages for slow subscribers

A slow optional notification subscriber must not stall the manager lock, while recovery and sweep work requests still need reliable local delivery to keep daemon progress deterministic. Treating all subscribers as blocking makes best-effort fanout a backpressure source for unrelated required work.

Make reservation fanout nonblocking for slow subscribers, keep recovery and sweep notifications on a required cancellation-aware delivery path, and cover both slow best-effort subscribers and required subscribers in manager tests.
---
 notifications/manager.go      |  18 +++++-
 notifications/manager_test.go | 113 +++++++++++++++++++++++++++++++++-
 2 files changed, 127 insertions(+), 4 deletions(-)

diff --git a/notifications/manager.go b/notifications/manager.go
index 3eda5d65..b87052b3 100644
--- a/notifications/manager.go
+++ b/notifications/manager.go
@@ -332,7 +332,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 +351,10 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerStaticLoopInSweepNotification)
 
-			recvChan <- staticLoopInSweepRequestNtfn
+			select {
+			case recvChan <- staticLoopInSweepRequestNtfn:
+			case <-sub.subCtx.Done():
+			}
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
@@ -359,7 +368,10 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerUnfinishedSwapNotification)
 
-			recvChan <- unfinishedSwapNtfn
+			select {
+			case recvChan <- unfinishedSwapNtfn:
+			case <-sub.subCtx.Done():
+			}
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_HtlcConfirmed:
diff --git a/notifications/manager_test.go b/notifications/manager_test.go
index 44300d7a..fdddef56 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,117 @@ 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[:],
+			},
+		},
+	}
+}
+
+// TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full
+// notification channel does not block delivery to other subscribers.
+func TestManager_SlowSubscriberDoesNotBlock(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)
+}
+
+// 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()
+
+	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(unfinishedSwapNotification(swapHashA))
+
+	done := make(chan struct{})
+	go func() {
+		mgr.handleNotification(unfinishedSwapNotification(swapHashB))
+		close(done)
+	}()
+
+	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 <-done:
+
+	case <-time.After(time.Second):
+		t.Fatal("second unfinished swap notification did not unblock")
+	}
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHashB[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("second unfinished swap notification was dropped")
+	}
+}
+
 // TestManager_Backoff verifies that repeated failures in
 // subscribeNotifications cause the Manager to space out subscription attempts
 // via a predictable incremental backoff.