X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fpeer_handler.rs;h=48bad2d69e16d59c5dd3d3a1436dc94436254f2a;hb=84d3630dffd31c133ff1b7380505597812f6c12f;hp=a701200be19af3824835b937f9faf6a413be2ddd;hpb=4703d4e72565ddfd150b9368ea036f4973fd7590;p=rust-lightning

diff --git a/lightning/src/ln/peer_handler.rs b/lightning/src/ln/peer_handler.rs
index a701200b..48bad2d6 100644
--- a/lightning/src/ln/peer_handler.rs
+++ b/lightning/src/ln/peer_handler.rs
@@ -21,27 +21,41 @@ use ln::features::InitFeatures;
 use ln::msgs;
 use ln::msgs::{ChannelMessageHandler, LightningError, RoutingMessageHandler};
 use ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager};
-use util::ser::{VecWriter, Writeable};
+use util::ser::{VecWriter, Writeable, Writer};
 use ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep};
 use ln::wire;
-use ln::wire::Encode;
 use util::byte_utils;
 use util::events::{MessageSendEvent, MessageSendEventsProvider};
 use util::logger::Logger;
 use routing::network_graph::NetGraphMsgHandler;
 
 use prelude::*;
+use io;
 use alloc::collections::LinkedList;
-use std::sync::{Arc, Mutex};
+use sync::{Arc, Mutex};
 use core::sync::atomic::{AtomicUsize, Ordering};
 use core::{cmp, hash, fmt, mem};
 use core::ops::Deref;
-use std::error;
+use core::convert::Infallible;
+#[cfg(feature = "std")] use std::error;
 
 use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::hashes::sha256::HashEngine as Sha256Engine;
 use bitcoin::hashes::{HashEngine, Hash};
 
+/// Handler for BOLT1-compliant messages.
+pub trait CustomMessageHandler: wire::CustomMessageReader {
+	/// Called with the message type that was received and the buffer to be read.
+	/// Can return a `MessageHandlingError` if the message could not be handled.
+	fn handle_custom_message(&self, msg: Self::CustomMessage, sender_node_id: &PublicKey) -> Result<(), LightningError>;
+
+	/// Gets the list of pending messages which were generated by the custom message
+	/// handler, clearing the list in the process. The first tuple element must
+	/// correspond to the intended recipients node ids. If no connection to one of the
+	/// specified node does not exist, the message is simply not sent to it.
+	fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)>;
+}
+
 /// A dummy struct which implements `RoutingMessageHandler` without storing any routing information
 /// or doing any processing. You can provide one of these as the route_handler in a MessageHandler.
 pub struct IgnoringMessageHandler{}
@@ -52,7 +66,6 @@ impl RoutingMessageHandler for IgnoringMessageHandler {
 	fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result<bool, LightningError> { Ok(false) }
 	fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result<bool, LightningError> { Ok(false) }
 	fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result<bool, LightningError> { Ok(false) }
-	fn handle_htlc_fail_channel_update(&self, _update: &msgs::HTLCFailChannelUpdate) {}
 	fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) ->
 		Vec<(msgs::ChannelAnnouncement, Option<msgs::ChannelUpdate>, Option<msgs::ChannelUpdate>)> { Vec::new() }
 	fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<msgs::NodeAnnouncement> { Vec::new() }
@@ -67,6 +80,35 @@ impl Deref for IgnoringMessageHandler {
 	fn deref(&self) -> &Self { self }
 }
 
+// Implement Type for Infallible, note that it cannot be constructed, and thus you can never call a
+// method that takes self for it.
+impl wire::Type for Infallible {
+	fn type_id(&self) -> u16 {
+		unreachable!();
+	}
+}
+impl Writeable for Infallible {
+	fn write<W: Writer>(&self, _: &mut W) -> Result<(), io::Error> {
+		unreachable!();
+	}
+}
+
+impl wire::CustomMessageReader for IgnoringMessageHandler {
+	type CustomMessage = Infallible;
+	fn read<R: io::Read>(&self, _message_type: u16, _buffer: &mut R) -> Result<Option<Self::CustomMessage>, msgs::DecodeError> {
+		Ok(None)
+	}
+}
+
+impl CustomMessageHandler for IgnoringMessageHandler {
+	fn handle_custom_message(&self, _msg: Infallible, _sender_node_id: &PublicKey) -> Result<(), LightningError> {
+		// Since we always return `None` in the read the handle method should never be called.
+		unreachable!();
+	}
+
+	fn get_and_clear_pending_msg(&self) -> Vec<(PublicKey, Self::CustomMessage)> { Vec::new() }
+}
+
 /// A dummy struct which implements `ChannelMessageHandler` without having any channels.
 /// You can provide one of these as the route_handler in a MessageHandler.
 pub struct ErroringMessageHandler {
@@ -160,10 +202,15 @@ pub struct MessageHandler<CM: Deref, RM: Deref> where
 		CM::Target: ChannelMessageHandler,
 		RM::Target: RoutingMessageHandler {
 	/// A message handler which handles messages specific to channels. Usually this is just a
-	/// ChannelManager object or a ErroringMessageHandler.
+	/// [`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 NetGraphMsgHandlerMonitor object or an IgnoringMessageHandler.
+	/// graph. Usually this is just a [`NetGraphMsgHandler`] object or an
+	/// [`IgnoringMessageHandler`].
+	///
+	/// [`NetGraphMsgHandler`]: crate::routing::network_graph::NetGraphMsgHandler
 	pub route_handler: RM,
 }
 
@@ -173,29 +220,35 @@ pub struct MessageHandler<CM: Deref, RM: Deref> where
 ///
 /// For efficiency, Clone should be relatively cheap for this type.
 ///
-/// You probably want to just extend an int and put a file descriptor in a struct and implement
-/// send_data. Note that if you are using a higher-level net library that may call close() itself,
-/// be careful to ensure you don't have races whereby you might register a new connection with an
-/// fd which is the same as a previous one which has yet to be removed via
-/// PeerManager::socket_disconnected().
+/// Two descriptors may compare equal (by [`cmp::Eq`] and [`hash::Hash`]) as long as the original
+/// has been disconnected, the [`PeerManager`] has been informed of the disconnection (either by it
+/// having triggered the disconnection or a call to [`PeerManager::socket_disconnected`]), and no
+/// further calls to the [`PeerManager`] related to the original socket occur. This allows you to
+/// use a file descriptor for your SocketDescriptor directly, however for simplicity you may wish
+/// to simply use another value which is guaranteed to be globally unique instead.
 pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone {
 	/// Attempts to send some data from the given slice to the peer.
 	///
 	/// Returns the amount of data which was sent, possibly 0 if the socket has since disconnected.
-	/// Note that in the disconnected case, socket_disconnected must still fire and further write
-	/// attempts may occur until that time.
+	/// Note that in the disconnected case, [`PeerManager::socket_disconnected`] must still be
+	/// called and further write attempts may occur until that time.
 	///
-	/// If the returned size is smaller than data.len(), a write_available event must
-	/// trigger the next time more data can be written. Additionally, until the a send_data event
-	/// completes fully, no further read_events should trigger on the same peer!
+	/// If the returned size is smaller than `data.len()`, a
+	/// [`PeerManager::write_buffer_space_avail`] call must be made the next time more data can be
+	/// written. Additionally, until a `send_data` event completes fully, no further
+	/// [`PeerManager::read_event`] calls should be made for the same peer! Because this is to
+	/// prevent denial-of-service issues, you should not read or buffer any data from the socket
+	/// until then.
 	///
-	/// If a read_event on this descriptor had previously returned true (indicating that read
-	/// events should be paused to prevent DoS in the send buffer), resume_read may be set
-	/// indicating that read events on this descriptor should resume. A resume_read of false does
-	/// *not* imply that further read events should be paused.
+	/// If a [`PeerManager::read_event`] call on this descriptor had previously returned true
+	/// (indicating that read events should be paused to prevent DoS in the send buffer),
+	/// `resume_read` may be set indicating that read events on this descriptor should resume. A
+	/// `resume_read` of false carries no meaning, and should not cause any action.
 	fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize;
 	/// Disconnect the socket pointed to by this SocketDescriptor.
-	/// No [`PeerManager::socket_disconnected`] call need be generated as a result of this call.
+	///
+	/// You do *not* need to call [`PeerManager::socket_disconnected`] with this socket after this
+	/// call (doing so is a noop).
 	fn disconnect_socket(&mut self);
 }
 
@@ -218,6 +271,8 @@ impl fmt::Display for PeerHandleError {
 		formatter.write_str("Peer Sent Invalid Data")
 	}
 }
