diff --git a/notifications/manager.go b/notifications/manager.go index b87052b3..2ff6b087 100644 --- a/notifications/manager.go +++ b/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) } } diff --git a/notifications/manager_test.go b/notifications/manager_test.go index fdddef56..1768b004 100644 --- a/notifications/manager_test.go +++ b/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) } }