Skip to content

Commit 8d7b498

Browse files
SWvheerdenTheBlueMatt
SWvheerden
authored and
TheBlueMatt
committed
Handle-initial_routing_sync-requests-from-peers-in-their-Init-messages
1 parent 74f59a2 commit 8d7b498

File tree

4 files changed

+248
-43
lines changed

4 files changed

+248
-43
lines changed

‎src/ln/msgs.rs

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -382,6 +382,7 @@ impl NetAddress {
382382
}
383383
}
384384

385+
#[derive(Clone)]
385386
// Only exposed as broadcast of node_announcement should be filtered by node_id
386387
/// The unsigned part of a node_announcement
387388
pub struct UnsignedNodeAnnouncement {
@@ -398,6 +399,7 @@ pub struct UnsignedNodeAnnouncement {
398399
pub(crate) excess_address_data: Vec<u8>,
399400
pub(crate) excess_data: Vec<u8>,
400401
}
402+
#[derive(Clone)]
401403
/// A node_announcement message to be sent or received from a peer
402404
pub struct NodeAnnouncement {
403405
pub(crate) signature: Signature,
@@ -588,6 +590,14 @@ pub trait RoutingMessageHandler : Send + Sync {
588590
fn handle_channel_update(&self, msg: &ChannelUpdate) -> Result<bool, HandleError>;
589591
/// Handle some updates to the route graph that we learned due to an outbound failed payment.
590592
fn handle_htlc_fail_channel_update(&self, update: &HTLCFailChannelUpdate);
593+
/// Gets a subset of the channel announcements and updates required to dump our routing table
594+
/// to a remote node, starting at the short_channel_id indicated by starting_point and
595+
/// including batch_amount entries.
596+
fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(ChannelAnnouncement, ChannelUpdate, ChannelUpdate)>;
597+
/// Gets a subset of the node announcements required to dump our routing table to a remote node,
598+
/// starting at the node *after* the provided publickey and including batch_amount entries.
599+
/// If None is provided for starting_point, we start at the first node.
600+
fn get_next_node_announcements(&self, starting_point: Option<&PublicKey>, batch_amount: u8) -> Vec<NodeAnnouncement>;
591601
}
592602

593603
pub(crate) struct OnionRealm0HopData {

‎src/ln/peer_handler.rs

Lines changed: 108 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,12 @@ impl error::Error for PeerHandleError {
8888
}
8989
}
9090

91+
enum InitSyncTracker{
92+
NoSyncRequested,
93+
ChannelsSyncing(u64),
94+
NodesSyncing(PublicKey),
95+
}
96+
9197
struct Peer {
9298
channel_encryptor: PeerChannelEncryptor,
9399
outbound: bool,
@@ -102,6 +108,24 @@ struct Peer {
102108
pending_read_buffer: Vec<u8>,
103109
pending_read_buffer_pos: usize,
104110
pending_read_is_header: bool,
111+
112+
sync_status: InitSyncTracker,
113+
}
114+
115+
impl Peer {
116+
/// Returns true if the the channel announcements/updates for the given channel should be
117+
/// forwarded to this peer.
118+
/// If we are sending our routing table to this peer and we have not yet sent channel
119+
/// announcements/updates for the given channel_id then we will send it when we get to that
120+
/// point and we shouldn't send it yet to avoid sending duplicate updates. If we've already
121+
/// sent the old versions, we should send the update, and so return true here.
122+
fn should_forward_channel(&self, channel_id: u64)->bool{
123+
match self.sync_status {
124+
InitSyncTracker::NoSyncRequested => true,
125+
InitSyncTracker::ChannelsSyncing(i) => i < channel_id,
126+
InitSyncTracker::NodesSyncing(_) => true,
127+
}
128+
}
105129
}
106130

107131
struct PeerHolder<Descriptor: SocketDescriptor> {
@@ -221,6 +245,8 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
221245
pending_read_buffer: pending_read_buffer,
222246
pending_read_buffer_pos: 0,
223247
pending_read_is_header: false,
248+
249+
sync_status: InitSyncTracker::NoSyncRequested,
224250
}).is_some() {
225251
panic!("PeerManager driver duplicated descriptors!");
226252
};
@@ -255,21 +281,74 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
255281
pending_read_buffer: pending_read_buffer,
256282
pending_read_buffer_pos: 0,
257283
pending_read_is_header: false,
284+
285+
sync_status: InitSyncTracker::NoSyncRequested,
258286
}).is_some() {
259287
panic!("PeerManager driver duplicated descriptors!");
260288
};
261289
Ok(())
262290
}
263291

