net: Drain active channels on shutdown

This should solve hanging open connections/sockets on ^C

bin/fud/fud/src/proto.rs

@@ -533,7 +533,7 @@ impl ProtocolFud {
 impl ProtocolBase for ProtocolFud {
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "fud::ProtocolFud::start()", "START");
-        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().start(executor.clone()).await?;
         self.jobsman.clone().spawn(self.clone().handle_fud_ping_request(), executor.clone()).await;
         self.jobsman
             .clone()

example/dchat/dchatd/src/protocol_dchat.rs

@@ -68,7 +68,7 @@ impl net::ProtocolBase for ProtocolDchat {
     // ANCHOR: start
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "dchat", "ProtocolDchat::ProtocolBase::start() [START]");
-        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().start(executor.clone()).await?;
         self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
         debug!(target: "dchat", "ProtocolDchat::ProtocolBase::start() [STOP]");
         Ok(())

example/p2pdebug/src/proto/debugmsg.rs

@@ -80,7 +80,7 @@ impl net::ProtocolBase for ProtocolDebugmsg {
     /// waits for pong reply. Waits for ping and replies with a pong.
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "p2pdbg", "Protocoldebugmsg::start() [START]");
-        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().start(executor.clone()).await?;
         self.jobsman.clone().spawn(self.clone().handle_receive_debugmsg(), executor.clone()).await;
         debug!(target: "p2pdbg", "ProtocolDebugmsg::start() [END]");
         Ok(())

src/event_graph/proto.rs

@@ -382,7 +382,7 @@ pub struct ProtocolEventGraph {
 #[async_trait]
 impl ProtocolBase for ProtocolEventGraph {
     async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
-        self.jobsman.clone().start(ex.clone());
+        self.jobsman.clone().start(ex.clone()).await?;
         self.jobsman.clone().spawn(self.clone().handle_event_put(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_static_put(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_event_req(), ex.clone()).await;

src/net/channel.rs

@@ -33,7 +33,7 @@ use rand::{rngs::OsRng, Rng};
 use smol::{
     io::{self, AsyncRead, AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf},
     lock::{Mutex as AsyncMutex, OnceCell},
-    Executor,
+    Executor, Task,
 };
 use tracing::{debug, trace};
 use url::Url;
@@ -95,8 +95,12 @@ pub struct Channel {
     stop_publisher: PublisherPtr<Error>,
     /// Task that is listening for the stop signal
     receive_task: StoppableTaskPtr,
+    /// Cleanup jobs that must finish before the channel is fully stopped.
+    cleanup_tasks: AsyncMutex<Vec<Task<()>>>,
     /// A boolean marking if this channel is stopped
     stopped: AtomicBool,
+    /// A boolean marking if the receive task has been started.
+    started: AtomicBool,
     /// Weak pointer to respective session
     pub(in crate::net) session: SessionWeakPtr,
     /// The version message of the node we are connected to.
@@ -139,7 +143,9 @@ impl Channel {
             message_subsystem,
             stop_publisher: Publisher::new(),
             receive_task: StoppableTask::new(),
+            cleanup_tasks: AsyncMutex::new(Vec::new()),
             stopped: AtomicBool::new(false),
+            started: AtomicBool::new(false),
             session,
             version: OnceCell::new(),
             info,
@@ -159,9 +165,19 @@ impl Channel {
 
     /// Starts the channel. Runs a receive loop to start receiving messages
     /// or handles a network failure.
-    pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
+    pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::channel::start", "START {self:?}");
 
+        if self.is_stopped() || self.started.swap(true, SeqCst) {
+            return Err(Error::ChannelStopped)
+        }
+
+        if !self.p2p().track_channel(&self) {
+            self.started.store(false, SeqCst);
+            self.stopped.store(true, SeqCst);
+            return Err(Error::NetworkServiceStopped)
+        }
+
         let self_ = self.clone();
         self.receive_task.clone().start(
             self.clone().main_receive_loop(),
@@ -171,16 +187,45 @@ impl Channel {
         );
 
         debug!(target: "net::channel::start", "END {self:?}");
+        Ok(())
     }
 
     /// Stops the channel.
     /// Notifies all publishers that the channel has been closed in `handle_stop()`.
     pub async fn stop(&self) {
         debug!(target: "net::channel::stop", "START {self:?}");
+        if !self.started.load(SeqCst) {
+            return
+        }
         self.receive_task.stop().await;
         debug!(target: "net::channel::stop", "END {self:?}");
     }
 
+    /// Registers cleanup work that must complete before [`Channel::stop`]
+    /// returns. If shutdown has already begun, waits for the cleanup directly.
+    pub(crate) async fn add_cleanup_task(&self, task: Task<()>) -> Result<()> {
+        let mut cleanup_tasks = self.cleanup_tasks.lock().await;
+        if self.is_stopped() {
+            drop(cleanup_tasks);
+            task.await;
+            return Err(Error::ChannelStopped)
+        }
+
+        cleanup_tasks.push(task);
+        Ok(())
+    }
+
+    /// Requests channel shutdown without waiting for cleanup. Used to make
+    /// cancellation of an in-progress channel setup safe.
+    pub(crate) fn stop_nowait(&self) {
+        self.receive_task.stop_nowait();
+    }
+
+    #[cfg(test)]
+    pub(crate) async fn cleanup_task_count(&self) -> usize {
+        self.cleanup_tasks.lock().await.len()
+    }
+
     /// Creates a subscription to a stopped signal.
     /// If the channel is stopped then this will return a ChannelStopped error.
     pub async fn subscribe_stop(&self) -> Result<Subscription<Error>> {
@@ -405,6 +450,13 @@ impl Channel {
             }
         }
 
+        let cleanup_tasks = std::mem::take(&mut *self.cleanup_tasks.lock().await);
+        for task in cleanup_tasks {
+            task.await;
+        }
+
+        self.p2p().untrack_channel(self.info.id);
+
         debug!(target: "net::channel::handle_stop", "[END] {self:?}");
     }
 

src/net/p2p.rs

@@ -16,19 +16,23 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Weak,
+    },
 };
 
 use futures::{stream::FuturesUnordered, TryFutureExt};
 use futures_rustls::rustls::crypto::{ring, CryptoProvider};
+use parking_lot::Mutex;
 use smol::{fs, lock::RwLock as AsyncRwLock, stream::StreamExt};
 use tracing::{debug, error, info};
 use url::Url;
 
 use super::{
-    channel::ChannelPtr,
+    channel::{Channel, ChannelPtr},
     dnet::DnetEvent,
     hosts::{Hosts, HostsPtr},
     message::{Message, SerializedMessage},
@@ -78,6 +82,10 @@ pub struct P2p {
     pub dnet_enabled: AtomicBool,
     /// The publisher for which we can give dnet info over
     dnet_publisher: PublisherPtr<DnetEvent>,
+    /// Prevents channel registration while shutdown is in progress.
+    stopping: AtomicBool,
+    /// All started channels, including those still performing their handshake.
+    channels: Mutex<HashMap<u32, Weak<Channel>>>,
 }
 
 impl P2p {
@@ -118,6 +126,8 @@ impl P2p {
             session_direct: DirectSession::new(p2p.clone()),
             dnet_enabled: AtomicBool::new(false),
             dnet_publisher: Publisher::new(),
+            stopping: AtomicBool::new(false),
+            channels: Mutex::new(HashMap::new()),
         });
 
         register_default_protocols(self_.clone()).await;
@@ -127,6 +137,8 @@ impl P2p {
 
     /// Starts inbound, outbound, and manual sessions.
     pub async fn start(self: Arc<Self>) -> Result<()> {
+        self.stopping.store(false, Ordering::SeqCst);
+
         debug!(target: "net::p2p::start", "P2P::start() [BEGIN] [magic_bytes={:?}]",
                self.settings.read().await.magic_bytes.0);
         info!(target: "net::p2p::start", "[P2P] Starting P2P subsystem");
@@ -169,13 +181,48 @@ impl P2p {
 
     /// Stop the running P2P subsystem
     pub async fn stop(&self) {
-        // Stop the sessions
-        self.session_manual().stop().await;
+        self.stopping.store(true, Ordering::SeqCst);
+
+        // Stop connection producers before draining established channels.
         self.session_inbound().stop().await;
+        self.session_manual().stop().await;
         self.session_seedsync().stop().await;
         self.session_outbound().stop().await;
         self.session_refine().stop().await;
         self.session_direct().stop().await;
+
+        let channels = self.tracked_channels();
+        let stops = FuturesUnordered::new();
+        for channel in channels {
+            stops.push(async move { channel.stop().await });
+        }
+        stops.collect::<Vec<_>>().await;
+
+        debug_assert!(self.tracked_channels().is_empty(), "P2P stopped with active channels");
+        debug_assert!(self.hosts.channels().is_empty(), "P2P stopped with registered channels");
+    }
+
+    pub(crate) fn is_stopping(&self) -> bool {
+        self.stopping.load(Ordering::SeqCst)
+    }
+
+    pub(crate) fn track_channel(&self, channel: &ChannelPtr) -> bool {
+        let mut channels = self.channels.lock();
+        if self.is_stopping() {
+            return false
+        }
+
+        channels.retain(|_, channel| channel.strong_count() > 0);
+        channels.insert(channel.info.id, Arc::downgrade(channel));
+        true
+    }
+
+    pub(crate) fn untrack_channel(&self, channel_id: u32) {
+        self.channels.lock().remove(&channel_id);
+    }
+
+    fn tracked_channels(&self) -> Vec<ChannelPtr> {
+        self.channels.lock().values().filter_map(Weak::upgrade).collect()
     }
 
     /// Broadcasts a message concurrently across all active peers.

src/net/protocol/protocol_address.rs

@@ -295,7 +295,7 @@ impl ProtocolBase for ProtocolAddress {
         let getaddrs_max = settings.getaddrs_max;
         drop(settings);
 
-        self.jobsman.clone().start(ex.clone());
+        self.jobsman.clone().start(ex.clone()).await?;
 
         self.jobsman.clone().spawn(self.clone().send_my_addrs(), ex.clone()).await;
 

src/net/protocol/protocol_generic.rs

@@ -297,7 +297,7 @@ impl<M: Message + Clone, R: Message + Clone + Debug> ProtocolGeneric<M, R> {
 impl<M: Message + Clone, R: Message + Clone + Debug> ProtocolBase for ProtocolGeneric<M, R> {
     async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::protocol_generic::start", "START");
-        self.jobsman.clone().start(ex.clone());
+        self.jobsman.clone().start(ex.clone()).await?;
         self.jobsman.clone().spawn(self.clone().handle_receive_message(), ex).await;
         debug!(target: "net::protocol_generic::start", "END");
         Ok(())

src/net/protocol/protocol_holepunch.rs

@@ -546,7 +546,7 @@ impl ProtocolHolepunch {
 impl ProtocolBase for ProtocolHolepunch {
     async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::protocol_holepunch", "Starting on {}", self.channel.display_address());
-        self.jobsman.clone().start(ex.clone());
+        self.jobsman.clone().start(ex.clone()).await?;
         self.jobsman.clone().spawn(self.clone().handle_relay_requests(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().handle_connect_instructions(), ex.clone()).await;
         self.jobsman.spawn(self.clone().nonce_cleanup_loop(), ex).await;

src/net/protocol/protocol_jobs_manager.rs

@@ -16,7 +16,10 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
 
 use smol::{future::Future, lock::Mutex, Executor, Task};
 use tracing::{debug, trace};
@@ -31,12 +34,13 @@ pub struct ProtocolJobsManager {
     name: &'static str,
     channel: ChannelPtr,
     tasks: Mutex<Vec<Task<Result<()>>>>,
+    stopped: AtomicBool,
 }
 
 impl ProtocolJobsManager {
     /// Create a new protocol jobs manager
     pub fn new(name: &'static str, channel: ChannelPtr) -> ProtocolJobsManagerPtr {
-        Arc::new(Self { name, channel, tasks: Mutex::new(vec![]) })
+        Arc::new(Self { name, channel, tasks: Mutex::new(vec![]), stopped: AtomicBool::new(false) })
     }
 
     /// Returns configured name
@@ -45,8 +49,9 @@ impl ProtocolJobsManager {
     }
 
     /// Runs the task on an executor
-    pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
-        executor.spawn(self.handle_stop()).detach()
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        let channel = self.channel.clone();
+        channel.add_cleanup_task(executor.spawn(self.handle_stop())).await
     }
 
     /// Spawns a new task and adds it to the internal queue
@@ -54,7 +59,14 @@ impl ProtocolJobsManager {
     where
         F: Future<Output = Result<()>> + Send + 'a,
     {
-        self.tasks.lock().await.push(executor.spawn(future))
+        let task = executor.spawn(future);
+        let mut tasks = self.tasks.lock().await;
+        if self.stopped.load(Ordering::SeqCst) || self.channel.is_stopped() {
+            drop(tasks);
+            let _ = task.cancel().await;
+            return
+        }
+        tasks.push(task)
     }
 
     /// Waits for a stop signal, then closes all tasks.
@@ -68,11 +80,12 @@ impl ProtocolJobsManager {
             stop_sub.receive().await;
         }
 
+        self.stopped.store(true, Ordering::SeqCst);
         self.close_all_tasks().await
     }
 
     /// Closes all open tasks. Takes all the tasks from the internal queue.
-    async fn close_all_tasks(self: Arc<Self>) {
+    async fn close_all_tasks(&self) {
         debug!(
             target: "net::protocol_jobs_manager",
             "ProtocolJobsManager::close_all_tasks() [START, name={}, addr={}]",

src/net/protocol/protocol_ping.rs

@@ -187,7 +187,7 @@ impl ProtocolBase for ProtocolPing {
     /// waits for pong reply. Waits for ping and replies with a pong.
     async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::protocol_ping::start", "START => address={}", self.channel.display_address());
-        self.jobsman.clone().start(ex.clone());
+        self.jobsman.clone().start(ex.clone()).await?;
         self.jobsman.clone().spawn(self.clone().run_ping_pong(), ex.clone()).await;
         self.jobsman.clone().spawn(self.clone().reply_to_ping(), ex).await;
         debug!(target: "net::protocol_ping::start", "END => address={}", self.channel.display_address());

src/net/session/direct_session.rs

@@ -98,9 +98,13 @@ impl DirectSession {
     pub async fn stop(&self) {
         self.peer_discovery.clone().stop().await;
 
-        for (_, task) in self.retries_tasks.lock().await.iter() {
+        let retries_tasks = std::mem::take(&mut *self.retries_tasks.lock().await);
+        for (_, task) in retries_tasks {
             task.stop().await;
         }
+
+        self.tasks.lock().await.clear();
+        self.channels_usage.lock().await.clear();
     }
 
     /// Notify the peer discovery task to start it.
@@ -116,6 +120,10 @@ impl DirectSession {
     /// return it (even if the channel was not created by the direct session).
     /// Otherwise it will create a new channel to `addr` in the direct session.
     pub async fn get_channel(self: Arc<Self>, addr: &Url) -> Result<ChannelPtr> {
+        if self.p2p().is_stopping() {
+            return Err(Error::NetworkServiceStopped)
+        }
+
         // Check existing channels
         let channels = self.p2p().hosts().channels();
         if let Some(channel) =
@@ -123,7 +131,7 @@ impl DirectSession {
         {
             let mut channels_usage = self.channels_usage.lock().await;
             if channel.is_stopped() {
-                channel.clone().start(self.p2p().executor());
+                channel.clone().start(self.p2p().executor())?;
             }
             if channel.session_type_id() & SESSION_DIRECT != 0 {
                 channels_usage.entry(channel.info.id).and_modify(|count| *count += 1).or_insert(1);

src/net/session/inbound_session.rs

@@ -116,15 +116,14 @@ impl InboundSession {
     pub async fn stop(&self) {
         if self.p2p().settings().read().await.inbound_addrs.is_empty() {
             verbose!(target: "net::inbound_session", "[P2P] Stopping inbound session.");
-            return
         }
 
-        let acceptors = &*self.acceptors.lock().await;
+        let acceptors = std::mem::take(&mut *self.acceptors.lock().await);
         for acceptor in acceptors {
             acceptor.stop().await;
         }
 
-        let accept_tasks = &*self.accept_tasks.lock().await;
+        let accept_tasks = std::mem::take(&mut *self.accept_tasks.lock().await);
         for accept_task in accept_tasks {
             accept_task.stop().await;
         }

src/net/session/manual_session.rs

@@ -88,10 +88,10 @@ impl ManualSession {
 
     /// Stops the manual session.
     pub async fn stop(&self) {
-        let slots = &*self.slots.lock().await;
+        let slots = std::mem::take(&mut *self.slots.lock().await);
         let mut futures = FuturesUnordered::new();
 
-        for slot in slots {
+        for slot in &slots {
             futures.push(slot.stop());
         }
 

src/net/session/mod.rs

@@ -61,6 +61,26 @@ pub const SESSION_ALL: SessionBitFlag = 0b111111;
 
 pub type SessionWeakPtr = Weak<dyn Session + Send + Sync + 'static>;
 
+struct ChannelSetupGuard(Option<ChannelPtr>);
+
+impl ChannelSetupGuard {
+    fn new(channel: ChannelPtr) -> Self {
+        Self(Some(channel))
+    }
+
+    fn disarm(&mut self) {
+        self.0 = None;
+    }
+}
+
+impl Drop for ChannelSetupGuard {
+    fn drop(&mut self) {
+        if let Some(channel) = self.0.take() {
+            channel.stop_nowait();
+        }
+    }
+}
+
 /// Removes channel from the list of connected channels when a stop signal
 /// is received.
 pub async fn remove_sub_on_stop(
@@ -154,6 +174,10 @@ pub trait Session: Sync {
     ) -> Result<()> {
         trace!(target: "net::session::register_channel", "[START]");
 
+        if self.p2p().is_stopping() {
+            return Err(Error::NetworkServiceStopped)
+        }
+
         // Protocols should all be initialized but not started.
         // We do this so that the protocols can begin receiving and buffering
         // messages while the handshake protocol is ongoing. They are currently
@@ -173,7 +197,8 @@ pub trait Session: Sync {
             self.perform_handshake_protocols(protocol_version, channel.clone(), executor.clone());
 
         // Switch on the channel
-        channel.clone().start(executor.clone());
+        channel.clone().start(executor.clone())?;
+        let mut setup_guard = ChannelSetupGuard::new(channel.clone());
 
         // Wait for handshake to finish.
         match handshake_task.await {
@@ -185,6 +210,7 @@ pub trait Session: Sync {
                 debug!(target: "net::session::register_channel",
                 "Handshake error {e} {}", channel.clone().display_address());
 
+                channel.stop().await;
                 return Err(e)
             }
         }
@@ -196,9 +222,14 @@ pub trait Session: Sync {
         // Now start all the protocols. They are responsible for managing their own
         // lifetimes and correctly selfdestructing when the channel ends.
         for protocol in protocols {
-            protocol.start(executor.clone()).await?;
+            if let Err(e) = protocol.start(executor.clone()).await {
+                channel.stop().await;
+                return Err(e)
+            }
         }
 
+        setup_guard.disarm();
+
         trace!(target: "net::session::register_channel", "[END]");
 
         Ok(())
@@ -235,12 +266,20 @@ pub trait Session: Sync {
                 }
 
                 // Attempt to add channel to registry
+                if self.p2p().is_stopping() {
+                    return Err(Error::NetworkServiceStopped)
+                }
+
                 self.p2p().hosts().register_channel(channel.clone()).await;
 
                 // Subscribe to stop, so we can remove from registry
-                executor
-                    .spawn(remove_sub_on_stop(self.p2p(), channel, self.type_id(), stop_sub))
-                    .detach();
+                let cleanup_task = executor.spawn(remove_sub_on_stop(
+                    self.p2p(),
+                    channel.clone(),
+                    self.type_id(),
+                    stop_sub,
+                ));
+                channel.add_cleanup_task(cleanup_task).await?;
 
                 // Channel is ready for use
                 Ok(())

src/net/session/outbound_session.rs

@@ -105,7 +105,7 @@ impl OutboundSession {
     /// Stops the outbound session.
     pub(crate) async fn stop(&self) {
         debug!(target: "net::outbound_session", "Stopping outbound session..");
-        let slots = &*self.slots.lock().await;
+        let slots = std::mem::take(&mut *self.slots.lock().await);
         let mut futures = FuturesUnordered::new();
 
         for slot in slots {

src/net/session/refine_session.rs

@@ -132,7 +132,9 @@ impl RefineSession {
                     self.perform_handshake_protocols(proto_ver, channel.clone(), p2p.executor());
 
                 debug!(target: "net::refinery::handshake_node", "Starting channel {url}");
-                channel.clone().start(p2p.executor());
+                if channel.clone().start(p2p.executor()).is_err() {
+                    return false
+                }
 
                 // Ensure the channel gets stopped by adding a timeout to the handshake. Otherwise if
                 // the handshake does not finish channel.stop() will never get called, resulting in

src/net/session/seedsync_session.rs

@@ -118,7 +118,7 @@ impl SeedSyncSession {
     /// Stop the seedsync session.
     pub(crate) async fn stop(&self) {
         debug!(target: "net::seedsync_session", "Stopping seed sync session...");
-        let slots = &*self.slots.lock().await;
+        let slots = std::mem::take(&mut *self.slots.lock().await);
         let mut futures = FuturesUnordered::new();
 
         for slot in slots {

src/net/tests.rs

@@ -662,3 +662,65 @@ async fn p2p_inbound_slots_survive_rapid_disconnects_real(ex: Arc<Executor<'stat
 
     p2p.stop().await;
 }
+
+#[test]
+fn p2p_shutdown_drains_channels_across_restarts() {
+    test_body!(p2p_shutdown_drains_channels_across_restarts_real, 2);
+}
+
+async fn p2p_shutdown_drains_channels_across_restarts_real(ex: Arc<Executor<'static>>) {
+    const LIFECYCLE_COUNT: usize = 3;
+
+    let port = get_random_available_port();
+    let listen_url = Url::parse(&format!("tcp://127.0.0.1:{port}")).unwrap();
+    let server_settings = Settings {
+        localnet: true,
+        inbound_addrs: vec![listen_url.clone()],
+        inbound_connections: 8,
+        outbound_connections: 0,
+        active_profiles: vec!["tcp".to_string()],
+        ..Default::default()
+    };
+    let client_settings = Settings {
+        localnet: true,
+        peers: vec![listen_url],
+        inbound_connections: 0,
+        outbound_connections: 0,
+        active_profiles: vec!["tcp".to_string()],
+        ..Default::default()
+    };
+
+    let server = P2p::new(server_settings, ex.clone()).await.unwrap();
+    let client = P2p::new(client_settings, ex).await.unwrap();
+
+    for _ in 0..LIFECYCLE_COUNT {
+        server.clone().start().await.unwrap();
+        client.clone().start().await.unwrap();
+
+        timeout(Duration::from_secs(5), async {
+            while server.hosts().channels().is_empty() || client.hosts().channels().is_empty() {
+                Timer::after(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("manual connection was not established");
+
+        let channels = server
+            .hosts()
+            .channels()
+            .into_iter()
+            .chain(client.hosts().channels())
+            .collect::<Vec<_>>();
+
+        client.stop().await;
+        server.stop().await;
+
+        assert!(client.hosts().channels().is_empty());
+        assert!(server.hosts().channels().is_empty());
+        assert_eq!(server.session_inbound().connection_count().await, 0);
+        for channel in channels {
+            assert!(channel.is_stopped());
+            assert_eq!(channel.cleanup_task_count().await, 0);
+        }
+    }
+}