Add the ability to broadcast gossip msgs via the event pipeline

When we process gossip messages asynchronously we may find that we
want to forward a gossip message to a peer after we've returned
from the existing `handle_*` method. In order to do so, we need to
be able to send arbitrary loose gossip messages back to the
`PeerManager` via `MessageSendEvent`.

This commit modifies `MessageSendEvent` in order to support this.

Author: TheBlueMatt

lightning/src/ln/channelmanager.rs

@@ -5044,7 +5044,7 @@ where
 					), chan),
 					// Note that announcement_signatures fails if the channel cannot be announced,
 					// so get_channel_update_for_broadcast will never fail by the time we get here.
-					update_msg: self.get_channel_update_for_broadcast(chan.get()).unwrap(),
+					update_msg: Some(self.get_channel_update_for_broadcast(chan.get()).unwrap()),
 				});
 			},
 			hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close(format!("Got a message for a channel from the wrong node! No such channel for the passed counterparty_node_id {}", counterparty_node_id), msg.channel_id))
@@ -5970,7 +5970,7 @@ where
 										msg: announcement,
 										// Note that announcement_signatures fails if the channel cannot be announced,
 										// so get_channel_update_for_broadcast will never fail by the time we get here.
-										update_msg: self.get_channel_update_for_broadcast(channel).unwrap(),
+										update_msg: Some(self.get_channel_update_for_broadcast(channel).unwrap()),
 									});
 								}
 							}
@@ -6286,6 +6286,7 @@ where
 						&events::MessageSendEvent::SendChannelAnnouncement { .. } => false,
 						&events::MessageSendEvent::BroadcastChannelAnnouncement { .. } => true,
 						&events::MessageSendEvent::BroadcastChannelUpdate { .. } => true,
+						&events::MessageSendEvent::BroadcastNodeAnnouncement { .. } => true,
 						&events::MessageSendEvent::SendChannelUpdate { .. } => false,
 						&events::MessageSendEvent::HandleError { .. } => false,
 						&events::MessageSendEvent::SendChannelRangeQuery { .. } => false,

lightning/src/ln/functional_test_utils.rs

@@ -621,6 +621,9 @@ pub fn remove_first_msg_event_to_node(msg_node_id: &PublicKey, msg_events: &Vec<
 		MessageSendEvent::BroadcastChannelUpdate { .. } => {
 			false
 		},
+		MessageSendEvent::BroadcastNodeAnnouncement { .. } => {
+			false
+		},
 		MessageSendEvent::SendChannelUpdate { node_id, .. } => {
 			node_id == msg_node_id
 		},
@@ -1010,7 +1013,7 @@ pub fn create_chan_between_nodes_with_value_b<'a, 'b, 'c>(node_a: &Node<'a, 'b,
 	assert_eq!(events_7.len(), 1);
 	let (announcement, bs_update) = match events_7[0] {
 		MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
-			(msg, update_msg)
+			(msg, update_msg.clone().unwrap())
 		},
 		_ => panic!("Unexpected event"),
 	};
@@ -1021,6 +1024,7 @@ pub fn create_chan_between_nodes_with_value_b<'a, 'b, 'c>(node_a: &Node<'a, 'b,
 	let as_update = match events_8[0] {
 		MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
 			assert!(*announcement == *msg);
+			let update_msg = update_msg.clone().unwrap();
 			assert_eq!(update_msg.contents.short_channel_id, announcement.contents.short_channel_id);
 			assert_eq!(update_msg.contents.short_channel_id, bs_update.contents.short_channel_id);
 			update_msg
@@ -1031,7 +1035,7 @@ pub fn create_chan_between_nodes_with_value_b<'a, 'b, 'c>(node_a: &Node<'a, 'b,
 	*node_a.network_chan_count.borrow_mut() += 1;
 
 	expect_channel_ready_event(&node_b, &node_a.node.get_our_node_id());
-	((*announcement).clone(), (*as_update).clone(), (*bs_update).clone())
+	((*announcement).clone(), as_update, bs_update)
 }
 
 pub fn create_announced_chan_between_nodes<'a, 'b, 'c, 'd>(nodes: &'a Vec<Node<'b, 'c, 'd>>, a: usize, b: usize) -> (msgs::ChannelUpdate, msgs::ChannelUpdate, [u8; 32], Transaction) {

lightning/src/ln/peer_handler.rs

@@ -1711,10 +1711,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 								self.forward_broadcast_msg(peers, &wire::Message::ChannelAnnouncement(msg), None),
 							_ => {},
 						}
