diff --git a/discovery/sync_manager.go b/discovery/sync_manager.go index e5cace1a4..6ffea00cb 100644 --- a/discovery/sync_manager.go +++ b/discovery/sync_manager.go @@ -40,6 +40,11 @@ const ( // they'll be refilled at this rate. DefaultMsgBytesPerSecond = 1000 * 1_024 + // DefaultPeerMsgBytesPerSecond is the max bytes/s we'll permit for + // outgoing messages for a single peer. Once tokens (bytes) have been + // taken from the bucket, they'll be refilled at this rate. + DefaultPeerMsgBytesPerSecond = 50 * 1_024 + // assumedMsgSize is the assumed size of a message if we can't compute // its serialized size. This comes out to 1 KB. assumedMsgSize = 1_024 diff --git a/discovery/syncer.go b/discovery/syncer.go index a5074494f..671b75e04 100644 --- a/discovery/syncer.go +++ b/discovery/syncer.go @@ -17,6 +17,7 @@ import ( graphdb "github.com/lightningnetwork/lnd/graph/db" "github.com/lightningnetwork/lnd/lnpeer" "github.com/lightningnetwork/lnd/lnwire" + "golang.org/x/time/rate" ) // SyncerType encapsulates the different types of syncing mechanisms for a @@ -295,6 +296,11 @@ type gossipSyncerCfg struct { // timestampQueueSize is the size of the timestamp range queue. If not // set, defaults to the global timestampQueueSize constant. timestampQueueSize int + + // msgBytesPerSecond is the allotted bandwidth rate, expressed in + // bytes/second that this gossip syncer can consume. Once we exceed this + // rate, message sending will block until we're below the rate. + msgBytesPerSecond uint64 } // GossipSyncer is a struct that handles synchronizing the channel graph state @@ -407,6 +413,10 @@ type GossipSyncer struct { // allows contexts that either block or cancel on those depending on // the use case. cg *fn.ContextGuard + + // rateLimiter dictates the frequency with which we will reply to gossip + // queries to this peer. + rateLimiter *rate.Limiter } // newGossipSyncer returns a new instance of the GossipSyncer populated using @@ -418,6 +428,17 @@ func newGossipSyncer(cfg gossipSyncerCfg, sema chan struct{}) *GossipSyncer { queueSize = defaultTimestampQueueSize } + bytesPerSecond := cfg.msgBytesPerSecond + if bytesPerSecond == 0 { + bytesPerSecond = DefaultPeerMsgBytesPerSecond + } + bytesBurst := 2 * bytesPerSecond + + // We'll use this rate limiter to limit this single peer. + rateLimiter := rate.NewLimiter( + rate.Limit(bytesPerSecond), int(bytesBurst), + ) + return &GossipSyncer{ cfg: cfg, syncTransitionReqs: make(chan *syncTransitionReq), @@ -427,8 +448,9 @@ func newGossipSyncer(cfg gossipSyncerCfg, sema chan struct{}) *GossipSyncer { timestampRangeQueue: make( chan *lnwire.GossipTimestampRange, queueSize, ), - syncerSema: sema, - cg: fn.NewContextGuard(), + syncerSema: sema, + cg: fn.NewContextGuard(), + rateLimiter: rateLimiter, } }