staticaddr/loopin: wait for risk decisions

Subscribe to static loop-in confirmation-risk notifications, start the payment deadline only after server acceptance or legacy confirmation fallback, and cancel the swap invoice when the server rejects the risk wait.

Author: hieblmi

notifications/manager.go

@@ -26,6 +26,14 @@ const (
 	// static loop in sweep requests.
 	NotificationTypeStaticLoopInSweepRequest
 
+	// NotificationTypeStaticLoopInRiskAccepted is the notification type for
+	// 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
@@ -92,6 +100,12 @@ type Manager struct {
 	hasL402 bool
 
 	subscribers map[NotificationType][]subscriber
+
+	staticLoopInRiskAccepted map[lntypes.Hash]*swapserverrpc.
+					ServerStaticLoopInRiskAcceptedNotification
+
+	staticLoopInRiskRejected map[lntypes.Hash]*swapserverrpc.
+					ServerStaticLoopInRiskRejectedNotification
 }
 
 // NewManager creates a new notification manager.
@@ -107,13 +121,22 @@ func NewManager(cfg *Config) *Manager {
 	return &Manager{
 		cfg:         cfg,
 		subscribers: make(map[NotificationType][]subscriber),
+		staticLoopInRiskAccepted: make(
+			map[lntypes.Hash]*swapserverrpc.
+				ServerStaticLoopInRiskAcceptedNotification,
+		),
+		staticLoopInRiskRejected: make(
+			map[lntypes.Hash]*swapserverrpc.
+				ServerStaticLoopInRiskRejectedNotification,
+		),
 	}
 }
 
 type subscriber struct {
 	subCtx   context.Context
 	recvChan any
 	enqueue  func(any)
+	swapHash *lntypes.Hash
 }
 
 // newNotificationQueue creates a per-subscriber FIFO delivery function.
@@ -271,6 +294,80 @@ 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,
+		swapHash: &swapHash,
+	}
+
+	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
+}
+
+// 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 {
@@ -476,6 +573,94 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
 			queueNotification(sub, recvChan, staticLoopInSweepRequestNtfn)
 		}
 
+	case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInRiskAccepted: // nolint: lll
+		// 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 {
+			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
+				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)
+
+			select {
+			case recvChan <- riskAcceptedNtfn:
+			case <-sub.subCtx.Done():
+			default:
+				log.Debugf("Dropping static loop in risk " +
+					"accepted notification for slow subscriber")
+			}
+		}
+
+	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.