review: Collapse OutboundQueue/Transport traits

Reduce trait complexity in the OutboundQueue and Transport objects.
Splitting the objects into separate traits allowed for cleaner separation
of responsibilities. But in practice, this led to complexity in the
PeerManager function signatures and confusion in why they needed to be separated.

To move this module to a more maintainable state for the core
development team, collapse PayloadQueuer/SocketDescriptorFlusher into a
single IOutboundQueue trait.

Also, remove MessageQueuer in favor of just passing in the entire
Transport object.

This makes the code look more similar to the rest of the codebase while
still leveraging traits in the OutboundQueue and Transport layer that
allow for test doubles and real unit tests.

Author: julianknutsen

lightning/src/ln/peers/handler.rs

@@ -39,15 +39,15 @@ use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::hashes::sha256::HashEngine as Sha256Engine;
 use bitcoin::hashes::{HashEngine, Hash};
 use ln::peers::outbound_queue::OutboundQueue;
-use ln::peers::transport::{PayloadQueuer, Transport};
+use ln::peers::transport::Transport;
 use std::collections::hash_map::IterMut;
 use std::iter::Filter;
 
 // Number of items that can exist in the OutboundQueue before Sync message flow control is triggered
 const OUTBOUND_QUEUE_SIZE: usize = 10;
 
 /// Interface PeerManager uses to interact with the Transport object