+
+#[cfg(feature = "std")]
 impl error::Error for PeerHandleError {
 	fn description(&self) -> &str {
 		"Peer Sent Invalid Data"
@@ -230,6 +285,10 @@ enum InitSyncTracker{
 	NodesSyncing(PublicKey),
 }
 
+/// The ratio between buffer sizes at which we stop sending initial sync messages vs when we stop
+/// forwarding gossip messages to peers altogether.
+const FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO: usize = 2;
+
 /// When the outbound buffer has this many messages, we'll stop reading bytes from the peer until
 /// 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,
@@ -237,7 +296,29 @@ enum InitSyncTracker{
 const OUTBOUND_BUFFER_LIMIT_READ_PAUSE: usize = 10;
 /// 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 = 20;
+const OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP: usize = OUTBOUND_BUFFER_LIMIT_READ_PAUSE * FORWARD_INIT_SYNC_BUFFER_LIMIT_RATIO;
+
+/// If we've sent a ping, and are still awaiting a response, we may need to churn our way through
+/// the socket receive buffer before receiving the ping.
+///
+/// On a fairly old Arm64 board, with Linux defaults, this can take as long as 20 seconds, not
+/// including any network delays, outbound traffic, or the same for messages from other peers.
+///
+/// Thus, to avoid needlessly disconnecting a peer, we allow a peer to take this many timer ticks
+/// per connected peer to respond to a ping, as long as they send us at least one message during
+/// each tick, ensuring we aren't actually just disconnected.
+/// With a timer tick interval of five seconds, this translates to about 30 seconds per connected
+/// peer.
+///
+/// When we improve parallelism somewhat we should reduce this to e.g. this many timer ticks per
+/// two connected peers, assuming most LDK-running systems have at least two cores.
+const MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER: i8 = 6;
+
+/// This is the minimum number of messages we expect a peer to be able to handle within one timer
+/// 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.
+const BUFFER_DRAIN_MSGS_PER_TICK: usize = 32;
 
 struct Peer {
 	channel_encryptor: PeerChannelEncryptor,
@@ -254,7 +335,9 @@ struct Peer {
 
 	sync_status: InitSyncTracker,
 
-	awaiting_pong: bool,
+	msgs_sent_since_pong: usize,
+	awaiting_pong_tick_intervals: i8,
+	received_message_since_timer_tick: bool,
 }
 
 impl Peer {
@@ -299,7 +382,7 @@ fn _check_usize_is_32_or_64() {
 /// lifetimes). Other times you can afford a reference, which is more efficient, in which case
 /// SimpleRefPeerManager is the more appropriate type. Defining these type aliases prevents
 /// issues such as overly long function definitions.
-pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<C>, Arc<L>>>, Arc<L>>;
+pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<C>, Arc<L>>>, Arc<L>, Arc<IgnoringMessageHandler>>;
 
 /// 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
@@ -307,24 +390,37 @@ pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArc
 /// usage of lightning-net-tokio (since tokio::spawn requires parameters with static lifetimes).
 /// But if this is not necessary, using a reference is more efficient. Defining these type aliases
 /// helps with issues such as long function definitions.
-pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L>, &'f L>;
+pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L>, &'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 MessageHandlers.
+/// A PeerManager manages a set of peers, described by their [`SocketDescriptor`] and marshalls
+/// socket events into messages which it passes on to its [`MessageHandler`].
+///
+/// Locks are taken internally, so you must never assume that reentrancy from a
+/// [`SocketDescriptor`] call back into [`PeerManager`] methods will not deadlock.
+///
+/// Calls to [`read_event`] will decode relevant messages and pass them to the
+/// [`ChannelMessageHandler`], likely doing message processing in-line. Thus, the primary form of
+/// parallelism in Rust-Lightning is in calls to [`read_event`]. Note, however, that calls to any
+/// [`PeerManager`] functions related to the same connection must occur only in serial, making new
+/// calls only after previous ones have returned.
 ///
 /// Rather than using a plain PeerManager, it is preferable to use either a SimpleArcPeerManager
 /// a SimpleRefPeerManager, for conciseness. See their documentation for more details, but
 /// essentially you should default to using a SimpleRefPeerManager, and use a
 /// SimpleArcPeerManager when you require a PeerManager with a static lifetime, such as when
 /// you're using lightning-net-tokio.
-pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> where
+///
+/// [`read_event`]: PeerManager::read_event
+pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> where
 		CM::Target: ChannelMessageHandler,
 		RM::Target: RoutingMessageHandler,
-		L::Target: Logger {
+		L::Target: Logger,
+		CMH::Target: CustomMessageHandler {
 	message_handler: MessageHandler<CM, RM>,
 	peers: Mutex<PeerHolder<Descriptor>>,
 	our_node_secret: SecretKey,
 	ephemeral_key_midstate: Sha256Engine,
+	custom_message_handler: CMH,
 
 	// Usize needs to be at least 32 bits to avoid overflowing both low and high. If usize is 64
 	// bits we will never realistically count into high:
@@ -359,7 +455,7 @@ macro_rules! encode_msg {
 	}}
 }
 
-impl<Descriptor: SocketDescriptor, CM: Deref, L: Deref> PeerManager<Descriptor, CM, IgnoringMessageHandler, L> where
+impl<Descriptor: SocketDescriptor, CM: Deref, L: Deref> PeerManager<Descriptor, CM, IgnoringMessageHandler, L, IgnoringMessageHandler> where
 		CM::Target: ChannelMessageHandler,
 		L::Target: Logger {
 	/// Constructs a new PeerManager with the given ChannelMessageHandler. No routing message
@@ -373,11 +469,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, L: Deref> PeerManager<Descriptor,
 		Self::new(MessageHandler {
 			chan_handler: channel_message_handler,
 			route_handler: IgnoringMessageHandler{},
-		}, our_node_secret, ephemeral_random_data, logger)
+		}, our_node_secret, ephemeral_random_data, logger, IgnoringMessageHandler{})
 	}
 }
 
-impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref> PeerManager<Descriptor, ErroringMessageHandler, RM, L> where
+impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref> PeerManager<Descriptor, ErroringMessageHandler, RM, L, IgnoringMessageHandler> where
 		RM::Target: RoutingMessageHandler,
 		L::Target: Logger {
 	/// Constructs a new PeerManager with the given RoutingMessageHandler. No channel message
@@ -393,20 +489,19 @@ impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref> PeerManager<Descriptor,
 		Self::new(MessageHandler {
 			chan_handler: ErroringMessageHandler::new(),
 			route_handler: routing_message_handler,
-		}, our_node_secret, ephemeral_random_data, logger)
+		}, our_node_secret, ephemeral_random_data, logger, IgnoringMessageHandler{})
 	}
 }
 
