discovery: introduce rate limiter to GossipSyncer

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

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,
 	}
 }