discovery: bound channel range reply buffering

In this commit, we cap each QueryChannelRange response at 100,000 SCIDs
across all streamed replies. The existing reply-count limit did not track
the aggregate decoded working set, so memory use varied with the encoding
and composition of the reply stream.

We count raw SCIDs before timestamp filtering, charge replies using the
received encoding type, and release all accumulated range state on any
error. This bounds both memory and CPU work while still leaving headroom
above the current graph.

(cherry picked from commit ceff94fadd)

discovery/syncer.go

@@ -7,6 +7,7 @@ import (
 	"iter"
 	"math"
 	"math/rand"
+	"slices"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -171,6 +172,10 @@ const (
 	// the maximum number of replies allowed for zlib encoded replies.
 	maxQueryChanRangeRepliesZlibFactor = 4
 
+	// maxChanRangeReplySCIDs is the maximum number of short channel IDs
+	// we'll process for a single QueryChannelRange request.
+	maxChanRangeReplySCIDs = 100_000
+
 	// chanRangeQueryBuffer is the number of blocks back that we'll go when
 	// asking the remote peer for their any channels they know of beyond
 	// our highest known channel ID.
@@ -379,6 +384,10 @@ type GossipSyncer struct {
 	// within the waitingQueryChanReply state.
 	numChanRangeRepliesRcvd uint32
 
+	// numChanRangeReplySCIDsRcvd tracks the total number of short channel
+	// IDs received as part of a QueryChannelRange response.
+	numChanRangeReplySCIDsRcvd uint32
+
 	// newChansToQuery is used to pass the set of channels we should query
 	// for from the waitingQueryChanReply state to the queryNewChannels
 	// state.
@@ -920,9 +929,41 @@ func isLegacyReplyChannelRange(query *lnwire.QueryChannelRange,
 // processChanRangeReply is called each time the GossipSyncer receives a new
 // reply to the initial range query to discover new channels that it didn't
 // previously know of.
-func (g *GossipSyncer) processChanRangeReply(_ context.Context,
+func (g *GossipSyncer) processChanRangeReply(ctx context.Context,
 	msg *lnwire.ReplyChannelRange) error {
 
+	// Any error here terminates the range sync, so we release whatever we
+	// accumulated to stop the peer from pinning it by deliberately forcing
+	// an error. Our caller exits the state machine on any error we return,
+	// and nothing prunes a syncer until its peer disconnects, so otherwise
+	// the buffer stays reachable from a syncer that will never run again.
+	err := g.bufferChanRangeReply(ctx, msg)
+	if err != nil {
+		g.resetChanRangeReplyState()
+	}
+
+	return err
+}
+
+// bufferChanRangeReply validates a single ReplyChannelRange against the query
+// that prompted it, buffers the channels it announces, and advances the
+// syncer's state once the reply stream is complete.
+func (g *GossipSyncer) bufferChanRangeReply(_ context.Context,
+	msg *lnwire.ReplyChannelRange) error {
+
+	// A reply only means anything in the context of the query that
+	// prompted it, and every check below reads that query. Today this is
+	// unreachable, as we only accept a reply in waitingQueryRangeReply and
+	// we always set the query before entering that state. It is worth
+	// guarding anyway: an error leaves the syncer sitting in
+	// waitingQueryRangeReply with the query cleared, so any future change
+	// that recovers the handler instead of tearing it down would turn this
+	// into a remote panic.
+	if g.curQueryRangeMsg == nil {
+		return fmt.Errorf("received channel range reply without an " +
+			"active query")
+	}
+
 	// isStale returns whether the timestamp is too far into the past.
 	isStale := func(timestamp time.Time) bool {
 		return time.Since(timestamp) > graph.DefaultChannelPruneExpiry
@@ -975,8 +1016,44 @@ func (g *GossipSyncer) processChanRangeReply(_ context.Context,
 		}
 	}
 
+	// Charge the reply budget using the encoding that was actually
+	// received. The configured encoding is a local preference and does
+	// not describe the responder's message.
+	var replyCount uint32
+	switch msg.EncodingType {
+	case lnwire.EncodingSortedPlain:
+		replyCount = 1
+
+	case lnwire.EncodingSortedZlib:
+		replyCount = maxQueryChanRangeRepliesZlibFactor
+
+	default:
+		return fmt.Errorf(
+			"unhandled encoding type %v", msg.EncodingType,
+		)
+	}
+
+	numReplySCIDs := uint32(len(msg.ShortChanIDs))
+	if g.numChanRangeReplySCIDsRcvd > maxChanRangeReplySCIDs ||
+		numReplySCIDs > maxChanRangeReplySCIDs-
+			g.numChanRangeReplySCIDsRcvd {
+
+		return fmt.Errorf("channel range reply exceeds maximum "+
+			"number of short channel IDs: max=%v",
+			maxChanRangeReplySCIDs)
+	}
+
+	g.numChanRangeRepliesRcvd += replyCount
+	g.numChanRangeReplySCIDsRcvd += numReplySCIDs
 	g.prevReplyChannelRange = msg
 
+	// Reserve room for this reply in one shot instead of letting append
+	// grow the buffer an element at a time. Over a full reply stream this
+	// cuts the number of reallocations by about 3x.
+	g.bufferedChanRangeReplies = slices.Grow(
+		g.bufferedChanRangeReplies, int(numReplySCIDs),
+	)
+
 	for i, scid := range msg.ShortChanIDs {
 		info := graphdb.NewV1ChannelUpdateInfo(
 			scid, time.Time{}, time.Time{},
@@ -1022,15 +1099,6 @@ func (g *GossipSyncer) processChanRangeReply(_ context.Context,
 		)
 	}
 
-	switch g.cfg.encodingType {
-	case lnwire.EncodingSortedPlain:
-		g.numChanRangeRepliesRcvd++
-	case lnwire.EncodingSortedZlib:
-		g.numChanRangeRepliesRcvd += maxQueryChanRangeRepliesZlibFactor
-	default:
-		return fmt.Errorf("unhandled encoding type %v", g.cfg.encodingType)
-	}
-
 	log.Infof("GossipSyncer(%x): buffering chan range reply of size=%v",
 		g.cfg.peerPub[:], len(msg.ShortChanIDs))
 
@@ -1077,10 +1145,7 @@ func (g *GossipSyncer) processChanRangeReply(_ context.Context,
 	// As we've received the entirety of the reply, we no longer need to
 	// hold on to the set of buffered replies or the original query that
 	// prompted the replies, so we'll let that be garbage collected now.
-	g.curQueryRangeMsg = nil
-	g.prevReplyChannelRange = nil
-	g.bufferedChanRangeReplies = nil
-	g.numChanRangeRepliesRcvd = 0
+	g.resetChanRangeReplyState()
 
 	// If there aren't any channels that we don't know of, then we can
 	// switch straight to our terminal state.
@@ -1108,6 +1173,16 @@ func (g *GossipSyncer) processChanRangeReply(_ context.Context,
 	return nil
 }
 
+// resetChanRangeReplyState releases all state accumulated while processing a
+// ReplyChannelRange stream.
+func (g *GossipSyncer) resetChanRangeReplyState() {
+	g.curQueryRangeMsg = nil
+	g.prevReplyChannelRange = nil
+	g.bufferedChanRangeReplies = nil
+	g.numChanRangeRepliesRcvd = 0
+	g.numChanRangeReplySCIDsRcvd = 0
+}
+
 // genChanRangeQuery generates the initial message we'll send to the remote
 // party when we're kicking off the channel graph synchronization upon
 // connection. The historicalQuery boolean can be used to generate a query from

discovery/syncer_test.go

@@ -2571,6 +2571,183 @@ func TestGossipSyncerMaxChannelRangeReplies(t *testing.T) {
 	}, nil))
 }
 
+// TestGossipSyncerMaxChannelRangeSCIDs ensures that a gossip syncer rejects a
+// range response once the aggregate number of short channel IDs exceeds its
+// resource limit.
+func TestGossipSyncerMaxChannelRangeSCIDs(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	_, syncer, _ := newTestSyncer(
+		lnwire.ShortChannelID{BlockHeight: latestKnownHeight},
+		defaultEncoding, defaultChunkSize,
+	)
+
+	query, err := syncer.genChanRangeQuery(ctx, true)
+	require.NoError(t, err)
+
+	scids := make([]lnwire.ShortChannelID, defaultChunkSize)
+	for i := range scids {
+		scids[i] = lnwire.NewShortChanIDFromInt(uint64(i))
+	}
+
+	reply := &lnwire.ReplyChannelRange{
+		ChainHash:        query.ChainHash,
+		FirstBlockHeight: query.FirstBlockHeight,
+		NumBlocks:        query.NumBlocks,
+		EncodingType:     lnwire.EncodingSortedPlain,
+		ShortChanIDs:     scids,
+	}
+
+	numFullReplies := maxChanRangeReplySCIDs / len(scids)
+	for i := 0; i < numFullReplies; i++ {
+		require.NoError(t, syncer.processChanRangeReply(ctx, reply))
+	}
+
+	require.Len(
+		t, syncer.bufferedChanRangeReplies,
+		numFullReplies*len(scids),
+	)
+
+	numRemaining := maxChanRangeReplySCIDs -
+		numFullReplies*len(scids)
+	reply.ShortChanIDs = scids[:numRemaining]
+	require.NoError(t, syncer.processChanRangeReply(ctx, reply))
+	require.Len(
+		t, syncer.bufferedChanRangeReplies,
+		maxChanRangeReplySCIDs,
+	)
+
+	reply.ShortChanIDs = []lnwire.ShortChannelID{
+		lnwire.NewShortChanIDFromInt(uint64(len(scids))),
+	}
+	err = syncer.processChanRangeReply(ctx, reply)
+	require.ErrorContains(
+		t, err, "exceeds maximum number of short channel IDs",
+	)
+	require.Empty(t, syncer.bufferedChanRangeReplies)
+	require.Zero(t, syncer.numChanRangeReplySCIDsRcvd)
+	require.Nil(t, syncer.curQueryRangeMsg)
+}
+
+// TestGossipSyncerChanRangeReplyNoQuery ensures that a range reply which
+// arrives without an active query is rejected rather than dereferencing the
+// nil query.
+func TestGossipSyncerChanRangeReplyNoQuery(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	_, syncer, _ := newTestSyncer(
+		lnwire.ShortChannelID{BlockHeight: latestKnownHeight},
+		defaultEncoding, defaultChunkSize,
+	)
+
+	// Note that we deliberately skip genChanRangeQuery here, so
+	// curQueryRangeMsg is still nil.
+	require.Nil(t, syncer.curQueryRangeMsg)
+
+	err := syncer.processChanRangeReply(ctx, &lnwire.ReplyChannelRange{
+		FirstBlockHeight: 0,
+		NumBlocks:        100,
+		EncodingType:     lnwire.EncodingSortedPlain,
+		ShortChanIDs: []lnwire.ShortChannelID{
+			lnwire.NewShortChanIDFromInt(1),
+		},
+	})
+	require.ErrorContains(t, err, "without an active query")
+}
+
+// TestGossipSyncerCountsReceivedEncoding ensures that compressed range
+// replies consume the larger reply budget even when the local syncer uses
+// plain encoding.
+func TestGossipSyncerCountsReceivedEncoding(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+
+	_, syncer, _ := newTestSyncer(
+		lnwire.ShortChannelID{BlockHeight: latestKnownHeight},
+		defaultEncoding, defaultChunkSize,
+	)
+
+	query, err := syncer.genChanRangeQuery(ctx, true)
+	require.NoError(t, err)
+
+	reply := &lnwire.ReplyChannelRange{
+		ChainHash:        query.ChainHash,
+		FirstBlockHeight: query.FirstBlockHeight,
+		NumBlocks:        query.NumBlocks,
+		EncodingType:     lnwire.EncodingSortedZlib,
+	}
+	require.NoError(t, syncer.processChanRangeReply(ctx, reply))
+	require.Equal(
+		t, uint32(maxQueryChanRangeRepliesZlibFactor),
+		syncer.numChanRangeRepliesRcvd,
+	)
+}
+
+// deliverOverBudgetRangeReply waits for the syncer to send its initial range
+// query, then answers it with a single reply that overruns the aggregate SCID
+// budget. Sending the query is what populates curQueryRangeMsg and moves the
+// syncer into waitingQueryRangeReply, both of which ProcessQueryMsg requires.
+func deliverOverBudgetRangeReply(t *testing.T, syncer *GossipSyncer,
+	msgChan chan []lnwire.Message) {
+
+	t.Helper()
+
+	var query *lnwire.QueryChannelRange
+	select {
+	case msgs := <-msgChan:
+		require.Len(t, msgs, 1)
+
+		q, ok := msgs[0].(*lnwire.QueryChannelRange)
+		require.True(t, ok)
+		query = q
+
+	case <-time.After(time.Second):
+		t.Fatal("expected query channel range request msg")
+	}
+
+	scids := make([]lnwire.ShortChannelID, maxChanRangeReplySCIDs+1)
+	for i := range scids {
+		scids[i] = lnwire.NewShortChanIDFromInt(uint64(i))
+	}
+
+	// Complete is set so that, absent the budget check, this reply would be
+	// taken as the final one and carry on to the completion path. That is
+	// what lets assertRangeSyncAborted tell the two apart.
+	reply := &lnwire.ReplyChannelRange{
+		ChainHash:        query.ChainHash,
+		FirstBlockHeight: query.FirstBlockHeight,
+		NumBlocks:        query.NumBlocks,
+		Complete:         1,
+		EncodingType:     lnwire.EncodingSortedPlain,
+		ShortChanIDs:     scids,
+	}
+	require.NoError(t, syncer.ProcessQueryMsg(reply, nil))
+}
+
+// assertRangeSyncAborted asserts that the syncer bailed out of its range sync
+// rather than treating the reply stream as complete. Reaching the completion
+// path would filter the buffered SCIDs against our local graph, so the absence
+// of that request is what tells us the sync was torn down instead.
+//
+// NOTE: we cannot instead wait on the syncer's wait group, as ContextGuard
+// holds a reference on it until the syncer is signalled to quit.
+func assertRangeSyncAborted(t *testing.T, syncer *GossipSyncer) {
+	t.Helper()
+
+	series, ok := syncer.cfg.channelSeries.(*mockChannelGraphTimeSeries)
+	require.True(t, ok)
+
+	select {
+	case <-series.filterReq:
+		t.Fatal("syncer treated an over-budget reply stream as a " +
+			"completed response")
+
+	default:
+	}
+}
+
 // TestGossipSyncerStateHandlerErrors tests that errors in state handlers cause
 // the channelGraphSyncer goroutine to exit cleanly without endless retry loops.
 // This is a table-driven test covering various error types and states.
@@ -2583,6 +2760,16 @@ func TestGossipSyncerStateHandlerErrors(t *testing.T) {
 		setupState  func(*GossipSyncer)
 		chunkSize   int32
 		injectedErr error
+
+		// deliverMsg, if set, is run after the syncer has been started
+		// and is used to drive the syncer into an error through the
+		// public message path rather than through sendMsg injection.
+		deliverMsg func(*testing.T, *GossipSyncer,
+			chan []lnwire.Message)
+
+		// assertOutcome, if set, asserts the terminal state the syncer
+		// is left in once its goroutine has stopped.
+		assertOutcome func(*testing.T, *GossipSyncer)
 	}{
 		{
 			name:        "context cancel during syncingChans",
@@ -2623,6 +2810,41 @@ func TestGossipSyncerStateHandlerErrors(t *testing.T) {
 				}
 			},
 		},
+		{
+			// Unlike the cases above, this one drives the error in
+			// through ProcessQueryMsg so that we exercise the
+			// syncer's lifecycle rather than calling
+			// processChanRangeReply directly. The syncer starts in
+			// syncingChans and moves itself into
+			// waitingQueryRangeReply once it has sent its query.
+			name:        "SCID budget exceeded while waiting",
+			state:       syncingChans,
+			chunkSize:   defaultChunkSize,
+			injectedErr: nil,
+			setupState:  func(s *GossipSyncer) {},
+			deliverMsg:  deliverOverBudgetRangeReply,
+			assertOutcome: func(t *testing.T, s *GossipSyncer) {
+				// The budget check must abort the sync rather
+				// than let the partial stream be taken as a
+				// completed response.
+				//
+				// NOTE: the release of the buffered reply
+				// state is asserted by
+				// TestGossipSyncerMaxChannelRangeSCIDs, which
+				// can read those fields directly without
+				// racing the syncer's own goroutine.
+				assertRangeSyncAborted(t, s)
+
+				// NOTE: the syncer is left in
+				// waitingQueryRangeReply with no live handler.
+				// That matches how every other terminal error
+				// in this state machine behaves today.
+				require.Equal(
+					t, waitingQueryRangeReply,
+					s.syncState(),
+				)
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -2632,7 +2854,7 @@ func TestGossipSyncerStateHandlerErrors(t *testing.T) {
 
 			// Create syncer with error injection capability.
 			hID := lnwire.NewShortChanIDFromInt(10)
-			syncer, errInj, _ := newErrorInjectingSyncer(
+			syncer, errInj, msgChan := newErrorInjectingSyncer(
 				hID, tt.chunkSize,
 			)
 
@@ -2648,6 +2870,12 @@ func TestGossipSyncerStateHandlerErrors(t *testing.T) {
 			// goroutine.
 			syncer.Start()
 
+			// If this case drives its error in over the wire, do
+			// so now that the goroutine is running.
+			if tt.deliverMsg != nil {
+				tt.deliverMsg(t, syncer, msgChan)
+			}
+
 			// Wait long enough that an endless loop would
 			// accumulate many attempts. With the fix, we should
 			// only see 1-3 attempts. Without the fix, we'd see
@@ -2669,6 +2897,12 @@ func TestGossipSyncerStateHandlerErrors(t *testing.T) {
 				attemptCount,
 			)
 
+			// Verify the terminal state, if this case cares about
+			// it, before we signal the syncer to quit.
+			if tt.assertOutcome != nil {
+				tt.assertOutcome(t, syncer)
+			}
+
 			// Verify the syncer exits cleanly without hanging.
 			assertSyncerExitsCleanly(t, syncer, 2*time.Second)
 		})

docs/release-notes/release-notes-0.21.2.md

@@ -43,6 +43,16 @@
   v0.20-era mandatory version so the v0.21 waiting proof migration runs
   without replaying older migrations against an already-initialized database.
 
+* [Bounded the memory used while syncing the channel
+  graph](https://github.com/lightningnetwork/lnd/pull/10992). A peer replying
+  to our `query_channel_range` could previously make us buffer an
+  unpredictable number of short channel IDs, as the only limit was a coarse
+  67MB cap on the bytes a single zlib-compressed reply could decompress to.
+  Replies are now capped at a precise number of short channel IDs, both
+  per-message and in aggregate across a single query, and the accumulated
+  reply state is released as soon as any reply fails validation so that a
+  peer cannot pin it by deliberately forcing an error.
+
 # New Features
 
 ## Functional Enhancements
@@ -107,3 +117,4 @@
 
 * bitromortac
 * Jared Tobin
+* Olaoluwa Osuntokun