264-
fn do_attempt_write_data(descriptor: &mut Descriptor, peer: &mut Peer) {
292+
fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
293+
macro_rules! encode_and_send_msg {
294+
($msg: expr, $msg_code: expr) => {
295+
{
296+
log_trace!(self, "Encoding and sending sync update message of type {} to {}", $msg_code, log_pubkey!(peer.their_node_id.unwrap()));
297+
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
298+
}
299+
}
300+
}
301+
const MSG_BUFF_SIZE: usize = 10;
265302
while !peer.awaiting_write_event {
303+
if peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE {
304+
match peer.sync_status {
305+
InitSyncTracker::NoSyncRequested => {},
306+
InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => {
307+
let steps = ((MSG_BUFF_SIZE - peer.pending_outbound_buffer.len() + 2) / 3) as u8;
308+
let all_messages = self.message_handler.route_handler.get_next_channel_announcements(0, steps);
309+
for &(ref announce, ref update_a, ref update_b) in all_messages.iter() {
310+
encode_and_send_msg!(announce, 256);
311+
encode_and_send_msg!(update_a, 258);
312+
encode_and_send_msg!(update_b, 258);
313+
peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
314+
}
315+
if all_messages.is_empty() || all_messages.len() != steps as usize {
316+
peer.sync_status = InitSyncTracker::ChannelsSyncing(0xffff_ffff_ffff_ffff);
317+
}
318+
},
319+
InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => {
320+
let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
321+
let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps);
322+
for msg in all_messages.iter() {
323+
encode_and_send_msg!(msg, 256);
324+
peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
325+
}
326+
if all_messages.is_empty() || all_messages.len() != steps as usize {
327+
peer.sync_status = InitSyncTracker::NoSyncRequested;
328+
}
329+
},
330+
InitSyncTracker::ChannelsSyncing(_) => unreachable!(),
331+
InitSyncTracker::NodesSyncing(key) => {
332+
let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8;
333+
let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps);
334+
for msg in all_messages.iter() {
335+
encode_and_send_msg!(msg, 256);
336+
peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
337+
}
338+
if all_messages.is_empty() || all_messages.len() != steps as usize {
339+
peer.sync_status = InitSyncTracker::NoSyncRequested;
340+
}
341+
},
342+
}
343+
}
344+
266345
if {
267346
let next_buff = match peer.pending_outbound_buffer.front() {
268347
None => return,
269348
Some(buff) => buff,
270349
};
271-
let should_be_reading = peer.pending_outbound_buffer.len() < 10;
272350

351+
let should_be_reading = peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE;
273352
let data_sent = descriptor.send_data(next_buff, peer.pending_outbound_buffer_first_msg_offset, should_be_reading);
274353
peer.pending_outbound_buffer_first_msg_offset += data_sent;
275354
if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() { true } else { false }
@@ -297,7 +376,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
297376
None => panic!("Descriptor for write_event is not already known to PeerManager"),
298377
Some(peer) => {
299378
peer.awaiting_write_event = false;
300-
Self::do_attempt_write_data(descriptor, peer);
379+
self.do_attempt_write_data(descriptor, peer);
301380
}
302381
};
303382
Ok(())
@@ -522,6 +601,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
522601
if msg.local_features.supports_unknown_bits() { "present" } else { "none" },
523602
if msg.global_features.supports_unknown_bits() { "present" } else { "none" });
524603

604+
if msg.local_features.initial_routing_sync() {
605+
peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
606+
peers.peers_needing_send.insert(peer_descriptor.clone());
607+
}
525608
peer.their_global_features = Some(msg.global_features);
526609
peer.their_local_features = Some(msg.local_features);
527610

@@ -531,6 +614,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
531614
self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel);
532615
local_features.set_initial_routing_sync();
533616
}
617+
534618
encode_and_send_msg!(msgs::Init {
535619
global_features: msgs::GlobalFeatures::new(),
536620
local_features,
@@ -678,7 +762,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
678762
}
679763
}
680764

681-
Self::do_attempt_write_data(peer_descriptor, peer);
765+
self.do_attempt_write_data(peer_descriptor, peer);
682766

