X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=a9784327cba1a96df1a3194be92741a48b5b2164;hb=0052b2c5c3017aec1a80b6476fad441b63a67a8c;hp=5f1a8dc4034174830d851e4a4867db36f7241586;hpb=812f255d7c53c92656493fe103ad4129d1363a90;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index 5f1a8dc4..a9784327 100644 --- a/src/ln/peer_handler.rs +++ b/src/ln/peer_handler.rs @@ -15,7 +15,7 @@ use util::byte_utils; use util::events::{MessageSendEvent}; use util::logger::Logger; -use std::collections::{HashMap,hash_map,LinkedList}; +use std::collections::{HashMap,hash_map,HashSet,LinkedList}; use std::sync::{Arc, Mutex}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::{cmp,error,hash,fmt}; @@ -88,6 +88,12 @@ impl error::Error for PeerHandleError { } } +enum InitSyncTracker{ + NoSyncRequested, + ChannelsSyncing(u64), + NodesSyncing(PublicKey), +} + struct Peer { channel_encryptor: PeerChannelEncryptor, outbound: bool, @@ -102,21 +108,44 @@ struct Peer { pending_read_buffer: Vec, pending_read_buffer_pos: usize, pending_read_is_header: bool, + + sync_status: InitSyncTracker, +} + +impl Peer { + /// Returns true if the channel announcements/updates for the given channel should be + /// forwarded to this peer. + /// If we are sending our routing table to this peer and we have not yet sent channel + /// announcements/updates for the given channel_id then we will send it when we get to that + /// point and we shouldn't send it yet to avoid sending duplicate updates. If we've already + /// sent the old versions, we should send the update, and so return true here. + fn should_forward_channel(&self, channel_id: u64)->bool{ + match self.sync_status { + InitSyncTracker::NoSyncRequested => true, + InitSyncTracker::ChannelsSyncing(i) => i < channel_id, + InitSyncTracker::NodesSyncing(_) => true, + } + } } struct PeerHolder { peers: HashMap, + /// Added to by do_read_event for cases where we pushed a message onto the send buffer but + /// didn't call do_attempt_write_data to avoid reentrancy. Cleared in process_events() + peers_needing_send: HashSet, /// Only add to this set when noise completes: node_id_to_descriptor: HashMap, } struct MutPeerHolder<'a, Descriptor: SocketDescriptor + 'a> { peers: &'a mut HashMap, + peers_needing_send: &'a mut HashSet, node_id_to_descriptor: &'a mut HashMap, } impl PeerHolder { fn borrow_parts(&mut self) -> MutPeerHolder { MutPeerHolder { peers: &mut self.peers, + peers_needing_send: &mut self.peers_needing_send, node_id_to_descriptor: &mut self.node_id_to_descriptor, } } @@ -162,7 +191,11 @@ impl PeerManager { pub fn new(message_handler: MessageHandler, our_node_secret: SecretKey, logger: Arc) -> PeerManager { PeerManager { message_handler: message_handler, - peers: Mutex::new(PeerHolder { peers: HashMap::new(), node_id_to_descriptor: HashMap::new() }), + peers: Mutex::new(PeerHolder { + peers: HashMap::new(), + peers_needing_send: HashSet::new(), + node_id_to_descriptor: HashMap::new() + }), our_node_secret: our_node_secret, initial_syncs_sent: AtomicUsize::new(0), logger, @@ -188,7 +221,7 @@ impl PeerManager { /// Note that if an Err is returned here you MUST NOT call disconnect_event for the new /// descriptor but must disconnect the connection immediately. /// - /// Returns some bytes to send to the remote node. + /// Returns a small number of bytes to send to the remote node (currently always 50). /// /// Panics if descriptor is duplicative with some other descriptor which has not yet has a /// disconnect_event. @@ -201,7 +234,7 @@ impl PeerManager { if peers.peers.insert(descriptor, Peer { channel_encryptor: peer_encryptor, outbound: true, - their_node_id: Some(their_node_id), + their_node_id: None, their_global_features: None, their_local_features: None, @@ -212,6 +245,8 @@ impl PeerManager { pending_read_buffer: pending_read_buffer, pending_read_buffer_pos: 0, pending_read_is_header: false, + + sync_status: InitSyncTracker::NoSyncRequested, }).is_some() { panic!("PeerManager driver duplicated descriptors!"); }; @@ -246,21 +281,74 @@ impl PeerManager { pending_read_buffer: pending_read_buffer, pending_read_buffer_pos: 0, pending_read_is_header: false, + + sync_status: InitSyncTracker::NoSyncRequested, }).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 sync update 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)[..])); + } + } + } + const MSG_BUFF_SIZE: usize = 10; while !peer.awaiting_write_event { + if peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE { + match peer.sync_status { + InitSyncTracker::NoSyncRequested => {}, + InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => { + let steps = ((MSG_BUFF_SIZE - peer.pending_outbound_buffer.len() + 2) / 3) as u8; + let all_messages = self.message_handler.route_handler.get_next_channel_announcements(0, steps); + for &(ref announce, ref update_a, ref update_b) in all_messages.iter() { + encode_and_send_msg!(announce, 256); + encode_and_send_msg!(update_a, 258); + encode_and_send_msg!(update_b, 258); + peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1); + } + if all_messages.is_empty() || all_messages.len() != steps as usize { + peer.sync_status = InitSyncTracker::ChannelsSyncing(0xffff_ffff_ffff_ffff); + } + }, + InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => { + let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8; + let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps); + for msg in all_messages.iter() { + encode_and_send_msg!(msg, 256); + peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id); + } + if all_messages.is_empty() || all_messages.len() != steps as usize { + peer.sync_status = InitSyncTracker::NoSyncRequested; + } + }, + InitSyncTracker::ChannelsSyncing(_) => unreachable!(), + InitSyncTracker::NodesSyncing(key) => { + let steps = (MSG_BUFF_SIZE - peer.pending_outbound_buffer.len()) as u8; + let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps); + for msg in all_messages.iter() { + encode_and_send_msg!(msg, 256); + peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id); + } + if all_messages.is_empty() || all_messages.len() != steps as usize { + peer.sync_status = InitSyncTracker::NoSyncRequested; + } + }, + } + } + if { 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 should_be_reading = peer.pending_outbound_buffer.len() < MSG_BUFF_SIZE; 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 } @@ -288,7 +376,7 @@ impl PeerManager { 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(()) @@ -298,16 +386,12 @@ impl PeerManager { /// /// May return an Err to indicate that the connection should be closed. /// - /// Will very likely call send_data on the descriptor passed in (or a descriptor handed into - /// new_*_connection) before returning. Thus, be very careful with reentrancy issues! The - /// invariants around calling write_event in case a write did not fully complete must still - /// hold. Note that this function will often call send_data on many peers before returning, not - /// just this peer! + /// Will *not* call back into send_data on any descriptors to avoid reentrancy complexity. + /// Thus, however, you almost certainly want to call process_events() after any read_event to + /// generate send_data calls to handle responses. /// /// If Ok(true) is returned, further read_events should not be triggered until a write_event on - /// this file descriptor has resume_read set (preventing DoS issues in the send buffer). Note - /// that this must be true even if a send_data call with resume_read=true was made during the - /// course of this function! + /// this file descriptor has resume_read set (preventing DoS issues in the send buffer). /// /// Panics if the descriptor was not previously registered in a new_*_connection event. pub fn read_event(&self, peer_descriptor: &mut Descriptor, data: Vec) -> Result { @@ -347,6 +431,7 @@ impl PeerManager { { 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)[..])); + peers.peers_needing_send.insert(peer_descriptor.clone()); } } } @@ -390,11 +475,17 @@ impl PeerManager { match e { msgs::DecodeError::UnknownVersion => return Err(PeerHandleError{ no_connection_possible: false }), msgs::DecodeError::UnknownRequiredFeature => { - log_debug!(self, "Got a channel/node announcement with an known required feature flag, you may want to udpate!"); + log_debug!(self, "Got a channel/node announcement with an known required feature flag, you may want to update!"); continue; }, - msgs::DecodeError::InvalidValue => return Err(PeerHandleError{ no_connection_possible: false }), - msgs::DecodeError::ShortRead => return Err(PeerHandleError{ no_connection_possible: false }), + msgs::DecodeError::InvalidValue => { + log_debug!(self, "Got an invalid value while deserializing message"); + return Err(PeerHandleError{ no_connection_possible: false }); + }, + msgs::DecodeError::ShortRead => { + log_debug!(self, "Deserialization failed due to shortness of message"); + return Err(PeerHandleError{ no_connection_possible: false }); + }, msgs::DecodeError::ExtraAddressesPerType => { log_debug!(self, "Error decoding message, ignoring due to lnd spec incompatibility. See https://github.com/lightningnetwork/lnd/issues/1407"); continue; @@ -411,10 +502,14 @@ impl PeerManager { () => { match peers.node_id_to_descriptor.entry(peer.their_node_id.unwrap()) { hash_map::Entry::Occupied(_) => { + log_trace!(self, "Got second connection with {}, closing", log_pubkey!(peer.their_node_id.unwrap())); peer.their_node_id = None; // Unset so that we don't generate a peer_disconnected event return Err(PeerHandleError{ no_connection_possible: false }) }, - hash_map::Entry::Vacant(entry) => entry.insert(peer_descriptor.clone()), + hash_map::Entry::Vacant(entry) => { + log_trace!(self, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap())); + entry.insert(peer_descriptor.clone()) + }, }; } } @@ -427,11 +522,12 @@ impl PeerManager { peer.pending_read_buffer = [0; 66].to_vec(); // act three is 66 bytes long }, NextNoiseStep::ActTwo => { - let act_three = try_potential_handleerror!(peer.channel_encryptor.process_act_two(&peer.pending_read_buffer[..], &self.our_node_secret)).to_vec(); - peer.pending_outbound_buffer.push_back(act_three); + let (act_three, their_node_id) = try_potential_handleerror!(peer.channel_encryptor.process_act_two(&peer.pending_read_buffer[..], &self.our_node_secret)); + peer.pending_outbound_buffer.push_back(act_three.to_vec()); peer.pending_read_buffer = [0; 18].to_vec(); // Message length header is 18 bytes peer.pending_read_is_header = true; + peer.their_node_id = Some(their_node_id); insert_node_id!(); let mut local_features = msgs::LocalFeatures::new(); if self.initial_syncs_sent.load(Ordering::Acquire) < INITIAL_SYNCS_TO_SEND { @@ -471,6 +567,7 @@ impl PeerManager { log_trace!(self, "Received message of type {} from {}", msg_type, log_pubkey!(peer.their_node_id.unwrap())); if msg_type != 16 && peer.their_global_features.is_none() { // Need an init message as first message + log_trace!(self, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap())); return Err(PeerHandleError{ no_connection_possible: false }); } let mut reader = ::std::io::Cursor::new(&msg_data[2..]); @@ -505,6 +602,10 @@ impl PeerManager { if msg.local_features.supports_unknown_bits() { "present" } else { "none" }, if msg.global_features.supports_unknown_bits() { "present" } else { "none" }); + if msg.local_features.initial_routing_sync() { + peer.sync_status = InitSyncTracker::ChannelsSyncing(0); + peers.peers_needing_send.insert(peer_descriptor.clone()); + } peer.their_global_features = Some(msg.global_features); peer.their_local_features = Some(msg.local_features); @@ -514,15 +615,14 @@ impl PeerManager { self.initial_syncs_sent.fetch_add(1, Ordering::AcqRel); local_features.set_initial_routing_sync(); } + encode_and_send_msg!(msgs::Init { global_features: msgs::GlobalFeatures::new(), local_features, }, 16); } - for msg in self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap()) { - encode_and_send_msg!(msg, 136); - } + self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap()); }, 17 => { let msg = try_potential_decodeerror!(msgs::ErrorMessage::read(&mut reader)); @@ -559,11 +659,11 @@ impl PeerManager { // Channel control: 32 => { let msg = try_potential_decodeerror!(msgs::OpenChannel::read(&mut reader)); - try_potential_handleerror!(self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), &msg)); + try_potential_handleerror!(self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), peer.their_local_features.clone().unwrap(), &msg)); }, 33 => { let msg = try_potential_decodeerror!(msgs::AcceptChannel::read(&mut reader)); - try_potential_handleerror!(self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), &msg)); + try_potential_handleerror!(self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), peer.their_local_features.clone().unwrap(), &msg)); }, 34 => { @@ -607,33 +707,11 @@ impl PeerManager { 132 => { let msg = try_potential_decodeerror!(msgs::CommitmentSigned::read(&mut reader)); - let resps = try_potential_handleerror!(self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg)); - encode_and_send_msg!(resps.0, 133); - if let Some(resp) = resps.1 { - encode_and_send_msg!(resp, 132); - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg)); }, 133 => { let msg = try_potential_decodeerror!(msgs::RevokeAndACK::read(&mut reader)); - let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.update_add_htlcs { - encode_and_send_msg!(resp, 128); - } - for resp in resps.update_fulfill_htlcs { - encode_and_send_msg!(resp, 130); - } - for resp in resps.update_fail_htlcs { - encode_and_send_msg!(resp, 131); - } - if let Some(resp) = resps.update_fee { - encode_and_send_msg!(resp, 134); - } - encode_and_send_msg!(resps.commitment_signed, 132); - }, - None => {}, - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg)); }, 134 => { let msg = try_potential_decodeerror!(msgs::UpdateFee::read(&mut reader)); @@ -641,45 +719,7 @@ impl PeerManager { }, 136 => { let msg = try_potential_decodeerror!(msgs::ChannelReestablish::read(&mut reader)); - let (funding_locked, revoke_and_ack, commitment_update, order) = try_potential_handleerror!(self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg)); - if let Some(lock_msg) = funding_locked { - encode_and_send_msg!(lock_msg, 36); - } - macro_rules! handle_raa { () => { - if let Some(revoke_msg) = revoke_and_ack { - encode_and_send_msg!(revoke_msg, 133); - } - } } - macro_rules! handle_cu { () => { - match commitment_update { - Some(resps) => { - for resp in resps.update_add_htlcs { - encode_and_send_msg!(resp, 128); - } - for resp in resps.update_fulfill_htlcs { - encode_and_send_msg!(resp, 130); - } - for resp in resps.update_fail_htlcs { - encode_and_send_msg!(resp, 131); - } - if let Some(resp) = resps.update_fee { - encode_and_send_msg!(resp, 134); - } - encode_and_send_msg!(resps.commitment_signed, 132); - }, - None => {}, - } - } } - match order { - msgs::RAACommitmentOrder::RevokeAndACKFirst => { - handle_raa!(); - handle_cu!(); - }, - msgs::RAACommitmentOrder::CommitmentFirst => { - handle_cu!(); - handle_raa!(); - }, - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg)); }, // Routing control: @@ -723,7 +763,7 @@ impl PeerManager { } } - Self::do_attempt_write_data(peer_descriptor, peer); + self.do_attempt_write_data(peer_descriptor, peer); peer.pending_outbound_buffer.len() > 10 // pause_read } @@ -732,13 +772,12 @@ impl PeerManager { pause_read }; - self.process_events(); - Ok(pause_read) } - /// Checks for any events generated by our handlers and processes them. May be needed after eg - /// calls to ChannelManager::process_pending_htlc_forward. + /// Checks for any events generated by our handlers and processes them. Includes sending most + /// response messages as well as messages generated by calls to handler functions directly (eg + /// functions like ChannelManager::process_pending_htlc_forward or send_payment). pub fn process_events(&self) { { // TODO: There are some DoS attacks here where you can flood someone's outbound send @@ -746,7 +785,8 @@ impl PeerManager { // drop optional-ish messages when send buffers get full! let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_msg_events(); - let mut peers = self.peers.lock().unwrap(); + let mut peers_lock = self.peers.lock().unwrap(); + let peers = peers_lock.borrow_parts(); for event in events_generated.drain(..) { macro_rules! get_peer_for_forwarding { ($node_id: expr, $handle_no_such_peer: block) => { @@ -780,7 +820,7 @@ impl PeerManager { //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 {}", @@ -790,7 +830,7 @@ impl PeerManager { //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 {})", @@ -802,7 +842,7 @@ impl PeerManager { //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 {}", @@ -813,7 +853,7 @@ impl PeerManager { //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 {}", @@ -823,7 +863,7 @@ impl PeerManager { //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 {})", @@ -834,7 +874,7 @@ impl PeerManager { //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 {}", @@ -862,7 +902,7 @@ impl PeerManager { 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 {}", @@ -872,7 +912,7 @@ impl PeerManager { //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 {}", @@ -882,7 +922,7 @@ impl PeerManager { //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 {}", @@ -892,7 +932,17 @@ impl PeerManager { //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 {}", + log_pubkey!(node_id), + log_bytes!(msg.channel_id)); + let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, { + //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); }, 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); @@ -901,7 +951,8 @@ impl PeerManager { let encoded_update_msg = encode_msg!(update_msg, 258); for (ref descriptor, ref mut peer) in peers.peers.iter_mut() { - if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() { + if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() || + !peer.should_forward_channel(msg.contents.short_channel_id) { continue } match peer.their_node_id { @@ -914,7 +965,7 @@ impl PeerManager { } 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); } } }, @@ -924,11 +975,12 @@ impl PeerManager { let encoded_msg = encode_msg!(msg, 258); for (ref descriptor, ref mut peer) in peers.peers.iter_mut() { - if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() { + if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_global_features.is_none() || + !peer.should_forward_channel(msg.contents.short_channel_id) { 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); } } }, @@ -940,6 +992,7 @@ impl PeerManager { match *action { msgs::ErrorAction::DisconnectPeer { ref msg } => { if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) { + peers.peers_needing_send.remove(&descriptor); if let Some(mut peer) = peers.peers.remove(&descriptor) { if let Some(ref msg) = *msg { log_trace!(self, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}", @@ -948,7 +1001,7 @@ impl PeerManager { 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)); } @@ -966,7 +1019,7 @@ impl PeerManager { //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 { @@ -975,6 +1028,13 @@ impl PeerManager { } } } + + 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), + None => panic!("Inconsistent peers set state!"), + } + } } } @@ -990,6 +1050,7 @@ impl PeerManager { fn disconnect_event_internal(&self, descriptor: &Descriptor, no_connection_possible: bool) { let mut peers = self.peers.lock().unwrap(); + peers.peers_needing_send.remove(descriptor); let peer_option = peers.peers.remove(descriptor); match peer_option { None => panic!("Descriptor for disconnect_event is not already known to PeerManager"), @@ -1036,7 +1097,6 @@ mod tests { } fn create_network(peer_count: usize) -> Vec> { - let secp_ctx = Secp256k1::new(); let mut peers = Vec::new(); let mut rng = thread_rng(); let logger : Arc = Arc::new(test_utils::TestLogger::new()); @@ -1047,7 +1107,7 @@ mod tests { let node_id = { let mut key_slice = [0;32]; rng.fill_bytes(&mut key_slice); - SecretKey::from_slice(&secp_ctx, &key_slice).unwrap() + SecretKey::from_slice(&key_slice).unwrap() }; let msg_handler = MessageHandler { chan_handler: Arc::new(chan_handler), route_handler: Arc::new(router) }; let peer = PeerManager::new(msg_handler, node_id, Arc::clone(&logger)); @@ -1068,7 +1128,7 @@ mod tests { #[test] fn test_disconnect_peer() { // Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and - // push an DisconnectPeer event to remove the node flagged by id + // push a DisconnectPeer event to remove the node flagged by id let mut peers = create_network(2); establish_connection(&peers[0], &peers[1]); assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);