diff --git a/docs/release-notes/release-notes-next.md b/docs/release-notes/release-notes-next.md index ff66c270f..7ebced487 100644 --- a/docs/release-notes/release-notes-next.md +++ b/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. diff --git a/notifications/manager.go b/notifications/manager.go index eb47c1ed5..f6ae03d91 100644 --- a/notifications/manager.go +++ b/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 diff --git a/notifications/risk_decision_bound_test.go b/notifications/risk_decision_bound_test.go new file mode 100644 index 000000000..e8c4adfe0 --- /dev/null +++ b/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 +} diff --git a/notifications/risk_decision_payload_test.go b/notifications/risk_decision_payload_test.go new file mode 100644 index 000000000..d5d6a2ff9 --- /dev/null +++ b/notifications/risk_decision_payload_test.go @@ -0,0 +1,159 @@ +package notifications + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/lightninglabs/loop/swapserverrpc" + "github.com/lightningnetwork/lnd/lntypes" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" +) + +type riskDecisionPayloadNotification interface { + proto.Message + GetSwapHash() []byte +} + +// TestManager_RiskDecisionPayloadBounded verifies that the replay cache and +// subscriber channels retain only an independently owned hash, regardless of +// unknown protobuf fields or excess capacity in the incoming hash slice. +func TestManager_RiskDecisionPayloadBounded(t *testing.T) { + t.Parallel() + + for _, accepted := range []bool{true, false} { + t.Run(staticLoopInRiskDecisionName(accepted), func(t *testing.T) { + t.Parallel() + + for _, testCase := range []struct { + name string + persistErr error + }{ + {name: "persisted"}, + { + name: "unknown swap", + persistErr: errors.New("swap not found"), + }, + } { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + hash := lntypes.Hash{0x12, 0x34} + persistCalls := 0 + mgr := NewManager(&Config{ + PersistStaticLoopInRiskDecision: func( + _ context.Context, gotHash lntypes.Hash, + gotAccepted bool) error { + + persistCalls++ + require.Equal(t, hash, gotHash) + require.Equal(t, accepted, gotAccepted) + return testCase.persistErr + }, + }) + + // A valid hash can alias a much larger allocation. + backing := make([]byte, 1<<20) + copy(backing, hash[:]) + var ( + response *swapserverrpc.SubscribeNotificationsResponse + source riskDecisionPayloadNotification + liveAccepted <-chan *swapserverrpc. + ServerStaticLoopInRiskAcceptedNotification + liveRejected <-chan *swapserverrpc. + ServerStaticLoopInRiskRejectedNotification + ) + if accepted { + response = staticLoopInRiskAcceptedNotification(hash) + message := response.GetStaticLoopInRiskAccepted() + message.SwapHash = backing[:len(hash)] + source = message + liveAccepted = mgr.SubscribeStaticLoopInRiskAccepted( + t.Context(), hash, + ) + } else { + response = staticLoopInRiskRejectedNotification(hash) + message := response.GetStaticLoopInRiskRejected() + message.SwapHash = backing[:len(hash)] + source = message + liveRejected = mgr.SubscribeStaticLoopInRiskRejected( + t.Context(), hash, + ) + } + + // Protobuf preserves unrecognized fields, including + // arbitrarily large length-delimited values. + unknown := protowire.AppendTag(nil, 100, + protowire.BytesType) + unknown = protowire.AppendBytes(unknown, + make([]byte, 1<<20)) + source.ProtoReflect().SetUnknown(unknown) + + mgr.handleNotification(t.Context(), response) + require.Equal(t, 1, persistCalls) + require.Equal(t, unknown, + []byte(source.ProtoReflect().GetUnknown())) + + var cached riskDecisionPayloadNotification + if accepted { + cached = mgr.staticLoopInRiskAccepted[hash] + } else { + cached = mgr.staticLoopInRiskRejected[hash] + } + live := receiveRiskDecisionPayload(t, + liveAccepted, liveRejected) + + // Mutating the original message must not alter the + // cached or already delivered notification. + backing[0] ^= 0xff + assertRiskDecisionPayload(t, cached, hash) + assertRiskDecisionPayload(t, live, hash) + + if accepted { + liveAccepted = mgr.SubscribeStaticLoopInRiskAccepted( + t.Context(), hash, + ) + } else { + liveRejected = mgr.SubscribeStaticLoopInRiskRejected( + t.Context(), hash, + ) + } + replayed := receiveRiskDecisionPayload(t, + liveAccepted, liveRejected) + assertRiskDecisionPayload(t, replayed, hash) + }) + } + }) + } +} + +func receiveRiskDecisionPayload(t *testing.T, + accepted <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification, + rejected <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification, +) riskDecisionPayloadNotification { + + t.Helper() + + select { + case notification := <-accepted: + return notification + case notification := <-rejected: + return notification + case <-time.After(time.Second): + t.Fatal("risk decision notification was not delivered") + return nil + } +} + +func assertRiskDecisionPayload(t *testing.T, + notification riskDecisionPayloadNotification, hash lntypes.Hash) { + + t.Helper() + + require.Equal(t, hash[:], notification.GetSwapHash()) + require.Equal(t, len(hash), cap(notification.GetSwapHash())) + require.Empty(t, notification.ProtoReflect().GetUnknown()) +}