diff --git a/discovery/syncer.go b/discovery/syncer.go index 7ffe7f650..bb8b1d1e0 100644 --- a/discovery/syncer.go +++ b/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 diff --git a/discovery/syncer_test.go b/discovery/syncer_test.go index 1fd007295..617c0a9ae 100644 --- a/discovery/syncer_test.go +++ b/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) }) diff --git a/docs/release-notes/release-notes-0.21.2.md b/docs/release-notes/release-notes-0.21.2.md index cc96ab69b..82c8d420e 100644 --- a/docs/release-notes/release-notes-0.21.2.md +++ b/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