From c6c218f3844867ab641c61f3b54462eaa66d3326 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 3 Feb 2023 01:32:46 +0800 Subject: [PATCH 1/5] discovery: fix confusing loggings --- discovery/gossiper.go | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index eb13d77ea..7843b61e5 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -944,9 +944,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, - message.source.SerializeCompressed(), - msg.MsgType()) + "peer=%v, msg=%s", message.peer, msg.MsgType()) return } @@ -1282,10 +1280,8 @@ func (d *AuthenticatedGossiper) networkHandler() { case announcement := <-d.networkMsgs: log.Tracef("Received network message: "+ - "peer=%v, source=%x, msg=%s, is_remote=%v", - announcement.peer, - announcement.source.SerializeCompressed(), - announcement.msg.MsgType(), + "peer=%v, msg=%s, is_remote=%v", + announcement.peer, announcement.msg.MsgType(), announcement.isRemote) switch announcement.msg.(type) { @@ -2629,9 +2625,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, - nMsg.peer, nMsg.source.SerializeCompressed(), - nMsg.msg.MsgType(), nMsg.isRemote, + "peer=%v, msg=%s, is_remote=%v", shortChanID, + nMsg.peer, nMsg.msg.MsgType(), nMsg.isRemote, ) nMsg.err <- nil From c3d1d3c4f1074573778c3ddb841cedfa3f6d8c3d Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 2 Feb 2023 23:02:53 +0800 Subject: [PATCH 2/5] multi: make `SubBatchDelay` configurable This commit adds a new config option `--gossip.sub-batch-delay` so we can speed up our itest. --- config.go | 1 + discovery/gossiper.go | 4 ++++ lncfg/gossip.go | 2 ++ lntemp/node/config.go | 4 ++++ sample-lnd.conf | 4 ++++ server.go | 2 +- 6 files changed, 16 insertions(+), 1 deletion(-) diff --git a/config.go b/config.go index dca79551f..be346c84a 100644 --- a/config.go +++ b/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, diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 7843b61e5..f00d18823 100644 --- a/discovery/gossiper.go +++ b/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. diff --git a/lncfg/gossip.go b/lncfg/gossip.go index 0eb3e7763..a2c82a7aa 100644 --- a/lncfg/gossip.go +++ b/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. diff --git a/lntemp/node/config.go b/lntemp/node/config.go index 0c228fef8..00b183354 100644 --- a/lntemp/node/config.go +++ b/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...) diff --git a/sample-lnd.conf b/sample-lnd.conf index 96ac7e015..7d51b647d 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -1423,6 +1423,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] diff --git a/server.go b/server.go index d63d9644d..211955423 100644 --- a/server.go +++ b/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, From b73cfc59982c7f2e8195a9ead9e633948197cbcd Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 2 Feb 2023 18:57:20 +0800 Subject: [PATCH 3/5] discovery: split `sendBatch` into local and remote This commit refactors the method `sendBatch` into `sendLocalBatch` and `sendRemoteBatch` for clarity. The batch size calculation is also moved into `splitAnnouncementBatches`. --- discovery/gossiper.go | 72 ++++++++++++++++++++------------------ discovery/gossiper_test.go | 18 +++++++--- 2 files changed, 52 insertions(+), 38 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index f00d18823..ae6de9f71 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -1112,8 +1112,8 @@ func calculateSubBatchSize(totalDelay, subBatchDelay time.Duration, return batchSize } - subBatchSize := (int(batchSize)*int(subBatchDelay) + int(totalDelay) - 1) / - int(totalDelay) + subBatchSize := (batchSize*int(subBatchDelay) + + int(totalDelay) - 1) / int(totalDelay) if subBatchSize < minimumBatchSize { return minimumBatchSize @@ -1122,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) { @@ -1136,7 +1146,9 @@ func splitAnnouncementBatches(subBatchSize int, append(splitAnnouncementBatch, announcementBatch[0:subBatchSize:subBatchSize]) } - splitAnnouncementBatch = append(splitAnnouncementBatch, announcementBatch) + splitAnnouncementBatch = append( + splitAnnouncementBatch, announcementBatch, + ) return splitAnnouncementBatch } @@ -1148,16 +1160,7 @@ func splitAnnouncementBatches(subBatchSize int, func (d *AuthenticatedGossiper) splitAndSendAnnBatch(annBatch []msgWithSenders, isLocal bool) { - // Next, If we have new things to announce then broadcast them to all - // our immediately connected peers. - subBatchSize := calculateSubBatchSize( - d.cfg.TrickleDelay, d.cfg.SubBatchDelay, - d.cfg.MinimumBatchSize, len(annBatch), - ) - - splitAnnouncementBatch := splitAnnouncementBatches( - subBatchSize, annBatch, - ) + splitAnnouncementBatch := d.splitAnnouncementBatches(annBatch) d.wg.Add(1) go func() { @@ -1168,7 +1171,11 @@ func (d *AuthenticatedGossiper) splitAndSendAnnBatch(annBatch []msgWithSenders, len(splitAnnouncementBatch), isLocal) for _, announcementBatch := range splitAnnouncementBatch { - d.sendBatch(announcementBatch, isLocal) + if isLocal { + d.sendLocalBatch(announcementBatch) + } else { + d.sendRemoteBatch(announcementBatch) + } select { case <-time.After(d.cfg.SubBatchDelay): @@ -1179,28 +1186,25 @@ func (d *AuthenticatedGossiper) splitAndSendAnnBatch(annBatch []msgWithSenders, }() } -// sendBatch broadcasts a list of announcements to our peers. -func (d *AuthenticatedGossiper) sendBatch(annBatch []msgWithSenders, - isLocal bool) { +// sendLocalBatch broadcasts a list of locally generated announcements to our +// peers. For local announcements, we skip the filter and dedup logic and just +// 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 - // skip the filter and dedup logic below, and just send the - // announcements out to all our coonnected peers. - 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 + err := d.cfg.Broadcast(nil, msgsToSend...) + if err != nil { + log.Errorf("Unable to send local batch announcements: %v", err) } +} +// 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 diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 07a2b4bc0..c8ed6a14d 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -3531,7 +3531,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 +3544,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 +3559,16 @@ func TestSplitAnnouncementsCorrectSubBatches(t *testing.T) { var batchSize = announcementBatchSizes[testIndex] announcementBatch := make([]msgWithSenders, batchSize) - splitAnnouncementBatch := splitAnnouncementBatches( - subBatchSize, announcementBatch, + splitAnnouncementBatch := ctx.gossiper.splitAnnouncementBatches( + announcementBatch, ) lengthMiniBatches := len(splitAnnouncementBatch) if lengthMiniBatches != expectedNumberMiniBatches[testIndex] { t.Fatalf("Expecting %d mini batches, actual %d", - expectedNumberMiniBatches[testIndex], lengthMiniBatches) + expectedNumberMiniBatches[testIndex], + lengthMiniBatches) } } } From e34a08860854490656ac10a088c65e58e487b07f Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 2 Feb 2023 19:05:56 +0800 Subject: [PATCH 4/5] discovery: send both local and remote anns in the same goroutine This commit changes the sending of anns from using separate goroutines to always sending both local and remote announcements in the same goroutine. In addition, the local announcements are always sent first. This change is to fix the following case: 1. Alice and Bob have a channel 2. Alice receives Bob's NodeAnnouncement 3. Alice goes to broadcast the channel 4. The broadcast is split into a local and remote broadcast due to PR #7239. Bob's NodeAnnouncement is in the remote batch. Everything else (ChannelAnnouncement, ChannelUpdate x2, and Alice's NodeAnnouncement) is in the local batch. 5. The remote batch (containing Bob's NodeAnnouncement) runs before the local batch since they are spawned in separate goroutines. This means that Alice sends Carol the NodeAnnouncement before Carol knows of the channel. In step 2), Bob's NodeAnnouncement (isRemote = true) replaces Bob's NodeAnnouncement that Alice was going to relay (isRemote = false) after processing the AnnouncementSignatures. --- discovery/gossiper.go | 61 ++++++++++++++++++++++---------------- discovery/gossiper_test.go | 6 ++-- 2 files changed, 39 insertions(+), 28 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index ae6de9f71..e3bca5e1f 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -1154,34 +1154,48 @@ func (d *AuthenticatedGossiper) splitAnnouncementBatches( } // 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 -// isLocal is true, then we'll send them to all peers and skip any gossip -// filter checks. -func (d *AuthenticatedGossiper) splitAndSendAnnBatch(annBatch []msgWithSenders, - isLocal bool) { +// split size, and then sends out all items to the set of target peers. Locally +// generated announcements are always sent before remotely generated +// announcements. +func (d *AuthenticatedGossiper) splitAndSendAnnBatch( + annBatch msgsToBroadcast) { - splitAnnouncementBatch := d.splitAnnouncementBatches(annBatch) + // delayNextBatch is a helper closure that blocks for `SubBatchDelay` + // duration to delay the sending of next announcement batch. + delayNextBatch := func() { + select { + case <-time.After(d.cfg.SubBatchDelay): + case <-d.quit: + return + } + } + + // Fetch the local and remote announcements. + 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 "+ - "batches (local=%v)", len(annBatch), - len(splitAnnouncementBatch), isLocal) + log.Debugf("Broadcasting %v new local announcements in %d "+ + "sub batches", len(annBatch.localMsgs), + len(localBatches)) - for _, announcementBatch := range splitAnnouncementBatch { - if isLocal { - d.sendLocalBatch(announcementBatch) - } else { - d.sendRemoteBatch(announcementBatch) - } + // Send out the local announcements first. + for _, annBatch := range localBatches { + d.sendLocalBatch(annBatch) + delayNextBatch() + } - select { - case <-time.After(d.cfg.SubBatchDelay): - case <-d.quit: - return - } + log.Debugf("Broadcasting %v new remote announcements in %d "+ + "sub batches", len(annBatch.remoteMsgs), + len(remoteBatches)) + + // Now send the remote announcements. + for _, annBatch := range remoteBatches { + d.sendRemoteBatch(annBatch) + delayNextBatch() } }() } @@ -1354,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( - announcementBatch.localMsgs, true, - ) - d.splitAndSendAnnBatch( - announcementBatch.remoteMsgs, false, - ) + d.splitAndSendAnnBatch(announcementBatch) // The retransmission timer has ticked which indicates that we // should check if we need to prune or re-broadcast any of our diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index c8ed6a14d..322cd9564 100644 --- a/discovery/gossiper_test.go +++ b/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 } } From 9b61e09ca792f9b343ba6652f45632b9a8c583df Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 19 Dec 2022 21:03:09 +0800 Subject: [PATCH 5/5] docs: update release notes for gossip fix This commit also puts relevant docs closer to each other. --- docs/release-notes/release-notes-0.16.0.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/release-notes/release-notes-0.16.0.md b/docs/release-notes/release-notes-0.16.0.md index 80ad5fa71..1ff56fb24 100644 --- a/docs/release-notes/release-notes-0.16.0.md +++ b/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