staticaddr/loopin: handle risk rejection notification

Add client handling for the server's static loop-in risk-rejected notification.
If the server aborts confirmation-risk waiting before payment, the client fails
the local swap instead of waiting for a payment deadline that will never start.

Cache rejected notifications by swap hash using the same replay path as accepted
notifications, and clear the opposite cached state when a final risk decision is
received. This keeps reconnect and subscription-order races from stranding the
client in the risk wait.

Author: hieblmi

notifications/manager.go

@@ -30,6 +30,10 @@ const (
 	// static loop in confirmation risk acceptance.
 	NotificationTypeStaticLoopInRiskAccepted
 
+	// NotificationTypeStaticLoopInRiskRejected is the notification type for
+	// static loop in confirmation risk rejection.
+	NotificationTypeStaticLoopInRiskRejected
+
 	// NotificationTypeUnfinishedSwap is the notification type for unfinished
 	// swap notifications.
 	NotificationTypeUnfinishedSwap
@@ -83,6 +87,9 @@ type Manager struct {
 
 	staticLoopInRiskAccepted map[lntypes.Hash]*swapserverrpc.
 					ServerStaticLoopInRiskAcceptedNotification
+
+	staticLoopInRiskRejected map[lntypes.Hash]*swapserverrpc.
+					ServerStaticLoopInRiskRejectedNotification
 }
 
 // NewManager creates a new notification manager.
@@ -99,12 +106,17 @@ func NewManager(cfg *Config) *Manager {
 			map[lntypes.Hash]*swapserverrpc.
 				ServerStaticLoopInRiskAcceptedNotification,
 		),
+		staticLoopInRiskRejected: make(
+			map[lntypes.Hash]*swapserverrpc.
+				ServerStaticLoopInRiskRejectedNotification,
+		),
 	}
 }
 
 type subscriber struct {
 	subCtx   context.Context
 	recvChan any
+	swapHash *lntypes.Hash
 }
 
 // SubscribeReservations subscribes to the reservation notifications.
@@ -167,6 +179,7 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
 	sub := subscriber{
 		subCtx:   ctx,
 		recvChan: notifChan,
+		swapHash: &swapHash,
 	}
 
 	m.Lock()
@@ -190,6 +203,43 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
 	return notifChan
 }
 
+// SubscribeStaticLoopInRiskRejected subscribes to static loop in risk rejected
+// notifications.
+func (m *Manager) SubscribeStaticLoopInRiskRejected(ctx context.Context,
+	swapHash lntypes.Hash,
+) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification {
+
+	notifChan := make(
+		chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification, 1,
+	)
+
+	sub := subscriber{
+		subCtx:   ctx,
+		recvChan: notifChan,
+		swapHash: &swapHash,
+	}
+
+	m.Lock()
+	m.subscribers[NotificationTypeStaticLoopInRiskRejected] = append(
+		m.subscribers[NotificationTypeStaticLoopInRiskRejected], sub,
+	)
+	if ntfn, ok := m.staticLoopInRiskRejected[swapHash]; ok {
+		notifChan <- ntfn
+		delete(m.staticLoopInRiskRejected, swapHash)
+	}
+	m.Unlock()
+
+	context.AfterFunc(ctx, func() {
+		m.removeSubscriber(NotificationTypeStaticLoopInRiskRejected, sub)
+		m.Lock()
+		delete(m.staticLoopInRiskRejected, swapHash)
+		m.Unlock()
+		close(notifChan)
+	})
+
+	return notifChan
+}
+
 // SubscribeUnfinishedSwaps subscribes to the unfinished swap notifications.
 func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context,
 ) <-chan *swapserverrpc.ServerUnfinishedSwapNotification {
@@ -376,24 +426,37 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 		}
 
 	case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInRiskAccepted: // nolint: lll
-		// We'll forward the static loop in risk accepted notification to all
-		// subscribers.
+		// We'll forward the static loop in risk accepted notification to the
+		// subscriber for the matching swap.
 		riskAcceptedNtfn := ntfn.GetStaticLoopInRiskAccepted()
 		m.Lock()
 		defer m.Unlock()
 
