Merge pull request #7264 from yyforyongyu/gossip-resend-messages

discovery: fix message order in batch sending

config.go

@@ -629,6 +629,7 @@ func DefaultConfig() Config {
 		Gossip: &lncfg.Gossip{
 			MaxChannelUpdateBurst: discovery.DefaultMaxChannelUpdateBurst,
 			ChannelUpdateInterval: discovery.DefaultChannelUpdateInterval,
+			SubBatchDelay:         discovery.DefaultSubBatchDelay,
 		},
 		Invoices: &lncfg.Invoices{
 			HoldExpiryDelta: lncfg.DefaultHoldInvoiceExpiryDelta,

discovery/gossiper.go

@@ -47,6 +47,10 @@ const (
 	// updates that we'll hold onto.
 	maxPrematureUpdates = 100
 
+	// DefaultSubBatchDelay is the default delay we'll use when
+	// broadcasting the next announcement batch.
+	DefaultSubBatchDelay = 5 * time.Second
+
 	// maxRejectedUpdates tracks the max amount of rejected channel updates
 	// we'll maintain. This is the global size across all peers. We'll
 	// allocate ~3 MB max to the cache.
@@ -944,9 +948,7 @@ func (d *deDupedAnnouncements) addMsg(message networkMsg) {
 		// timestamp, then we'll just discard the message we got.
 		if oldTimestamp > msg.Timestamp {
 			log.Debugf("Ignored outdated network message: "+
-				"peer=%v, source=%x, msg=%s, ", message.peer,
+				"peer=%v, msg=%s", message.peer, msg.MsgType())
-				message.source.SerializeCompressed(),
-				msg.MsgType())
 			return
 		}
 
@@ -1110,8 +1112,8 @@ func calculateSubBatchSize(totalDelay, subBatchDelay time.Duration,
 		return batchSize
 	}
 
-	subBatchSize := (int(batchSize)*int(subBatchDelay) + int(totalDelay) - 1) /
+	subBatchSize := (batchSize*int(subBatchDelay) +
-		int(totalDelay)
+		int(totalDelay) - 1) / int(totalDelay)
 
 	if subBatchSize < minimumBatchSize {
 		return minimumBatchSize
@@ -1120,10 +1122,20 @@ func calculateSubBatchSize(totalDelay, subBatchDelay time.Duration,
 	return subBatchSize
 }
 
+// batchSizeCalculator maps to the function `calculateSubBatchSize`. We create
+// this variable so the function can be mocked in our test.
+var batchSizeCalculator = calculateSubBatchSize
+
 // splitAnnouncementBatches takes an exiting list of announcements and
 // decomposes it into sub batches controlled by the `subBatchSize`.
-func splitAnnouncementBatches(subBatchSize int,
+func (d *AuthenticatedGossiper) splitAnnouncementBatches(
 	announcementBatch []msgWithSenders) [][]msgWithSenders {
+
+	subBatchSize := batchSizeCalculator(
+		d.cfg.TrickleDelay, d.cfg.SubBatchDelay,
+		d.cfg.MinimumBatchSize, len(announcementBatch),
+	)
+
 	var splitAnnouncementBatch [][]msgWithSenders
 
 	for subBatchSize < len(announcementBatch) {
@@ -1134,71 +1146,79 @@ func splitAnnouncementBatches(subBatchSize int,
 			append(splitAnnouncementBatch,
 				announcementBatch[0:subBatchSize:subBatchSize])
 	}
-	splitAnnouncementBatch = append(splitAnnouncementBatch, announcementBatch)
+	splitAnnouncementBatch = append(
+		splitAnnouncementBatch, announcementBatch,
+	)
 
 	return splitAnnouncementBatch
 }
 
 // splitAndSendAnnBatch takes a batch of messages, computes the proper batch
-// split size, and then sends out all items to the set of target peers. If
+// split size, and then sends out all items to the set of target peers. Locally
-// isLocal is true, then we'll send them to all peers and skip any gossip
+// generated announcements are always sent before remotely generated
-// filter checks.
+// announcements.
-func (d *AuthenticatedGossiper) splitAndSendAnnBatch(annBatch []msgWithSenders,
+func (d *AuthenticatedGossiper) splitAndSendAnnBatch(
-	isLocal bool) {
+	annBatch msgsToBroadcast) {
 
-	// Next, If we have new things to announce then broadcast them to all
+	// delayNextBatch is a helper closure that blocks for `SubBatchDelay`
-	// our immediately connected peers.
+	// duration to delay the sending of next announcement batch.
-	subBatchSize := calculateSubBatchSize(
+	delayNextBatch := func() {
-		d.cfg.TrickleDelay, d.cfg.SubBatchDelay,
+		select {
-		d.cfg.MinimumBatchSize, len(annBatch),
+		case <-time.After(d.cfg.SubBatchDelay):
-	)
+		case <-d.quit:
+			return
+		}
+	}
 
-	splitAnnouncementBatch := splitAnnouncementBatches(
+	// Fetch the local and remote announcements.
-		subBatchSize, annBatch,
+	localBatches := d.splitAnnouncementBatches(annBatch.localMsgs)
-	)
+	remoteBatches := d.splitAnnouncementBatches(annBatch.remoteMsgs)
 
 	d.wg.Add(1)
 	go func() {
 		defer d.wg.Done()
 
-		log.Infof("Broadcasting %v new announcements in %d sub "+
+		log.Debugf("Broadcasting %v new local announcements in %d "+
-			"batches (local=%v)", len(annBatch),
+			"sub batches", len(annBatch.localMsgs),
-			len(splitAnnouncementBatch), isLocal)
+			len(localBatches))
 
-		for _, announcementBatch := range splitAnnouncementBatch {
+		// Send out the local announcements first.
-			d.sendBatch(announcementBatch, isLocal)
+		for _, annBatch := range localBatches {
+			d.sendLocalBatch(annBatch)
+			delayNextBatch()
+		}
 
-			select {
+		log.Debugf("Broadcasting %v new remote announcements in %d "+
-			case <-time.After(d.cfg.SubBatchDelay):
+			"sub batches", len(annBatch.remoteMsgs),
-			case <-d.quit:
+			len(remoteBatches))
-				return
+
-			}
+		// Now send the remote announcements.
+		for _, annBatch := range remoteBatches {
+			d.sendRemoteBatch(annBatch)
+			delayNextBatch()
 		}
 	}()
 }
 
-// sendBatch broadcasts a list of announcements to our peers.
+// sendLocalBatch broadcasts a list of locally generated announcements to our
-func (d *AuthenticatedGossiper) sendBatch(annBatch []msgWithSenders,
+// peers. For local announcements, we skip the filter and dedup logic and just
-	isLocal bool) {
+// send the announcements out to all our coonnected peers.
+func (d *AuthenticatedGossiper) sendLocalBatch(annBatch []msgWithSenders) {
+	msgsToSend := lnutils.Map(
+		annBatch, func(m msgWithSenders) lnwire.Message {
+			return m.msg
+		},
+	)
 
-	// If this is a batch of announcements created locally, then we can
+	err := d.cfg.Broadcast(nil, msgsToSend...)
-	// skip the filter and dedup logic below, and just send the
+	if err != nil {
-	// announcements out to all our coonnected peers.
+		log.Errorf("Unable to send local batch announcements: %v", err)
-	if isLocal {
-		msgsToSend := lnutils.Map(
-			annBatch, func(m msgWithSenders) lnwire.Message {
-				return m.msg
-			},
-		)
-		err := d.cfg.Broadcast(nil, msgsToSend...)
-		if err != nil {
-			log.Errorf("Unable to send local batch "+
-				"announcements: %v", err)
-		}
-
-		return
 	}
+}
 
+// sendRemoteBatch broadcasts a list of remotely generated announcements to our
+// peers.
+func (d *AuthenticatedGossiper) sendRemoteBatch(annBatch []msgWithSenders) {
 	syncerPeers := d.syncMgr.GossipSyncers()
 
 	// We'll first attempt to filter out this new message for all peers
@@ -1282,10 +1302,8 @@ func (d *AuthenticatedGossiper) networkHandler() {
 
 		case announcement := <-d.networkMsgs:
 			log.Tracef("Received network message: "+
-				"peer=%v, source=%x, msg=%s, is_remote=%v",
+				"peer=%v, msg=%s, is_remote=%v",
-				announcement.peer,
+				announcement.peer, announcement.msg.MsgType(),
-				announcement.source.SerializeCompressed(),
-				announcement.msg.MsgType(),
 				announcement.isRemote)
 
 			switch announcement.msg.(type) {
@@ -1350,12 +1368,7 @@ func (d *AuthenticatedGossiper) networkHandler() {
 			// announcements, we'll blast them out w/o regard for
 			// our peer's policies so we ensure they propagate
 			// properly.
-			d.splitAndSendAnnBatch(
+			d.splitAndSendAnnBatch(announcementBatch)
-				announcementBatch.localMsgs, true,
-			)
-			d.splitAndSendAnnBatch(
-				announcementBatch.remoteMsgs, false,
-			)
 
 		// The retransmission timer has ticked which indicates that we
 		// should check if we need to prune or re-broadcast any of our
@@ -2629,9 +2642,8 @@ func (d *AuthenticatedGossiper) handleChanUpdate(nMsg *networkMsg,
 	) {
 
 		log.Debugf("Ignored stale edge policy for short_chan_id(%v): "+
-			"peer=%v, source=%x, msg=%s, is_remote=%v", shortChanID,
+			"peer=%v, msg=%s, is_remote=%v", shortChanID,
-			nMsg.peer, nMsg.source.SerializeCompressed(),
+			nMsg.peer, nMsg.msg.MsgType(), nMsg.isRemote,
-			nMsg.msg.MsgType(), nMsg.isRemote,
 		)
 
 		nMsg.err <- nil

discovery/gossiper_test.go

@@ -784,7 +784,7 @@ func createTestCtx(t *testing.T, startHeight uint32) (*testCtx, error) {
 		HistoricalSyncTicker:  ticker.NewForce(DefaultHistoricalSyncInterval),
 		NumActiveSyncers:      3,
 		AnnSigner:             &mock.SingleSigner{Privkey: selfKeyPriv},
-		SubBatchDelay:         time.Second * 5,
+		SubBatchDelay:         1 * time.Millisecond,
 		MinimumBatchSize:      10,
 		MaxChannelUpdateBurst: DefaultMaxChannelUpdateBurst,
 		ChannelUpdateInterval: DefaultChannelUpdateInterval,
@@ -3371,7 +3371,9 @@ out:
 		case <-sentMsgs:
 		case err := <-notifyErr:
 			t.Fatal(err)
-		default:
+
+		// Give it 5 seconds to drain out.
+		case <-time.After(5 * time.Second):
 			break out
 		}
 	}
@@ -3531,7 +3533,10 @@ func assertMessage(t *testing.T, expected, got lnwire.Message) {
 // TestSplitAnnouncementsCorrectSubBatches checks that we split a given
 // sizes of announcement list into the correct number of batches.
 func TestSplitAnnouncementsCorrectSubBatches(t *testing.T) {
-	t.Parallel()
+	// Create our test harness.
+	const blockHeight = 100
+	ctx, err := createTestCtx(t, blockHeight)
+	require.NoError(t, err, "can't create context")
 
 	const subBatchSize = 10
 
@@ -3541,6 +3546,12 @@ func TestSplitAnnouncementsCorrectSubBatches(t *testing.T) {
 	lengthAnnouncementBatchSizes := len(announcementBatchSizes)
 	lengthExpectedNumberMiniBatches := len(expectedNumberMiniBatches)
 
+	batchSizeCalculator = func(totalDelay, subBatchDelay time.Duration,
+		minimumBatchSize, batchSize int) int {
+
+		return subBatchSize
+	}
+
 	if lengthAnnouncementBatchSizes != lengthExpectedNumberMiniBatches {
 		t.Fatal("Length of announcementBatchSizes and " +
 			"expectedNumberMiniBatches should be equal")
@@ -3550,15 +3561,16 @@ func TestSplitAnnouncementsCorrectSubBatches(t *testing.T) {
 		var batchSize = announcementBatchSizes[testIndex]
 		announcementBatch := make([]msgWithSenders, batchSize)
 
-		splitAnnouncementBatch := splitAnnouncementBatches(
+		splitAnnouncementBatch := ctx.gossiper.splitAnnouncementBatches(
-			subBatchSize, announcementBatch,
+			announcementBatch,
 		)
 
 		lengthMiniBatches := len(splitAnnouncementBatch)
 
 		if lengthMiniBatches != expectedNumberMiniBatches[testIndex] {
 			t.Fatalf("Expecting %d mini batches, actual %d",
-				expectedNumberMiniBatches[testIndex], lengthMiniBatches)
+				expectedNumberMiniBatches[testIndex],
+				lengthMiniBatches)
 		}
 	}
 }

docs/release-notes/release-notes-0.16.0.md

@@ -7,6 +7,10 @@ locally to all connected
 peers](https://github.com/lightningnetwork/lnd/pull/7239), regardless of their
 current gossip sync query status.
 
+There's also a [bug fix](https://github.com/lightningnetwork/lnd/pull/7186)
+that might lead to channel updates being missed, causing channel graph being
+incomplete. Aside from that, a potential announcement messages being sent out
+of order is also [fixed](https://github.com/lightningnetwork/lnd/pull/7264).
 
 ## BOLT Specs
 

lncfg/gossip.go

@@ -16,6 +16,8 @@ type Gossip struct {
 	MaxChannelUpdateBurst int `long:"max-channel-update-burst" description:"The maximum number of updates for a specific channel and direction that lnd will accept over the channel update interval."`
 
 	ChannelUpdateInterval time.Duration `long:"channel-update-interval" description:"The interval used to determine how often lnd should allow a burst of new updates for a specific channel and direction."`
+
+	SubBatchDelay time.Duration `long:"sub-batch-delay" description:"The duration to wait before sending the next announcement batch if there are multiple. Use a small value if there are a lot announcements and they need to be broadcast quickly."`
 }
 
 // Parse the pubkeys for the pinned syncers.

lntemp/node/config.go

@@ -187,6 +187,10 @@ func (cfg *BaseNodeConfig) GenArgs() []string {
 		// Use a small batch window so we can broadcast our sweep
 		// transactions faster.
 		"--sweeper.batchwindowduration=5s",
+
+		// Use a small batch delay so we can broadcast the
+		// announcements quickly in the tests.
+		"--gossip.sub-batch-delay=5ms",
 	}
 	args = append(args, nodeArgs...)
 

sample-lnd.conf

@@ -1429,6 +1429,10 @@ litecoin.node=ltcd
 ; gossip.max-channel-update-burst=10
 ; gossip.channel-update-interval=1m
 
+; The duration to wait before sending the next announcement batch if there are
+; multiple. Use a small value if there are a lot announcements and they need to
+; be broadcast quickly.
+; gossip.sub-batch-delay=5s
 
 [invoices]
 

server.go

@@ -1013,7 +1013,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		HistoricalSyncTicker:    ticker.New(cfg.HistoricalSyncInterval),
 		NumActiveSyncers:        cfg.NumGraphSyncPeers,
 		MinimumBatchSize:        10,
-		SubBatchDelay:           time.Second * 5,
+		SubBatchDelay:           cfg.Gossip.SubBatchDelay,
 		IgnoreHistoricalFilters: cfg.IgnoreHistoricalGossipFilters,
 		PinnedSyncers:           cfg.Gossip.PinnedSyncers,
 		MaxChannelUpdateBurst:   cfg.Gossip.MaxChannelUpdateBurst,