Pipe channel node information through to forward_broadcast_msg

Sadly, the lightning gossip protocol operates by always flooding
peers with all the latest gossip they receive. For nodes with many
peers, this can result in lots of duplicative gossip as they
receive every message from every peer. As a results, some lightning
implementations disable gossip with new peers after some threshold.
This should mostly work as these peers expect to receive the latest
gossip from their many other peers.

However, in some cases an LDK node may wish to open public
channels but only has a single connection to the bulk of the  rest
of the network - with one such peer which requests that it not
receive any gossip. In that case, LDK would dutifully never send
any gossip to its only connection to the outside world. We would
still send gossip for channels with that peer as it would be sent
as unicast gossip, but if we then open another connection to
another peer which doesn't have any connection to the outside world
any information on that channel wouldn't propagate.

We've seen this setup on some LSPs, where they have a public node
and then an LSP which only connects through that public node, but
expects to open public channels to its LSP clients.

In the next commit we'll start forwarding such gossip to all peers
by ignoring peer gossip limitations about all of our own channels.
Here we do the final prep work, passing the two sides of each
channel back from `handle_channel_update` calls and into
`forward_broadcast_msg`.

Author: TheBlueMatt

lightning-net-tokio/src/lib.rs

@@ -660,8 +660,8 @@ mod tests {
 		}
 		fn handle_channel_update(
 			&self, _their_node_id: Option<PublicKey>, _msg: &ChannelUpdate,
-		) -> Result<bool, LightningError> {
-			Ok(false)
+		) -> Result<Option<(NodeId, NodeId)>, LightningError> {
+			Ok(None)
 		}
 		fn get_next_channel_announcement(
 			&self, _starting_point: u64,

lightning/src/ln/msgs.rs

@@ -2199,13 +2199,13 @@ pub trait RoutingMessageHandler: BaseMessageHandler {
 	fn handle_channel_announcement(
 		&self, their_node_id: Option<PublicKey>, msg: &ChannelAnnouncement,
 	) -> Result<bool, LightningError>;
-	/// Handle an incoming `channel_update` message, returning true if it should be forwarded on,
-	/// `false` or returning an `Err` otherwise.
+	/// Handle an incoming `channel_update` message, returning the node IDs of the channel
+	/// participants if the message should be forwarded on, `None` or returning an `Err` otherwise.
 	///
 	/// If `their_node_id` is `None`, the message was generated by our own local node.
 	fn handle_channel_update(
 		&self, their_node_id: Option<PublicKey>, msg: &ChannelUpdate,
-	) -> Result<bool, LightningError>;
+	) -> Result<Option<(NodeId, NodeId)>, LightningError>;
 	/// Gets channel announcements and updates required to dump our routing table to a remote node,
 	/// starting at the `short_channel_id` indicated by `starting_point` and including announcements
 	/// for a single channel.

lightning/src/ln/peer_handler.rs

@@ -157,8 +157,8 @@ impl RoutingMessageHandler for IgnoringMessageHandler {
 	}
 	fn handle_channel_update(
 		&self, _their_node_id: Option<PublicKey>, _msg: &msgs::ChannelUpdate,
-	) -> Result<bool, LightningError> {
-		Ok(false)
+	) -> Result<Option<(NodeId, NodeId)>, LightningError> {
+		Ok(None)
 	}
 	fn get_next_channel_announcement(
 		&self, _starting_point: u64,
@@ -611,6 +611,19 @@ where
 	pub send_only_message_handler: SM,
 }
 
+/// A gossip message to be forwarded to all peers.
+enum BroadcastGossipMessage {
+	ChannelAnnouncement(msgs::ChannelAnnouncement),
+	NodeAnnouncement(msgs::NodeAnnouncement),
+	ChannelUpdate {
+		msg: msgs::ChannelUpdate,
+		/// One of the two channel endpoints.
+		node_id_1: NodeId,
+		/// One of the two channel endpoints.
+		node_id_2: NodeId,
+	},
+}
+
 /// Provides an object which can be used to send data to and which uniquely identifies a connection
 /// to a remote host. You will need to be able to generate multiple of these which meet Eq and
 /// implement Hash to meet the PeerManager API.
@@ -2045,10 +2058,7 @@ where
 		message: wire::Message<
 			<<CMH as Deref>::Target as wire::CustomMessageReader>::CustomMessage,
 		>,
-	) -> Result<
-		Option<wire::Message<<<CMH as Deref>::Target as wire::CustomMessageReader>::CustomMessage>>,
-		MessageHandlingError,
-	> {
+	) -> Result<Option<BroadcastGossipMessage>, MessageHandlingError> {
 		let their_node_id = peer_lock
 			.their_node_id
 			.expect("We know the peer's public key by the time we receive messages")
@@ -2390,10 +2400,7 @@ where
 			<<CMH as Deref>::Target as wire::CustomMessageReader>::CustomMessage,
 		>,
 		their_node_id: PublicKey, logger: &WithContext<'a, L>,
-	) -> Result<
-		Option<wire::Message<<<CMH as Deref>::Target as wire::CustomMessageReader>::CustomMessage>>,
-		MessageHandlingError,
-	> {
+	) -> Result<Option<BroadcastGossipMessage>, MessageHandlingError> {
 		if is_gossip_msg(message.type_id()) {
 			log_gossip!(logger, "Received message {:?} from {}", message, their_node_id);
 		} else {
@@ -2575,7 +2582,7 @@ where
 					.handle_channel_announcement(Some(their_node_id), &msg)
 					.map_err(|e| -> MessageHandlingError { e.into() })?
 				{
-					should_forward = Some(wire::Message::ChannelAnnouncement(msg));
+					should_forward = Some(BroadcastGossipMessage::ChannelAnnouncement(msg));
 				}
 				self.update_gossip_backlogged();
 			},
@@ -2585,7 +2592,7 @@ where
 					.handle_node_announcement(Some(their_node_id), &msg)
 					.map_err(|e| -> MessageHandlingError { e.into() })?
 				{
-					should_forward = Some(wire::Message::NodeAnnouncement(msg));
+					should_forward = Some(BroadcastGossipMessage::NodeAnnouncement(msg));
 				}
 				self.update_gossip_backlogged();
 			},
