Always forward gossip for all our channels to all our peers

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.

Here we finally forward all gossip that is about either side of any
of our channels to all our peers, irrespective of their requested
gossip filtering.

Author: TheBlueMatt

lightning/src/ln/peer_handler.rs

@@ -1114,6 +1114,7 @@ pub struct PeerManager<
 	gossip_processing_backlog_lifted: AtomicBool,
 
 	node_signer: NS,
+	our_node_id: NodeId,
 
 	logger: L,
 	secp_ctx: Secp256k1<secp256k1::SignOnly>,
@@ -1328,6 +1329,9 @@ where
 		let ephemeral_hash = Sha256::from_engine(ephemeral_key_midstate.clone()).to_byte_array();
 		secp_ctx.seeded_randomize(&ephemeral_hash);
 
+		let our_node_pubkey =
+			node_signer.get_node_id(Recipient::Node).expect("node_id must be available");
+
 		PeerManager {
 			message_handler,
 			peers: FairRwLock::new(new_hash_map()),
@@ -1339,6 +1343,7 @@ where
 			gossip_processing_backlog_lifted: AtomicBool::new(false),
 			last_node_announcement_serial: AtomicU32::new(current_time),
 			logger,
+			our_node_id: NodeId::from_pubkey(&our_node_pubkey),
 			node_signer,
 			secp_ctx,
 		}
@@ -2667,12 +2672,16 @@ where
 			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);