683767
peer.pending_outbound_buffer.len() > 10 // pause_read
684768
}
@@ -735,7 +819,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
735819
//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
736820
});
737821
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 33)));
738-
Self::do_attempt_write_data(&mut descriptor, peer);
822+
self.do_attempt_write_data(&mut descriptor, peer);
739823
},
740824
MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
741825
log_trace!(self, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
@@ -745,7 +829,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
745829
//TODO: Drop the pending channel? (or just let it timeout, but that sucks)
746830
});
747831
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 32)));
748-
Self::do_attempt_write_data(&mut descriptor, peer);
832+
self.do_attempt_write_data(&mut descriptor, peer);
749833
},
750834
MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
751835
log_trace!(self, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
@@ -757,7 +841,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
757841
//they should just throw away this funding transaction
758842
});
759843
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 34)));
760-
Self::do_attempt_write_data(&mut descriptor, peer);
844+
self.do_attempt_write_data(&mut descriptor, peer);
761845
},
762846
MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
763847
log_trace!(self, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
@@ -768,7 +852,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
768852
//they should just throw away this funding transaction
769853
});
770854
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 35)));
771-
Self::do_attempt_write_data(&mut descriptor, peer);
855+
self.do_attempt_write_data(&mut descriptor, peer);
772856
},
773857
MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
774858
log_trace!(self, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
@@ -778,7 +862,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
778862
//TODO: Do whatever we're gonna do for handling dropped messages
779863
});
780864
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 36)));
781-
Self::do_attempt_write_data(&mut descriptor, peer);
865+
self.do_attempt_write_data(&mut descriptor, peer);
782866
},
783867
MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
784868
log_trace!(self, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
@@ -789,7 +873,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
789873
//they should just throw away this funding transaction
790874
});
791875
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 259)));
792-
Self::do_attempt_write_data(&mut descriptor, peer);
876+
self.do_attempt_write_data(&mut descriptor, peer);
793877
},
794878
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 } } => {
795879
log_trace!(self, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
@@ -817,7 +901,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
817901
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 134)));
818902
}
819903
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed, 132)));
820-
Self::do_attempt_write_data(&mut descriptor, peer);
904+
self.do_attempt_write_data(&mut descriptor, peer);
821905
},
822906
MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
823907
log_trace!(self, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
@@ -827,7 +911,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
827911
//TODO: Do whatever we're gonna do for handling dropped messages
828912
});
829913
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 133)));
830-
Self::do_attempt_write_data(&mut descriptor, peer);
914+
self.do_attempt_write_data(&mut descriptor, peer);
831915
},
832916
MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
833917
log_trace!(self, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
@@ -837,7 +921,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
837921
//TODO: Do whatever we're gonna do for handling dropped messages
838922
});
839923
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 39)));
840-
Self::do_attempt_write_data(&mut descriptor, peer);
924+
self.do_attempt_write_data(&mut descriptor, peer);
841925
},
842926
MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
843927
log_trace!(self, "Handling Shutdown event in peer_handler for node {} for channel {}",
@@ -847,7 +931,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
847931
//TODO: Do whatever we're gonna do for handling dropped messages
848932
});
849933
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 38)));
850-
Self::do_attempt_write_data(&mut descriptor, peer);
934+
self.do_attempt_write_data(&mut descriptor, peer);
851935
},
852936
MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
853937
log_trace!(self, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
@@ -857,7 +941,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
857941
//TODO: Do whatever we're gonna do for handling dropped messages
858942
});
859943
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 136)));
860-
Self::do_attempt_write_data(&mut descriptor, peer);
944+
self.do_attempt_write_data(&mut descriptor, peer);
861945
},
862946
MessageSendEvent::BroadcastChannelAnnouncement { ref msg, ref update_msg } => {
863947
log_trace!(self, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
@@ -866,7 +950,8 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
866950
let encoded_update_msg = encode_msg!(update_msg, 258);
867951

868952
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
869-
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() {
953+
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() ||
954+
!peer.should_forward_channel(msg.contents.short_channel_id) {
870955
continue
871956
}
872957
match peer.their_node_id {
@@ -879,7 +964,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
879964
}
880965
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
881966
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_update_msg[..]));
882-
Self::do_attempt_write_data(&mut (*descriptor).clone(), peer);
967+
self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
883968
}
884969
}
885970
},
@@ -889,11 +974,12 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
889974
let encoded_msg = encode_msg!(msg, 258);
890975

891976
for (ref descriptor, ref mut peer) in peers.peers.iter_mut() {
892-
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() {
977+
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() ||
978+
!peer.should_forward_channel(msg.contents.short_channel_id) {
893979
continue
894980
}
895981
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
896-
Self::do_attempt_write_data(&mut (*descriptor).clone(), peer);
982+
self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
897983
}
898984
}
899985
},
@@ -914,7 +1000,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
9141000
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
9151001
// This isn't guaranteed to work, but if there is enough free
9161002
// room in the send buffer, put the error message there...
917-
Self::do_attempt_write_data(&mut descriptor, &mut peer);
1003+
self.do_attempt_write_data(&mut descriptor, &mut peer);
9181004
} else {
9191005
log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id));
9201006
}
@@ -932,7 +1018,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
9321018
//TODO: Do whatever we're gonna do for handling dropped messages
9331019
});
9341020
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
935-
Self::do_attempt_write_data(&mut descriptor, peer);
1021+
self.do_attempt_write_data(&mut descriptor, peer);
9361022
},
9371023
}
9381024
} else {
@@ -944,7 +1030,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
9441030

9451031
for mut descriptor in peers.peers_needing_send.drain() {
9461032
match peers.peers.get_mut(&descriptor) {
947-
Some(peer) => Self::do_attempt_write_data(&mut descriptor, peer),
1033+
Some(peer) => self.do_attempt_write_data(&mut descriptor, peer),
9481034
None => panic!("Inconsistent peers set state!"),
9491035
}
9501036
}

0 commit comments

Comments
 (0)