-						match self.message_handler.route_handler.handle_channel_update(&update_msg) {
-							Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
-								self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(update_msg), None),
-							_ => {},
+						if let Some(msg) = update_msg {
+							match self.message_handler.route_handler.handle_channel_update(&msg) {
+								Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
+									self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(msg), None),
+								_ => {},
+							}
 						}
 					},
 					MessageSendEvent::BroadcastChannelUpdate { msg } => {
@@ -1725,6 +1727,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 							_ => {},
 						}
 					},
+					MessageSendEvent::BroadcastNodeAnnouncement { msg } => {
+						log_debug!(self.logger, "Handling BroadcastNodeAnnouncement event in peer_handler for node {}", msg.contents.node_id);
+						match self.message_handler.route_handler.handle_node_announcement(&msg) {
+							Ok(_) | Err(LightningError { action: msgs::ErrorAction::IgnoreDuplicateGossip, .. }) =>
+								self.forward_broadcast_msg(peers, &wire::Message::NodeAnnouncement(msg), None),
+							_ => {},
+						}
+					},
 					MessageSendEvent::SendChannelUpdate { ref node_id, ref msg } => {
 						log_trace!(self.logger, "Handling SendChannelUpdate event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id), msg.contents.short_channel_id);

lightning/src/ln/priv_short_conf_tests.rs

@@ -184,14 +184,14 @@ fn do_test_1_conf_open(connect_style: ConnectStyle) {
 		msg.clone()
 	} else { panic!("Unexpected event"); };
 	let (bs_announcement, bs_update) = if let MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } = bs_announce_events[1] {
-		(msg.clone(), update_msg.clone())
+		(msg.clone(), update_msg.clone().unwrap())
 	} else { panic!("Unexpected event"); };
 
 	nodes[0].node.handle_announcement_signatures(&nodes[1].node.get_our_node_id(), &bs_announcement_sigs);
 	let as_announce_events = nodes[0].node.get_and_clear_pending_msg_events();
 	assert_eq!(as_announce_events.len(), 1);
 	let (announcement, as_update) = if let MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } = as_announce_events[0] {
-		(msg.clone(), update_msg.clone())
+		(msg.clone(), update_msg.clone().unwrap())
 	} else { panic!("Unexpected event"); };
 	assert_eq!(announcement, bs_announcement);
 
@@ -757,7 +757,7 @@ fn test_public_0conf_channel() {
 	match bs_announcement[0] {
 		MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
 			announcement = msg.clone();
-			bs_update = update_msg.clone();
+			bs_update = update_msg.clone().unwrap();
 		},
 		_ => panic!("Unexpected event"),
 	};
@@ -767,6 +767,7 @@ fn test_public_0conf_channel() {
 	match as_announcement[0] {
 		MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
 			assert!(announcement == *msg);
+			let update_msg = update_msg.as_ref().unwrap();
 			assert_eq!(update_msg.contents.short_channel_id, scid);
 			assert_eq!(update_msg.contents.short_channel_id, announcement.contents.short_channel_id);
 			assert_eq!(update_msg.contents.short_channel_id, bs_update.contents.short_channel_id);

lightning/src/ln/reload_tests.rs

@@ -140,7 +140,7 @@ fn test_funding_peer_disconnect() {
 	assert_eq!(events_7.len(), 1);
 	let (chan_announcement, as_update) = match events_7[0] {
 		MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
-			(msg.clone(), update_msg.clone())
+			(msg.clone(), update_msg.clone().unwrap())
 		},
 		_ => panic!("Unexpected event {:?}", events_7[0]),
 	};
@@ -153,7 +153,7 @@ fn test_funding_peer_disconnect() {
 	let bs_update = match events_8[0] {
 		MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
 			assert_eq!(*msg, chan_announcement);
-			update_msg.clone()
+			update_msg.clone().unwrap()
 		},
 		_ => panic!("Unexpected event {:?}", events_8[0]),
 	};

lightning/src/util/events.rs

@@ -1612,13 +1612,18 @@ pub enum MessageSendEvent {
 		/// The channel_announcement which should be sent.
 		msg: msgs::ChannelAnnouncement,
 		/// The followup channel_update which should be sent.
-		update_msg: msgs::ChannelUpdate,
+		update_msg: Option<msgs::ChannelUpdate>,
 	},
 	/// Used to indicate that a channel_update should be broadcast to all peers.
 	BroadcastChannelUpdate {
 		/// The channel_update which should be sent.
 		msg: msgs::ChannelUpdate,
 	},
+	/// Used to indicate that a node_announcement should be broadcast to all peers.
+	BroadcastNodeAnnouncement {
+		/// The node_announcement which should be sent.
+		msg: msgs::NodeAnnouncement,
+	},
 	/// Used to indicate that a channel_update should be sent to a single peer.
 	/// In contrast to [`Self::BroadcastChannelUpdate`], this is used when the channel is a
 	/// private channel and we shouldn't be informing all of our peers of channel parameters.