staticaddr/loopin: wait for risk acceptance notification

Wait for the server's static loop-in risk-accepted notification before starting
the client payment deadline. The server may intentionally hold the swap at the
confirmation-risk gate after HTLC signing, and the client deadline should not
run while that server-side wait is still in progress.

Cache risk-accepted notifications by swap hash inside the local notification
manager and replay them to the per-swap subscriber. This covers both reconnects
and the internal race where the global notification stream receives the server
event before the static loop-in FSM registers its waiter.

Author: hieblmi

notifications/manager.go

@@ -26,6 +26,10 @@ const (
 	// static loop in sweep requests.
 	NotificationTypeStaticLoopInSweepRequest
 
+	// NotificationTypeStaticLoopInRiskAccepted is the notification type for
+	// static loop in confirmation risk acceptance.
+	NotificationTypeStaticLoopInRiskAccepted
+
 	// NotificationTypeUnfinishedSwap is the notification type for unfinished
 	// swap notifications.
 	NotificationTypeUnfinishedSwap
@@ -76,6 +80,9 @@ type Manager struct {
 	hasL402 bool
 
 	subscribers map[NotificationType][]subscriber
+
+	staticLoopInRiskAccepted map[lntypes.Hash]*swapserverrpc.
+					ServerStaticLoopInRiskAcceptedNotification
 }
 
 // NewManager creates a new notification manager.
@@ -88,6 +95,10 @@ func NewManager(cfg *Config) *Manager {
 	return &Manager{
 		cfg:         cfg,
 		subscribers: make(map[NotificationType][]subscriber),
+		staticLoopInRiskAccepted: make(
+			map[lntypes.Hash]*swapserverrpc.
+				ServerStaticLoopInRiskAcceptedNotification,
+		),
 	}
 }
 
@@ -143,6 +154,42 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context,
 	return notifChan
 }
 
+// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted
+// notifications.
+func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
+	swapHash lntypes.Hash,
+) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification {
+
+	notifChan := make(
+		chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification, 1,
+	)
+
+	sub := subscriber{
+		subCtx:   ctx,
+		recvChan: notifChan,
+	}
+
+	m.Lock()
+	m.subscribers[NotificationTypeStaticLoopInRiskAccepted] = append(
+		m.subscribers[NotificationTypeStaticLoopInRiskAccepted], sub,
+	)
+	if ntfn, ok := m.staticLoopInRiskAccepted[swapHash]; ok {
+		notifChan <- ntfn
+		delete(m.staticLoopInRiskAccepted, swapHash)
+	}
+	m.Unlock()
+
+	context.AfterFunc(ctx, func() {
+		m.removeSubscriber(NotificationTypeStaticLoopInRiskAccepted, sub)
+		m.Lock()
+		delete(m.staticLoopInRiskAccepted, swapHash)
+		m.Unlock()
+		close(notifChan)
+	})
+
+	return notifChan
+}
+
 // SubscribeUnfinishedSwaps subscribes to the unfinished swap notifications.
 func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context,
 ) <-chan *swapserverrpc.ServerUnfinishedSwapNotification {
@@ -328,6 +375,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.
+		riskAcceptedNtfn := ntfn.GetStaticLoopInRiskAccepted()
+		m.Lock()
+		defer m.Unlock()
+
+		if riskAcceptedNtfn != nil {
+			swapHash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash)
+			if err != nil {
+				log.Warnf("Received invalid static loop in risk "+
+					"accepted notification: %v", err)
+			} else {
+				m.staticLoopInRiskAccepted[swapHash] =
+					riskAcceptedNtfn
+			}
+		}
+
+		for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskAccepted] { // nolint: lll
+			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")
+			}
+		}
+
 	case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
 		// We'll forward the unfinished swap notification to all
 		// subscribers.

notifications/manager_test.go

@@ -362,6 +362,76 @@ func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) {
 	}
 }
 
+// TestManager_StaticLoopInRiskAcceptedNotification tests that the Manager
+// forwards static loop in risk accepted notifications to subscribers.
+func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	swapHash := lntypes.Hash{0x04, 0x05}
+
+	subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash)
+
+	mgr.handleNotification(
+		&swapserverrpc.SubscribeNotificationsResponse{
+			Notification: &swapserverrpc.
+				SubscribeNotificationsResponse_StaticLoopInRiskAccepted{
+				StaticLoopInRiskAccepted: &swapserverrpc.
+					ServerStaticLoopInRiskAcceptedNotification{
+					SwapHash: swapHash[:],
+				},
+			},
+		},
+	)
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHash[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("did not receive risk accepted notification")
+	}
+}
+
+// TestManager_StaticLoopInRiskAcceptedNotificationReplay tests that the Manager
+// replays a risk accepted notification that arrives before the swap-specific
+// subscriber is registered.
+func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+
+	swapHash := lntypes.Hash{0x06, 0x07}
+	mgr.handleNotification(
+		&swapserverrpc.SubscribeNotificationsResponse{
+			Notification: &swapserverrpc.
+				SubscribeNotificationsResponse_StaticLoopInRiskAccepted{
+				StaticLoopInRiskAccepted: &swapserverrpc.
+					ServerStaticLoopInRiskAcceptedNotification{
+					SwapHash: swapHash[:],
+				},
+			},
+		},
+	)
+
+	subCtx, subCancel := context.WithCancel(t.Context())
+	defer subCancel()
+
+	subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash)
+
+	select {
+	case received := <-subChan:
+		require.Equal(t, swapHash[:], received.SwapHash)
+
+	case <-time.After(time.Second):
+		t.Fatal("did not replay risk accepted notification")
+	}
+}
+
 // TestManager_Backoff verifies that repeated failures in
 // subscribeNotifications cause the Manager to space out subscription attempts
 // via a predictable incremental backoff.