-pub(super) trait ITransport: MessageQueuer {
+pub(super) trait ITransport {
 	/// Instantiate the new outbound Transport
 	fn new_outbound(initiator_static_private_key: &SecretKey, responder_static_public_key: &PublicKey, initiator_ephemeral_private_key: &SecretKey) -> Self;
 
@@ -59,29 +59,51 @@ pub(super) trait ITransport: MessageQueuer {
 
 	/// Process input data similar to reading it off a descriptor directly. Returns true on the first call
 	/// that results in the transport being newly connected.
-	fn process_input(&mut self, input: &[u8], output_buffer: &mut impl PayloadQueuer) -> Result<bool, String>;
+	fn process_input(&mut self, input: &[u8], outbound_queue: &mut impl IOutboundQueue) -> Result<bool, String>;
 
 	/// Returns true if the connection is established and encrypted messages can be sent.
 	fn is_connected(&self) -> bool;
 
 	/// Returns the node_id of the remote node. Panics if not connected.
 	fn get_their_node_id(&self) -> PublicKey;
 
+	/// Encodes, encrypts, and enqueues a message to the outbound queue. Panics if the connection is
+	/// not established yet.
+	fn enqueue_message<M: Encode + Writeable, L: Deref>(&mut self, message: &M, outbound_queue: &mut impl IOutboundQueue, logger: L) where L::Target: Logger;
+
 	/// Returns all Messages that have been received and can be successfully parsed by the Transport
 	fn drain_messages<L: Deref>(&mut self, logger: L) -> Result<Vec<Message>, PeerHandleError> where L::Target: Logger;
 }
 
-/// Interface PeerManager uses to queue message to send. Implemented by Transport to handle
-/// encryption/decryption post-NOISE.
-pub(super) trait MessageQueuer {
-	/// Encodes, encrypts, and enqueues a message to the outbound queue. Panics if the connection is
-	/// not established yet.
-	fn enqueue_message<M: Encode + Writeable, Q: PayloadQueuer, L: Deref>(&mut self, message: &M, output_buffer: &mut Q, logger: L) where L::Target: Logger;
-}
+/// The OutboundQueue is a container for unencrypted payloads during the NOISE handshake and
+/// encrypted Messages post-NOISE. This trait abstracts the behavior to push items to a queue, flush
+/// them through a SocketDescriptor, and handle flow control. Each Peer owns a separate OutboundQueue.
+///
+/// A trait is used to enable tests to use test doubles that implement a subset of the api with
+/// cleaner test validation.
+pub(super) trait IOutboundQueue {
+
+	//  ____            _       __  __      _   _               _
+	// |  _ \ _   _ ___| |__   |  \/  | ___| |_| |__   ___   __| |___
+	// | |_) | | | / __| '_ \  | |\/| |/ _ \ __| '_ \ / _ \ / _` / __|
+	// |  __/| |_| \__ \ | | | | |  | |  __/ |_| | | | (_) | (_| \__ \
+	// |_|    \__,_|___/_| |_| |_|  |_|\___|\__|_| |_|\___/ \__,_|___/
+
+	/// Unconditionally queue item. May increase queue above soft limit.
+	fn push_back(&mut self, item: Vec<u8>);
+
+	/// Returns true if the queue is empty
+	fn is_empty(&self) -> bool;
+
+	/// Returns the amount of free space in the queue before the soft limit
+	fn queue_space(&self) -> usize;
+
+	//  _____ _           _       __  __      _   _               _
+	// |  ___| |_   _ ___| |__   |  \/  | ___| |_| |__   ___   __| |___
+	// | |_  | | | | / __| '_ \  | |\/| |/ _ \ __| '_ \ / _ \ / _` / __|
+	// |  _| | | |_| \__ \ | | | | |  | |  __/ |_| | | | (_) | (_| \__ \
+	// |_|   |_|\__,_|___/_| |_| |_|  |_|\___|\__|_| |_|\___/ \__,_|___/
 
-/// Trait representing a container that can try to flush data through a SocketDescriptor. Used by the
-/// PeerManager to handle flushing the outbound queue and flow control.
-pub(super) trait SocketDescriptorFlusher {
 	/// Write previously enqueued data to the SocketDescriptor. A return of false indicates the
 	/// underlying SocketDescriptor could not fulfill the send_data() call and the blocked state
 	/// has been set. Use unblock() when the SocketDescriptor may have more room.
@@ -583,11 +605,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, TransportImpl
 
 	// Fill remaining slots in output queue with sync messages, updating the sync state when
 	// appropriate
-	fn fill_outbound_queue_with_sync<Q: PayloadQueuer + SocketDescriptorFlusher>(
+	fn fill_outbound_queue_with_sync(
 		&self,
 		sync_status: &mut InitSyncTracker,
-		message_queuer: &mut impl MessageQueuer,
-		outbound_queue: &mut Q) {
+		transport: &mut TransportImpl,
+		outbound_queue: &mut OutboundQueue) {
 
 		let queue_space = outbound_queue.queue_space();
 		if queue_space > 0 {
@@ -597,12 +619,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, TransportImpl
 					let steps = ((queue_space + 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() {
-						message_queuer.enqueue_message(announce, outbound_queue, &*self.logger);
+						transport.enqueue_message(announce, outbound_queue, &*self.logger);
 						if let &Some(ref update_a) = update_a_option {
-							message_queuer.enqueue_message(update_a, outbound_queue, &*self.logger);
+							transport.enqueue_message(update_a, outbound_queue, &*self.logger);
 						}
 						if let &Some(ref update_b) = update_b_option {
-							message_queuer.enqueue_message(update_b, outbound_queue, &*self.logger);
+							transport.enqueue_message(update_b, outbound_queue, &*self.logger);
 						}
 						*sync_status = InitSyncTracker::ChannelsSyncing(announce.contents.short_channel_id + 1);
 					}
@@ -614,7 +636,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, TransportImpl
 					let steps = queue_space as u8;
 					let all_messages = self.message_handler.route_handler.get_next_node_announcements(None, steps);
 					for msg in all_messages.iter() {
-						message_queuer.enqueue_message(msg, outbound_queue, &*self.logger);
+						transport.enqueue_message(msg, outbound_queue, &*self.logger);
 						*sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
 					}
 					if all_messages.is_empty() || all_messages.len() != steps as usize {
@@ -626,7 +648,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, TransportImpl
 					let steps = queue_space as u8;
 					let all_messages = self.message_handler.route_handler.get_next_node_announcements(Some(&key), steps);
 					for msg in all_messages.iter() {
-						message_queuer.enqueue_message(msg, outbound_queue, &*self.logger);
+						transport.enqueue_message(msg, outbound_queue, &*self.logger);
 						*sync_status = InitSyncTracker::NodesSyncing(msg.contents.node_id);
 					}
 					if all_messages.is_empty() || all_messages.len() != steps as usize {
@@ -637,18 +659,18 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, TransportImpl
 		}
 	}
 
-	fn do_attempt_write_data<Q: PayloadQueuer + SocketDescriptorFlusher>(
+	fn do_attempt_write_data(
 		&self,
 		descriptor: &mut Descriptor,
 		post_init_state: &mut Option<PostInitState>,
-		message_queuer: &mut impl MessageQueuer,
-		outbound_queue: &mut Q) {
+		transport: &mut TransportImpl,
+		outbound_queue: &mut OutboundQueue) {
 
 		while !outbound_queue.is_blocked() {
 			// If connected, fill output queue with sync messages
 			match post_init_state {
 				None => {},
-				&mut Some(ref mut state) => self.fill_outbound_queue_with_sync(&mut state.sync_status, message_queuer, outbound_queue)
+				&mut Some(ref mut state) => self.fill_outbound_queue_with_sync(&mut state.sync_status, transport, outbound_queue)
 			}
 
 			// No messages to send

lightning/src/ln/peers/outbound_queue.rs

@@ -10,8 +10,7 @@
 /// Abstracts the buffer used to write data through a SocketDescriptor handling partial writes and
 /// flow control.
 
-use ln::peers::handler::{SocketDescriptor, SocketDescriptorFlusher};
-use ln::peers::transport::PayloadQueuer;
+use ln::peers::handler::{IOutboundQueue, SocketDescriptor};
 use std::collections::LinkedList;
 use std::cmp;
 
@@ -22,24 +21,31 @@ pub(super) struct OutboundQueue {
 	buffer_first_msg_offset: usize,
 }
 
-impl PayloadQueuer for OutboundQueue {
-	/// Unconditionally queue item. May increase queue above soft limit.
+impl OutboundQueue {
+	pub(super) fn new(soft_limit: usize) -> Self {
+		Self {
+			blocked: false,
+			soft_limit,
+			buffer: LinkedList::new(),
+			buffer_first_msg_offset: 0,
+		}
+	}
+}
+
+impl IOutboundQueue for OutboundQueue {
+
 	fn push_back(&mut self, item: Vec<u8>) {
 		self.buffer.push_back(item);
 	}
 
-	/// Returns true if the queue is empty
 	fn is_empty(&self) -> bool {
 		self.buffer.is_empty()
 	}
 
-	/// Returns the amount of free space in the queue before the soft limit
 	fn queue_space(&self) -> usize {
 		self.soft_limit - cmp::min(self.soft_limit, self.buffer.len())
 	}
-}
 
-impl SocketDescriptorFlusher for OutboundQueue {
 	fn try_flush_one(&mut self, descriptor: &mut impl SocketDescriptor) -> bool {
 		// Exit early if  a previous full write failed and haven't heard that there may be more
 		// room available
@@ -77,20 +83,6 @@ impl SocketDescriptorFlusher for OutboundQueue {
 	}
 }
 
-impl OutboundQueue {
-
-	/// Create a new writer with a soft limit that is used to notify the SocketDescriptor when
-	/// it is OK to resume reading if it was paused
-	pub(super) fn new(soft_limit: usize) -> Self {
-		Self {
-			blocked: false,
-			soft_limit,
-			buffer: LinkedList::new(),
-			buffer_first_msg_offset: 0,
-		}
-	}
-}
-
 #[cfg(test)]
 mod tests {
 	use super::*;

lightning/src/ln/peers/test_util.rs

@@ -13,8 +13,8 @@ use bitcoin::secp256k1;
 use bitcoin::secp256k1::key::{PublicKey, SecretKey};
 
 use ln::peers::conduit::Conduit;
-use ln::peers::handler::{SocketDescriptor, ITransport, PeerHandleError, MessageQueuer};
-use ln::peers::transport::{IPeerHandshake, PayloadQueuer};
+use ln::peers::handler::{SocketDescriptor, IOutboundQueue, ITransport, PeerHandleError};
+use ln::peers::transport::IPeerHandshake;
 
 use std::rc::Rc;
 use std::cell::{RefCell};
@@ -223,8 +223,9 @@ impl Hash for SocketDescriptorMock {
 	}
 }
 
-/// Implement PayloadQueuer for Vec<Vec<u8>> so it can be used as a Spy in tests
-impl PayloadQueuer for Vec<Vec<u8>> {
+/// Implement IOutboundQueue for Vec<Vec<u8>> so it can be used as a Spy in tests. This only implements
+/// a subset of the push methods needed for the tests.
+impl IOutboundQueue for Vec<Vec<u8>> {
 	fn push_back(&mut self, item: Vec<u8>) {
 		self.push(item)
 	}
@@ -236,6 +237,18 @@ impl PayloadQueuer for Vec<Vec<u8>> {
 	fn queue_space(&self) -> usize {
 		unimplemented!()
 	}
+
+	fn try_flush_one(&mut self, _descriptor: &mut impl SocketDescriptor) -> bool {
+		unimplemented!()
+	}
+
+	fn unblock(&mut self) {
+		unimplemented!()
+	}
+
+	fn is_blocked(&self) -> bool {
+		unimplemented!()
+	}
 }
 
 // Builder for TransportTestStub that allows tests to easily construct the Transport layer they
@@ -296,8 +309,8 @@ impl<'a> ITransport for &'a RefCell<TransportStub> {
 		unimplemented!()
 	}
 
-	fn process_input(&mut self, input: &[u8], output_buffer: &mut impl PayloadQueuer) -> Result<bool, String> {
-		self.borrow_mut().process_input(input, output_buffer)
+	fn process_input(&mut self, input: &[u8], outbound_queue: &mut impl IOutboundQueue) -> Result<bool, String> {
+		self.borrow_mut().process_input(input, outbound_queue)
 	}
 
 	fn is_connected(&self) -> bool {
@@ -308,14 +321,12 @@ impl<'a> ITransport for &'a RefCell<TransportStub> {
 		self.borrow().get_their_node_id()
 	}
 
-	fn drain_messages<L: Deref>(&mut self, logger: L) -> Result<Vec<Message>, PeerHandleError> where L::Target: Logger {
-		self.borrow_mut().drain_messages(logger)
+	fn enqueue_message<M: Encode + Writeable, L: Deref>(&mut self, message: &M, outbound_queue: &mut impl IOutboundQueue, logger: L) where L::Target: Logger {
+		self.borrow_mut().enqueue_message(message, outbound_queue, logger)
 	}
-}
 
-impl<'a> MessageQueuer for &'a RefCell<TransportStub> {
-	fn enqueue_message<M: Encode + Writeable, Q: PayloadQueuer, L: Deref>(&mut self, message: &M, output_buffer: &mut Q, logger: L) where L::Target: Logger {
-		self.borrow_mut().enqueue_message(message, output_buffer, logger)
+	fn drain_messages<L: Deref>(&mut self, logger: L) -> Result<Vec<Message>, PeerHandleError> where L::Target: Logger {
+		self.borrow_mut().drain_messages(logger)
 	}
 }
 
@@ -345,7 +356,7 @@ impl ITransport for TransportStub {
 		unimplemented!()
 	}
 
-	fn process_input(&mut self, _input: &[u8], _output_buffer: &mut impl PayloadQueuer) -> Result<bool, String> {
+	fn process_input(&mut self, _input: &[u8], _outbound_queue: &mut impl IOutboundQueue) -> Result<bool, String> {
 		if self.process_returns_error {
 			Err("Oh no!".to_string())
 		} else {
@@ -369,16 +380,14 @@ impl ITransport for TransportStub {
 		self.their_node_id.unwrap()
 	}
 
-	fn drain_messages<L: Deref>(&mut self, _logger: L) -> Result<Vec<Message>, PeerHandleError> where L::Target: Logger {
-		Ok(self.messages.drain(..).collect())
-	}
-}
-
-impl MessageQueuer for TransportStub {
-	fn enqueue_message<M: Encode + Writeable, Q: PayloadQueuer, L: Deref>(&mut self, message: &M, output_buffer: &mut Q, _logger: L)
+	fn enqueue_message<M: Encode + Writeable, L: Deref>(&mut self, message: &M, outbound_queue: &mut impl IOutboundQueue, _logger: L)
 		where L::Target: Logger {
 		let mut buffer = VecWriter(Vec::new());
 		wire::write(message, &mut buffer).unwrap();
-		output_buffer.push_back(buffer.0);
+		outbound_queue.push_back(buffer.0);
+	}
+
+	fn drain_messages<L: Deref>(&mut self, _logger: L) -> Result<Vec<Message>, PeerHandleError> where L::Target: Logger {
+		Ok(self.messages.drain(..).collect())
 	}
-}
+}