-/// Manages and reacts to connection events. You probably want to use file descriptors as PeerIds.
-/// PeerIds may repeat, but only after socket_disconnected() has been called.
-impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<Descriptor, CM, RM, L> where
+impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, CMH: Deref> PeerManager<Descriptor, CM, RM, L, CMH> where
 		CM::Target: ChannelMessageHandler,
 		RM::Target: RoutingMessageHandler,
-		L::Target: Logger {
+		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<CM, RM>, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L) -> Self {
+	pub fn new(message_handler: MessageHandler<CM, RM>, 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);
 
@@ -421,6 +516,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 			peer_counter_low: AtomicUsize::new(0),
 			peer_counter_high: AtomicUsize::new(0),
 			logger,
+			custom_message_handler,
 		}
 	}
 
@@ -458,8 +554,10 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 	///
 	/// 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 had a
-	/// socket_disconnected().
+	/// Panics if descriptor is duplicative with some other descriptor which has not yet been
+	/// [`socket_disconnected()`].
+	///
+	/// [`socket_disconnected()`]: PeerManager::socket_disconnected
 	pub fn new_outbound_connection(&self, their_node_id: PublicKey, descriptor: Descriptor) -> Result<Vec<u8>, 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();
@@ -481,7 +579,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 			sync_status: InitSyncTracker::NoSyncRequested,
 
-			awaiting_pong: false,
+			msgs_sent_since_pong: 0,
+			awaiting_pong_tick_intervals: 0,
+			received_message_since_timer_tick: false,
 		}).is_some() {
 			panic!("PeerManager driver duplicated descriptors!");
 		};
@@ -495,8 +595,10 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 	/// call socket_disconnected for the new descriptor but must disconnect the connection
 	/// immediately.
 	///
-	/// Panics if descriptor is duplicative with some other descriptor which has not yet had
-	/// socket_disconnected called.
+	/// 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) -> Result<(), PeerHandleError> {
 		let peer_encryptor = PeerChannelEncryptor::new_inbound(&self.our_node_secret);
 		let pending_read_buffer = [0; 50].to_vec(); // Noise act one is 50 bytes
@@ -517,7 +619,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 			sync_status: InitSyncTracker::NoSyncRequested,
 
-			awaiting_pong: false,
+			msgs_sent_since_pong: 0,
+			awaiting_pong_tick_intervals: 0,
+			received_message_since_timer_tick: false,
 		}).is_some() {
 			panic!("PeerManager driver duplicated descriptors!");
 		};
