feat(net): Add shutdown function

Author: Frando

src/net.rs

@@ -27,7 +27,7 @@ use n0_future::{
 use rand::rngs::StdRng;
 use rand_core::SeedableRng;
 use serde::{Deserialize, Serialize};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, error_span, trace, warn, Instrument};
 
@@ -162,6 +162,15 @@ impl ProtocolHandler for Gossip {
             Ok(())
         })
     }
+
+    fn shutdown(&self) -> BoxFuture<()> {
+        let this = self.clone();
+        Box::pin(async move {
+            if let Err(err) = this.shutdown().await {
+                warn!("error while shutting down gossip: {err:#}");
+            }
+        })
+    }
 }
 
 /// Builder to configure and construct [`Gossip`].
@@ -316,6 +325,17 @@ impl Gossip {
     ) -> EventStream {
         self.inner.subscribe_with_stream(topic_id, options, updates)
     }
+
+    /// Shutdown the gossip instance.
+    ///
+    /// This leaves all topics, sending `Disconnect` messages to peers, and then
+    /// stops the gossip actor loop and drops all state and connections.
+    pub async fn shutdown(&self) -> anyhow::Result<()> {
+        let (reply, reply_rx) = oneshot::channel();
+        self.inner.send(ToActor::Shutdown { reply }).await?;
+        reply_rx.await?;
+        Ok(())
+    }
 }
 
 impl Inner {
@@ -450,6 +470,9 @@ enum ToActor {
         topic: TopicId,
         receiver_id: ReceiverId,
     },
+    Shutdown {
+        reply: oneshot::Sender<()>,
+    },
 }
 
 /// Actor that sends and handles messages between the connection and main state loops
@@ -580,7 +603,18 @@ impl Actor {
                 trace!(?i, "tick: to_actor_rx");
                 inc!(Metrics, actor_tick_rx);
                 match msg {
-                    Some(msg) => self.handle_to_actor_msg(msg, Instant::now()).await?,
+                    Some(msg) => {
+                        if let ToActor::Shutdown { reply } = msg {
+                            debug!("received shutdown message, quit all topics");
+                            self.quit_queue.extend(self.topics.keys().copied());
+                            self.process_quit_queue().await.ok();
+                            debug!("all topics quit, stop gossip actor");
+                            reply.send(()).ok();
+                            return Ok(None)
+                        } else {
+                            self.handle_to_actor_msg(msg, Instant::now()).await?;
+                        }
+                    }
                     None => {
                         debug!("all gossip handles dropped, stop gossip actor");
                         return Ok(None)
@@ -808,6 +842,7 @@ impl Actor {
             ToActor::ReceiverGone { topic, receiver_id } => {
                 self.handle_receiver_gone(topic, receiver_id).await?;
             }
+            ToActor::Shutdown { .. } => unreachable!("handled in main loop"),
         }
         Ok(())
     }