Merge pull request #10530 from NishantBansal2003/gossip-race

discovery: fix race on remoteUpdateHorizon in GossipSyncer

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")
+	}
+}

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)