mirror/lnd
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

discovery: adhere to proper channel chunk splitting for ReplyChannelRange

Browse Source
This commit is contained in:
Wilmer Paulino committed 2020-12-08 15:18:07 -08:00
1 parent c5fc7334a4
commit a4f33ae63c
7 files changed
+340 -145
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

discovery/syncer.go
+128 -87
View File
@@ -4,6 +4,8 @@ import (
"errors"
"fmt"
"math"
"math/rand"
"sort"
"sync"
"sync/atomic"
"time"
@@ -128,6 +130,14 @@ const (
// maxUndelayedQueryReplies queries.
DefaultDelayedQueryReplyInterval = 5 * time.Second
// maxQueryChanRangeReplies specifies the default limit of replies to
// process for a single QueryChannelRange request.
maxQueryChanRangeReplies = 500
// maxQueryChanRangeRepliesZlibFactor specifies the factor applied to
// the maximum number of replies allowed for zlib encoded replies.
maxQueryChanRangeRepliesZlibFactor = 4
// chanRangeQueryBuffer is the number of blocks back that we'll go when
// asking the remote peer for their any channels they know of beyond
// our highest known channel ID.
@@ -240,6 +250,10 @@ type gossipSyncerCfg struct {
// bestHeight returns the latest height known of the chain.
bestHeight func() uint32
// maxQueryChanRangeReplies is the maximum number of replies we'll allow
// for a single QueryChannelRange request.
maxQueryChanRangeReplies uint32
}
// GossipSyncer is a struct that handles synchronizing the channel graph state
@@ -316,6 +330,11 @@ type GossipSyncer struct {
// buffer all the chunked response to our query.
bufferedChanRangeReplies []lnwire.ShortChannelID
// numChanRangeRepliesRcvd is used to track the number of replies
// received as part of a QueryChannelRange. This field is primarily used
// within the waitingQueryChanReply state.
numChanRangeRepliesRcvd uint32
// newChansToQuery is used to pass the set of channels we should query
// for from the waitingQueryChanReply state to the queryNewChannels
// state.
@@ -741,17 +760,27 @@ func (g *GossipSyncer) processChanRangeReply(msg *lnwire.ReplyChannelRange) erro
g.bufferedChanRangeReplies = append(
g.bufferedChanRangeReplies, msg.ShortChanIDs...,
)
switch g.cfg.encodingType {
case lnwire.EncodingSortedPlain:
g.numChanRangeRepliesRcvd++
case lnwire.EncodingSortedZlib:
g.numChanRangeRepliesRcvd += maxQueryChanRangeRepliesZlibFactor
default:
return fmt.Errorf("unhandled encoding type %v", g.cfg.encodingType)
}
log.Infof("GossipSyncer(%x): buffering chan range reply of size=%v",
g.cfg.peerPub[:], len(msg.ShortChanIDs))
// If this isn't the last response, then we can exit as we've already
// buffered the latest portion of the streaming reply.
// If this isn't the last response and we can continue to receive more,
// then we can exit as we've already buffered the latest portion of the
// streaming reply.
maxReplies := g.cfg.maxQueryChanRangeReplies
switch {
// If we're communicating with a legacy node, we'll need to look at the
// complete field.
case isLegacyReplyChannelRange(g.curQueryRangeMsg, msg):
if msg.Complete == 0 {
if msg.Complete == 0 && g.numChanRangeRepliesRcvd < maxReplies {
return nil
}
@@ -763,7 +792,8 @@ func (g *GossipSyncer) processChanRangeReply(msg *lnwire.ReplyChannelRange) erro
// TODO(wilmer): This might require some padding if the remote
// node is not aware of the last height we sent them, i.e., is
// behind a few blocks from us.
if replyLastHeight < queryLastHeight {
if replyLastHeight < queryLastHeight &&
g.numChanRangeRepliesRcvd < maxReplies {
return nil
}
}
@@ -786,6 +816,7 @@ func (g *GossipSyncer) processChanRangeReply(msg *lnwire.ReplyChannelRange) erro
g.curQueryRangeMsg = nil
g.prevReplyChannelRange = nil
g.bufferedChanRangeReplies = nil
g.numChanRangeRepliesRcvd = 0
// If there aren't any channels that we don't know of, then we can
// switch straight to our terminal state.
@@ -930,7 +961,7 @@ func (g *GossipSyncer) replyChanRangeQuery(query *lnwire.QueryChannelRange) erro
// channel ID's that match their query.
startBlock := query.FirstBlockHeight
endBlock := query.LastBlockHeight()
channelRange, err := g.cfg.channelSeries.FilterChannelRange(
channelRanges, err := g.cfg.channelSeries.FilterChannelRange(
query.ChainHash, startBlock, endBlock,
)
if err != nil {
@@ -940,102 +971,98 @@ func (g *GossipSyncer) replyChanRangeQuery(query *lnwire.QueryChannelRange) erro
// TODO(roasbeef): means can't send max uint above?
// * or make internal 64
// In the base case (no actual response) the first block and last block
// will match those of the query. In the loop below, we'll update these
// two variables incrementally with each chunk to properly compute the
// starting block for each response and the number of blocks in a
// response.
firstBlockHeight := startBlock
lastBlockHeight := endBlock
// We'll send our response in a streaming manner, chunk-by-chunk. We do
// this as there's a transport message size limit which we'll need to
// adhere to. We also need to make sure all of our replies cover the
// expected range of the query.
sendReplyForChunk := func(channelChunk []lnwire.ShortChannelID,
firstHeight, lastHeight uint32, finalChunk bool) error {
numChannels := int32(len(channelRange))
numChansSent := int32(0)
for {
// We'll send our this response in a streaming manner,
// chunk-by-chunk. We do this as there's a transport message
// size limit which we'll need to adhere to.
var channelChunk []lnwire.ShortChannelID
// We know this is the final chunk, if the difference between
// the total number of channels, and the number of channels
// we've sent is less-than-or-equal to the chunk size.
isFinalChunk := (numChannels - numChansSent) <= g.cfg.chunkSize
// If this is indeed the last chunk, then we'll send the
// remainder of the channels.
if isFinalChunk {
channelChunk = channelRange[numChansSent:]
log.Infof("GossipSyncer(%x): sending final chan "+
"range chunk, size=%v", g.cfg.peerPub[:],
len(channelChunk))
} else {
// Otherwise, we'll only send off a fragment exactly
// sized to the proper chunk size.
channelChunk = channelRange[numChansSent : numChansSent+g.cfg.chunkSize]
log.Infof("GossipSyncer(%x): sending range chunk of "+
"size=%v", g.cfg.peerPub[:], len(channelChunk))
}
// If we have any channels at all to return, then we need to
// update our pointers to the first and last blocks for each
// response.
if len(channelChunk) > 0 {
// If this is the first response we'll send, we'll point
// the first block to the first block in the query.
// Otherwise, we'll continue from the block we left off
// at.
if numChansSent == 0 {
firstBlockHeight = startBlock
} else {
firstBlockHeight = lastBlockHeight
}
// If this is the last response we'll send, we'll point
// the last block to the last block of the query.
// Otherwise, we'll set it to the height of the last
// channel in the chunk.
if isFinalChunk {
lastBlockHeight = endBlock
} else {
lastBlockHeight = channelChunk[len(channelChunk)-1].BlockHeight
}
}
// The number of blocks contained in this response (the total
// span) is the difference between the last channel ID and the
// first in the range. We add one as even if all channels
// The number of blocks contained in the current chunk (the
// total span) is the difference between the last channel ID and
// the first in the range. We add one as even if all channels
// returned are in the same block, we need to count that.
numBlocksInResp := lastBlockHeight - firstBlockHeight + 1
numBlocks := lastHeight - firstHeight + 1
complete := uint8(0)
if finalChunk {
complete = 1
}
// With our chunk assembled, we'll now send to the remote peer
// the current chunk.
replyChunk := lnwire.ReplyChannelRange{
return g.cfg.sendToPeerSync(&lnwire.ReplyChannelRange{
QueryChannelRange: lnwire.QueryChannelRange{
ChainHash: query.ChainHash,
NumBlocks: numBlocksInResp,
FirstBlockHeight: firstBlockHeight,
NumBlocks: numBlocks,
FirstBlockHeight: firstHeight,
},
Complete: 0,
Complete: complete,
EncodingType: g.cfg.encodingType,
ShortChanIDs: channelChunk,
})
}
var (
firstHeight = query.FirstBlockHeight
lastHeight uint32
channelChunk []lnwire.ShortChannelID
)
for _, channelRange := range channelRanges {
channels := channelRange.Channels
numChannels := int32(len(channels))
numLeftToAdd := g.cfg.chunkSize - int32(len(channelChunk))
// Include the current block in the ongoing chunk if it can fit
// and move on to the next block.
if numChannels <= numLeftToAdd {
channelChunk = append(channelChunk, channels...)
continue
}
if isFinalChunk {
replyChunk.Complete = 1
}
if err := g.cfg.sendToPeerSync(&replyChunk); err != nil {
// Otherwise, we need to send our existing channel chunk as is
// as its own reply and start a new one for the current block.
// We'll mark the end of our current chunk as the height before
// the current block to ensure the whole query range is replied
// to.
log.Infof("GossipSyncer(%x): sending range chunk of size=%v",
g.cfg.peerPub[:], len(channelChunk))
lastHeight = channelRange.Height - 1
err := sendReplyForChunk(
channelChunk, firstHeight, lastHeight, false,
)
if err != nil {
return err
}
// If this was the final chunk, then we'll exit now as our
// response is now complete.
if isFinalChunk {
return nil
// With the reply constructed, we'll start tallying channels for
// our next one keeping in mind our chunk size. This may result
// in channels for this block being left out from the reply, but
// this isn't an issue since we'll randomly shuffle them and we
// assume a historical gossip sync is performed at a later time.
firstHeight = channelRange.Height
chunkSize := numChannels
exceedsChunkSize := numChannels > g.cfg.chunkSize
if exceedsChunkSize {
rand.Shuffle(len(channels), func(i, j int) {
channels[i], channels[j] = channels[j], channels[i]
})
chunkSize = g.cfg.chunkSize
}
channelChunk = channels[:chunkSize]
numChansSent += int32(len(channelChunk))
// Sort the chunk once again if we had to shuffle it.
if exceedsChunkSize {
sort.Slice(channelChunk, func(i, j int) bool {
return channelChunk[i].ToUint64() <
channelChunk[j].ToUint64()
})
}
}
// Send the remaining chunk as the final reply.
log.Infof("GossipSyncer(%x): sending final chan range chunk, size=%v",
g.cfg.peerPub[:], len(channelChunk))
return sendReplyForChunk(
channelChunk, firstHeight, query.LastBlockHeight(), true,
)
}
// replyShortChanIDs will be dispatched in response to a query by the remote
@@ -1285,11 +1312,23 @@ func (g *GossipSyncer) FilterGossipMsgs(msgs ...msgWithSenders) {
// ProcessQueryMsg is used by outside callers to pass new channel time series
// queries to the internal processing goroutine.
func (g *GossipSyncer) ProcessQueryMsg(msg lnwire.Message, peerQuit <-chan struct{}) {
func (g *GossipSyncer) ProcessQueryMsg(msg lnwire.Message, peerQuit <-chan struct{}) error {
var msgChan chan lnwire.Message
switch msg.(type) {
case *lnwire.QueryChannelRange, *lnwire.QueryShortChanIDs:
msgChan = g.queryMsgs
// Reply messages should only be expected in states where we're waiting
// for a reply.
case *lnwire.ReplyChannelRange, *lnwire.ReplyShortChanIDsEnd:
syncState := g.syncState()
if syncState != waitingQueryRangeReply &&
syncState != waitingQueryChanReply {
return fmt.Errorf("received unexpected query reply "+
"message %T", msg)
}
msgChan = g.gossipMsgs
default:
msgChan = g.gossipMsgs
}
@@ -1299,6 +1338,8 @@ func (g *GossipSyncer) ProcessQueryMsg(msg lnwire.Message, peerQuit <-chan struc
case <-peerQuit:
case <-g.quit:
}
return nil
}
// setSyncState sets the gossip syncer's state to the given state.