Skip forwarding gossip messages to peers if their buffer is over-full

Author: TheBlueMatt

lightning/src/ln/peer_handler.rs

@@ -233,6 +233,15 @@ enum InitSyncTracker{
 	NodesSyncing(PublicKey),
 }
 
+/// When the outbound buffer has this many messages, we'll stop reading bytes from the peer until
+/// we manage to send messages until we reach this limit.
+/// We also use this as the target number of outbound gossip messages to keep in the write buffer,
+/// refilled as we send bytes.
+const OUTBOUND_BUFFER_LIMIT_READ_PAUSE: usize = 10;
+/// When the outbound buffer has this many messages, we'll simply skip relaying gossip messages to
+/// the peer.
+const OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP: usize = 20;
+
 struct Peer {
 	channel_encryptor: PeerChannelEncryptor,
 	their_node_id: Option<PublicKey>,
@@ -531,13 +540,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 				}
 			}
 		}
-		const MSG_BUFF_SIZE: usize = 10;
 		while !peer.awaiting_write_event {
-			if peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE {
+			if peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE {
 				match peer.sync_status {
 					InitSyncTracker::NoSyncRequested => {},
 					InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
-						let steps = ((MSG_BUFF_SIZE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
+						let steps = ((OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
 						let all_messages = self.message_handler.route_handler.get_next_channel_announcements(c, steps);
 						for &(ref announce, ref update_a_option, ref update_b_option) in all_messages.iter() {
 							encode_and_send_msg!(announce);
@@ -554,7 +562,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 						}
 					},
 					InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => {
-						let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
+						let steps = (OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len()) as u8;
 						let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps);
 						for msg in all_messages.iter() {
 							encode_and_send_msg!(msg);
@@ -566,7 +574,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 					},
 					InitSyncTracker::ChannelsSyncing(_) => unreachable!(),
 					InitSyncTracker::NodesSyncing(key) => {
-						let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
+						let steps = (OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len()) as u8;
 						let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps);
 						for msg in all_messages.iter() {
 							encode_and_send_msg!(msg);
@@ -585,7 +593,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 					Some(buff) => buff,
 				};
 
-				let should_be_reading = peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE;
+				let should_be_reading = peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE;
 				let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..];
 				let data_sent = descriptor.send_data(pending, should_be_reading);
 				peer.pending_outbound_buffer_first_msg_offset += data_sent;
@@ -814,7 +822,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 						}
 					}
 
-					peer.pending_outbound_buffer.len() > 10 // pause_read
+					peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_READ_PAUSE // pause_read
 				}
 			};
 
@@ -1027,6 +1035,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
 						continue
 					}
+					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+						continue;
+					}
 					if peer.their_node_id.as_ref() == Some(&msg.contents.node_id_1) ||
 					   peer.their_node_id.as_ref() == Some(&msg.contents.node_id_2) {
 						continue;
@@ -1046,6 +1057,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							!peer.should_forward_node_announcement(msg.contents.node_id) {
 						continue
 					}
+					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+						continue;
+					}
 					if peer.their_node_id.as_ref() == Some(&msg.contents.node_id) {
 						continue;
 					}
@@ -1064,6 +1078,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id)  {
 						continue
 					}
+					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+						continue;
+					}
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}