Handle-initial_routing_sync-requests-from-peers-in-their-Init-messages

Author: SWvheerden

src/ln/msgs.rs

@@ -333,7 +333,7 @@ pub struct AnnouncementSignatures {
 }
 
 /// An address which can be used to connect to a remote peer
-#[derive(Clone)]
+#[derive(Clone, PartialEq)]
 pub enum NetAddress {
 	/// An IPv4 address/port on which the peer is listenting.
 	IPv4 {
@@ -380,7 +380,7 @@ impl NetAddress {
 		}
 	}
 }
-
+#[derive(Clone, PartialEq)]
 // Only exposed as broadcast of node_announcement should be filtered by node_id
 /// The unsigned part of a node_announcement
 pub struct UnsignedNodeAnnouncement {
@@ -397,6 +397,7 @@ pub struct UnsignedNodeAnnouncement {
 	pub(crate) excess_address_data: Vec<u8>,
 	pub(crate) excess_data: Vec<u8>,
 }
+#[derive(Clone, PartialEq)]
 /// A node_announcement message to be sent or received from a peer
 pub struct NodeAnnouncement {
 	pub(crate) signature: Signature,
@@ -585,6 +586,9 @@ pub trait RoutingMessageHandler : Send + Sync {
 	fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result<bool, HandleError>;
 	/// Handle some updates to the route graph that we learned due to an outbound failed payment.
 	fn handle_htlc_fail_channel_update(&self, update: &HTLCFailChannelUpdate);
+	///Gets a subset of the messages required to dump our routing table to a remote node, starting at the short_channel_id indicated by starting_point and including batch_amount entries
+	/// This function will start iterating at 0 if the starting_point is < 0.
+	fn get_next_announcements(&self, starting_point: u64, batch_amount: u8)->(Vec<(ChannelAnnouncement, NodeAnnouncement, NodeAnnouncement,ChannelUpdate,ChannelUpdate)>, i64);
 }
 
 pub(crate) struct OnionRealm0HopData {

src/ln/peer_handler.rs

@@ -102,6 +102,7 @@ struct Peer {
 	pending_read_buffer: Vec<u8>,
 	pending_read_buffer_pos: usize,
 	pending_read_is_header: bool,
+	last_synced_channel : i64,
 }
 
 struct PeerHolder<Descriptor: SocketDescriptor> {
@@ -221,6 +222,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 			pending_read_buffer: pending_read_buffer,
 			pending_read_buffer_pos: 0,
 			pending_read_is_header: false,
+			last_synced_channel : -1,
 		}).is_some() {
 			panic!("PeerManager driver duplicated descriptors!");
 		};
@@ -255,22 +257,42 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 			pending_read_buffer: pending_read_buffer,
 			pending_read_buffer_pos: 0,
 			pending_read_is_header: false,
+			last_synced_channel : -1,
 		}).is_some() {
 			panic!("PeerManager driver duplicated descriptors!");
 		};
 		Ok(())
 	}
 