+		var (
+			swapHash    lntypes.Hash
+			hasSwapHash bool
+		)
 		if riskAcceptedNtfn != nil {
-			swapHash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash)
+			hash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash)
 			if err != nil {
 				log.Warnf("Received invalid static loop in risk "+
 					"accepted notification: %v", err)
 			} else {
-				m.staticLoopInRiskAccepted[swapHash] =
+				swapHash = hash
+				hasSwapHash = true
+				m.staticLoopInRiskAccepted[hash] =
 					riskAcceptedNtfn
+				delete(m.staticLoopInRiskRejected, hash)
 			}
 		}
 
 		for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskAccepted] { // nolint: lll
+			if !hasSwapHash || sub.swapHash == nil ||
+				*sub.swapHash != swapHash {
+
+				continue
+			}
+
 			recvChan := sub.recvChan.(chan *swapserverrpc.
 				ServerStaticLoopInRiskAcceptedNotification)
 
@@ -406,6 +469,50 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			}
 		}
 
+	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()
+		m.Lock()
+		defer m.Unlock()
+
+		var (
+			swapHash    lntypes.Hash
+			hasSwapHash bool
+		)
+		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
+				m.staticLoopInRiskRejected[hash] =
+					riskRejectedNtfn
+				delete(m.staticLoopInRiskAccepted, hash)
+			}
+		}
+
+		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")
+			}
+		}
+
 	case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
 		// We'll forward the unfinished swap notification to all
 		// subscribers.

notifications/manager_test.go

@@ -202,6 +202,86 @@ func staticLoopInSweepNotification(
 	}
 }
 
+func staticLoopInRiskAcceptedNotification(
+	swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse {
+
+	return &swapserverrpc.SubscribeNotificationsResponse{
+		Notification: &swapserverrpc.
+			SubscribeNotificationsResponse_StaticLoopInRiskAccepted{
+			StaticLoopInRiskAccepted: &swapserverrpc.
+				ServerStaticLoopInRiskAcceptedNotification{
+				SwapHash: swapHash[:],
+			},
+		},
+	}
+}
+
+func staticLoopInRiskRejectedNotification(
+	swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse {
+
+	return &swapserverrpc.SubscribeNotificationsResponse{
+		Notification: &swapserverrpc.
+			SubscribeNotificationsResponse_StaticLoopInRiskRejected{
+			StaticLoopInRiskRejected: &swapserverrpc.
+				ServerStaticLoopInRiskRejectedNotification{
+				SwapHash: swapHash[:],
+			},
+		},
+	}
+}
+
+type staticLoopInRiskNotification interface {
+	GetSwapHash() []byte
+}
+
+func assertStaticLoopInRiskNotificationSwapScoped[
+	T staticLoopInRiskNotification](t *testing.T,
+	subscribe func(*Manager, context.Context, lntypes.Hash) <-chan T,
+	notification func(lntypes.Hash) *swapserverrpc.
+		SubscribeNotificationsResponse, label string,
+	swapHashA, swapHashB lntypes.Hash) {
+
+	t.Helper()
+
+	mgr := NewManager(&Config{})
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	subChanA := subscribe(mgr, subCtx, swapHashA)
+	subChanB := subscribe(mgr, subCtx, swapHashB)
+
+	mgr.handleNotification(notification(swapHashA))
+
+	select {
+	case received := <-subChanA:
+		require.Equal(t, swapHashA[:], received.GetSwapHash())
+
+	case <-time.After(time.Second):
+		t.Fatalf("did not receive first swap risk %s notification",
+			label)
+	}
+
+	select {
+	case received := <-subChanB:
+		t.Fatalf("second swap received wrong notification: %x",
+			received.GetSwapHash())
+
+	default:
+	}
+
+	mgr.handleNotification(notification(swapHashB))
+
+	select {
+	case received := <-subChanB:
+		require.Equal(t, swapHashB[:], received.GetSwapHash())
+
+	case <-time.After(time.Second):
+		t.Fatalf("did not receive second swap risk %s notification",
+			label)
+	}
+}
+
 // TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full
 // notification channel does not block delivery to other subscribers.
 func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) {
@@ -335,6 +415,22 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) {
 	}
 }
 
+// TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped verifies that a
+// notification for one swap does not occupy another swap's subscriber channel.
+func TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped(t *testing.T) {
+	t.Parallel()
+
+	assertStaticLoopInRiskNotificationSwapScoped(
+		t, func(m *Manager, ctx context.Context,
+			swapHash lntypes.Hash) <-chan *swapserverrpc.
+			ServerStaticLoopInRiskAcceptedNotification {
+
+			return m.SubscribeStaticLoopInRiskAccepted(ctx, swapHash)
+		}, staticLoopInRiskAcceptedNotification, "accepted",
+		lntypes.Hash{0x04, 0x05}, lntypes.Hash{0x06, 0x07},
+	)
+}
+
 // TestManager_StaticLoopInRiskAcceptedNotificationReplay tests that the Manager
 // replays a risk accepted notification that arrives before the swap-specific
 // subscriber is registered.
@@ -370,6 +466,92 @@ func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) {
 	}
 }
 
+// TestManager_StaticLoopInRiskRejectedNotification tests that the Manager
+// forwards static loop in risk rejected notifications to subscribers.
+func TestManager_StaticLoopInRiskRejectedNotification(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	swapHash := lntypes.Hash{0x08, 0x09}
+
+	subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash)
+
+	mgr.handleNotification(
+		&swapserverrpc.SubscribeNotificationsResponse{
+			Notification: &swapserverrpc.
+				SubscribeNotificationsResponse_StaticLoopInRiskRejected{
+				StaticLoopInRiskRejected: &swapserverrpc.
+					ServerStaticLoopInRiskRejectedNotification{
+					SwapHash: swapHash[:],
+				},
+			},
+		},
+	)
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHash[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("did not receive risk rejected notification")
+	}
+}
+
+// TestManager_StaticLoopInRiskRejectedNotificationSwapScoped verifies that a
+// notification for one swap does not occupy another swap's subscriber channel.
+func TestManager_StaticLoopInRiskRejectedNotificationSwapScoped(t *testing.T) {
+	t.Parallel()
+
+	assertStaticLoopInRiskNotificationSwapScoped(
+		t, func(m *Manager, ctx context.Context,
+			swapHash lntypes.Hash) <-chan *swapserverrpc.
+			ServerStaticLoopInRiskRejectedNotification {
+
+			return m.SubscribeStaticLoopInRiskRejected(ctx, swapHash)
+		}, staticLoopInRiskRejectedNotification, "rejected",
+		lntypes.Hash{0x08, 0x09}, lntypes.Hash{0x0a, 0x0b},
+	)
+}
+
+// TestManager_StaticLoopInRiskRejectedNotificationReplay tests that the Manager
+// replays a risk rejected notification that arrives before the swap-specific
+// subscriber is registered.
+func TestManager_StaticLoopInRiskRejectedNotificationReplay(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+
+	swapHash := lntypes.Hash{0x0a, 0x0b}
+	mgr.handleNotification(
+		&swapserverrpc.SubscribeNotificationsResponse{
+			Notification: &swapserverrpc.
+				SubscribeNotificationsResponse_StaticLoopInRiskRejected{
+				StaticLoopInRiskRejected: &swapserverrpc.
+					ServerStaticLoopInRiskRejectedNotification{
+					SwapHash: swapHash[:],
+				},
+			},
+		},
+	)
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash)
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHash[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("did not replay risk rejected notification")
+	}
+}
+
 // TestManager_Backoff verifies that repeated failures in
 // subscribeNotifications cause the Manager to space out subscription attempts
 // via a predictable incremental backoff.