Implement buffering onion messages for peers.

In this commit, we check if a peer's outbound buffer has room for onion
messages, and if so pulls them from an implementer of a new trait,
OnionMessageProvider.

Makes sure channel messages are prioritized over OMs. We treat OMs the same as
gossip, but prioritize them over gossip.

The onion_message module remains private until further rate limiting is added.

Author: valentinewallace

lightning/src/ln/peer_handler.rs

@@ -401,10 +401,10 @@ impl Peer {
 		self.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE
 	}
 
-	/// Determines if we should push additional gossip messages onto a peer's outbound buffer for
-	/// backfilling gossip data to the peer. This is checked every time the peer's buffer may have
-	/// been drained.
-	fn should_buffer_gossip_backfill(&self) -> bool {
+	/// Determines if we should push additional messages onto a peer's outbound buffer for backfilling
+	/// onion messages and gossip data to the peer. This is checked every time the peer's buffer may
+	/// have been drained.
+	fn should_buffer_message_backfill(&self) -> bool {
 		self.pending_outbound_buffer.is_empty() &&
 			self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
 	}
@@ -755,44 +755,54 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 
 	fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
 		while !peer.awaiting_write_event {
-			if peer.should_buffer_gossip_backfill() {
-				match peer.sync_status {
-					InitSyncTracker::NoSyncRequested => {},
-					InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
-						if let Some((announce, update_a_option, update_b_option)) =
-							self.message_handler.route_handler.get_next_channel_announcement(c)
-						{
-							self.enqueue_message(peer, &announce);
-							if let Some(update_a) = update_a_option {
-								self.enqueue_message(peer, &update_a);
+			if peer.should_buffer_message_backfill() {
+				let next_onion_message_opt = if let Some(peer_node_id) = peer.their_node_id {
+					self.message_handler.onion_message_handler.next_onion_message_for_peer(peer_node_id)
+				} else { None };
+
+				// For now, let onion messages starve gossip.
+				if let Some(next_onion_message) = next_onion_message_opt {
+					self.enqueue_message(peer, &next_onion_message);
+				} else {
+					match peer.sync_status {
+						InitSyncTracker::NoSyncRequested => {},
+						InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
+							if let Some((announce, update_a_option, update_b_option)) =
+								self.message_handler.route_handler.get_next_channel_announcement(c)
+							{
+								self.enqueue_message(peer, &announce);
+								if let Some(update_a) = update_a_option {
+									self.enqueue_message(peer, &update_a);
+								}
+								if let Some(update_b) = update_b_option {
+									self.enqueue_message(peer, &update_b);
+								}
+								peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
+							} else {
+								peer.sync_status = InitSyncTracker::ChannelsSyncing(0xffff_ffff_ffff_ffff);
 							}
-							if let Some(update_b) = update_b_option {
-								self.enqueue_message(peer, &update_b);
+						},
+						InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => {
+							if let Some(msg) = self.message_handler.route_handler.get_next_node_announcement(None) {
+								self.enqueue_message(peer, &msg);
+								peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
+							} else {
+								peer.sync_status = InitSyncTracker::NoSyncRequested;
 							}
-							peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
-						} else {
-							peer.sync_status = InitSyncTracker::ChannelsSyncing(0xffff_ffff_ffff_ffff);
-						}
-					},
-					InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => {
-						if let Some(msg) = self.message_handler.route_handler.get_next_node_announcement(None) {
-							self.enqueue_message(peer, &msg);
-							peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
-						} else {
-							peer.sync_status = InitSyncTracker::NoSyncRequested;
-						}
-					},
-					InitSyncTracker::ChannelsSyncing(_) => unreachable!(),
-					InitSyncTracker::NodesSyncing(key) => {
-						if let Some(msg) = self.message_handler.route_handler.get_next_node_announcement(Some(&key)) {
-							self.enqueue_message(peer, &msg);
-							peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
-						} else {
-							peer.sync_status = InitSyncTracker::NoSyncRequested;
-						}
-					},
+						},
+						InitSyncTracker::ChannelsSyncing(_) => unreachable!(),
+						InitSyncTracker::NodesSyncing(key) => {
+							if let Some(msg) = self.message_handler.route_handler.get_next_node_announcement(Some(&key)) {
+								self.enqueue_message(peer, &msg);
+								peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
+							} else {
+								peer.sync_status = InitSyncTracker::NoSyncRequested;
+							}
+						},
+					}
 				}
 			}
+
 			if peer.msgs_sent_since_pong >= BUFFER_DRAIN_MSGS_PER_TICK {
 				self.maybe_send_extra_ping(peer);
 			}