(3/4) Conduit fixups from 494

lightning-net-tokio/src/lib.rs

@@ -38,7 +38,7 @@
 //! type ChainFilter = dyn lightning::chain::Filter;
 //! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemoryChannelKeys, Arc<ChainFilter>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>>;
 //! type ChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>;
-//! type PeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger>;
+//! type PeerManager = lightning::ln::peers::handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger>;
 //!
 //! // Connect to node with pubkey their_node_id at addr:
 //! async fn connect_to_node(peer_manager: PeerManager, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, their_node_id: PublicKey, addr: SocketAddr) {
@@ -78,8 +78,8 @@ use tokio::{io, time};
 use tokio::sync::mpsc;
 use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
-use lightning::ln::peer_handler;
-use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait;
+use lightning::ln::peers::handler;
+use lightning::ln::peers::handler::SocketDescriptor as LnSocketTrait;
 use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
 use lightning::util::logger::Logger;
 
@@ -134,7 +134,7 @@ impl Connection {
 			_ => panic!()
 		}
 	}
-	async fn schedule_read<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
+	async fn schedule_read<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
 			CMH: ChannelMessageHandler + 'static,
 			RMH: RoutingMessageHandler + 'static,
 			L: Logger + 'static + ?Sized {
@@ -190,17 +190,20 @@ impl Connection {
 					Ok(len) => {
 						prepare_read_write_call!();
 						let read_res = peer_manager.read_event(&mut our_descriptor, &buf[0..len]);
-						let mut us_lock = us.lock().unwrap();
-						match read_res {
-							Ok(pause_read) => {
-								if pause_read {
-									us_lock.read_paused = true;
-								}
-								Self::event_trigger(&mut us_lock);
-							},
-							Err(e) => shutdown_socket!(e, Disconnect::CloseConnection),
+						{
+							let mut us_lock = us.lock().unwrap();
+							match read_res {
+								Ok(pause_read) => {
+									if pause_read {
+										us_lock.read_paused = true;
+									}
+									Self::event_trigger(&mut us_lock);
+								},
+								Err(e) => shutdown_socket!(e, Disconnect::CloseConnection),
+							}
+							us_lock.block_disconnect_socket = false;
 						}
-						us_lock.block_disconnect_socket = false;
+						peer_manager.process_events();
 					},
 					Err(e) => shutdown_socket!(e, Disconnect::PeerDisconnected),
 				},
@@ -247,7 +250,7 @@ impl Connection {
 /// not need to poll the provided future in order to make progress.
 ///
 /// See the module-level documentation for how to handle the event_notify mpsc::Sender.
-pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
+pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, stream: TcpStream) -> impl std::future::Future<Output=()> where
 		CMH: ChannelMessageHandler + 'static,
 		RMH: RoutingMessageHandler + 'static,
 		L: Logger + 'static + ?Sized {
@@ -289,7 +292,7 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
 /// not need to poll the provided future in order to make progress.
 ///
 /// See the module-level documentation for how to handle the event_notify mpsc::Sender.
-pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
+pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, stream: TcpStream) -> impl std::future::Future<Output=()> where
 		CMH: ChannelMessageHandler + 'static,
 		RMH: RoutingMessageHandler + 'static,
 		L: Logger + 'static + ?Sized {
@@ -361,7 +364,7 @@ pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<S
 /// make progress.
 ///
 /// See the module-level documentation for how to handle the event_notify mpsc::Sender.
-pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
+pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, event_notify: mpsc::Sender<()>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
 		CMH: ChannelMessageHandler + 'static,
 		RMH: RoutingMessageHandler + 'static,
 		L: Logger + 'static + ?Sized {
@@ -412,7 +415,7 @@ impl SocketDescriptor {
 		Self { conn, id }
 	}
 }
-impl peer_handler::SocketDescriptor for SocketDescriptor {
+impl handler::SocketDescriptor for SocketDescriptor {
 	fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize {
 		// To send data, we take a lock on our Connection to access the WriteHalf of the TcpStream,
 		// writing to it if there's room in the kernel buffer, or otherwise create a new Waker with
@@ -504,7 +507,7 @@ impl Hash for SocketDescriptor {
 mod tests {
 	use lightning::ln::features::*;
 	use lightning::ln::msgs::*;
-	use lightning::ln::peer_handler::{MessageHandler, PeerManager};
+	use lightning::ln::peers::handler::{MessageHandler, PeerManager};
 	use lightning::util::events::*;
 	use bitcoin::secp256k1::{Secp256k1, SecretKey, PublicKey};
 

lightning/src/ln/features.rs

@@ -491,6 +491,10 @@ impl<T: sealed::StaticRemoteKey> Features<T> {
 	pub(crate) fn requires_static_remote_key(&self) -> bool {
 		<T as sealed::StaticRemoteKey>::requires_feature(&self.flags)
 	}
+	#[cfg(test)]
+	pub(crate) fn clear_requires_static_remote_key(&mut self) {
+		<T as sealed::StaticRemoteKey>::clear_bits(&mut self.flags)
+	}
 }
 
 impl<T: sealed::InitialRoutingSync> Features<T> {

lightning/src/ln/mod.rs

@@ -20,16 +20,11 @@
 
 pub mod channelmanager;
 pub mod msgs;
-pub mod peer_handler;
+pub mod peers;
 pub mod chan_utils;
 pub mod features;
 pub(crate) mod onchaintx;
 
-#[cfg(feature = "fuzztarget")]
-pub mod peer_channel_encryptor;
-#[cfg(not(feature = "fuzztarget"))]
-pub(crate) mod peer_channel_encryptor;
-
 mod channel;
 mod onion_utils;
 mod wire;
@@ -53,4 +48,3 @@ mod reorg_tests;
 #[allow(unused_mut)]
 mod onion_route_tests;
 
-pub use self::peer_channel_encryptor::LN_MAX_MSG_LEN;

lightning/src/ln/msgs.rs

@@ -672,6 +672,7 @@ pub enum ErrorAction {
 }
 
 /// An Err type for failure to process messages.
+#[derive(Clone)]
 pub struct LightningError {
 	/// A human-readable message describing the error
 	pub err: String,