diff --git a/config.go b/config.go index 41fe291ba..e49bbc909 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 eb13d77ea..e3bca5e1f 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. @@ -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, - message.source.SerializeCompressed(), - msg.MsgType()) + "peer=%v, msg=%s", message.peer, msg.MsgType()) return } @@ -1110,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 @@ -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 -// 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) { - // 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), - ) + // 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 + } + } - splitAnnouncementBatch := splitAnnouncementBatches( - subBatchSize, annBatch, - ) + // 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 { - d.sendBatch(announcementBatch, isLocal) + // 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() } }() } -// 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 @@ -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", - 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) { @@ -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( - 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 @@ -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, - 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 diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 07a2b4bc0..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 } } @@ -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( - 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) } } } diff --git a/docs/release-notes/release-notes-0.16.0.md b/docs/release-notes/release-notes-0.16.0.md index aebee3add..bb2b9a9d7 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 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 805db4202..1b64e0c1b 100644 --- a/sample-lnd.conf +++ b/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] 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,