@@ -525,28 +629,20 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 	}
 
 	fn do_attempt_write_data(&self, descriptor: &mut Descriptor, peer: &mut Peer) {
-		macro_rules! encode_and_send_msg {
-			($msg: expr) => {
-				{
-					log_trace!(self.logger, "Encoding and sending sync update message of type {} to {}", $msg.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
-					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg)[..]));
-				}
-			}
-		}
 		while !peer.awaiting_write_event {
-			if peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE {
+			if peer.pending_outbound_buffer.len() < OUTBOUND_BUFFER_LIMIT_READ_PAUSE && peer.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK {
 				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() {
-							encode_and_send_msg!(announce);
+							self.enqueue_message(peer, announce);
 							if let &Some(ref update_a) = update_a_option {
-								encode_and_send_msg!(update_a);
+								self.enqueue_message(peer, update_a);
 							}
 							if let &Some(ref update_b) = update_b_option {
-								encode_and_send_msg!(update_b);
+								self.enqueue_message(peer, update_b);
 							}
 							peer.sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
 						}
@@ -558,7 +654,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 						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() {
-							encode_and_send_msg!(msg);
+							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 {
@@ -570,7 +666,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 						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() {
-							encode_and_send_msg!(msg);
+							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 {
@@ -579,6 +675,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 					},
 				}
 			}
+			if peer.msgs_sent_since_pong >= BUFFER_DRAIN_MSGS_PER_TICK {
+				self.maybe_send_extra_ping(peer);
+			}
 
 			if {
 				let next_buff = match peer.pending_outbound_buffer.front() {
@@ -604,12 +703,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 	///
 	/// May return an Err to indicate that the connection should be closed.
 	///
-	/// Will most likely call send_data on the descriptor passed in (or the descriptor handed into
-	/// new_*\_connection) before returning. Thus, be very careful with reentrancy issues! The
-	/// invariants around calling write_buffer_space_avail in case a write did not fully complete
-	/// must still hold - be ready to call write_buffer_space_avail again if a write call generated
-	/// here isn't sufficient! Panics if the descriptor was not previously registered in a
-	/// new_\*_connection event.
+	/// May call [`send_data`] on the descriptor passed in (or an equal descriptor) before
+	/// returning. Thus, be very careful with reentrancy issues! The invariants around calling
+	/// [`write_buffer_space_avail`] in case a write did not fully complete must still hold - be
+	/// ready to call `[write_buffer_space_avail`] again if a write call generated here isn't
+	/// sufficient!
+	///
+	/// [`send_data`]: SocketDescriptor::send_data
+	/// [`write_buffer_space_avail`]: PeerManager::write_buffer_space_avail
 	pub fn write_buffer_space_avail(&self, descriptor: &mut Descriptor) -> Result<(), PeerHandleError> {
 		let mut peers = self.peers.lock().unwrap();
 		match peers.peers.get_mut(descriptor) {
@@ -631,33 +732,41 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 	///
 	/// May return an Err to indicate that the connection should be closed.
 	///
-	/// 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.
+	/// Will *not* call back into [`send_data`] on any descriptors to avoid reentrancy complexity.
+	/// Thus, however, you should 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 send_data call
-	/// on this file descriptor has resume_read set (preventing DoS issues in the send buffer).
+	/// If `Ok(true)` is returned, further read_events should not be triggered until a
+	/// [`send_data`] call on this descriptor has `resume_read` set (preventing DoS issues in the
+	/// send buffer).
 	///
+	/// [`send_data`]: SocketDescriptor::send_data
+	/// [`process_events`]: PeerManager::process_events
 	pub fn read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result<bool, PeerHandleError> {
 		match self.do_read_event(peer_descriptor, data) {
 			Ok(res) => Ok(res),
 			Err(e) => {
+				log_trace!(self.logger, "Peer sent invalid data or we decided to disconnect due to a protocol error");
 				self.disconnect_event_internal(peer_descriptor, e.no_connection_possible);
 				Err(e)
 			}
 		}
 	}
 
-	/// Append a message to a peer's pending outbound/write buffer, and update the map of peers needing sends accordingly.
-	fn enqueue_message<M: Encode + Writeable>(&self, peer: &mut Peer, message: &M) {
-		let mut buffer = VecWriter(Vec::new());
-		wire::write(message, &mut buffer).unwrap(); // crash if the write failed
-		let encoded_message = buffer.0;
-
-		log_trace!(self.logger, "Enqueueing message of type {} to {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+	/// Append a message to a peer's pending outbound/write buffer
+	fn enqueue_encoded_message(&self, peer: &mut Peer, encoded_message: &Vec<u8>) {
+		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<M: wire::Type>(&self, peer: &mut Peer, message: &M) {
+		let mut buffer = VecWriter(Vec::with_capacity(2048));
+		wire::write(message, &mut buffer).unwrap(); // crash if the write failed
+		log_trace!(self.logger, "Enqueueing message {:?} to {}", message, log_pubkey!(peer.their_node_id.unwrap()));
+		self.enqueue_encoded_message(peer, &buffer.0);
+	}
+
 	fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: &[u8]) -> Result<bool, PeerHandleError> {
 		let pause_read = {
 			let mut peers_lock = self.peers.lock().unwrap();
@@ -695,21 +804,25 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 											match e.action {
 												msgs::ErrorAction::DisconnectPeer { msg: _ } => {
 													//TODO: Try to push msg
-													log_trace!(self.logger, "Got Err handling message, disconnecting peer because {}", e.err);
+													log_debug!(self.logger, "Error handling message; disconnecting peer with: {}", e.err);
 													return Err(PeerHandleError{ no_connection_possible: false });
 												},
+												msgs::ErrorAction::IgnoreAndLog(level) => {
+													log_given_level!(self.logger, level, "Error handling message; ignoring: {}", e.err);
+													continue
+												},
 												msgs::ErrorAction::IgnoreError => {
-													log_trace!(self.logger, "Got Err handling message, ignoring because {}", e.err);
+													log_debug!(self.logger, "Error handling message; ignoring: {}", e.err);
 													continue;
 												},
 												msgs::ErrorAction::SendErrorMessage { msg } => {
-													log_trace!(self.logger, "Got Err handling message, sending Error message because {}", e.err);
+													log_debug!(self.logger, "Error handling message; sending error message with: {}", e.err);
 													self.enqueue_message(peer, &msg);
 													continue;
 												},
 											}
 										}
-									};
+									}
 								}
 							}
 
@@ -722,7 +835,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 											return Err(PeerHandleError{ no_connection_possible: false })
 										},
 										hash_map::Entry::Vacant(entry) => {
-											log_trace!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap()));
+											log_debug!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap()));
 											entry.insert(peer_descriptor.clone())
 										},
 									};
@@ -775,15 +888,15 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 										peer.pending_read_buffer = [0; 18].to_vec();
 										peer.pending_read_is_header = true;
 
-										let mut reader = ::std::io::Cursor::new(&msg_data[..]);
-										let message_result = wire::read(&mut reader);
+										let mut reader = io::Cursor::new(&msg_data[..]);
+										let message_result = wire::read(&mut reader, &*self.custom_message_handler);
 										let message = match message_result {
 											Ok(x) => x,
 											Err(e) => {
 												match e {
 													msgs::DecodeError::UnknownVersion => return Err(PeerHandleError { no_connection_possible: false }),
 													msgs::DecodeError::UnknownRequiredFeature => {
-														log_debug!(self.logger, "Got a channel/node announcement with an known required feature flag, you may want to update!");
+														log_trace!(self.logger, "Got a channel/node announcement with an known required feature flag, you may want to update!");
 														continue;
 													}
 													msgs::DecodeError::InvalidValue => {
@@ -797,7 +910,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 													msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError { no_connection_possible: false }),
 													msgs::DecodeError::Io(_) => return Err(PeerHandleError { no_connection_possible: false }),
 													msgs::DecodeError::UnsupportedCompression => {
-														log_debug!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
+														log_trace!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
 														continue;
 													}
 												}
@@ -839,13 +952,18 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 	/// Process an incoming message and return a decision (ok, lightning error, peer handling error) regarding the next action with the peer
 	/// Returns the message back if it needs to be broadcasted to all other peers.
-	fn handle_message(&self, peer: &mut Peer, message: wire::Message) -> Result<Option<wire::Message>, MessageHandlingError> {
-		log_trace!(self.logger, "Received message of type {} from {}", message.type_id(), log_pubkey!(peer.their_node_id.unwrap()));
+	fn handle_message(
+		&self,
+		peer: &mut Peer,
+		message: wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>
+	) -> Result<Option<wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>>, MessageHandlingError> {
+		log_trace!(self.logger, "Received message {:?} from {}", message, log_pubkey!(peer.their_node_id.unwrap()));
+		peer.received_message_since_timer_tick = true;
 
 		// Need an Init as first message
 		if let wire::Message::Init(_) = message {
 		} else if peer.their_features.is_none() {
-			log_trace!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
+			log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
 			return Err(PeerHandleError{ no_connection_possible: false }.into());
 		}
 
@@ -855,22 +973,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 			// Setup and Control messages:
 			wire::Message::Init(msg) => {
 				if msg.features.requires_unknown_bits() {
-					log_info!(self.logger, "Peer features required unknown version bits");
+					log_debug!(self.logger, "Peer features required unknown version bits");
 					return Err(PeerHandleError{ no_connection_possible: true }.into());
 				}
 				if peer.their_features.is_some() {
 					return Err(PeerHandleError{ no_connection_possible: false }.into());
 				}
 
-				log_info!(
-					self.logger, "Received peer Init message: data_loss_protect: {}, initial_routing_sync: {}, upfront_shutdown_script: {}, gossip_queries: {}, static_remote_key: {}, unknown flags (local and global): {}",
-					if msg.features.supports_data_loss_protect() { "supported" } else { "not supported"},
-					if msg.features.initial_routing_sync() { "requested" } else { "not requested" },
-					if msg.features.supports_upfront_shutdown_script() { "supported" } else { "not supported"},
-					if msg.features.supports_gossip_queries() { "supported" } else { "not supported" },
-					if msg.features.supports_static_remote_key() { "supported" } else { "not supported"},
-					if msg.features.supports_unknown_bits() { "present" } else { "none" }
-				);
+				log_info!(self.logger, "Received peer Init message: {}", msg.features);
 
 				if msg.features.initial_routing_sync() {
 					peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
@@ -912,7 +1022,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 				}
 			},
 			wire::Message::Pong(_msg) => {
-				peer.awaiting_pong = false;
+				peer.awaiting_pong_tick_intervals = 0;
+				peer.msgs_sent_since_pong = 0;
 			},
 
 			// Channel messages:
@@ -1007,21 +1118,25 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 			},
 
 			// Unknown messages:
-			wire::Message::Unknown(msg_type) if msg_type.is_even() => {
-				log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", msg_type);
+			wire::Message::Unknown(type_id) if message.is_even() => {
+				log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", type_id);
 				// Fail the channel if message is an even, unknown type as per BOLT #1.
 				return Err(PeerHandleError{ no_connection_possible: true }.into());
 			},
-			wire::Message::Unknown(msg_type) => {
-				log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", msg_type);
-			}
+			wire::Message::Unknown(type_id) => {
+				log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", type_id);
+			},
+			wire::Message::Custom(custom) => {
+				self.custom_message_handler.handle_custom_message(custom, &peer.their_node_id.unwrap())?;
+			},
 		};
 		Ok(should_forward)
 	}
 
-	fn forward_broadcast_msg(&self, peers: &mut PeerHolder<Descriptor>, msg: &wire::Message, except_node: Option<&PublicKey>) {
+	fn forward_broadcast_msg(&self, peers: &mut PeerHolder<Descriptor>, msg: &wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>, except_node: Option<&PublicKey>) {
 		match msg {
 			wire::Message::ChannelAnnouncement(ref msg) => {
+				log_trace!(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) in peers.peers.iter_mut() {
@@ -1029,7 +1144,10 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
 						continue
 					}
-					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+					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_SIZE_LIMIT
+					{
+						log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
 					}
 					if peer.their_node_id.as_ref() == Some(&msg.contents.node_id_1) ||
@@ -1039,10 +1157,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}
-					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+					self.enqueue_encoded_message(peer, &encoded_msg);
 				}
 			},
 			wire::Message::NodeAnnouncement(ref msg) => {
+				log_trace!(self.logger, "Sending message to all peers except {:?} or the announced node: {:?}", except_node, msg);
 				let encoded_msg = encode_msg!(msg);
 
 				for (_, peer) in peers.peers.iter_mut() {
@@ -1050,7 +1169,10 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							!peer.should_forward_node_announcement(msg.contents.node_id) {
 						continue
 					}
-					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+					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_SIZE_LIMIT
+					{
+						log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
 						continue;
 					}
 					if peer.their_node_id.as_ref() == Some(&msg.contents.node_id) {
@@ -1059,10 +1181,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 					if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
 						continue;
 					}
-					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+					self.enqueue_encoded_message(peer, &encoded_msg);
 				}
 			},
 			wire::Message::ChannelUpdate(ref msg) => {
+				log_trace!(self.logger, "Sending message to all peers except {:?}: {:?}", except_node, msg);
 				let encoded_msg = encode_msg!(msg);
 
 				for (_, peer) in peers.peers.iter_mut() {
@@ -1070,13 +1193,16 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							!peer.should_forward_channel_announcement(msg.contents.short_channel_id)  {
 						continue
 					}
-					if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+					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_SIZE_LIMIT
+					{
+						log_trace!(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;
 					}
-					peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+					self.enqueue_encoded_message(peer, &encoded_msg);
 				}
 			},
 			_ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"),
@@ -1085,7 +1211,17 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 	/// 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).
+	/// functions like [`ChannelManager::process_pending_htlc_forwards`] or [`send_payment`]).
+	///
+	/// May call [`send_data`] on [`SocketDescriptor`]s. Thus, be very careful with reentrancy
+	/// issues!
+	///
+	/// You don't have to call this function explicitly if you are using [`lightning-net-tokio`]
+	/// or one of the other clients provided in our language bindings.
+	///
+	/// [`send_payment`]: crate::ln::channelmanager::ChannelManager::send_payment
+	/// [`ChannelManager::process_pending_htlc_forwards`]: crate::ln::channelmanager::ChannelManager::process_pending_htlc_forwards
+	/// [`send_data`]: SocketDescriptor::send_data
 	pub fn process_events(&self) {
 		{
 			// TODO: There are some DoS attacks here where you can flood someone's outbound send
@@ -1096,75 +1232,69 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 			let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_msg_events();
 			events_generated.append(&mut self.message_handler.route_handler.get_and_clear_pending_msg_events());
 			let peers = &mut *peers_lock;
-			for event in events_generated.drain(..) {
-				macro_rules! get_peer_for_forwarding {
-					($node_id: expr) => {
-						{
-							match peers.node_id_to_descriptor.get($node_id) {
-								Some(descriptor) => match peers.peers.get_mut(&descriptor) {
-									Some(peer) => {
-										if peer.their_features.is_none() {
-											continue;
-										}
-										peer
-									},
-									None => panic!("Inconsistent peers set state!"),
-								},
-								None => {
-									continue;
+			macro_rules! get_peer_for_forwarding {
+				($node_id: expr) => {
+					{
+						match peers.node_id_to_descriptor.get($node_id) {
+							Some(descriptor) => match peers.peers.get_mut(&descriptor) {
+								Some(peer) => {
+									if peer.their_features.is_none() {
+										continue;
+									}
+									peer
 								},
-							}
+								None => panic!("Inconsistent peers set state!"),
+							},
+							None => {
+								continue;
+							},
 						}
 					}
 				}
+			}
+			for event in events_generated.drain(..) {
 				match event {
 					MessageSendEvent::SendAcceptChannel { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.temporary_channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendOpenChannel { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.temporary_channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendFundingCreated { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
+						log_debug!(self.logger, "Handling SendFundingCreated event in peer_handler for node {} for channel {} (which becomes {})",
 								log_pubkey!(node_id),
 								log_bytes!(msg.temporary_channel_id),
 								log_funding_channel_id!(msg.funding_txid, msg.funding_output_index));
 						// TODO: If the peer is gone we should generate a DiscardFunding event
 						// indicating to the wallet that they should just throw away this funding transaction
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendFundingSigned { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendFundingSigned event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendFundingLocked event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendAnnouncementSignatures { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
+						log_debug!(self.logger, "Handling SendAnnouncementSignatures event in peer_handler for node {} for channel {})",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					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.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
+						log_debug!(self.logger, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
 								log_pubkey!(node_id),
 								update_add_htlcs.len(),
 								update_fulfill_htlcs.len(),
@@ -1172,71 +1302,70 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 								log_bytes!(commitment_signed.channel_id));
 						let peer = get_peer_for_forwarding!(node_id);
 						for msg in update_add_htlcs {
-							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+							self.enqueue_message(peer, msg);
 						}
 						for msg in update_fulfill_htlcs {
-							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+							self.enqueue_message(peer, msg);
 						}
 						for msg in update_fail_htlcs {
-							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+							self.enqueue_message(peer, msg);
 						}
 						for msg in update_fail_malformed_htlcs {
-							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+							self.enqueue_message(peer, msg);
 						}
 						if let &Some(ref msg) = update_fee {
-							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+							self.enqueue_message(peer, msg);
 						}
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed)));
+						self.enqueue_message(peer, commitment_signed);
 					},
 					MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendClosingSigned { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendClosingSigned event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendShutdown { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling Shutdown event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
-						log_trace!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
+						log_debug!(self.logger, "Handling SendChannelReestablish event in peer_handler for node {} for channel {}",
 								log_pubkey!(node_id),
 								log_bytes!(msg.channel_id));
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::BroadcastChannelAnnouncement { msg, update_msg } => {
-						log_trace!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+						log_debug!(self.logger, "Handling BroadcastChannelAnnouncement event in peer_handler for short channel id {}", msg.contents.short_channel_id);
 						if self.message_handler.route_handler.handle_channel_announcement(&msg).is_ok() && self.message_handler.route_handler.handle_channel_update(&update_msg).is_ok() {
 							self.forward_broadcast_msg(peers, &wire::Message::ChannelAnnouncement(msg), None);
 							self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(update_msg), None);
 						}
 					},
 					MessageSendEvent::BroadcastNodeAnnouncement { msg } => {
-						log_trace!(self.logger, "Handling BroadcastNodeAnnouncement event in peer_handler");
+						log_debug!(self.logger, "Handling BroadcastNodeAnnouncement event in peer_handler");
 						if self.message_handler.route_handler.handle_node_announcement(&msg).is_ok() {
 							self.forward_broadcast_msg(peers, &wire::Message::NodeAnnouncement(msg), None);
 						}
 					},
 					MessageSendEvent::BroadcastChannelUpdate { msg } => {
-						log_trace!(self.logger, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
+						log_debug!(self.logger, "Handling BroadcastChannelUpdate event in peer_handler for short channel id {}", msg.contents.short_channel_id);
 						if self.message_handler.route_handler.handle_channel_update(&msg).is_ok() {
 							self.forward_broadcast_msg(peers, &wire::Message::ChannelUpdate(msg), None);
 						}
 					},
-					MessageSendEvent::PaymentFailureNetworkUpdate { ref update } => {
-						self.message_handler.route_handler.handle_htlc_fail_channel_update(update);
+					MessageSendEvent::SendChannelUpdate { ref node_id, ref msg } => {
+						log_trace!(self.logger, "Handling SendChannelUpdate event in peer_handler for node {} for channel {}",
+								log_pubkey!(node_id), msg.contents.short_channel_id);
+						let peer = get_peer_for_forwarding!(node_id);
+						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
 					},
 					MessageSendEvent::HandleError { ref node_id, ref action } => {
 						match *action {
@@ -1247,7 +1376,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 											log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
 													log_pubkey!(node_id),
 													msg.data);
-											peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+											self.enqueue_message(&mut peer, msg);
 											// 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);
@@ -1259,23 +1388,25 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 									self.message_handler.chan_handler.peer_disconnected(&node_id, false);
 								}
 							},
-							msgs::ErrorAction::IgnoreError => {},
+							msgs::ErrorAction::IgnoreAndLog(level) => {
+								log_given_level!(self.logger, level, "Received a HandleError event to be ignored for node {}", log_pubkey!(node_id));
+							},
+							msgs::ErrorAction::IgnoreError => {
+								log_debug!(self.logger, "Received a HandleError event to be ignored for node {}", log_pubkey!(node_id));
+							},
 							msgs::ErrorAction::SendErrorMessage { ref msg } => {
 								log_trace!(self.logger, "Handling SendErrorMessage HandleError event in peer_handler for node {} with message {}",
 										log_pubkey!(node_id),
 										msg.data);
-								let peer = get_peer_for_forwarding!(node_id);
-								peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+								self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 							},
 						}
 					},
 					MessageSendEvent::SendChannelRangeQuery { ref node_id, ref msg } => {
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					},
 					MessageSendEvent::SendShortIdsQuery { ref node_id, ref msg } => {
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					}
 					MessageSendEvent::SendReplyChannelRange { ref node_id, ref msg } => {
 						log_trace!(self.logger, "Handling SendReplyChannelRange event in peer_handler for node {} with num_scids={} first_blocknum={} number_of_blocks={}, sync_complete={}",
@@ -1284,12 +1415,15 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 							msg.first_blocknum,
 							msg.number_of_blocks,
 							msg.sync_complete);
-						let peer = get_peer_for_forwarding!(node_id);
-						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg)));
+						self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
 					}
 				}
 			}
 
+			for (node_id, msg) in self.custom_message_handler.get_and_clear_pending_msg() {
+				self.enqueue_message(get_peer_for_forwarding!(&node_id), &msg);
+			}
+
 			for (descriptor, peer) in peers.peers.iter_mut() {
 				self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
 			}
@@ -1297,10 +1431,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 	}
 
 	/// Indicates that the given socket descriptor's connection is now closed.
-	///
-	/// This need only be called if the socket has been disconnected by the peer or your own
-	/// decision to disconnect it and may be skipped in any case where other parts of this library
-	/// (eg PeerHandleError, explicit disconnect_socket calls) instruct you to disconnect the peer.
 	pub fn socket_disconnected(&self, descriptor: &Descriptor) {
 		self.disconnect_event_internal(descriptor, false);
 	}
@@ -1317,6 +1447,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 			Some(peer) => {
 				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 { "" });
 						peers.node_id_to_descriptor.remove(&node_id);
 						self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible);
 					},
@@ -1328,11 +1461,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 	/// Disconnect a peer given its node id.
 	///
-	/// Set no_connection_possible to true to prevent any further connection with this peer,
+	/// Set `no_connection_possible` to true to prevent any further connection with this peer,
 	/// force-closing any channels we have with it.
 	///
-	/// If a peer is connected, this will call `disconnect_socket` on the descriptor for the peer,
-	/// so be careful about reentrancy issues.
+	/// If a peer is connected, this will call [`disconnect_socket`] on the descriptor for the
+	/// peer. Thus, be very careful about reentrancy issues.
+	///
+	/// [`disconnect_socket`]: SocketDescriptor::disconnect_socket
 	pub fn disconnect_by_node_id(&self, node_id: PublicKey, no_connection_possible: bool) {
 		let mut peers_lock = self.peers.lock().unwrap();
 		if let Some(mut descriptor) = peers_lock.node_id_to_descriptor.remove(&node_id) {
@@ -1343,10 +1478,48 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 		}
 	}
 
-	/// This function should be called roughly once every 30 seconds.
-	/// It will send pings to each peer and disconnect those which did not respond to the last round of pings.
+	/// Disconnects all currently-connected peers. This is useful on platforms where there may be
+	/// an indication that TCP sockets have stalled even if we weren't around to time them out
+	/// using regular ping/pongs.
+	pub fn disconnect_all_peers(&self) {
+		let mut peers_lock = self.peers.lock().unwrap();
+		let peers = &mut *peers_lock;
+		for (mut descriptor, peer) in peers.peers.drain() {
+			if let Some(node_id) = peer.their_node_id {
+				log_trace!(self.logger, "Disconnecting peer with id {} due to client request to disconnect all peers", node_id);
+				peers.node_id_to_descriptor.remove(&node_id);
+				self.message_handler.chan_handler.peer_disconnected(&node_id, false);
+			}
+			descriptor.disconnect_socket();
+		}
+		debug_assert!(peers.node_id_to_descriptor.is_empty());
+	}
+
+	/// This is called when we're blocked on sending additional gossip messages until we receive a
+	/// pong. If we aren't waiting on a pong, we take this opportunity to send a ping (setting
+	/// `awaiting_pong_tick_intervals` to a special flag value to indicate this).
+	fn maybe_send_extra_ping(&self, peer: &mut Peer) {
+		if peer.awaiting_pong_tick_intervals == 0 {
+			peer.awaiting_pong_tick_intervals = -1;
+			let ping = msgs::Ping {
+				ponglen: 0,
+				byteslen: 64,
+			};
+			self.enqueue_message(peer, &ping);
+		}
+	}
 
-	/// Will most likely call send_data on all of the registered descriptors, thus, be very careful with reentrancy issues!
+	/// Send pings to each peer and disconnect those which did not respond to the last round of
+	/// pings.
+	///
+	/// This may be called on any timescale you want, however, roughly once every five to ten
+	/// seconds is preferred. The call rate determines both how often we send a ping to our peers
+	/// and how much time they have to respond before we disconnect them.
+	///
+	/// May call [`send_data`] on all [`SocketDescriptor`]s. Thus, be very careful with reentrancy
+	/// issues!
+	///
+	/// [`send_data`]: SocketDescriptor::send_data
 	pub fn timer_tick_occurred(&self) {
 		let mut peers_lock = self.peers.lock().unwrap();
 		{
@@ -1354,9 +1527,18 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 			let node_id_to_descriptor = &mut peers.node_id_to_descriptor;
 			let peers = &mut peers.peers;
 			let mut descriptors_needing_disconnect = Vec::new();
+			let peer_count = peers.len();
 
 			peers.retain(|descriptor, peer| {
-				if peer.awaiting_pong {
+				if !peer.channel_encryptor.is_ready_for_encryption() {
+					// The peer needs to complete its handshake before we can exchange messages
+					return true;
+				}
+
+				if (peer.awaiting_pong_tick_intervals > 0 && !peer.received_message_since_timer_tick)
+					|| peer.awaiting_pong_tick_intervals as u64 >
+						MAX_BUFFER_DRAIN_TICK_INTERVALS_PER_PEER as u64 * peer_count as u64
+				{
 					descriptors_needing_disconnect.push(descriptor.clone());
 					match peer.their_node_id {
 						Some(node_id) => {
@@ -1373,21 +1555,26 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 					return false;
 				}
 
-				if !peer.channel_encryptor.is_ready_for_encryption() {
-					// The peer needs to complete its handshake before we can exchange messages
+				peer.received_message_since_timer_tick = false;
+				if peer.awaiting_pong_tick_intervals == -1 {
+					// Magic value set in `maybe_send_extra_ping`.
+					peer.awaiting_pong_tick_intervals = 1;
 					return true;
 				}
 
+				if peer.awaiting_pong_tick_intervals > 0 {
+					peer.awaiting_pong_tick_intervals += 1;
+					return true;
+				}
+
+				peer.awaiting_pong_tick_intervals = 1;
 				let ping = msgs::Ping {
 					ponglen: 0,
 					byteslen: 64,
 				};
-				peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(&ping)));
-
-				let mut descriptor_clone = descriptor.clone();
-				self.do_attempt_write_data(&mut descriptor_clone, peer);
+				self.enqueue_message(peer, &ping);
+				self.do_attempt_write_data(&mut (descriptor.clone()), &mut *peer);
 
-				peer.awaiting_pong = true;
 				true
 			});
 
@@ -1400,7 +1587,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 #[cfg(test)]
 mod tests {
-	use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
+	use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler};
 	use ln::msgs;
 	use util::events;
 	use util::test_utils;
@@ -1409,7 +1596,7 @@ mod tests {
 	use bitcoin::secp256k1::key::{SecretKey, PublicKey};
 
 	use prelude::*;
-	use std::sync::{Arc, Mutex};
+	use sync::{Arc, Mutex};
 	use core::sync::atomic::Ordering;
 
 	#[derive(Clone)]
@@ -1459,20 +1646,20 @@ mod tests {
 		cfgs
 	}
 
-	fn create_network<'a>(peer_count: usize, cfgs: &'a Vec<PeerManagerCfg>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger>> {
+	fn create_network<'a>(peer_count: usize, cfgs: &'a Vec<PeerManagerCfg>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger, IgnoringMessageHandler>> {
 		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 peer = PeerManager::new(msg_handler, node_secret, &ephemeral_bytes, &cfgs[i].logger);
+			let peer = PeerManager::new(msg_handler, node_secret, &ephemeral_bytes, &cfgs[i].logger, IgnoringMessageHandler {});
 			peers.push(peer);
 		}
 
 		peers
 	}
 
-	fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger>) -> (FileDescriptor, FileDescriptor) {
+	fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger, IgnoringMessageHandler>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger, IgnoringMessageHandler>) -> (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())) };
@@ -1546,11 +1733,23 @@ mod tests {
 		// than can fit into a peer's buffer).
 		let (mut fd_a, mut fd_b) = establish_connection(&peers[0], &peers[1]);
 
-		// Make each peer to read the messages that the other peer just wrote to them.
-		peers[0].process_events();
-		peers[1].read_event(&mut fd_b, &fd_a.outbound_data.lock().unwrap().split_off(0)).unwrap();
-		peers[1].process_events();
-		peers[0].read_event(&mut fd_a, &fd_b.outbound_data.lock().unwrap().split_off(0)).unwrap();
+		// Make each peer to read the messages that the other peer just wrote to them. Note that
+		// due to the max-messagse-before-bing limits this may take a few iterations to complete.
+		for _ in 0..150/super::BUFFER_DRAIN_MSGS_PER_TICK + 1 {
+			peers[0].process_events();
+			let b_read_data = fd_a.outbound_data.lock().unwrap().split_off(0);
+			assert!(!b_read_data.is_empty());
+
+			peers[1].read_event(&mut fd_b, &b_read_data).unwrap();
+			peers[1].process_events();
+
+			let a_read_data = fd_b.outbound_data.lock().unwrap().split_off(0);
+			assert!(!a_read_data.is_empty());
+			peers[0].read_event(&mut fd_a, &a_read_data).unwrap();
+
+			peers[1].process_events();
+			assert_eq!(fd_b.outbound_data.lock().unwrap().len(), 0, "Until B receives data, it shouldn't send more messages");
+		}
 
 		// Check that each peer has received the expected number of channel updates and channel
 		// announcements.