@@ -2594,11 +2601,12 @@ where
 				chan_handler.handle_channel_update(their_node_id, &msg);
 
 				let route_handler = &self.message_handler.route_handler;
-				if route_handler
+				if let Some((node_id_1, node_id_2)) = route_handler
 					.handle_channel_update(Some(their_node_id), &msg)
 					.map_err(|e| -> MessageHandlingError { e.into() })?
 				{
-					should_forward = Some(wire::Message::ChannelUpdate(msg));
+					should_forward =
+						Some(BroadcastGossipMessage::ChannelUpdate { msg, node_id_1, node_id_2 });
 				}
 				self.update_gossip_backlogged();
 			},
@@ -2652,12 +2660,11 @@ where
 	/// unless `allow_large_buffer` is set, in which case the message will be treated as critical
 	/// and delivered no matter the available buffer space.
 	fn forward_broadcast_msg(
-		&self, peers: &HashMap<Descriptor, Mutex<Peer>>,
-		msg: &wire::Message<<<CMH as Deref>::Target as wire::CustomMessageReader>::CustomMessage>,
+		&self, peers: &HashMap<Descriptor, Mutex<Peer>>, msg: &BroadcastGossipMessage,
 		except_node: Option<&PublicKey>, allow_large_buffer: bool,
 	) {
 		match msg {
-			wire::Message::ChannelAnnouncement(ref msg) => {
+			BroadcastGossipMessage::ChannelAnnouncement(ref msg) => {
 				log_gossip!(self.logger, "Sending message to all peers except {:?} or the announced channel's counterparties: {:?}", except_node, msg);
 				let encoded_msg = encode_msg!(msg);
 
@@ -2696,7 +2703,7 @@ where
 					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
-			wire::Message::NodeAnnouncement(ref msg) => {
+			BroadcastGossipMessage::NodeAnnouncement(ref msg) => {
 				log_gossip!(
 					self.logger,
 					"Sending message to all peers except {:?} or the announced node: {:?}",
@@ -2738,7 +2745,7 @@ where
 					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
-			wire::Message::ChannelUpdate(ref msg) => {
+			BroadcastGossipMessage::ChannelUpdate { msg, node_id_1: _, node_id_2: _ } => {
 				log_gossip!(
 					self.logger,
 					"Sending message to all peers except {:?}: {:?}",
@@ -2775,9 +2782,6 @@ where
 					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
-			_ => {
-				debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages")
-			},
 		}
 	}
 
@@ -3129,13 +3133,15 @@ where
 						},
 						MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
 							log_debug!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+							let node_id_1 = msg.contents.node_id_1;
+							let node_id_2 = msg.contents.node_id_2;
 							match route_handler.handle_channel_announcement(None, &msg) {
 								Ok(_)
 								| Err(LightningError {
 									action: msgs::ErrorAction::IgnoreDuplicateGossip,
 									..
 								}) => {
-									let forward = wire::Message::ChannelAnnouncement(msg);
+									let forward = BroadcastGossipMessage::ChannelAnnouncement(msg);
 									self.forward_broadcast_msg(
 										peers,
 										&forward,
@@ -3152,7 +3158,11 @@ where
 										action: msgs::ErrorAction::IgnoreDuplicateGossip,
 										..
 									}) => {
-										let forward = wire::Message::ChannelUpdate(msg);
+										let forward = BroadcastGossipMessage::ChannelUpdate {
+											msg,
+											node_id_1,
+											node_id_2,
+										};
 										self.forward_broadcast_msg(
 											peers,
 											&forward,
@@ -3164,15 +3174,19 @@ where
 								}
 							}
 						},
-						MessageSendEvent::BroadcastChannelUpdate { msg, .. } => {
+						MessageSendEvent::BroadcastChannelUpdate { msg, node_id_1, node_id_2 } => {
 							log_debug!(self.logger, "Handling BroadcastChannelUpdate event in peer_handler for contents {:?}", msg.contents);
 							match route_handler.handle_channel_update(None, &msg) {
 								Ok(_)
 								| Err(LightningError {
 									action: msgs::ErrorAction::IgnoreDuplicateGossip,
 									..
 								}) => {
-									let forward = wire::Message::ChannelUpdate(msg);
+									let forward = BroadcastGossipMessage::ChannelUpdate {
+										msg,
+										node_id_1,
+										node_id_2,
+									};
 									self.forward_broadcast_msg(
 										peers,
 										&forward,
@@ -3191,7 +3205,7 @@ where
 									action: msgs::ErrorAction::IgnoreDuplicateGossip,
 									..
 								}) => {
-									let forward = wire::Message::NodeAnnouncement(msg);
+									let forward = BroadcastGossipMessage::NodeAnnouncement(msg);
 									self.forward_broadcast_msg(
 										peers,
 										&forward,
@@ -3668,7 +3682,7 @@ where
 		let _ = self.message_handler.route_handler.handle_node_announcement(None, &msg);
 		self.forward_broadcast_msg(
 			&*self.peers.read().unwrap(),
-			&wire::Message::NodeAnnouncement(msg),
+			&BroadcastGossipMessage::NodeAnnouncement(msg),
 			None,
 			true,
 		);

lightning/src/routing/gossip.rs

@@ -556,9 +556,12 @@ where
 
 	fn handle_channel_update(
 		&self, _their_node_id: Option<PublicKey>, msg: &msgs::ChannelUpdate,
-	) -> Result<bool, LightningError> {
-		self.network_graph.update_channel(msg)?;
-		Ok(msg.contents.excess_data.len() <= MAX_EXCESS_BYTES_FOR_RELAY)
+	) -> Result<Option<(NodeId, NodeId)>, LightningError> {
+		match self.network_graph.update_channel(msg) {
+			Ok(nodes) if msg.contents.excess_data.len() <= MAX_EXCESS_BYTES_FOR_RELAY => Ok(nodes),
+			Ok(_) => Ok(None),
+			Err(e) => Err(e),
+		}
 	}
 
 	fn get_next_channel_announcement(
@@ -2433,7 +2436,11 @@ where
 	///
 	/// If not built with `std`, any updates with a timestamp more than two weeks in the past or
 	/// materially in the future will be rejected.
-	pub fn update_channel(&self, msg: &msgs::ChannelUpdate) -> Result<(), LightningError> {
+	///
+	/// Returns the [`NodeId`]s of both sides of the channel if it was applied.
+	pub fn update_channel(
+		&self, msg: &msgs::ChannelUpdate,
+	) -> Result<Option<(NodeId, NodeId)>, LightningError> {
 		self.update_channel_internal(&msg.contents, Some(&msg), Some(&msg.signature), false)
 	}
 
@@ -2443,9 +2450,11 @@ where
 	///
 	/// If not built with `std`, any updates with a timestamp more than two weeks in the past or
 	/// materially in the future will be rejected.
+	///
+	/// Returns the [`NodeId`]s of both sides of the channel if it was applied.
 	pub fn update_channel_unsigned(
 		&self, msg: &msgs::UnsignedChannelUpdate,
-	) -> Result<(), LightningError> {
+	) -> Result<Option<(NodeId, NodeId)>, LightningError> {
 		self.update_channel_internal(msg, None, None, false)
 	}
 
@@ -2456,13 +2465,14 @@ where
 	/// If not built with `std`, any updates with a timestamp more than two weeks in the past or
 	/// materially in the future will be rejected.
 	pub fn verify_channel_update(&self, msg: &msgs::ChannelUpdate) -> Result<(), LightningError> {
-		self.update_channel_internal(&msg.contents, Some(&msg), Some(&msg.signature), true)
+		self.update_channel_internal(&msg.contents, Some(&msg), Some(&msg.signature), true)?;
+		Ok(())
 	}
 
 	fn update_channel_internal(
 		&self, msg: &msgs::UnsignedChannelUpdate, full_msg: Option<&msgs::ChannelUpdate>,
 		sig: Option<&secp256k1::ecdsa::Signature>, only_verify: bool,
-	) -> Result<(), LightningError> {
+	) -> Result<Option<(NodeId, NodeId)>, LightningError> {
 		let chan_enabled = msg.channel_flags & (1 << 1) != (1 << 1);
 
 		if msg.chain_hash != self.chain_hash {
@@ -2602,7 +2612,7 @@ where
 		}
 
 		if only_verify {
-			return Ok(());
+			return Ok(None);
 		}
 
 		let mut channels = self.channels.write().unwrap();
@@ -2633,9 +2643,11 @@ where
 			} else {
 				channel.one_to_two = new_channel_info;
 			}
-		}
 
-		Ok(())
+			Ok(Some((channel.node_one, channel.node_two)))
+		} else {
+			Ok(None)
+		}
 	}
 
 	fn remove_channel_in_nodes_callback<RM: FnMut(IndexedMapOccupiedEntry<NodeId, NodeInfo>)>(
@@ -3180,7 +3192,7 @@ pub(crate) mod tests {
 		let valid_channel_update = get_signed_channel_update(|_| {}, node_1_privkey, &secp_ctx);
 		network_graph.verify_channel_update(&valid_channel_update).unwrap();
 		match gossip_sync.handle_channel_update(Some(node_1_pubkey), &valid_channel_update) {
-			Ok(res) => assert!(res),
+			Ok(res) => assert!(res.is_some()),
 			_ => panic!(),
 		};
 
@@ -3202,9 +3214,9 @@ pub(crate) mod tests {
 			node_1_privkey,
 			&secp_ctx,
 		);
-		// Return false because contains excess data
+		// Update is accepted but won't be relayed because contains excess data
 		match gossip_sync.handle_channel_update(Some(node_1_pubkey), &valid_channel_update) {
-			Ok(res) => assert!(!res),
+			Ok(res) => assert!(res.is_none()),
 			_ => panic!(),
 		};
 

lightning/src/routing/test_utils.rs

@@ -111,7 +111,7 @@ pub(crate) fn update_channel(
 	};
 
 	match gossip_sync.handle_channel_update(Some(node_pubkey), &valid_channel_update) {
-		Ok(res) => assert!(res),
+		Ok(res) => assert!(res.is_some()),
 		Err(e) => panic!("{e:?}")
 	};
 }

lightning/src/util/test_utils.rs

@@ -1522,9 +1522,9 @@ impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
 	}
 	fn handle_channel_update(
 		&self, _their_node_id: Option<PublicKey>, _msg: &msgs::ChannelUpdate,
-	) -> Result<bool, msgs::LightningError> {
+	) -> Result<Option<(NodeId, NodeId)>, msgs::LightningError> {
 		self.chan_upds_recvd.fetch_add(1, Ordering::AcqRel);
-		Ok(true)
+		Ok(Some((NodeId::from_slice(&[2; 33]).unwrap(), NodeId::from_slice(&[3; 33]).unwrap())))
 	}
 	fn get_next_channel_announcement(
 		&self, starting_point: u64,