Merge pull request #1368 from TheBlueMatt/2022-03-fix-post-start-sync

Send a gossip_timestamp_filter on connect to enable gossip sync

Author: TheBlueMatt

lightning-net-tokio/src/lib.rs

@@ -496,7 +496,7 @@ mod tests {
 		fn handle_channel_update(&self, _msg: &ChannelUpdate) -> Result<bool, LightningError> { Ok(false) }
 		fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) -> Vec<(ChannelAnnouncement, Option<ChannelUpdate>, Option<ChannelUpdate>)> { Vec::new() }
 		fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<NodeAnnouncement> { Vec::new() }
-		fn sync_routing_table(&self, _their_node_id: &PublicKey, _init_msg: &Init) { }
+		fn peer_connected(&self, _their_node_id: &PublicKey, _init_msg: &Init) { }
 		fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
 		fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
 		fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: QueryChannelRange) -> Result<(), LightningError> { Ok(()) }

lightning/src/ln/channelmanager.rs

@@ -5899,6 +5899,7 @@ impl<Signer: Sign, M: Deref , T: Deref , K: Deref , F: Deref , L: Deref >
 					&events::MessageSendEvent::SendChannelRangeQuery { .. } => false,
 					&events::MessageSendEvent::SendShortIdsQuery { .. } => false,
 					&events::MessageSendEvent::SendReplyChannelRange { .. } => false,
+					&events::MessageSendEvent::SendGossipTimestampFilter { .. } => false,
 				}
 			});
 		}

lightning/src/ln/msgs.rs

@@ -886,7 +886,7 @@ pub trait RoutingMessageHandler : MessageSendEventsProvider {
 	/// Called when a connection is established with a peer. This can be used to
 	/// perform routing table synchronization using a strategy defined by the
 	/// implementor.
-	fn sync_routing_table(&self, their_node_id: &PublicKey, init: &Init);
+	fn peer_connected(&self, their_node_id: &PublicKey, init: &Init);
 	/// Handles the reply of a query we initiated to learn about channels
 	/// for a given range of blocks. We can expect to receive one or more
 	/// replies to a single query.

lightning/src/ln/peer_handler.rs

@@ -69,7 +69,7 @@ impl RoutingMessageHandler for IgnoringMessageHandler {
 	fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) ->
 		Vec<(msgs::ChannelAnnouncement, Option<msgs::ChannelUpdate>, Option<msgs::ChannelUpdate>)> { Vec::new() }
 	fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<msgs::NodeAnnouncement> { Vec::new() }
-	fn sync_routing_table(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {}
+	fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {}
 	fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyChannelRange) -> Result<(), LightningError> { Ok(()) }
 	fn handle_reply_short_channel_ids_end(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyShortChannelIdsEnd) -> Result<(), LightningError> { Ok(()) }
 	fn handle_query_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::QueryChannelRange) -> Result<(), LightningError> { Ok(()) }
@@ -1018,7 +1018,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 					return Err(PeerHandleError{ no_connection_possible: true }.into());
 				}
 
-				self.message_handler.route_handler.sync_routing_table(&peer.their_node_id.unwrap(), &msg);
+				self.message_handler.route_handler.peer_connected(&peer.their_node_id.unwrap(), &msg);
 
 				self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap(), &msg);
 				peer.their_features = Some(msg.features);
@@ -1477,6 +1477,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> P
 							msg.sync_complete);
 						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					}
+					MessageSendEvent::SendGossipTimestampFilter { ref node_id, ref msg } => {
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
+					}
 				}
 			}
 

lightning/src/routing/network_graph.rs

@@ -25,7 +25,7 @@ use chain;
 use chain::Access;
 use ln::features::{ChannelFeatures, NodeFeatures};
 use ln::msgs::{DecodeError, ErrorAction, Init, LightningError, RoutingMessageHandler, NetAddress, MAX_VALUE_MSAT};
-use ln::msgs::{ChannelAnnouncement, ChannelUpdate, NodeAnnouncement, OptionalField};
+use ln::msgs::{ChannelAnnouncement, ChannelUpdate, NodeAnnouncement, OptionalField, GossipTimestampFilter};
 use ln::msgs::{QueryChannelRange, ReplyChannelRange, QueryShortChannelIds, ReplyShortChannelIdsEnd};
 use ln::msgs;
 use util::ser::{Writeable, Readable, Writer};
@@ -395,13 +395,28 @@ where C::Target: chain::Access, L::Target: Logger
 	/// to request gossip messages for each channel. The sync is considered complete
 	/// when the final reply_scids_end message is received, though we are not
 	/// tracking this directly.
