From caa24791be51ade6b833eef32fe86b7f534a1b8c Mon Sep 17 00:00:00 2001 From: Nishant Bansal Date: Thu, 29 Jan 2026 22:06:59 +0530 Subject: [PATCH] discovery: fix race on remoteUpdateHorizon in GossipSyncer Guard access to remoteUpdateHorizon to prevent a race when the gossiper is flushing a pending batch of announcements while concurrently processing a GossipTimestampRange message from a peer. Signed-off-by: Nishant Bansal --- discovery/gossiper_test.go | 74 ++++++++++++++++++++++++++++++++++++++ discovery/syncer.go | 12 ++++--- 2 files changed, 81 insertions(+), 5 deletions(-) diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 018eb3b00..efcedb0e9 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -940,7 +940,11 @@ func createTestCtx(t *testing.T, startHeight uint32, isChanPeer bool) ( return lnwire.ShortChannelID{}, fmt.Errorf("no peer alias") } + hID := lnwire.ShortChannelID{BlockHeight: startHeight} + channelSeries := newMockChannelGraphTimeSeries(hID) + gossiper := New(Config{ + ChanSeries: channelSeries, ChainIO: chain, ChainParams: &chaincfg.MainNetParams, Notifier: notifier, @@ -5297,3 +5301,73 @@ func TestGossiperShutdownWrongChainAnnouncement(t *testing.T) { // is blocked trying to send to the error channel a second time. require.NoError(t, tCtx.gossiper.Stop()) } + +// TestGossipSyncerRace verifies that there is no race when the gossiper flushes +// a pending batch of new announcements to the network while concurrently +// processing a GossipTimestampRange message from a peer. +func TestGossipSyncerRace(t *testing.T) { + t.Parallel() + + tCtx, err := createTestCtx(t, 0, false) + require.NoError(t, err) + + nodePeer := &mockPeer{remoteKeyPriv1.PubKey(), nil, nil, atomic.Bool{}} + + // Connect the remote peer so it can send us a GossipTimestampRange + // message. + tCtx.gossiper.InitSyncState(nodePeer) + + errCh := make(chan error, 1) + + go func() { + // Wait for the trickle delay to elapse before sending the + // GossipTimestampRange message. + time.Sleep(trickleDelay) + + gossipTimestampRange := &lnwire.GossipTimestampRange{ + ChainHash: tCtx.gossiper.syncMgr.cfg.ChainHash, + FirstTimestamp: uint32(time.Now().Unix()), + TimestampRange: 3600, + } + + select { + case err := <-tCtx.gossiper.ProcessRemoteAnnouncement( + t.Context(), gossipTimestampRange, nodePeer, + ): + errCh <- err + case <-time.After(2 * time.Second): + errCh <- fmt.Errorf("gossip message not processed") + } + }() + + // Send a channel announcement from the remote peer, which will be + // flushed to the network after the trickle delay. + ca, err := tCtx.createRemoteChannelAnnouncement(0) + require.NoError(t, err) + + select { + case err := <-tCtx.gossiper.ProcessRemoteAnnouncement( + t.Context(), ca, nodePeer, + ): + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("remote announcement not processed") + } + + // After the trickle delay, the channel announcement is flushed to the + // network. At the same time, the peer sends a GossipTimestampRange + // message, which could trigger a race. + select { + case <-tCtx.broadcastedMessage: + case <-time.After(2 * trickleDelay): + t.Fatal("announcement was not broadcast") + } + + // Ensure the goroutine completed successfully. + select { + case err := <-errCh: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for gossip message processing") + } +} diff --git a/discovery/syncer.go b/discovery/syncer.go index ce970eeef..37f670545 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -1576,9 +1576,13 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context, func (g *GossipSyncer) FilterGossipMsgs(ctx context.Context, msgs ...msgWithSenders) { + g.Lock() + filter := g.remoteUpdateHorizon + g.Unlock() + // If the peer doesn't have an update horizon set, then we won't send // it any new update messages. - if g.remoteUpdateHorizon == nil { + if filter == nil { log.Tracef("GossipSyncer(%x): skipped due to nil "+ "remoteUpdateHorizon", g.cfg.peerPub[:]) return @@ -1619,12 +1623,10 @@ func (g *GossipSyncer) FilterGossipMsgs(ctx context.Context, // We'll construct a helper function that we'll us below to determine // if a given messages passes the gossip msg filter. - g.Lock() - startTime := time.Unix(int64(g.remoteUpdateHorizon.FirstTimestamp), 0) + startTime := time.Unix(int64(filter.FirstTimestamp), 0) endTime := startTime.Add( - time.Duration(g.remoteUpdateHorizon.TimestampRange) * time.Second, + time.Duration(filter.TimestampRange) * time.Second, ) - g.Unlock() passesFilter := func(timeStamp uint32) bool { t := time.Unix(int64(timeStamp), 0)