+				let our_channel = self.our_node_id == msg.contents.node_id_1
+					|| self.our_node_id == msg.contents.node_id_2;
 
 				for (_, peer_mutex) in peers.iter() {
 					let mut peer = peer_mutex.lock().unwrap();
-					if !peer.handshake_complete()
-						|| !peer.should_forward_channel_announcement(msg.contents.short_channel_id)
-					{
+					if !peer.handshake_complete() {
+						continue;
+					}
+					let scid = msg.contents.short_channel_id;
+					if !our_channel && !peer.should_forward_channel_announcement(scid) {
 						continue;
 					}
 					debug_assert!(peer.their_node_id.is_some());
@@ -2711,12 +2720,15 @@ where
 					msg
 				);
 				let encoded_msg = encode_msg!(msg);
+				let our_announcement = self.our_node_id == msg.contents.node_id;
 
 				for (_, peer_mutex) in peers.iter() {
 					let mut peer = peer_mutex.lock().unwrap();
-					if !peer.handshake_complete()
-						|| !peer.should_forward_node_announcement(msg.contents.node_id)
-					{
+					if !peer.handshake_complete() {
+						continue;
+					}
+					let node_id = msg.contents.node_id;
+					if !our_announcement && !peer.should_forward_node_announcement(node_id) {
 						continue;
 					}
 					debug_assert!(peer.their_node_id.is_some());
@@ -2745,20 +2757,23 @@ where
 					peer.gossip_broadcast_buffer.push_back(encoded_message);
 				}
 			},
-			BroadcastGossipMessage::ChannelUpdate { msg, node_id_1: _, node_id_2: _ } => {
+			BroadcastGossipMessage::ChannelUpdate { msg, node_id_1, node_id_2 } => {
 				log_gossip!(
 					self.logger,
 					"Sending message to all peers except {:?}: {:?}",
 					except_node,
 					msg
 				);
 				let encoded_msg = encode_msg!(msg);
+				let our_channel = self.our_node_id == *node_id_1 || self.our_node_id == *node_id_2;
 
 				for (_, peer_mutex) in peers.iter() {
 					let mut peer = peer_mutex.lock().unwrap();
-					if !peer.handshake_complete()
-						|| !peer.should_forward_channel_announcement(msg.contents.short_channel_id)
-					{
+					if !peer.handshake_complete() {
+						continue;
+					}
+					let scid = msg.contents.short_channel_id;
+					if !our_channel && !peer.should_forward_channel_announcement(scid) {
 						continue;
 					}
 					debug_assert!(peer.their_node_id.is_some());
@@ -4890,6 +4905,90 @@ mod tests {
 		assert_eq!(filter_addresses(None), None);
 	}
 
+	#[test]
+	fn test_forward_gossip_for_our_channels_ignores_peer_filter() {
+		// Tests that gossip for channels where we are one of the endpoints is forwarded to all
+		// peers, regardless of any gossip filters they may have set. This ensures that updates
+		// for our own channels always propagate to all connected peers.
+
+		let cfgs = create_peermgr_cfgs(2);
+		let peers = create_network(2, &cfgs);
+
+		let id_0 = peers[0].node_signer.get_node_id(Recipient::Node).unwrap();
+		let addr_0 = SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 1000 };
+		let addr_1 = SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 1001 };
+
+		let mut fd_0_1 = FileDescriptor::new(1);
+		let mut fd_1_0 = FileDescriptor::new(2);
+
+		// Connect the peers and exchange the initial connection handshake.
+		let initial_data =
+			peers[1].new_outbound_connection(id_0, fd_1_0.clone(), Some(addr_0)).unwrap();
+		peers[0].new_inbound_connection(fd_0_1.clone(), Some(addr_1)).unwrap();
+		peers[0].read_event(&mut fd_0_1, &initial_data).unwrap();
+
+		peers[0].process_events();
+		let data_0_1 = fd_0_1.outbound_data.lock().unwrap().split_off(0);
+		peers[1].read_event(&mut fd_1_0, &data_0_1).unwrap();
+
+		peers[1].process_events();
+		let data_1_0 = fd_1_0.outbound_data.lock().unwrap().split_off(0);
+		peers[0].read_event(&mut fd_0_1, &data_1_0).unwrap();
+
+		peers[0].process_events();
+		let data_0_1 = fd_0_1.outbound_data.lock().unwrap().split_off(0);
+		peers[1].read_event(&mut fd_1_0, &data_0_1).unwrap();
+
+		// Once peer 1 receives the Init message in the last read_event, it'll generate a
+		// `GossipTimestampFilter` which will request gossip. Instead we drop it here.
+		cfgs[1]
+			.routing_handler
+			.pending_events
+			.lock()
+			.unwrap()
+			.retain(|ev| !matches!(ev, MessageSendEvent::SendGossipTimestampFilter { .. }));
+
+		peers[1].process_events();
+		let data_1_0 = fd_1_0.outbound_data.lock().unwrap().split_off(0);
+		peers[0].read_event(&mut fd_0_1, &data_1_0).unwrap();
+
+		peers[0].process_events();
+		assert!(fd_0_1.outbound_data.lock().unwrap().is_empty());
+		assert!(fd_1_0.outbound_data.lock().unwrap().is_empty());
+
+		let mut check_message_received = |expected_received: bool| {
+			let initial_count = cfgs[1].routing_handler.chan_upds_recvd.load(Ordering::Acquire);
+
+			peers[0].process_events();
+			let data_0_1 = fd_0_1.outbound_data.lock().unwrap().split_off(0);
+			assert_eq!(data_0_1.is_empty(), !expected_received);
+			peers[1].read_event(&mut fd_1_0, &data_0_1).unwrap();
+
+			let final_count = cfgs[1].routing_handler.chan_upds_recvd.load(Ordering::Acquire);
+			assert_eq!(final_count > initial_count, expected_received);
+		};
+
+		// Broadcast a gossip message that is unrelated to us and check that it doesn't get relayed
+		let unrelated_msg_ev = MessageSendEvent::BroadcastChannelUpdate {
+			msg: test_utils::get_dummy_channel_update(43),
+			node_id_1: NodeId::from_slice(&[2; 33]).unwrap(),
+			node_id_2: NodeId::from_slice(&[3; 33]).unwrap(),
+		};
+		cfgs[0].routing_handler.pending_events.lock().unwrap().push(unrelated_msg_ev);
+
+		check_message_received(false);
+
+		// Broadcast a gossip message that we're a party to and check that its relayed
+		let our_channel_msg_ev = MessageSendEvent::BroadcastChannelUpdate {
+			msg: test_utils::get_dummy_channel_update(43),
+			node_id_1: NodeId::from_pubkey(&id_0),
+			node_id_2: NodeId::from_slice(&[3; 33]).unwrap(),
+		};
+		cfgs[0].routing_handler.pending_events.lock().unwrap().push(our_channel_msg_ev);
+
+		check_message_received(true);
+	}
+
 	#[test]
 	#[cfg(feature = "std")]
 	fn test_process_events_multithreaded() {