-	fn sync_routing_table(&self, their_node_id: &PublicKey, init_msg: &Init) {
-
+	fn peer_connected(&self, their_node_id: &PublicKey, init_msg: &Init) {
 		// We will only perform a sync with peers that support gossip_queries.
 		if !init_msg.features.supports_gossip_queries() {
 			return ();
 		}
 
+		// Send a gossip_timestamp_filter to enable gossip message receipt. Note that we have to
+		// use a "all timestamps" filter as sending the current timestamp would result in missing
+		// gossip messages that are simply sent late. We could calculate the intended filter time
+		// by looking at the current time and subtracting two weeks (before which we'll reject
+		// messages), but there's not a lot of reason to bother - our peers should be discarding
+		// the same messages.
+		let mut pending_events = self.pending_events.lock().unwrap();
+		pending_events.push(MessageSendEvent::SendGossipTimestampFilter {
+			node_id: their_node_id.clone(),
+			msg: GossipTimestampFilter {
+				chain_hash: self.network_graph.genesis_hash,
+				first_timestamp: 0,
+				timestamp_range: u32::max_value(),
+			},
+		});
+
 		// Check if we need to perform a full synchronization with this peer
 		if !self.should_request_full_sync(&their_node_id) {
 			return ();
@@ -410,7 +425,6 @@ where C::Target: chain::Access, L::Target: Logger
 		let first_blocknum = 0;
 		let number_of_blocks = 0xffffffff;
 		log_debug!(self.logger, "Sending query_channel_range peer={}, first_blocknum={}, number_of_blocks={}", log_pubkey!(their_node_id), first_blocknum, number_of_blocks);
-		let mut pending_events = self.pending_events.lock().unwrap();
 		pending_events.push(MessageSendEvent::SendChannelRangeQuery {
 			node_id: their_node_id.clone(),
 			msg: QueryChannelRange {
@@ -2271,18 +2285,27 @@ mod tests {
 		// It should ignore if gossip_queries feature is not enabled
 		{
 			let init_msg = Init { features: InitFeatures::known().clear_gossip_queries() };
-			net_graph_msg_handler.sync_routing_table(&node_id_1, &init_msg);
+			net_graph_msg_handler.peer_connected(&node_id_1, &init_msg);
 			let events = net_graph_msg_handler.get_and_clear_pending_msg_events();
 			assert_eq!(events.len(), 0);
 		}
 
 		// It should send a query_channel_message with the correct information
 		{
 			let init_msg = Init { features: InitFeatures::known() };
-			net_graph_msg_handler.sync_routing_table(&node_id_1, &init_msg);
+			net_graph_msg_handler.peer_connected(&node_id_1, &init_msg);
 			let events = net_graph_msg_handler.get_and_clear_pending_msg_events();
-			assert_eq!(events.len(), 1);
+			assert_eq!(events.len(), 2);
 			match &events[0] {
+				MessageSendEvent::SendGossipTimestampFilter{ node_id, msg } => {
+					assert_eq!(node_id, &node_id_1);
+					assert_eq!(msg.chain_hash, chain_hash);
+					assert_eq!(msg.first_timestamp, 0);
+					assert_eq!(msg.timestamp_range, u32::max_value());
+				},
+				_ => panic!("Expected MessageSendEvent::SendChannelRangeQuery")
+			};
+			match &events[1] {
 				MessageSendEvent::SendChannelRangeQuery{ node_id, msg } => {
 					assert_eq!(node_id, &node_id_1);
 					assert_eq!(msg.chain_hash, chain_hash);
@@ -2303,12 +2326,14 @@ mod tests {
 			for n in 1..7 {
 				let node_privkey = &SecretKey::from_slice(&[n; 32]).unwrap();
 				let node_id = PublicKey::from_secret_key(&secp_ctx, node_privkey);
-				net_graph_msg_handler.sync_routing_table(&node_id, &init_msg);
+				net_graph_msg_handler.peer_connected(&node_id, &init_msg);
 				let events = net_graph_msg_handler.get_and_clear_pending_msg_events();
 				if n <= 5 {
-					assert_eq!(events.len(), 1);
+					assert_eq!(events.len(), 2);
 				} else {
-					assert_eq!(events.len(), 0);
+					// Even after the we stop sending the explicit query, we should still send a
+					// gossip_timestamp_filter on each new connection.
+					assert_eq!(events.len(), 1);
 				}
 
 			}

lightning/src/util/events.rs

@@ -909,7 +909,15 @@ pub enum MessageSendEvent {
 		node_id: PublicKey,
 		/// The reply_channel_range which should be sent.
 		msg: msgs::ReplyChannelRange,
-	}
+	},
+	/// Sends a timestamp filter for inbound gossip. This should be sent on each new connection to
+	/// enable receiving gossip messages from the peer.
+	SendGossipTimestampFilter {
+		/// The node_id of this message recipient
+		node_id: PublicKey,
+		/// The gossip_timestamp_filter which should be sent.
+		msg: msgs::GossipTimestampFilter,
+	},
 }
 
 /// A trait indicating an object may generate message send events

lightning/src/util/test_utils.rs

@@ -384,7 +384,7 @@ impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
 		Vec::new()
 	}
 
-	fn sync_routing_table(&self, _their_node_id: &PublicKey, _init_msg: &msgs::Init) {}
+	fn peer_connected(&self, _their_node_id: &PublicKey, _init_msg: &msgs::Init) {}
 
 	fn handle_reply_channel_range(&self, _their_node_id: &PublicKey, _msg: msgs::ReplyChannelRange) -> Result<(), msgs::LightningError> {
 		Ok(())