Fix P2P restarts

* implement command channel draining in event loop

* remove redundant shutdown controller

* Remove events drain until needed

---------

Co-authored-by: Aleksandar Terentić <[email protected]>

Author: sh3ll3x3c

client/src/cli.rs

@@ -1,6 +1,6 @@
 use avail_light_core::{
 	network::{Network, ServiceMode},
-	types::NetworkMode,
+	types::{KademliaMode, NetworkMode},
 };
 use clap::{command, ArgAction, Parser};
 use tracing::Level;
@@ -85,4 +85,8 @@ pub struct CliOpts {
 	/// AutoNAT behaviour mode: client, server, or disabled
 	#[arg(long, value_enum, default_value = "server")]
 	pub auto_nat_mode: ServiceMode,
+
+	/// Kademlia operation mode: client or server
+	#[arg(long, value_enum, default_value = "client")]
+	pub operation_mode: Option<KademliaMode>,
 }

client/src/main.rs

@@ -529,6 +529,10 @@ pub fn load_runtime_config(opts: &CliOpts) -> Result<RuntimeConfig> {
 		));
 	}
 
+	if let Some(operation_mode) = opts.operation_mode {
+		cfg.libp2p.kademlia.operation_mode = operation_mode;
+	}
+
 	Ok(cfg)
 }
 

core/src/network/p2p/client.rs

@@ -1,7 +1,7 @@
 #[cfg(not(feature = "multiproof"))]
 use avail_core::kate::{CHUNK_SIZE, COMMITMENT_SIZE};
 use color_eyre::{
-	eyre::{eyre, OptionExt, WrapErr},
+	eyre::{eyre, OptionExt},
 	Result,
 };
 use futures::future::join_all;
@@ -112,9 +112,13 @@ impl Client {
 			}
 		})?;
 
-		response_receiver
-			.await
-			.wrap_err("sender should not be dropped")?
+		response_receiver.await.map_err(|_| {
+			if self.command_sender.is_closed() {
+				eyre!("Event loop shut down before command could be processed")
+			} else {
+				eyre!("Command handler failed to send response")
+			}
+		})?
 	}
 
 	/// Starts listening on provided multiaddresses and saves the listener IDs

core/src/network/p2p/event_loop.rs

@@ -148,6 +148,7 @@ impl EventLoop {
 	}
 
 	pub async fn run(mut self) {
+		info!("Running P2P event loop");
 		// shutdown will wait as long as this token is not dropped
 		let _delay_token = self
 			.shutdown
@@ -223,6 +224,7 @@ impl EventLoop {
 			}
 		}
 		self.disconnect_peers();
+		info!("Exiting P2P event loop");
 	}
 
 	fn disconnect_peers(&mut self) {

core/src/network/p2p/restart.rs

@@ -45,7 +45,7 @@ pub async fn init_and_start_p2p_client(
 	.await?;
 
 	// Start the new event loop
-	spawn_in_span(shutdown.with_cancel(p2p_event_loop.run()));
+	spawn_in_span(p2p_event_loop.run());
 
 	let addrs = vec![libp2p_cfg.tcp_multiaddress()];
 

core/src/types.rs

@@ -127,7 +127,7 @@ impl TryFrom<(&AvailHeader, Option<f64>)> for BlockVerified {
 	}
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq)]
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, clap::ValueEnum)]
 #[serde(try_from = "String")]
 pub enum KademliaMode {
 	Client,