Merge pull request #10103 from yyforyongyu/rate-limit-peer

Rate limit outgoing gossip bandwidth by peer

config.go

@@ -721,6 +721,7 @@ func DefaultConfig() Config {
 			MsgBurstBytes:         discovery.DefaultMsgBytesBurst,
 			FilterConcurrency:     discovery.DefaultFilterConcurrency,
 			BanThreshold:          discovery.DefaultBanThreshold,
+			PeerMsgRateBytes:      discovery.DefaultPeerMsgBytesPerSecond,
 		},
 		Invoices: &lncfg.Invoices{
 			HoldExpiryDelta: lncfg.DefaultHoldInvoiceExpiryDelta,

discovery/gossiper.go

@@ -407,6 +407,10 @@ type Config struct {
 	// BanThreshold is the score used to decide whether a given peer is
 	// banned or not.
 	BanThreshold uint64
+
+	// PeerMsgRateBytes is the rate limit for the number of bytes per second
+	// that we'll allocate to outbound gossip messages for a single peer.
+	PeerMsgRateBytes uint64
 }
 
 // processedNetworkMsg is a wrapper around networkMsg and a boolean. It is
@@ -609,6 +613,7 @@ func New(cfg Config, selfKeyDesc *keychain.KeyDescriptor) *AuthenticatedGossiper
 		AllotedMsgBytesPerSecond: cfg.MsgRateBytes,
 		AllotedMsgBytesBurst:     cfg.MsgBurstBytes,
 		FilterConcurrency:        cfg.FilterConcurrency,
+		PeerMsgBytesPerSecond:    cfg.PeerMsgRateBytes,
 	})
 
 	gossiper.reliableSender = newReliableSender(&reliableSenderCfg{

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
@@ -141,6 +146,11 @@ type SyncManagerCfg struct {
 	// FilterConcurrency is the maximum number of concurrent gossip filter
 	// applications that can be processed. If not set, defaults to 5.
 	FilterConcurrency int
+
+	// PeerMsgBytesPerSecond is the allotted bandwidth rate, expressed in
+	// bytes/second that a single gossip syncer can consume. Once we exceed
+	// this rate, message sending will block until we're below the rate.
+	PeerMsgBytesPerSecond uint64
 }
 
 // SyncManager is a subsystem of the gossiper that manages the gossip syncers
@@ -200,7 +210,7 @@ type SyncManager struct {
 	gossipFilterSema chan struct{}
 
 	// rateLimiter dictates the frequency with which we will reply to gossip
-	// queries from a peer. This is used to delay responses to peers to
+	// queries from all peers. This is used to delay responses to peers to
 	// prevent DOS vulnerabilities if they are spamming with an unreasonable
 	// number of queries.
 	rateLimiter *rate.Limiter
@@ -554,8 +564,8 @@ func (m *SyncManager) isPinnedSyncer(s *GossipSyncer) bool {
 
 // deriveRateLimitReservation will take the current message and derive a
 // reservation that can be used to wait on the rate limiter.
-func (m *SyncManager) deriveRateLimitReservation(msg lnwire.Message,
-) (*rate.Reservation, error) {
+func deriveRateLimitReservation(rl *rate.Limiter,
+	msg lnwire.Message) (*rate.Reservation, error) {
 
 	var (
 		msgSize uint32
@@ -575,12 +585,12 @@ func (m *SyncManager) deriveRateLimitReservation(msg lnwire.Message,
 		msgSize = assumedMsgSize
 	}
 
-	return m.rateLimiter.ReserveN(time.Now(), int(msgSize)), nil
+	return rl.ReserveN(time.Now(), int(msgSize)), nil
 }
 
 // waitMsgDelay takes a delay, and waits until it has finished.
-func (m *SyncManager) waitMsgDelay(ctx context.Context, peerPub [33]byte,
-	limitReservation *rate.Reservation) error {
+func waitMsgDelay(ctx context.Context, peerPub [33]byte,
+	limitReservation *rate.Reservation, quit <-chan struct{}) error {
 
 	// If we've already replied a handful of times, we will start to delay
 	// responses back to the remote peer. This can help prevent DOS attacks
@@ -602,7 +612,7 @@ func (m *SyncManager) waitMsgDelay(ctx context.Context, peerPub [33]byte,
 
 			return ErrGossipSyncerExiting
 
-		case <-m.quit:
+		case <-quit:
 			limitReservation.Cancel()
 
 			return ErrGossipSyncerExiting
@@ -614,15 +624,15 @@ func (m *SyncManager) waitMsgDelay(ctx context.Context, peerPub [33]byte,
 
 // maybeRateLimitMsg takes a message, and may wait a period of time to rate
 // limit the msg.
-func (m *SyncManager) maybeRateLimitMsg(ctx context.Context, peerPub [33]byte,
-	msg lnwire.Message) error {
+func maybeRateLimitMsg(ctx context.Context, rl *rate.Limiter, peerPub [33]byte,
+	msg lnwire.Message, quit <-chan struct{}) error {
 
-	delay, err := m.deriveRateLimitReservation(msg)
+	delay, err := deriveRateLimitReservation(rl, msg)
 	if err != nil {
 		return nil
 	}
 
-	return m.waitMsgDelay(ctx, peerPub, delay)
+	return waitMsgDelay(ctx, peerPub, delay, quit)
 }
 
 // sendMessages sends a set of messages to the remote peer.
@@ -630,9 +640,13 @@ func (m *SyncManager) sendMessages(ctx context.Context, sync bool,
 	peer lnpeer.Peer, nodeID route.Vertex, msgs ...lnwire.Message) error {
 
 	for _, msg := range msgs {
-		if err := m.maybeRateLimitMsg(ctx, nodeID, msg); err != nil {
+		err := maybeRateLimitMsg(
+			ctx, m.rateLimiter, nodeID, msg, m.quit,
+		)
+		if err != nil {
 			return err
 		}
+
 		if err := peer.SendMessageLazy(sync, msg); err != nil {
 			return err
 		}
@@ -654,15 +668,10 @@ func (m *SyncManager) createGossipSyncer(peer lnpeer.Peer) *GossipSyncer {
 		encodingType:  encoding,
 		chunkSize:     encodingTypeToChunkSize[encoding],
 		batchSize:     requestBatchSize,
-		sendToPeer: func(ctx context.Context,
+		sendMsg: func(ctx context.Context, sync bool,
 			msgs ...lnwire.Message) error {
 
-			return m.sendMessages(ctx, false, peer, nodeID, msgs...)
-		},
-		sendToPeerSync: func(ctx context.Context,
-			msgs ...lnwire.Message) error {
-
-			return m.sendMessages(ctx, true, peer, nodeID, msgs...)
+			return m.sendMessages(ctx, sync, peer, nodeID, msgs...)
 		},
 		ignoreHistoricalFilters:  m.cfg.IgnoreHistoricalFilters,
 		bestHeight:               m.cfg.BestHeight,
@@ -670,6 +679,7 @@ func (m *SyncManager) createGossipSyncer(peer lnpeer.Peer) *GossipSyncer {
 		maxQueryChanRangeReplies: maxQueryChanRangeReplies,
 		noTimestampQueryOption:   m.cfg.NoTimestampQueries,
 		isStillZombieChannel:     m.cfg.IsStillZombieChannel,
+		msgBytesPerSecond:        m.cfg.PeerMsgBytesPerSecond,
 	}, m.gossipFilterSema)
 
 	// Gossip syncers are initialized by default in a PassiveSync type

discovery/sync_manager_test.go

@@ -748,7 +748,7 @@ func TestDeriveRateLimitReservation(t *testing.T) {
 		}
 
 		// First message should have no delay as it fits within burst.
-		delay1, err := sm.deriveRateLimitReservation(msg)
+		delay1, err := deriveRateLimitReservation(sm.rateLimiter, msg)
 		require.NoError(t, err)
 		require.Equal(
 			t, time.Duration(0), delay1.Delay(), "first message "+
@@ -757,7 +757,7 @@ func TestDeriveRateLimitReservation(t *testing.T) {
 
 		// Second message should have a non-zero delay as the token
 		// bucket is now depleted.
-		delay2, err := sm.deriveRateLimitReservation(msg)
+		delay2, err := deriveRateLimitReservation(sm.rateLimiter, msg)
 		require.NoError(t, err)
 		require.True(
 			t, delay2.Delay() > 0, "second message should have "+
@@ -766,7 +766,7 @@ func TestDeriveRateLimitReservation(t *testing.T) {
 
 		// Third message should have an even longer delay since the
 		// token bucket is still refilling at a constant rate.
-		delay3, err := sm.deriveRateLimitReservation(msg)
+		delay3, err := deriveRateLimitReservation(sm.rateLimiter, msg)
 		require.NoError(t, err)
 		require.True(t, delay3.Delay() > delay2.Delay(), "third "+
 			"message should have longer delay than second: %s > %s",
@@ -798,7 +798,7 @@ func TestDeriveRateLimitReservation(t *testing.T) {
 
 		// The error should propagate through
 		// deriveRateLimitReservation.
-		_, err := sm.deriveRateLimitReservation(msg)
+		_, err := deriveRateLimitReservation(sm.rateLimiter, msg)
 		require.Error(t, err)
 		require.Equal(
 			t, errMsg, err, "Error should be propagated unchanged",
@@ -815,7 +815,7 @@ func TestDeriveRateLimitReservation(t *testing.T) {
 		initialMsg := &TestSizeableMessage{
 			size: uint32(bytesBurst),
 		}
-		_, err := sm.deriveRateLimitReservation(initialMsg)
+		_, err := deriveRateLimitReservation(sm.rateLimiter, initialMsg)
 		require.NoError(t, err)
 
 		// Now send two messages of different sizes and compare their
@@ -828,18 +828,22 @@ func TestDeriveRateLimitReservation(t *testing.T) {
 		}
 
 		// Send the small message first.
-		smallDelay, err := sm.deriveRateLimitReservation(smallMsg)
+		smallDelay, err := deriveRateLimitReservation(
+			sm.rateLimiter, smallMsg,
+		)
 		require.NoError(t, err)
 
 		// Reset the limiter to the same state, then empty the bucket.
 		sm.rateLimiter = rate.NewLimiter(
 			rate.Limit(bytesPerSec), int(bytesBurst),
 		)
-		_, err = sm.deriveRateLimitReservation(initialMsg)
+		_, err = deriveRateLimitReservation(sm.rateLimiter, initialMsg)
 		require.NoError(t, err)
 
 		// Now send the large message.
-		largeDelay, err := sm.deriveRateLimitReservation(largeMsg)
+		largeDelay, err := deriveRateLimitReservation(
+			sm.rateLimiter, largeMsg,
+		)
 		require.NoError(t, err)
 
 		// The large message should have a longer delay than the small

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
@@ -245,15 +246,10 @@ type gossipSyncerCfg struct {
 	// the remote node in a single QueryShortChanIDs request.
 	batchSize int32
 
-	// sendToPeer sends a variadic number of messages to the remote peer.
-	// This method should not block while waiting for sends to be written
-	// to the wire.
-	sendToPeer func(context.Context, ...lnwire.Message) error
-
-	// sendToPeerSync sends a variadic number of messages to the remote
-	// peer, blocking until all messages have been sent successfully or a
-	// write error is encountered.
-	sendToPeerSync func(context.Context, ...lnwire.Message) error
+	// sendMsg sends a variadic number of messages to the remote peer.
+	// The boolean indicates whether this method should be blocked or not
+	// while waiting for sends to be written to the wire.
+	sendMsg func(context.Context, bool, ...lnwire.Message) error
 
 	// noSyncChannels will prevent the GossipSyncer from spawning a
 	// channelGraphSyncer, meaning we will not try to reconcile unknown
@@ -295,6 +291,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 +408,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 +423,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 +443,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,
 	}
 }
 
@@ -496,7 +513,7 @@ func (g *GossipSyncer) handleSyncingChans(ctx context.Context) {
 
 	// Send the msg to the remote peer, which is non-blocking as
 	// `sendToPeer` only queues the msg in Brontide.
-	err = g.cfg.sendToPeer(ctx, queryRangeMsg)
+	err = g.sendToPeer(ctx, queryRangeMsg)
 	if err != nil {
 		log.Errorf("Unable to send chan range query: %v", err)
 		return
@@ -783,11 +800,11 @@ func (g *GossipSyncer) sendGossipTimestampRange(ctx context.Context,
 		TimestampRange: timestampRange,
 	}
 
-	if err := g.cfg.sendToPeer(ctx, localUpdateHorizon); err != nil {
+	if err := g.sendToPeer(ctx, localUpdateHorizon); err != nil {
 		return err
 	}
 
-	if firstTimestamp == zeroTimestamp && timestampRange == 0 {
+	if firstTimestamp.Equal(zeroTimestamp) && timestampRange == 0 {
 		g.localUpdateHorizon = nil
 	} else {
 		g.localUpdateHorizon = localUpdateHorizon
@@ -839,7 +856,7 @@ func (g *GossipSyncer) synchronizeChanIDs(ctx context.Context) bool {
 
 	// With our chunk obtained, we'll send over our next query, then return
 	// false indicating that we're net yet fully synced.
-	err := g.cfg.sendToPeer(ctx, &lnwire.QueryShortChanIDs{
+	err := g.sendToPeer(ctx, &lnwire.QueryShortChanIDs{
 		ChainHash:    g.cfg.chainHash,
 		EncodingType: lnwire.EncodingSortedPlain,
 		ShortChanIDs: queryChunk,
@@ -1154,7 +1171,7 @@ func (g *GossipSyncer) replyChanRangeQuery(ctx context.Context,
 			"chain=%v, we're on chain=%v", query.ChainHash,
 			g.cfg.chainHash)
 
-		return g.cfg.sendToPeerSync(ctx, &lnwire.ReplyChannelRange{
+		return g.sendToPeerSync(ctx, &lnwire.ReplyChannelRange{
 			ChainHash:        query.ChainHash,
 			FirstBlockHeight: query.FirstBlockHeight,
 			NumBlocks:        query.NumBlocks,
@@ -1227,7 +1244,7 @@ func (g *GossipSyncer) replyChanRangeQuery(ctx context.Context,
 			)
 		}
 
-		return g.cfg.sendToPeerSync(ctx, &lnwire.ReplyChannelRange{
+		return g.sendToPeerSync(ctx, &lnwire.ReplyChannelRange{
 			ChainHash:        query.ChainHash,
 			NumBlocks:        numBlocks,
 			FirstBlockHeight: firstHeight,
@@ -1333,7 +1350,7 @@ func (g *GossipSyncer) replyShortChanIDs(ctx context.Context,
 			"chain=%v, we're on chain=%v", query.ChainHash,
 			g.cfg.chainHash)
 
-		return g.cfg.sendToPeerSync(ctx, &lnwire.ReplyShortChanIDsEnd{
+		return g.sendToPeerSync(ctx, &lnwire.ReplyShortChanIDsEnd{
 			ChainHash: query.ChainHash,
 			Complete:  0,
 		})
@@ -1364,7 +1381,7 @@ func (g *GossipSyncer) replyShortChanIDs(ctx context.Context,
 	// each one individually and synchronously to throttle the sends and
 	// perform buffering of responses in the syncer as opposed to the peer.
 	for _, msg := range replyMsgs {
-		err := g.cfg.sendToPeerSync(ctx, msg)
+		err := g.sendToPeerSync(ctx, msg)
 		if err != nil {
 			return err
 		}
@@ -1372,7 +1389,7 @@ func (g *GossipSyncer) replyShortChanIDs(ctx context.Context,
 
 	// Regardless of whether we had any messages to reply with, send over
 	// the sentinel message to signal that the stream has terminated.
-	return g.cfg.sendToPeerSync(ctx, &lnwire.ReplyShortChanIDsEnd{
+	return g.sendToPeerSync(ctx, &lnwire.ReplyShortChanIDsEnd{
 		ChainHash: query.ChainHash,
 		Complete:  1,
 	})
@@ -1462,7 +1479,7 @@ func (g *GossipSyncer) ApplyGossipFilter(ctx context.Context,
 		defer g.isSendingBacklog.Store(false)
 
 		for _, msg := range newUpdatestoSend {
-			err := g.cfg.sendToPeerSync(ctx, msg)
+			err := g.sendToPeerSync(ctx, msg)
 			switch {
 			case err == ErrGossipSyncerExiting:
 				return
@@ -1608,7 +1625,7 @@ func (g *GossipSyncer) FilterGossipMsgs(ctx context.Context,
 		return
 	}
 
-	if err = g.cfg.sendToPeer(ctx, msgsToSend...); err != nil {
+	if err = g.sendToPeer(ctx, msgsToSend...); err != nil {
 		log.Errorf("unable to send gossip msgs: %v", err)
 	}
 
@@ -1802,3 +1819,43 @@ func (g *GossipSyncer) handleHistoricalSync(req *historicalSyncReq) {
 	g.setSyncState(syncingChans)
 	close(req.doneChan)
 }
+
+// sendToPeer sends a variadic number of messages to the remote peer. This
+// method should not block while waiting for sends to be written to the wire.
+func (g *GossipSyncer) sendToPeer(ctx context.Context,
+	msgs ...lnwire.Message) error {
+
+	return g.sendMsgRateLimited(ctx, false, msgs...)
+}
+
+// sendToPeerSync sends a variadic number of messages to the remote peer,
+// blocking until all messages have been sent successfully or a write error is
+// encountered.
+func (g *GossipSyncer) sendToPeerSync(ctx context.Context,
+	msgs ...lnwire.Message) error {
+
+	return g.sendMsgRateLimited(ctx, true, msgs...)
+}
+
+// sendMsgRateLimited sends a variadic number of messages to the remote peer,
+// applying our per-peer rate limit before each send. The sync boolean
+// determines if the send is blocking or not.
+func (g *GossipSyncer) sendMsgRateLimited(ctx context.Context, sync bool,
+	msgs ...lnwire.Message) error {
+
+	for _, msg := range msgs {
+		err := maybeRateLimitMsg(
+			ctx, g.rateLimiter, g.cfg.peerPub, msg, g.cg.Done(),
+		)
+		if err != nil {
+			return err
+		}
+
+		err = g.cfg.sendMsg(ctx, sync, msg)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

discovery/syncer_atomic_test.go

@@ -26,9 +26,14 @@ func TestGossipSyncerSingleBacklogSend(t *testing.T) {
 	// Create a blocking sendToPeerSync function. We'll use this to simulate
 	// sending a large backlog.
 	blockingSendChan := make(chan struct{})
-	sendToPeerSync := func(_ context.Context,
+	mockSendMsg := func(_ context.Context, sync bool,
 		msgs ...lnwire.Message) error {
 
+		// Sync is only true when calling `sendToPeerSync`.
+		if !sync {
+			return nil
+		}
+
 		// Track that we're in a send goroutine.
 		count := activeGoroutines.Add(1)
 		totalGoroutinesLaunched.Add(1)
@@ -55,8 +60,8 @@ func TestGossipSyncerSingleBacklogSend(t *testing.T) {
 		defaultChunkSize, true, true, true,
 	)
 
-	// Override the sendToPeerSync to use our blocking version.
-	syncer.cfg.sendToPeerSync = sendToPeerSync
+	// Override the sendMsg to use our blocking version.
+	syncer.cfg.sendMsg = mockSendMsg
 	syncer.cfg.ignoreHistoricalFilters = false
 
 	syncer.Start()

discovery/syncer_test.go

@@ -200,13 +200,7 @@ func newTestSyncer(hID lnwire.ShortChannelID,
 		noSyncChannels:         !syncChannels,
 		noReplyQueries:         !replyQueries,
 		noTimestampQueryOption: !timestamps,
-		sendToPeer: func(_ context.Context,
-			msgs ...lnwire.Message) error {
-
-			msgChan <- msgs
-			return nil
-		},
-		sendToPeerSync: func(_ context.Context,
+		sendMsg: func(_ context.Context, _ bool,
 			msgs ...lnwire.Message) error {
 
 			msgChan <- msgs
@@ -392,16 +386,21 @@ func TestGossipSyncerFilterGossipMsgsAllInMemory(t *testing.T) {
 	// We'll then instruct the gossiper to filter this set of messages.
 	syncer.FilterGossipMsgs(ctx, msgs...)
 
-	// Out of all the messages we sent in, we should only get 2 of them
+	// Out of all the messages we sent in, we should only get 3 of them
 	// back.
-	select {
-	case <-time.After(time.Second * 15):
-		t.Fatalf("no msgs received")
+	msgReceived := make([]lnwire.Message, 0, 3)
+	for {
+		select {
+		case <-time.After(time.Second * 1):
+			t.Fatalf("timeout receiving msg, want 3 msgs, got %v "+
+				"messages: %v", len(msgReceived), msgReceived)
 
-	case msgs := <-msgChan:
-		if len(msgs) != 3 {
-			t.Fatalf("expected 3 messages instead got %v "+
-				"messages: %v", len(msgs), spew.Sdump(msgs))
+		case msgs := <-msgChan:
+			msgReceived = append(msgReceived, msgs...)
+		}
+
+		if len(msgReceived) == 3 {
+			break
 		}
 	}
 

docs/release-notes/release-notes-0.20.0.md

@@ -94,6 +94,10 @@ circuit. The indices are only available for forwarding events saved after v0.20.
   finish under this timeout value. Consider using a larger timeout value if you
   have a slow network.
 
+* The default value for `gossip.msg-rate-bytes` has been
+  [increased](https://github.com/lightningnetwork/lnd/pull/10096) from 100KB to
+  1MB, and `gossip.msg-burst-bytes` has been increased from 200KB to 2MB.
+
 * Added [`deletecanceledinvoices`](
   https://github.com/lightningnetwork/lnd/pull/9625) RPC to allow the removal of
   a canceled invoice. Supports deleting a canceled invoice by providing its
@@ -105,6 +109,14 @@ circuit. The indices are only available for forwarding events saved after v0.20.
   disconnected and banned. Setting the value to 0 effectively disables banning
   by setting the threshold to the maximum possible value. 
 
+* A [new config](https://github.com/lightningnetwork/lnd/pull/10103) value
+  `gossip.peer-msg-rate-bytes=102400` is introduced to allow limiting the
+  outgoing bandwidth used by each peer when processing gossip-related messages.
+  Note this is different from `gossip.msg-rate-bytes`, as this new config
+  controls the bandwidth per peer, while `msg-rate-bytes` controls the gossip as
+  a whole. This new config prevents a single misbehaving peer from using up all
+  the bandwidth.
+
 ## lncli Additions
 
 * [`lncli sendpayment` and `lncli queryroutes` now support the

lncfg/gossip.go

@@ -41,6 +41,8 @@ type Gossip struct {
 	FilterConcurrency int `long:"filter-concurrency" description:"The maximum number of concurrent gossip filter applications that can be processed. If not set, defaults to 5."`
 
 	BanThreshold uint64 `long:"ban-threshold" description:"The score at which a peer is banned. A peer's ban score is incremented for each invalid gossip message. Invalid messages include those with bad signatures, stale timestamps, excessive updates, or invalid chain data. Once the score reaches this threshold, the peer is banned. Set to 0 to disable banning."`
+
+	PeerMsgRateBytes uint64 `long:"peer-msg-rate-bytes" description:"The peer-specific rate of outbound gossip messages, expressed in bytes per second. This setting controls the long-term average speed of gossip traffic sent from your node. The rate limit is applied to each peer. If the rate of outgoing messages exceeds this value, lnd will start to queue and delay messages sending to that peer to stay within the limit."`
 }
 
 // Parse the pubkeys for the pinned syncers.
@@ -72,6 +74,17 @@ func (g *Gossip) Validate() error {
 			g.MsgBurstBytes, lnwire.MaxSliceLength)
 	}
 
+	if g.MsgBurstBytes <= g.MsgRateBytes {
+		return fmt.Errorf("msg-burst-bytes=%v must be greater than "+
+			"msg-rate-bytes=%v", g.MsgBurstBytes, g.MsgRateBytes)
+	}
+
+	if g.MsgRateBytes <= g.PeerMsgRateBytes {
+		return fmt.Errorf("msg-rate-bytes=%v must be greater than "+
+			"peer-msg-rate-bytes=%v", g.MsgRateBytes,
+			g.PeerMsgRateBytes)
+	}
+
 	return nil
 }
 

sample-lnd.conf

@@ -1814,6 +1814,13 @@
 ;
 ; gossip.ban-threshold=100
 
+; The peer-specific rate of outbound gossip messages, expressed in bytes per
+; second. This setting controls the long-term average speed of gossip traffic
+; sent from your node. The rate limit is applied to each peer. If the rate of
+; outgoing messages exceeds this value, lnd will start to queue and delay
+; messages sending to that peer to stay within the limit.
+; gossip.peer-msg-rate-bytes=51200
+
 [invoices]
 
 ; If a hold invoice has accepted htlcs that reach their expiry height and are

server.go

@@ -1103,6 +1103,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 		MsgBurstBytes:           cfg.Gossip.MsgBurstBytes,
 		FilterConcurrency:       cfg.Gossip.FilterConcurrency,
 		BanThreshold:            cfg.Gossip.BanThreshold,
+		PeerMsgRateBytes:        cfg.Gossip.PeerMsgRateBytes,
 	}, nodeKeyDesc)
 
 	accessCfg := &accessManConfig{