X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fpeer_handler.rs;h=f80c8984c1cb0437831e70510c6d9fa0e4e7e0fe;hb=fa1a0d8531612d16cedc9c1bdcea3e3989649c7c;hp=f771cf4ff2a4b36d25720f526ad577a182e8b1e0;hpb=eb17464e78ca3af0c80b5262b2d14fbebff40f10;p=rust-lightning diff --git a/lightning/src/ln/peer_handler.rs b/lightning/src/ln/peer_handler.rs index f771cf4f..f80c8984 100644 --- a/lightning/src/ln/peer_handler.rs +++ b/lightning/src/ln/peer_handler.rs @@ -12,23 +12,23 @@ //! Instead of actually servicing sockets ourselves we require that you implement the //! SocketDescriptor interface and use that to receive actions which you should perform on the //! socket, and call into PeerManager with bytes read from the socket. The PeerManager will then -//! call into the provided message handlers (probably a ChannelManager and NetGraphmsgHandler) with messages -//! they should handle, and encoding/sending response messages. +//! call into the provided message handlers (probably a ChannelManager and P2PGossipSync) with +//! messages they should handle, and encoding/sending response messages. -use bitcoin::secp256k1::key::{SecretKey,PublicKey}; +use bitcoin::secp256k1::{self, Secp256k1, SecretKey, PublicKey}; use ln::features::InitFeatures; use ln::msgs; -use ln::msgs::{ChannelMessageHandler, LightningError, NetAddress, RoutingMessageHandler}; +use ln::msgs::{ChannelMessageHandler, LightningError, NetAddress, OnionMessageHandler, RoutingMessageHandler}; use ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager}; use util::ser::{VecWriter, Writeable, Writer}; use ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep}; use ln::wire; use ln::wire::Encode; +use routing::gossip::{NetworkGraph, P2PGossipSync}; use util::atomic_counter::AtomicCounter; -use util::events::{MessageSendEvent, MessageSendEventsProvider}; +use util::events::{MessageSendEvent, MessageSendEventsProvider, OnionMessageProvider}; use util::logger::Logger; -use routing::network_graph::{NetworkGraph, NetGraphMsgHandler}; use prelude::*; use io; @@ -67,15 +67,23 @@ impl RoutingMessageHandler for IgnoringMessageHandler { fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result { Ok(false) } fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result { Ok(false) } fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result { Ok(false) } - fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) -> - Vec<(msgs::ChannelAnnouncement, Option, Option)> { Vec::new() } - fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec { Vec::new() } + fn get_next_channel_announcement(&self, _starting_point: u64) -> + Option<(msgs::ChannelAnnouncement, Option, Option)> { None } + fn get_next_node_announcement(&self, _starting_point: Option<&PublicKey>) -> Option { None } 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(()) } fn handle_query_short_channel_ids(&self, _their_node_id: &PublicKey, _msg: msgs::QueryShortChannelIds) -> Result<(), LightningError> { Ok(()) } } +impl OnionMessageProvider for IgnoringMessageHandler { + fn next_onion_message_for_peer(&self, _peer_node_id: PublicKey) -> Option { None } +} +impl OnionMessageHandler for IgnoringMessageHandler { + fn handle_onion_message(&self, _their_node_id: &PublicKey, _msg: &msgs::OnionMessage) {} + fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {} + fn peer_disconnected(&self, _their_node_id: &PublicKey, _no_connection_possible: bool) {} +} impl Deref for IgnoringMessageHandler { type Target = IgnoringMessageHandler; fn deref(&self) -> &Self { self } @@ -151,7 +159,7 @@ impl ChannelMessageHandler for ErroringMessageHandler { fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) { ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id); } - fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) { + fn handle_channel_ready(&self, their_node_id: &PublicKey, msg: &msgs::ChannelReady) { ErroringMessageHandler::push_error(self, their_node_id, msg.channel_id); } fn handle_shutdown(&self, their_node_id: &PublicKey, _their_features: &InitFeatures, msg: &msgs::Shutdown) { @@ -199,20 +207,25 @@ impl Deref for ErroringMessageHandler { } /// Provides references to trait impls which handle different types of messages. -pub struct MessageHandler where +pub struct MessageHandler where CM::Target: ChannelMessageHandler, - RM::Target: RoutingMessageHandler { + RM::Target: RoutingMessageHandler, + OM::Target: OnionMessageHandler, +{ /// A message handler which handles messages specific to channels. Usually this is just a /// [`ChannelManager`] object or an [`ErroringMessageHandler`]. /// /// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager pub chan_handler: CM, /// A message handler which handles messages updating our knowledge of the network channel - /// graph. Usually this is just a [`NetGraphMsgHandler`] object or an - /// [`IgnoringMessageHandler`]. + /// graph. Usually this is just a [`P2PGossipSync`] object or an [`IgnoringMessageHandler`]. /// - /// [`NetGraphMsgHandler`]: crate::routing::network_graph::NetGraphMsgHandler + /// [`P2PGossipSync`]: crate::routing::gossip::P2PGossipSync pub route_handler: RM, + + /// A message handler which handles onion messages. For now, this can only be an + /// [`IgnoringMessageHandler`]. + pub onion_message_handler: OM, } /// Provides an object which can be used to send data to and which uniquely identifies a connection @@ -258,8 +271,13 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone { /// descriptor. #[derive(Clone)] pub struct PeerHandleError { - /// Used to indicate that we probably can't make any future connections to this peer, implying - /// we should go ahead and force-close any channels we have with it. + /// Used to indicate that we probably can't make any future connections to this peer (e.g. + /// because we required features that our peer was missing, or vice versa). + /// + /// While LDK's [`ChannelManager`] will not do it automatically, you likely wish to force-close + /// any channels with this peer or check for new versions of LDK. + /// + /// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager pub no_connection_possible: bool, } impl fmt::Debug for PeerHandleError { @@ -294,7 +312,7 @@ const FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO: usize = 2; /// we have fewer than this many messages in the outbound buffer again. /// We also use this as the target number of outbound gossip messages to keep in the write buffer, /// refilled as we send bytes. -const OUTBOUND_BUFFER_LIMIT_READ_PAUSE: usize = 10; +const OUTBOUND_BUFFER_LIMIT_READ_PAUSE: usize = 12; /// When the outbound buffer has this many messages, we'll simply skip relaying gossip messages to /// the peer. const OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP: usize = OUTBOUND_BUFFER_LIMIT_READ_PAUSE * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO; @@ -319,6 +337,10 @@ const MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER: i8 = 4; /// tick. Once we have sent this many messages since the last ping, we send a ping right away to /// ensures we don't just fill up our send buffer and leave the peer with too many messages to /// process before the next ping. +/// +/// Note that we continue responding to other messages even after we've sent this many messages, so +/// it's more of a general guideline used for gossip backfill (and gossip forwarding, times +/// [`FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO`]) than a hard limit. const BUFFER_DRAIN_MSGS_PER_TICK: usize = 32; struct Peer { @@ -329,6 +351,9 @@ struct Peer { pending_outbound_buffer: LinkedList>, pending_outbound_buffer_first_msg_offset: usize, + // Queue gossip broadcasts separately from `pending_outbound_buffer` so we can easily prioritize + // channel messages over them. + gossip_broadcast_buffer: LinkedList>, awaiting_write_event: bool, pending_read_buffer: Vec, @@ -374,14 +399,42 @@ impl Peer { InitSyncTracker::NodesSyncing(pk) => pk < node_id, } } -} -struct PeerHolder { - /// Peer is under its own mutex for sending and receiving bytes, but note that we do *not* hold - /// this mutex while we're processing a message. This is fine as [`PeerManager::read_event`] - /// requires that there be no parallel calls for a given peer, so mutual exclusion of messages - /// handed to the `MessageHandler`s for a given peer is already guaranteed. - peers: HashMap>, + /// Returns whether we should be reading bytes from this peer, based on whether its outbound + /// buffer still has space and we don't need to pause reads to get some writes out. + fn should_read(&self) -> bool { + self.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE + } + + /// Determines if we should push additional gossip background sync (aka "backfill") onto a peer's + /// outbound buffer. This is checked every time the peer's buffer may have been drained. + fn should_buffer_gossip_backfill(&self) -> bool { + self.pending_outbound_buffer.is_empty() && self.gossip_broadcast_buffer.is_empty() + && self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK + } + + /// Determines if we should push an onion message onto a peer's outbound buffer. This is checked + /// every time the peer's buffer may have been drained. + fn should_buffer_onion_message(&self) -> bool { + self.pending_outbound_buffer.is_empty() + && self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK + } + + /// Determines if we should push additional gossip broadcast messages onto a peer's outbound + /// buffer. This is checked every time the peer's buffer may have been drained. + fn should_buffer_gossip_broadcast(&self) -> bool { + self.pending_outbound_buffer.is_empty() + && self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK + } + + /// Returns whether this peer's outbound buffers are full and we should drop gossip broadcasts. + fn buffer_full_drop_gossip_broadcast(&self) -> bool { + let total_outbound_buffered = + self.gossip_broadcast_buffer.len() + self.pending_outbound_buffer.len(); + + total_outbound_buffered > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP || + self.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO + } } /// SimpleArcPeerManager is useful when you need a PeerManager with a static lifetime, e.g. @@ -391,7 +444,7 @@ struct PeerHolder { /// issues such as overly long function definitions. /// /// (C-not exported) as Arcs don't make sense in bindings -pub type SimpleArcPeerManager = PeerManager>, Arc, Arc, Arc>>, Arc, Arc>; +pub type SimpleArcPeerManager = PeerManager>, Arc>>, Arc, Arc>>, IgnoringMessageHandler, Arc, Arc>; /// SimpleRefPeerManager is a type alias for a PeerManager reference, and is the reference /// counterpart to the SimpleArcPeerManager type alias. Use this type by default when you don't @@ -401,7 +454,7 @@ pub type SimpleArcPeerManager = PeerManager = PeerManager, &'e NetGraphMsgHandler<&'g NetworkGraph, &'h C, &'f L>, &'f L, IgnoringMessageHandler>; +pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, SD, M, T, F, C, L> = PeerManager, &'e P2PGossipSync<&'g NetworkGraph<&'f L>, &'h C, &'f L>, IgnoringMessageHandler, &'f L, IgnoringMessageHandler>; /// A PeerManager manages a set of peers, described by their [`SocketDescriptor`] and marshalls /// socket events into messages which it passes on to its [`MessageHandler`]. @@ -422,13 +475,22 @@ pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, SD, M, T, F, C, L> /// you're using lightning-net-tokio. /// /// [`read_event`]: PeerManager::read_event -pub struct PeerManager where +pub struct PeerManager where CM::Target: ChannelMessageHandler, RM::Target: RoutingMessageHandler, + OM::Target: OnionMessageHandler, L::Target: Logger, CMH::Target: CustomMessageHandler { - message_handler: MessageHandler, - peers: FairRwLock>, + message_handler: MessageHandler, + /// Connection state for each connected peer - we have an outer read-write lock which is taken + /// as read while we're doing processing for a peer and taken write when a peer is being added + /// or removed. + /// + /// The inner Peer lock is held for sending and receiving bytes, but note that we do *not* hold + /// it while we're processing a message. This is fine as [`PeerManager::read_event`] requires + /// that there be no parallel calls for a given peer, so mutual exclusion of messages handed to + /// the `MessageHandler`s for a given peer is already guaranteed. + peers: FairRwLock>>, /// Only add to this set when noise completes. /// Locked *after* peers. When an item is removed, it must be removed with the `peers` write /// lock held. Entries may be added with only the `peers` read lock held (though the @@ -450,6 +512,7 @@ pub struct PeerManager } enum MessageHandlingError { @@ -477,31 +540,34 @@ macro_rules! encode_msg { }} } -impl PeerManager where +impl PeerManager where CM::Target: ChannelMessageHandler, + OM::Target: OnionMessageHandler, L::Target: Logger { - /// Constructs a new PeerManager with the given ChannelMessageHandler. No routing message - /// handler is used and network graph messages are ignored. + /// Constructs a new `PeerManager` with the given `ChannelMessageHandler` and + /// `OnionMessageHandler`. No routing message handler is used and network graph messages are + /// ignored. /// /// ephemeral_random_data is used to derive per-connection ephemeral keys and must be /// cryptographically secure random bytes. /// /// (C-not exported) as we can't export a PeerManager with a dummy route handler - pub fn new_channel_only(channel_message_handler: CM, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L) -> Self { + pub fn new_channel_only(channel_message_handler: CM, onion_message_handler: OM, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L) -> Self { Self::new(MessageHandler { chan_handler: channel_message_handler, route_handler: IgnoringMessageHandler{}, + onion_message_handler, }, our_node_secret, ephemeral_random_data, logger, IgnoringMessageHandler{}) } } -impl PeerManager where +impl PeerManager where RM::Target: RoutingMessageHandler, L::Target: Logger { - /// Constructs a new PeerManager with the given RoutingMessageHandler. No channel message - /// handler is used and messages related to channels will be ignored (or generate error - /// messages). Note that some other lightning implementations time-out connections after some - /// time if no channel is built with the peer. + /// Constructs a new `PeerManager` with the given `RoutingMessageHandler`. No channel message + /// handler or onion message handler is used and onion and channel messages will be ignored (or + /// generate error messages). Note that some other lightning implementations time-out connections + /// after some time if no channel is built with the peer. /// /// ephemeral_random_data is used to derive per-connection ephemeral keys and must be /// cryptographically secure random bytes. @@ -511,6 +577,7 @@ impl PeerManager) -> Option { } } -impl PeerManager where +impl PeerManager where CM::Target: ChannelMessageHandler, RM::Target: RoutingMessageHandler, + OM::Target: OnionMessageHandler, L::Target: Logger, CMH::Target: CustomMessageHandler { /// Constructs a new PeerManager with the given message handlers and node_id secret key /// ephemeral_random_data is used to derive per-connection ephemeral keys and must be /// cryptographically secure random bytes. - pub fn new(message_handler: MessageHandler, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L, custom_message_handler: CMH) -> Self { + pub fn new(message_handler: MessageHandler, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L, custom_message_handler: CMH) -> Self { let mut ephemeral_key_midstate = Sha256::engine(); ephemeral_key_midstate.input(ephemeral_random_data); + let mut secp_ctx = Secp256k1::signing_only(); + let ephemeral_hash = Sha256::from_engine(ephemeral_key_midstate.clone()).into_inner(); + secp_ctx.seeded_randomize(&ephemeral_hash); + PeerManager { message_handler, - peers: FairRwLock::new(PeerHolder { - peers: HashMap::new(), - }), + peers: FairRwLock::new(HashMap::new()), node_id_to_descriptor: Mutex::new(HashMap::new()), event_processing_lock: Mutex::new(()), blocked_event_processors: AtomicBool::new(false), @@ -581,6 +651,7 @@ impl P peer_counter: AtomicCounter::new(), logger, custom_message_handler, + secp_ctx, } } @@ -591,7 +662,7 @@ impl P /// completed and we are sure the remote peer has the private key for the given node_id. pub fn get_peer_node_ids(&self) -> Vec { let peers = self.peers.read().unwrap(); - peers.peers.values().filter_map(|peer_mutex| { + peers.values().filter_map(|peer_mutex| { let p = peer_mutex.lock().unwrap(); if !p.channel_encryptor.is_ready_for_encryption() || p.their_features.is_none() { return None; @@ -614,8 +685,7 @@ impl P /// peer using the init message. /// The user should pass the remote network address of the host they are connected to. /// - /// Note that if an Err is returned here you MUST NOT call socket_disconnected for the new - /// descriptor but must disconnect the connection immediately. + /// If an `Err` is returned here you must disconnect the connection immediately. /// /// Returns a small number of bytes to send to the remote node (currently always 50). /// @@ -625,11 +695,11 @@ impl P /// [`socket_disconnected()`]: PeerManager::socket_disconnected pub fn new_outbound_connection(&self, their_node_id: PublicKey, descriptor: Descriptor, remote_network_address: Option) -> Result, PeerHandleError> { let mut peer_encryptor = PeerChannelEncryptor::new_outbound(their_node_id.clone(), self.get_ephemeral_key()); - let res = peer_encryptor.get_act_one().to_vec(); + let res = peer_encryptor.get_act_one(&self.secp_ctx).to_vec(); let pending_read_buffer = [0; 50].to_vec(); // Noise act two is 50 bytes let mut peers = self.peers.write().unwrap(); - if peers.peers.insert(descriptor, Mutex::new(Peer { + if peers.insert(descriptor, Mutex::new(Peer { channel_encryptor: peer_encryptor, their_node_id: None, their_features: None, @@ -637,6 +707,7 @@ impl P pending_outbound_buffer: LinkedList::new(), pending_outbound_buffer_first_msg_offset: 0, + gossip_broadcast_buffer: LinkedList::new(), awaiting_write_event: false, pending_read_buffer, @@ -663,20 +734,19 @@ impl P /// The user should pass the remote network address of the host they are connected to. /// /// May refuse the connection by returning an Err, but will never write bytes to the remote end - /// (outbound connector always speaks first). Note that if an Err is returned here you MUST NOT - /// call socket_disconnected for the new descriptor but must disconnect the connection - /// immediately. + /// (outbound connector always speaks first). If an `Err` is returned here you must disconnect + /// the connection immediately. /// /// Panics if descriptor is duplicative with some other descriptor which has not yet been /// [`socket_disconnected()`]. /// /// [`socket_disconnected()`]: PeerManager::socket_disconnected pub fn new_inbound_connection(&self, descriptor: Descriptor, remote_network_address: Option) -> Result<(), PeerHandleError> { - let peer_encryptor = PeerChannelEncryptor::new_inbound(&self.our_node_secret); + let peer_encryptor = PeerChannelEncryptor::new_inbound(&self.our_node_secret, &self.secp_ctx); let pending_read_buffer = [0; 50].to_vec(); // Noise act one is 50 bytes let mut peers = self.peers.write().unwrap(); - if peers.peers.insert(descriptor, Mutex::new(Peer { + if peers.insert(descriptor, Mutex::new(Peer { channel_encryptor: peer_encryptor, their_node_id: None, their_features: None, @@ -684,6 +754,7 @@ impl P pending_outbound_buffer: LinkedList::new(), pending_outbound_buffer_first_msg_offset: 0, + gossip_broadcast_buffer: LinkedList::new(), awaiting_write_event: false, pending_read_buffer, @@ -704,46 +775,52 @@ impl P fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) { while !peer.awaiting_write_event { - if peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE && peer.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK { + if peer.should_buffer_onion_message() { + if let Some(peer_node_id) = peer.their_node_id { + if let Some(next_onion_message) = + self.message_handler.onion_message_handler.next_onion_message_for_peer(peer_node_id) { + self.enqueue_message(peer, &next_onion_message); + } + } + } + if peer.should_buffer_gossip_broadcast() { + if let Some(msg) = peer.gossip_broadcast_buffer.pop_front() { + peer.pending_outbound_buffer.push_back(msg); + } + } + if peer.should_buffer_gossip_backfill() { match peer.sync_status { InitSyncTracker::NoSyncRequested => {}, InitSyncTracker::ChannelsSyncing(c) if c < 0xffff_ffff_ffff_ffff => { - let steps = ((OUTBOUND_BUFFER_LIMIT_READ_PAUSE - peer.pending_outbound_buffer.len() + 2) / 3) as u8; - let all_messages = self.message_handler.route_handler.get_next_channel_announcements(c, steps); - for &(ref announce, ref update_a_option, ref update_b_option) in all_messages.iter() { - self.enqueue_message(peer, announce); - if let &Some(ref update_a) = update_a_option { - self.enqueue_message(peer, update_a); + if let Some((announce, update_a_option, update_b_option)) = + self.message_handler.route_handler.get_next_channel_announcement(c) + { + self.enqueue_message(peer, &announce); + if let Some(update_a) = update_a_option { + self.enqueue_message(peer, &update_a); } - if let &Some(ref update_b) = update_b_option { - self.enqueue_message(peer, update_b); + if let Some(update_b) = update_b_option { + self.enqueue_message(peer, &update_b); } peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1); - } - if all_messages.is_empty() || all_messages.len() != steps as usize { + } else { peer.sync_status = InitSyncTracker::ChannelsSyncing(0xffff_ffff_ffff_ffff); } }, InitSyncTracker::ChannelsSyncing(c) if c == 0xffff_ffff_ffff_ffff => { - let steps = (OUTBOUND_BUFFER_LIMIT_READ_PAUSE - 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() { - self.enqueue_message(peer, msg); + if let Some(msg) = self.message_handler.route_handler.get_next_node_announcement(None) { + self.enqueue_message(peer, &msg); peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id); - } - if all_messages.is_empty() || all_messages.len() != steps as usize { + } else { peer.sync_status = InitSyncTracker::NoSyncRequested; } }, InitSyncTracker::ChannelsSyncing(_) => unreachable!(), InitSyncTracker::NodesSyncing(key) => { - let steps = (OUTBOUND_BUFFER_LIMIT_READ_PAUSE - 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() { - self.enqueue_message(peer, msg); + if let Some(msg) = self.message_handler.route_handler.get_next_node_announcement(Some(&key)) { + self.enqueue_message(peer, &msg); peer.sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id); - } - if all_messages.is_empty() || all_messages.len() != steps as usize { + } else { peer.sync_status = InitSyncTracker::NoSyncRequested; } }, @@ -753,18 +830,15 @@ impl P self.maybe_send_extra_ping(peer); } - if { - let next_buff = match peer.pending_outbound_buffer.front() { - None => return, - Some(buff) => buff, - }; + let next_buff = match peer.pending_outbound_buffer.front() { + None => return, + Some(buff) => buff, + }; - let should_be_reading = peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE; - let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..]; - let data_sent = descriptor.send_data(pending, 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 } - } { + let pending = &next_buff[peer.pending_outbound_buffer_first_msg_offset..]; + let data_sent = descriptor.send_data(pending, peer.should_read()); + peer.pending_outbound_buffer_first_msg_offset += data_sent; + if peer.pending_outbound_buffer_first_msg_offset == next_buff.len() { peer.pending_outbound_buffer_first_msg_offset = 0; peer.pending_outbound_buffer.pop_front(); } else { @@ -787,7 +861,7 @@ impl P /// [`write_buffer_space_avail`]: PeerManager::write_buffer_space_avail pub fn write_buffer_space_avail(&self, descriptor: &mut Descriptor) -> Result<(), PeerHandleError> { let peers = self.peers.read().unwrap(); - match peers.peers.get(descriptor) { + match peers.get(descriptor) { None => { // This is most likely a simple race condition where the user found that the socket // was writeable, then we told the user to `disconnect_socket()`, then they called @@ -828,12 +902,6 @@ impl P } } - /// Append a message to a peer's pending outbound/write buffer - fn enqueue_encoded_message(&self, peer: &mut Peer, encoded_message: &Vec) { - peer.msgs_sent_since_pong += 1; - peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..])); - } - /// Append a message to a peer's pending outbound/write buffer fn enqueue_message(&self, peer: &mut Peer, message: &M) { let mut buffer = VecWriter(Vec::with_capacity(2048)); @@ -844,7 +912,14 @@ impl P } else { log_trace!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap())) } - self.enqueue_encoded_message(peer, &buffer.0); + peer.msgs_sent_since_pong += 1; + peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&buffer.0[..])); + } + + /// Append a message to a peer's pending outbound/write gossip broadcast buffer + fn enqueue_encoded_gossip_broadcast(&self, peer: &mut Peer, encoded_message: &Vec) { + peer.msgs_sent_since_pong += 1; + peer.gossip_broadcast_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_message[..])); } fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result { @@ -852,7 +927,7 @@ impl P let peers = self.peers.read().unwrap(); let mut msgs_to_forward = Vec::new(); let mut peer_node_id = None; - match peers.peers.get(peer_descriptor) { + match peers.get(peer_descriptor) { None => { // This is most likely a simple race condition where the user read some bytes // from the socket, then we told the user to `disconnect_socket()`, then they @@ -937,14 +1012,16 @@ impl P let next_step = peer.channel_encryptor.get_noise_step(); match next_step { NextNoiseStep::ActOne => { - let act_two = try_potential_handleerror!(peer, - peer.channel_encryptor.process_act_one_with_keys(&peer.pending_read_buffer[..], &self.our_node_secret, self.get_ephemeral_key())).to_vec(); + let act_two = try_potential_handleerror!(peer, peer.channel_encryptor + .process_act_one_with_keys(&peer.pending_read_buffer[..], + &self.our_node_secret, self.get_ephemeral_key(), &self.secp_ctx)).to_vec(); peer.pending_outbound_buffer.push_back(act_two); peer.pending_read_buffer = [0; 66].to_vec(); // act three is 66 bytes long }, NextNoiseStep::ActTwo => { let (act_three, their_node_id) = try_potential_handleerror!(peer, - peer.channel_encryptor.process_act_two(&peer.pending_read_buffer[..], &self.our_node_secret)); + peer.channel_encryptor.process_act_two(&peer.pending_read_buffer[..], + &self.our_node_secret, &self.secp_ctx)); 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; @@ -1037,7 +1114,7 @@ impl P } } } - pause_read = peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_READ_PAUSE; + pause_read = !peer.should_read(); if let Some(message) = msg_to_handle { match self.handle_message(&peer_mutex, peer_lock, message) { @@ -1098,8 +1175,9 @@ impl P } self.message_handler.route_handler.peer_connected(&their_node_id, &msg); - self.message_handler.chan_handler.peer_connected(&their_node_id, &msg); + self.message_handler.onion_message_handler.peer_connected(&their_node_id, &msg); + peer_lock.their_features = Some(msg.features); return Ok(None); } else if peer_lock.their_features.is_none() { @@ -1198,8 +1276,8 @@ impl P wire::Message::FundingSigned(msg) => { self.message_handler.chan_handler.handle_funding_signed(&their_node_id, &msg); }, - wire::Message::FundingLocked(msg) => { - self.message_handler.chan_handler.handle_funding_locked(&their_node_id, &msg); + wire::Message::ChannelReady(msg) => { + self.message_handler.chan_handler.handle_channel_ready(&their_node_id, &msg); }, wire::Message::Shutdown(msg) => { @@ -1272,6 +1350,11 @@ impl P self.message_handler.route_handler.handle_reply_channel_range(&their_node_id, msg)?; }, + // Onion message: + wire::Message::OnionMessage(msg) => { + self.message_handler.onion_message_handler.handle_onion_message(&their_node_id, &msg); + }, + // Unknown messages: wire::Message::Unknown(type_id) if message.is_even() => { log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", type_id); @@ -1288,21 +1371,19 @@ impl P Ok(should_forward) } - fn forward_broadcast_msg(&self, peers: &PeerHolder, msg: &wire::Message<<::Target as wire::CustomMessageReader>::CustomMessage>, except_node: Option<&PublicKey>) { + fn forward_broadcast_msg(&self, peers: &HashMap>, msg: &wire::Message<<::Target as wire::CustomMessageReader>::CustomMessage>, except_node: Option<&PublicKey>) { match msg { wire::Message::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); - for (_, peer_mutex) in peers.peers.iter() { + for (_, peer_mutex) in peers.iter() { let mut peer = peer_mutex.lock().unwrap(); if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() || !peer.should_forward_channel_announcement(msg.contents.short_channel_id) { continue } - if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP - || peer.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO - { + if peer.buffer_full_drop_gossip_broadcast() { log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id); continue; } @@ -1313,22 +1394,20 @@ impl P if except_node.is_some() && peer.their_node_id.as_ref() == except_node { continue; } - self.enqueue_encoded_message(&mut *peer, &encoded_msg); + self.enqueue_encoded_gossip_broadcast(&mut *peer, &encoded_msg); } }, wire::Message::NodeAnnouncement(ref msg) => { log_gossip!(self.logger, "Sending message to all peers except {:?} or the announced node: {:?}", except_node, msg); let encoded_msg = encode_msg!(msg); - for (_, peer_mutex) in peers.peers.iter() { + for (_, peer_mutex) in peers.iter() { let mut peer = peer_mutex.lock().unwrap(); if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() || !peer.should_forward_node_announcement(msg.contents.node_id) { continue } - if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP - || peer.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO - { + if peer.buffer_full_drop_gossip_broadcast() { log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id); continue; } @@ -1338,29 +1417,27 @@ impl P if except_node.is_some() && peer.their_node_id.as_ref() == except_node { continue; } - self.enqueue_encoded_message(&mut *peer, &encoded_msg); + self.enqueue_encoded_gossip_broadcast(&mut *peer, &encoded_msg); } }, wire::Message::ChannelUpdate(ref msg) => { log_gossip!(self.logger, "Sending message to all peers except {:?}: {:?}", except_node, msg); let encoded_msg = encode_msg!(msg); - for (_, peer_mutex) in peers.peers.iter() { + for (_, peer_mutex) in peers.iter() { let mut peer = peer_mutex.lock().unwrap(); if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() || !peer.should_forward_channel_announcement(msg.contents.short_channel_id) { continue } - if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP - || peer.msgs_sent_since_pong > BUFFER_DRAIN_MSGS_PER_TICK * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO - { + if peer.buffer_full_drop_gossip_broadcast() { log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id); continue; } if except_node.is_some() && peer.their_node_id.as_ref() == except_node { continue; } - self.enqueue_encoded_message(&mut *peer, &encoded_msg); + self.enqueue_encoded_gossip_broadcast(&mut *peer, &encoded_msg); } }, _ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"), @@ -1426,7 +1503,7 @@ impl P } let descriptor_opt = self.node_id_to_descriptor.lock().unwrap().get($node_id).cloned(); match descriptor_opt { - Some(descriptor) => match peers.peers.get(&descriptor) { + Some(descriptor) => match peers.get(&descriptor) { Some(peer_mutex) => { let peer_lock = peer_mutex.lock().unwrap(); if peer_lock.their_features.is_none() { @@ -1475,8 +1552,8 @@ impl P log_bytes!(msg.channel_id)); self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg); }, - MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => { - log_debug!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}", + MessageSendEvent::SendChannelReady { ref node_id, ref msg } => { + log_debug!(self.logger, "Handling SendChannelReady event in peer_handler for node {} for channel {}", log_pubkey!(node_id), log_bytes!(msg.channel_id)); self.enqueue_message(&mut *get_peer_for_forwarding!(node_id), msg); @@ -1625,7 +1702,7 @@ impl P self.enqueue_message(&mut *get_peer_for_forwarding!(&node_id), &msg); } - for (descriptor, peer_mutex) in peers.peers.iter() { + for (descriptor, peer_mutex) in peers.iter() { self.do_attempt_write_data(&mut (*descriptor).clone(), &mut *peer_mutex.lock().unwrap()); } } @@ -1639,7 +1716,7 @@ impl P // lock). if let Some(mut descriptor) = self.node_id_to_descriptor.lock().unwrap().remove(&node_id) { - if let Some(peer_mutex) = peers.peers.remove(&descriptor) { + if let Some(peer_mutex) = peers.remove(&descriptor) { if let Some(msg) = msg { log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}", log_pubkey!(node_id), @@ -1655,6 +1732,7 @@ impl P } descriptor.disconnect_socket(); self.message_handler.chan_handler.peer_disconnected(&node_id, false); + self.message_handler.onion_message_handler.peer_disconnected(&node_id, false); } } } @@ -1667,7 +1745,7 @@ impl P fn disconnect_event_internal(&self, descriptor: &Descriptor, no_connection_possible: bool) { let mut peers = self.peers.write().unwrap(); - let peer_option = peers.peers.remove(descriptor); + let peer_option = peers.remove(descriptor); match peer_option { None => { // This is most likely a simple race condition where the user found that the socket @@ -1676,15 +1754,13 @@ impl P }, Some(peer_lock) => { let peer = peer_lock.lock().unwrap(); - match peer.their_node_id { - Some(node_id) => { - log_trace!(self.logger, - "Handling disconnection of peer {}, with {}future connection to the peer possible.", - log_pubkey!(node_id), if no_connection_possible { "no " } else { "" }); - self.node_id_to_descriptor.lock().unwrap().remove(&node_id); - self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible); - }, - None => {} + if let Some(node_id) = peer.their_node_id { + log_trace!(self.logger, + "Handling disconnection of peer {}, with {}future connection to the peer possible.", + log_pubkey!(node_id), if no_connection_possible { "no " } else { "" }); + self.node_id_to_descriptor.lock().unwrap().remove(&node_id); + self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible); + self.message_handler.onion_message_handler.peer_disconnected(&node_id, no_connection_possible); } } }; @@ -1703,8 +1779,9 @@ impl P let mut peers_lock = self.peers.write().unwrap(); if let Some(mut descriptor) = self.node_id_to_descriptor.lock().unwrap().remove(&node_id) { log_trace!(self.logger, "Disconnecting peer with id {} due to client request", node_id); - peers_lock.peers.remove(&descriptor); + peers_lock.remove(&descriptor); self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible); + self.message_handler.onion_message_handler.peer_disconnected(&node_id, no_connection_possible); descriptor.disconnect_socket(); } } @@ -1716,10 +1793,11 @@ impl P let mut peers_lock = self.peers.write().unwrap(); self.node_id_to_descriptor.lock().unwrap().clear(); let peers = &mut *peers_lock; - for (mut descriptor, peer) in peers.peers.drain() { + for (mut descriptor, peer) in peers.drain() { if let Some(node_id) = peer.lock().unwrap().their_node_id { log_trace!(self.logger, "Disconnecting peer with id {} due to client request to disconnect all peers", node_id); self.message_handler.chan_handler.peer_disconnected(&node_id, false); + self.message_handler.onion_message_handler.peer_disconnected(&node_id, false); } descriptor.disconnect_socket(); } @@ -1755,7 +1833,7 @@ impl P { let peers_lock = self.peers.read().unwrap(); - for (descriptor, peer_mutex) in peers_lock.peers.iter() { + for (descriptor, peer_mutex) in peers_lock.iter() { let mut peer = peer_mutex.lock().unwrap(); if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_node_id.is_none() { // The peer needs to complete its handshake before we can exchange messages. We @@ -1779,7 +1857,7 @@ impl P if (peer.awaiting_pong_timer_tick_intervals > 0 && !peer.received_message_since_timer_tick) || peer.awaiting_pong_timer_tick_intervals as u64 > - MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER as u64 * peers_lock.peers.len() as u64 + MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER as u64 * peers_lock.len() as u64 { descriptors_needing_disconnect.push(descriptor.clone()); continue; @@ -1805,11 +1883,12 @@ impl P { let mut peers_lock = self.peers.write().unwrap(); for descriptor in descriptors_needing_disconnect.iter() { - if let Some(peer) = peers_lock.peers.remove(&descriptor) { + if let Some(peer) = peers_lock.remove(descriptor) { if let Some(node_id) = peer.lock().unwrap().their_node_id { log_trace!(self.logger, "Disconnecting peer with id {} due to ping timeout", node_id); self.node_id_to_descriptor.lock().unwrap().remove(&node_id); self.message_handler.chan_handler.peer_disconnected(&node_id, false); + self.message_handler.onion_message_handler.peer_disconnected(&node_id, false); } } } @@ -1838,13 +1917,13 @@ fn is_gossip_msg(type_id: u16) -> bool { #[cfg(test)] mod tests { use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler, filter_addresses}; - use ln::msgs; + use ln::{msgs, wire}; use ln::msgs::NetAddress; use util::events; use util::test_utils; use bitcoin::secp256k1::Secp256k1; - use bitcoin::secp256k1::key::{SecretKey, PublicKey}; + use bitcoin::secp256k1::{SecretKey, PublicKey}; use prelude::*; use sync::{Arc, Mutex}; @@ -1897,12 +1976,12 @@ mod tests { cfgs } - fn create_network<'a>(peer_count: usize, cfgs: &'a Vec) -> Vec> { + fn create_network<'a>(peer_count: usize, cfgs: &'a Vec) -> Vec> { let mut peers = Vec::new(); for i in 0..peer_count { let node_secret = SecretKey::from_slice(&[42 + i as u8; 32]).unwrap(); let ephemeral_bytes = [i as u8; 32]; - let msg_handler = MessageHandler { chan_handler: &cfgs[i].chan_handler, route_handler: &cfgs[i].routing_handler }; + let msg_handler = MessageHandler { chan_handler: &cfgs[i].chan_handler, route_handler: &cfgs[i].routing_handler, onion_message_handler: IgnoringMessageHandler {} }; let peer = PeerManager::new(msg_handler, node_secret, &ephemeral_bytes, &cfgs[i].logger, IgnoringMessageHandler {}); peers.push(peer); } @@ -1910,7 +1989,7 @@ mod tests { peers } - fn establish_connection<'a>(peer_a: &PeerManager, peer_b: &PeerManager) -> (FileDescriptor, FileDescriptor) { + fn establish_connection<'a>(peer_a: &PeerManager, peer_b: &PeerManager) -> (FileDescriptor, FileDescriptor) { let secp_ctx = Secp256k1::new(); let a_id = PublicKey::from_secret_key(&secp_ctx, &peer_a.our_node_secret); let mut fd_a = FileDescriptor { fd: 1, outbound_data: Arc::new(Mutex::new(Vec::new())) }; @@ -1919,11 +1998,18 @@ mod tests { peer_a.new_inbound_connection(fd_a.clone(), None).unwrap(); assert_eq!(peer_a.read_event(&mut fd_a, &initial_data).unwrap(), false); peer_a.process_events(); - assert_eq!(peer_b.read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap(), false); + + let a_data = fd_a.outbound_data.lock().unwrap().split_off(0); + assert_eq!(peer_b.read_event(&mut fd_b, &a_data).unwrap(), false); + peer_b.process_events(); - assert_eq!(peer_a.read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap(), false); + let b_data = fd_b.outbound_data.lock().unwrap().split_off(0); + assert_eq!(peer_a.read_event(&mut fd_a, &b_data).unwrap(), false); + peer_a.process_events(); - assert_eq!(peer_b.read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap(), false); + let a_data = fd_a.outbound_data.lock().unwrap().split_off(0); + assert_eq!(peer_b.read_event(&mut fd_b, &a_data).unwrap(), false); + (fd_a.clone(), fd_b.clone()) } @@ -1935,7 +2021,7 @@ mod tests { let chan_handler = test_utils::TestChannelMessageHandler::new(); let mut peers = create_network(2, &cfgs); establish_connection(&peers[0], &peers[1]); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 1); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); let secp_ctx = Secp256k1::new(); let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret); @@ -1948,7 +2034,49 @@ mod tests { peers[0].message_handler.chan_handler = &chan_handler; peers[0].process_events(); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 0); + assert_eq!(peers[0].peers.read().unwrap().len(), 0); + } + + #[test] + fn test_send_simple_msg() { + // Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and + // push a message from one peer to another. + let cfgs = create_peermgr_cfgs(2); + let a_chan_handler = test_utils::TestChannelMessageHandler::new(); + let b_chan_handler = test_utils::TestChannelMessageHandler::new(); + let mut peers = create_network(2, &cfgs); + let (fd_a, mut fd_b) = establish_connection(&peers[0], &peers[1]); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); + + let secp_ctx = Secp256k1::new(); + let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret); + + let msg = msgs::Shutdown { channel_id: [42; 32], scriptpubkey: bitcoin::Script::new() }; + a_chan_handler.pending_events.lock().unwrap().push(events::MessageSendEvent::SendShutdown { + node_id: their_id, msg: msg.clone() + }); + peers[0].message_handler.chan_handler = &a_chan_handler; + + b_chan_handler.expect_receive_msg(wire::Message::Shutdown(msg)); + peers[1].message_handler.chan_handler = &b_chan_handler; + + peers[0].process_events(); + + let a_data = fd_a.outbound_data.lock().unwrap().split_off(0); + assert_eq!(peers[1].read_event(&mut fd_b, &a_data).unwrap(), false); + } + + #[test] + fn test_disconnect_all_peer() { + // Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and + // then calls disconnect_all_peers + let cfgs = create_peermgr_cfgs(2); + let peers = create_network(2, &cfgs); + establish_connection(&peers[0], &peers[1]); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); + + peers[0].disconnect_all_peers(); + assert_eq!(peers[0].peers.read().unwrap().len(), 0); } #[test] @@ -1957,17 +2085,17 @@ mod tests { let cfgs = create_peermgr_cfgs(2); let peers = create_network(2, &cfgs); establish_connection(&peers[0], &peers[1]); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 1); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); // peers[0] awaiting_pong is set to true, but the Peer is still connected peers[0].timer_tick_occurred(); peers[0].process_events(); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 1); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); // Since timer_tick_occurred() is called again when awaiting_pong is true, all Peers are disconnected peers[0].timer_tick_occurred(); peers[0].process_events(); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 0); + assert_eq!(peers[0].peers.read().unwrap().len(), 0); } #[test] @@ -2006,10 +2134,10 @@ mod tests { // Check that each peer has received the expected number of channel updates and channel // announcements. - assert_eq!(cfgs[0].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 100); - assert_eq!(cfgs[0].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 50); - assert_eq!(cfgs[1].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 100); - assert_eq!(cfgs[1].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 50); + assert_eq!(cfgs[0].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 108); + assert_eq!(cfgs[0].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 54); + assert_eq!(cfgs[1].routing_handler.chan_upds_recvd.load(Ordering::Acquire), 108); + assert_eq!(cfgs[1].routing_handler.chan_anns_recvd.load(Ordering::Acquire), 54); } #[test] @@ -2029,20 +2157,22 @@ mod tests { peers[0].new_inbound_connection(fd_a.clone(), None).unwrap(); // If we get a single timer tick before completion, that's fine - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 1); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); peers[0].timer_tick_occurred(); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 1); + assert_eq!(peers[0].peers.read().unwrap().len(), 1); assert_eq!(peers[0].read_event(&mut fd_a, &initial_data).unwrap(), false); peers[0].process_events(); - assert_eq!(peers[1].read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap(), false); + let a_data = fd_a.outbound_data.lock().unwrap().split_off(0); + assert_eq!(peers[1].read_event(&mut fd_b, &a_data).unwrap(), false); peers[1].process_events(); // ...but if we get a second timer tick, we should disconnect the peer peers[0].timer_tick_occurred(); - assert_eq!(peers[0].peers.read().unwrap().peers.len(), 0); + assert_eq!(peers[0].peers.read().unwrap().len(), 0); - assert!(peers[0].read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).is_err()); + let b_data = fd_b.outbound_data.lock().unwrap().split_off(0); + assert!(peers[0].read_event(&mut fd_a, &b_data).is_err()); } #[test]