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) } }