-	fn do_attempt_write_data(descriptor: &mut Descriptor, peer: &mut Peer) {
+	fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
+		macro_rules! encode_and_send_msg {
+			($msg: expr, $msg_code: expr) => {
+				{
+					log_trace!(self, "Encoding and sending message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
+					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
+				}
+			}
+		}
 		while !peer.awaiting_write_event {
 			if {
+				let should_be_reading = peer.pending_outbound_buffer.len() < 10;
+				if (peer.last_synced_channel > -1) &&(should_be_reading){
+					let all_messages_tuple = self.message_handler.route_handler.get_next_announcements(peer.last_synced_channel as u64,(10-peer.pending_outbound_buffer.len()) as u8);
+					for tuple in all_messages_tuple.0.iter(){
+						encode_and_send_msg!(tuple.0, 256);
+						encode_and_send_msg!(tuple.1, 257);
+						encode_and_send_msg!(tuple.2, 257);
+						encode_and_send_msg!(tuple.3, 258);
+						encode_and_send_msg!(tuple.4, 258);
+					}
+					peer.last_synced_channel = all_messages_tuple.1;
+				}
 				let next_buff = match peer.pending_outbound_buffer.front() {
 					None => return,
 					Some(buff) => buff,
 				};
-				let should_be_reading = peer.pending_outbound_buffer.len() < 10;
 
-				let data_sent = descriptor.send_data(next_buff, peer.pending_outbound_buffer_first_msg_offset, should_be_reading);
+				let data_sent = descriptor.send_data(&next_buff, peer.pending_outbound_buffer_first_msg_offset, should_be_reading);
 				peer.pending_outbound_buffer_first_msg_offset += data_sent;
 				if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() { true } else { false }
 			} {
@@ -297,7 +319,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 			None => panic!("Descriptor for write_event is not already known to PeerManager"),
 			Some(peer) => {
 				peer.awaiting_write_event = false;
-				Self::do_attempt_write_data(descriptor, peer);
+				self.do_attempt_write_data(descriptor, peer);
 			}
 		};
 		Ok(())
@@ -511,6 +533,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 													if msg.local_features.supports_unknown_bits() { "present" } else { "none" },
 													if msg.global_features.supports_unknown_bits() { "present" } else { "none" });
 
+												let do_they_require_sync = msg.local_features.initial_routing_sync();
 												peer.their_global_features = Some(msg.global_features);
 												peer.their_local_features = Some(msg.local_features);
 
@@ -520,6 +543,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 														self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
 														local_features.set_initial_routing_sync();
 													}
+
+													if do_they_require_sync {
+														peer.last_synced_channel = 0; //set to larger than -1, means next sent message will trigger slow trickle of sync data
+													}
 													encode_and_send_msg!(msgs::Init {
 														global_features: msgs::GlobalFeatures::new(),
 														local_features,
@@ -667,7 +694,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 						}
 					}
 
-					Self::do_attempt_write_data(peer_descriptor, peer);
+					self.do_attempt_write_data(peer_descriptor, peer);
 
 					peer.pending_outbound_buffer.len() > 10 // pause_read
 				}
@@ -724,7 +751,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 33)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
@@ -734,7 +761,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 32)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
@@ -746,7 +773,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//they should just throw away this funding transaction
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 34)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
@@ -757,7 +784,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//they should just throw away this funding transaction
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 35)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
@@ -767,7 +794,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Do whatever we're gonna do for handling dropped messages
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 36)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
@@ -778,7 +805,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//they should just throw away this funding transaction
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 259)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
 						log_trace!(self, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
@@ -806,7 +833,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 134)));
 						}
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed, 132)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
@@ -816,7 +843,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Do whatever we're gonna do for handling dropped messages
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 133)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
@@ -826,7 +853,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Do whatever we're gonna do for handling dropped messages
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 39)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
 						log_trace!(self, "Handling Shutdown event in peer_handler for node {} for channel {}",
@@ -836,7 +863,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Do whatever we're gonna do for handling dropped messages
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 38)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
 						log_trace!(self, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
@@ -846,7 +873,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								//TODO: Do whatever we're gonna do for handling dropped messages
 							});
 						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 136)));
-						Self::do_attempt_write_data(&mut descriptor, peer);
+						self.do_attempt_write_data(&mut descriptor, peer);
 					},
 					MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
 						log_trace!(self, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
@@ -868,7 +895,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 								}
 								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
 								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_update_msg[..]));
-								Self::do_attempt_write_data(&mut (*descriptor).clone(), peer);
+								self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
 							}
 						}
 					},
@@ -882,7 +909,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 									continue
 								}
 								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
-								Self::do_attempt_write_data(&mut (*descriptor).clone(), peer);
+								self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
 							}
 						}
 					},
@@ -903,7 +930,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 												peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
 												// This isn't guaranteed to work, but if there is enough free
 												// room in the send buffer, put the error message there...
-												Self::do_attempt_write_data(&mut descriptor, &mut peer);
+												self.do_attempt_write_data(&mut descriptor, &mut peer);
 											} else {
 												log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id));
 											}
@@ -921,7 +948,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 										//TODO: Do whatever we're gonna do for handling dropped messages
 									});
 									peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
-									Self::do_attempt_write_data(&mut descriptor, peer);
+									self.do_attempt_write_data(&mut descriptor, peer);
 								},
 							}
 						} else {
@@ -933,7 +960,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 
 			for mut descriptor in peers.peers_needing_send.drain() {
 				match peers.peers.get_mut(&descriptor) {
-					Some(peer) => Self::do_attempt_write_data(&mut descriptor, peer),
+					Some(peer) => self.do_attempt_write_data(&mut descriptor, peer),
 					None => panic!("Inconsistent peers set state!"),
 				}
 			}