notifications: bound pending risk decision replay

docs/release-notes/release-notes-next.md

@@ -11,6 +11,11 @@
 
 #### Bug Fixes
 
+* Bound the static Loop In risk decision replay cache to prevent unbounded
+  memory growth from swap server notifications for unknown swaps. Cached
+  decisions retain only a copy of the validated swap hash, without unknown
+  protobuf fields or oversized backing buffers.
+
 #### Maintenance
 - Update the LoopRPC requirement to the revision used by this release.
 

notifications/manager.go

@@ -57,6 +57,11 @@ const (
 	// we queue per subscriber before dropping new notifications.
 	defaultMaxQueuedNotifications = 1024
 
+	// maxPendingRiskDecisions bounds the replay cache across accepted and
+	// rejected decisions. The server chooses the hashes, including hashes
+	// for swaps that do not exist locally and cannot be persisted.
+	maxPendingRiskDecisions = 1024
+
 	// current_version is the current version of the notification listener.
 	current_version = swapserverrpc.SubscribeNotificationsRequest_V1
 )
@@ -91,8 +96,8 @@ type Config struct {
 
 	// PersistStaticLoopInRiskDecision durably records static loop-in
 	// confirmation-risk decisions. If this fails, the notification is still
-	// cached and forwarded so a later subscriber can process it after the swap
-	// row exists.
+	// forwarded and cached subject to maxPendingRiskDecisions so a later
+	// subscriber can process it after the swap row exists.
 	PersistStaticLoopInRiskDecision func(context.Context, lntypes.Hash,
 		bool) error
 }
@@ -124,7 +129,8 @@ type Manager struct {
 
 	// staticLoopInRiskPersisted records whether the cached risk decision for
 	// a swap hash was durably persisted. Unpersisted decisions remain cached
-	// after subscriber cancellation so they can be replayed.
+	// after subscriber cancellation so they can be replayed, subject to the
+	// shared maxPendingRiskDecisions limit.
 	staticLoopInRiskPersisted map[lntypes.Hash]bool
 }
 
@@ -607,6 +613,7 @@ func (m *Manager) handleStaticLoopInRiskDecision(ctx context.Context,
 	defer m.Unlock()
 
 	if hasSwapHash {
+		m.evictPendingRiskDecision(swapHash)
 		cacheDecision(swapHash, persisted)
 	}
 
@@ -621,6 +628,30 @@ func (m *Manager) handleStaticLoopInRiskDecision(ctx context.Context,
 	}
 }
 
+// evictPendingRiskDecision makes room for a new risk decision in the bounded
+// replay cache. Updating an existing hash does not need additional space. At
+// capacity, an arbitrary cached decision is evicted, even if persistence
+// failed.
+//
+// NOTE: The caller must hold the Manager mutex.
+func (m *Manager) evictPendingRiskDecision(swapHash lntypes.Hash) {
+	if _, ok := m.staticLoopInRiskPersisted[swapHash]; ok {
+		return
+	}
+
+	if len(m.staticLoopInRiskPersisted) < maxPendingRiskDecisions {
+		return
+	}
+
+	for hash := range m.staticLoopInRiskPersisted {
+		delete(m.staticLoopInRiskAccepted, hash)
+		delete(m.staticLoopInRiskRejected, hash)
+		delete(m.staticLoopInRiskPersisted, hash)
+
+		return
+	}
+}
+
 // handleNotification handles an incoming notification from the server,
 // forwarding it to the appropriate subscribers.
 func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc.
@@ -672,6 +703,12 @@ func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc.
 			ctx, swapHashBytes, true,
 			NotificationTypeStaticLoopInRiskAccepted,
 			func(swapHash lntypes.Hash, persisted bool) {
+				// Retain only the validated hash, without unknown
+				// protobuf fields or the original backing buffer.
+				riskAcceptedNtfn = &swapserverrpc.
+					ServerStaticLoopInRiskAcceptedNotification{
+					SwapHash: swapHash[:],
+				}
 				m.staticLoopInRiskAccepted[swapHash] =
 					riskAcceptedNtfn
 				m.staticLoopInRiskPersisted[swapHash] = persisted
@@ -700,6 +737,12 @@ func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc.
 			ctx, swapHashBytes, false,
 			NotificationTypeStaticLoopInRiskRejected,
 			func(swapHash lntypes.Hash, persisted bool) {
+				// Retain only the validated hash, without unknown
+				// protobuf fields or the original backing buffer.
+				riskRejectedNtfn = &swapserverrpc.
+					ServerStaticLoopInRiskRejectedNotification{
+					SwapHash: swapHash[:],
+				}
 				m.staticLoopInRiskRejected[swapHash] =
 					riskRejectedNtfn
 				m.staticLoopInRiskPersisted[swapHash] = persisted

notifications/risk_decision_bound_test.go

@@ -0,0 +1,205 @@
+package notifications
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/lightninglabs/loop/swapserverrpc"
+	"github.com/lightningnetwork/lnd/lntypes"
+	"github.com/stretchr/testify/require"
+)
+
+// TestManager_BoundedPendingRiskDecisions exercises the receive loop with a
+// stream of distinct hashes. The replay cache must stay bounded regardless of
+// whether a decision can be persisted.
+func TestManager_BoundedPendingRiskDecisions(t *testing.T) {
+	t.Parallel()
+
+	for _, testCase := range []struct {
+		name       string
+		persist    bool
+		persistErr error
+	}{
+		{
+			name:       "unknown swaps",
+			persist:    true,
+			persistErr: errors.New("static loop in swap not found"),
+		},
+		{
+			name:    "persisted decisions",
+			persist: true,
+		},
+		{
+			name: "no persistence callback",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			t.Parallel()
+
+			const numNotifications = 20 * maxPendingRiskDecisions
+			recvChan := make(
+				chan *swapserverrpc.SubscribeNotificationsResponse,
+				numNotifications,
+			)
+			for i := range numNotifications {
+				hash := riskDecisionTestHash(i)
+				ntfn := staticLoopInRiskAcceptedNotification(hash)
+				if i%2 != 0 {
+					ntfn = staticLoopInRiskRejectedNotification(hash)
+				}
+				recvChan <- ntfn
+			}
+			close(recvChan)
+
+			persistCalls := 0
+			cfg := &Config{
+				Client: &mockNotificationsClient{
+					mockStream: &mockSubscribeNotificationsClient{
+						recvChan: recvChan,
+					},
+				},
+			}
+			if testCase.persist {
+				cfg.PersistStaticLoopInRiskDecision = func(
+					_ context.Context, _ lntypes.Hash,
+					_ bool) error {
+
+					persistCalls++
+					return testCase.persistErr
+				}
+			}
+			mgr := NewManager(cfg)
+
+			require.ErrorIs(t, mgr.subscribeNotifications(t.Context()),
+				io.EOF)
+			if testCase.persist {
+				require.Equal(t, numNotifications, persistCalls)
+			}
+
+			mgr.Lock()
+			require.Len(t, mgr.staticLoopInRiskPersisted,
+				maxPendingRiskDecisions)
+			require.Equal(t, maxPendingRiskDecisions,
+				len(mgr.staticLoopInRiskAccepted)+
+					len(mgr.staticLoopInRiskRejected))
+			for hash, persisted := range mgr.staticLoopInRiskPersisted {
+				_, accepted := mgr.staticLoopInRiskAccepted[hash]
+				_, rejected := mgr.staticLoopInRiskRejected[hash]
+				require.NotEqual(t, accepted, rejected)
+				require.Equal(t, testCase.persistErr == nil,
+					persisted)
+			}
+			mgr.Unlock()
+
+			// A decision arriving before its subscriber still replays
+			// even after the cache has reached capacity.
+			acceptedHash := riskDecisionTestHash(numNotifications)
+			mgr.handleNotification(t.Context(),
+				staticLoopInRiskAcceptedNotification(acceptedHash))
+			accepted := mgr.SubscribeStaticLoopInRiskAccepted(
+				t.Context(), acceptedHash,
+			)
+			select {
+			case ntfn := <-accepted:
+				require.Equal(t, acceptedHash[:], ntfn.SwapHash)
+			case <-time.After(time.Second):
+				t.Fatal("accepted decision was not replayed at capacity")
+			}
+
+			rejectedHash := riskDecisionTestHash(numNotifications + 1)
+			mgr.handleNotification(t.Context(),
+				staticLoopInRiskRejectedNotification(rejectedHash))
+			rejected := mgr.SubscribeStaticLoopInRiskRejected(
+				t.Context(), rejectedHash,
+			)
+			select {
+			case ntfn := <-rejected:
+				require.Equal(t, rejectedHash[:], ntfn.SwapHash)
+			case <-time.After(time.Second):
+				t.Fatal("rejected decision was not replayed at capacity")
+			}
+		})
+	}
+}
+
+// TestManager_RiskDecisionUpdatesAtCapacity verifies that replacing a cached
+// decision does not evict another hash, including when its direction changes.
+func TestManager_RiskDecisionUpdatesAtCapacity(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{})
+	for i := range maxPendingRiskDecisions {
+		mgr.handleNotification(t.Context(),
+			staticLoopInRiskAcceptedNotification(riskDecisionTestHash(i)))
+	}
+
+	hash := riskDecisionTestHash(0)
+	for _, notification := range []*swapserverrpc.SubscribeNotificationsResponse{
+		staticLoopInRiskAcceptedNotification(hash),
+		staticLoopInRiskRejectedNotification(hash),
+		staticLoopInRiskRejectedNotification(hash),
+		staticLoopInRiskAcceptedNotification(hash),
+	} {
+		mgr.handleNotification(t.Context(), notification)
+		require.Len(t, mgr.staticLoopInRiskPersisted,
+			maxPendingRiskDecisions)
+		for i := range maxPendingRiskDecisions {
+			require.Contains(t, mgr.staticLoopInRiskPersisted,
+				riskDecisionTestHash(i))
+		}
+		_, accepted := mgr.staticLoopInRiskAccepted[hash]
+		_, rejected := mgr.staticLoopInRiskRejected[hash]
+		require.Equal(t, notification.GetStaticLoopInRiskAccepted() != nil,
+			accepted)
+		require.NotEqual(t, accepted, rejected)
+	}
+}
+
+// TestManager_RiskDecisionDeliveryAtCapacity ensures that cache eviction does
+// not interfere with forwarding decisions to matching active subscribers.
+func TestManager_RiskDecisionDeliveryAtCapacity(t *testing.T) {
+	t.Parallel()
+
+	mgr := NewManager(&Config{
+		PersistStaticLoopInRiskDecision: func(_ context.Context,
+			_ lntypes.Hash, _ bool) error {
+
+			return errors.New("static loop in swap not found")
+		},
+	})
+	for i := range maxPendingRiskDecisions {
+		mgr.handleNotification(t.Context(),
+			staticLoopInRiskAcceptedNotification(riskDecisionTestHash(i)))
+	}
+
+	hash := riskDecisionTestHash(maxPendingRiskDecisions)
+	accepted := mgr.SubscribeStaticLoopInRiskAccepted(t.Context(), hash)
+	rejected := mgr.SubscribeStaticLoopInRiskRejected(t.Context(), hash)
+	mgr.handleNotification(t.Context(),
+		staticLoopInRiskAcceptedNotification(hash))
+	mgr.handleNotification(t.Context(),
+		staticLoopInRiskRejectedNotification(hash))
+
+	select {
+	case ntfn := <-accepted:
+		require.Equal(t, hash[:], ntfn.SwapHash)
+	case <-time.After(time.Second):
+		t.Fatal("accepted decision was not delivered at capacity")
+	}
+	select {
+	case ntfn := <-rejected:
+		require.Equal(t, hash[:], ntfn.SwapHash)
+	case <-time.After(time.Second):
+		t.Fatal("rejected decision was not delivered at capacity")
+	}
+}
+
+func riskDecisionTestHash(index int) lntypes.Hash {
+	var hash lntypes.Hash
+	binary.BigEndian.PutUint64(hash[24:], uint64(index)+1)
+	return hash
+}