add async-scoped

src/core/player.rs

@@ -7,17 +7,23 @@
 //!
 //! A player actor is a serial handler of commands from a single player. It is preferable to run all per-player validations in the player actor,
 //! so that they don't overload the room actor.
-use std::collections::{HashMap, HashSet};
-use std::sync::RwLock;
+use std::{
+    collections::{HashMap, HashSet},
+    sync::{Arc, RwLock},
+};
 
-use futures_util::FutureExt;
 use prometheus::{IntGauge, Registry as MetricsRegistry};
 use serde::Serialize;
-use tokio::sync::mpsc::{channel, Receiver, Sender};
+use tokio::{
+    sync::mpsc::{channel, Receiver, Sender},
+    task::JoinHandle,
+};
 
-use crate::util::table::{AnonTable, Key as AnonKey};
-use crate::prelude::*;
-use crate::core::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
+use crate::{
+    core::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry},
+    prelude::*,
+    util::table::{AnonTable, Key as AnonKey},
+};
 
 /// Opaque player identifier. Cannot contain spaces, must be shorter than 32.
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@@ -102,7 +108,7 @@ impl PlayerConnection {
 }
 
 /// Handle to a player actor.
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct PlayerHandle {
     tx: Sender<PlayerCommand>,
 }
@@ -148,9 +154,7 @@ impl PlayerHandle {
     }
 
     async fn send(&self, command: PlayerCommand) {
-        if let Err(e) = self.tx.send(command).await {
-            log::warn!("Failed to send command to a player: {e:?}");
-        }
+        let _ = self.tx.send(command).await;
     }
 
     pub async fn update(&self, update: Updates) {
@@ -158,7 +162,6 @@ impl PlayerHandle {
     }
 }
 
-#[derive(Debug)]
 enum PlayerCommand {
     /** Commands from connections */
     AddConnection {
@@ -171,10 +174,8 @@ enum PlayerCommand {
     GetRooms(Promise<Vec<RoomInfo>>),
     /** Events from rooms */
     Update(Updates),
-    Stop,
 }
 
-#[derive(Debug)]
 pub enum Cmd {
     JoinRoom {
         room_id: RoomId,
@@ -196,7 +197,6 @@ pub enum Cmd {
     },
 }
 
-#[derive(Debug)]
 pub enum JoinResult {
     Success(RoomInfo),
     Banned,
@@ -227,75 +227,74 @@ pub enum Updates {
 }
 
 /// Handle to a player registry — a shared data structure containing information about players.
-pub struct PlayerRegistry<'a>(RwLock<PlayerRegistryInner<'a>>);
-impl<'a> PlayerRegistry<'a> {
+#[derive(Clone)]
+pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
+impl PlayerRegistry {
     pub fn empty(
-        room_registry: &'a RoomRegistry<'a>,
-        metrics: &MetricsRegistry,
-    ) -> Result<PlayerRegistry<'a>> {
+        room_registry: RoomRegistry,
+        metrics: &mut MetricsRegistry,
+    ) -> Result<PlayerRegistry> {
         let metric_active_players =
             IntGauge::new("chat_players_active", "Number of alive player actors")?;
         metrics.register(Box::new(metric_active_players.clone()))?;
-        let scope = unsafe { Scope::create() };
         let inner = PlayerRegistryInner {
             room_registry,
             players: HashMap::new(),
             metric_active_players,
-            scope,
         };
-        Ok(PlayerRegistry(RwLock::new(inner)))
+        Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
     }
 
-    pub async fn get_or_create_player(&self, id: PlayerId) -> PlayerHandle {
+    pub async fn get_or_create_player(&mut self, id: PlayerId) -> PlayerHandle {
         let mut inner = self.0.write().unwrap();
-        if let Some(handle) = inner.players.get(&id) {
+        if let Some((handle, _)) = inner.players.get(&id) {
             handle.clone()
         } else {
-            let handle = Player::launch(id.clone(), inner.room_registry, &mut inner.scope);
-            inner.players.insert(id, handle.clone());
+            let (handle, fiber) = Player::launch(id.clone(), inner.room_registry.clone());
+            inner.players.insert(id, (handle.clone(), fiber));
             inner.metric_active_players.inc();
             handle
         }
     }
 
-    pub async fn connect_to_player(&self, id: PlayerId) -> PlayerConnection {
+    pub async fn connect_to_player(&mut self, id: PlayerId) -> PlayerConnection {
         let player_handle = self.get_or_create_player(id).await;
         player_handle.subscribe().await
     }
 
-    pub async fn shutdown_all(self) -> Result<()> {
+    pub async fn shutdown_all(&mut self) -> Result<()> {
         let mut inner = self.0.write().unwrap();
-        for (i, k) in inner.players.drain() {
-            k.send(PlayerCommand::Stop).await;
+        let mut players = HashMap::new();
+        std::mem::swap(&mut players, &mut inner.players);
+        for (i, (k, j)) in inner.players.drain() {
             drop(k);
-            log::debug!("Stopping player #{i:?}")
+            j.await?;
+            log::debug!("Player stopped #{i:?}")
         }
-        let _ = inner.scope.collect().await;
         log::debug!("All players stopped");
         Ok(())
     }
 }
 
 /// The player registry state representation.
-struct PlayerRegistryInner<'a> {
-    room_registry: &'a RoomRegistry<'a>,
-    players: HashMap<PlayerId, PlayerHandle>,
+struct PlayerRegistryInner {
+    room_registry: RoomRegistry,
+    players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
     metric_active_players: IntGauge,
-    scope: Scope<'a>,
 }
 
 /// Player actor inner state representation.
-struct Player<'a> {
+struct Player {
     player_id: PlayerId,
     connections: AnonTable<Sender<Updates>>,
-    my_rooms: HashMap<RoomId, RoomHandle<'a>>,
+    my_rooms: HashMap<RoomId, RoomHandle>,
     banned_from: HashSet<RoomId>,
     rx: Receiver<PlayerCommand>,
     handle: PlayerHandle,
-    rooms: &'a RoomRegistry<'a>,
+    rooms: RoomRegistry,
 }
-impl<'a> Player<'a> {
-    fn launch(player_id: PlayerId, rooms: &'a RoomRegistry<'a>, scope: &mut Scope<'a>) -> PlayerHandle {
+impl Player {
+    fn launch(player_id: PlayerId, rooms: RoomRegistry) -> (PlayerHandle, JoinHandle<Player>) {
         let (tx, rx) = channel(32);
         let handle = PlayerHandle { tx };
         let handle_clone = handle.clone();
@@ -308,11 +307,11 @@ impl<'a> Player<'a> {
             handle,
             rooms,
         };
-        scope.spawn(player.main_loop().map(|_| ()));
-        handle_clone
+        let fiber = tokio::task::spawn(player.main_loop());
+        (handle_clone, fiber)
     }
 
-    async fn main_loop(mut self) -> Player<'a> {
+    async fn main_loop(mut self) -> Self {
         while let Some(cmd) = self.rx.recv().await {
             match cmd {
                 PlayerCommand::AddConnection { sender, promise } => {
@@ -349,7 +348,6 @@ impl<'a> Player<'a> {
                     }
                 }
                 PlayerCommand::Cmd(cmd, connection_id) => self.handle_cmd(cmd, connection_id).await,
-                PlayerCommand::Stop => { break; }
             }
         }
         log::debug!("Shutting down player actor #{:?}", self.player_id);

src/core/repo/mod.rs

@@ -1,6 +1,7 @@
 //! Storage and persistence logic.
 
 use std::str::FromStr;
+use std::sync::Arc;
 
 use serde::Deserialize;
 use sqlx::sqlite::SqliteConnectOptions;
@@ -14,8 +15,9 @@ pub struct StorageConfig {
     pub db_path: String,
 }
 
+#[derive(Clone)]
 pub struct Storage {
-    conn: Mutex<SqliteConnection>,
+    conn: Arc<Mutex<SqliteConnection>>,
 }
 impl Storage {
     pub async fn open(config: StorageConfig) -> Result<Storage> {
@@ -27,11 +29,11 @@ impl Storage {
         migrator.run(&mut conn).await?;
         log::info!("Migrations passed");
 
-        let conn = Mutex::new(conn);
+        let conn = Arc::new(Mutex::new(conn));
         Ok(Storage { conn })
     }
 
-    pub async fn retrieve_user_by_name(&self, name: &str) -> Result<Option<StoredUser>> {
+    pub async fn retrieve_user_by_name(&mut self, name: &str) -> Result<Option<StoredUser>> {
         let mut executor = self.conn.lock().await;
         let res = sqlx::query_as(
             "select u.id, u.name, c.password
@@ -45,7 +47,7 @@ impl Storage {
         Ok(res)
     }
 
-    pub async fn retrieve_room_by_name(&self, name: &str) -> Result<Option<StoredRoom>> {
+    pub async fn retrieve_room_by_name(&mut self, name: &str) -> Result<Option<StoredRoom>> {
         let mut executor = self.conn.lock().await;
         let res = sqlx::query_as(
             "select id, name, topic, message_count
@@ -59,7 +61,7 @@ impl Storage {
         Ok(res)
     }
 
-    pub async fn create_new_room(&self, name: &str, topic: &str) -> Result<u32> {
+    pub async fn create_new_room(&mut self, name: &str, topic: &str) -> Result<u32> {
         let mut executor = self.conn.lock().await;
         let (id,): (u32,) = sqlx::query_as(
             "insert into rooms(name, topic)
@@ -74,9 +76,9 @@ impl Storage {
         Ok(id)
     }
 
-    pub async fn insert_message(&self, room_id: u32, id: u32, content: &str) -> Result<()> {
+    pub async fn insert_message(&mut self, room_id: u32, id: u32, content: &str) -> Result<()> {
         let mut executor = self.conn.lock().await;
-        sqlx::query(
+        let (_,): (u32,) = sqlx::query_as(
             "insert into messages(room_id, id, content)
             values (?, ?, ?);
             update rooms set message_count = message_count + 1 where id = ?;",
@@ -85,14 +87,18 @@ impl Storage {
         .bind(id)
         .bind(content)
         .bind(room_id)
-        .execute(&mut *executor)
+        .fetch_one(&mut *executor)
         .await?;
 
         Ok(())
     }
 
-    pub async fn close(self) -> Result<()> {
-        let res = self.conn.into_inner();
+    pub async fn close(mut self) -> Result<()> {
+        let res = match Arc::try_unwrap(self.conn) {
+            Ok(e) => e,
+            Err(e) => return Err(fail("failed to acquire DB ownership on shutdown")),
+        };
+        let res = res.into_inner();
         res.close().await?;
         Ok(())
     }

src/core/room.rs

@@ -38,9 +38,10 @@ impl RoomId {
 }
 
 /// Shared datastructure for storing metadata about rooms.
-pub struct RoomRegistry<'a>(AsyncRwLock<RoomRegistryInner<'a>>);
-impl<'a> RoomRegistry<'a> {
-    pub fn new(metrics: &mut MetricRegistry, storage: &'a Storage) -> Result<RoomRegistry<'a>> {
+#[derive(Clone)]
+pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>);
+impl RoomRegistry {
+    pub fn new(metrics: &mut MetricRegistry, storage: Storage) -> Result<RoomRegistry> {
         let metric_active_rooms =
             IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
         metrics.register(Box::new(metric_active_rooms.clone()))?;
@@ -49,10 +50,10 @@ impl<'a> RoomRegistry<'a> {
             metric_active_rooms,
             storage,
         };
-        Ok(RoomRegistry(AsyncRwLock::new(inner)))
+        Ok(RoomRegistry(Arc::new(AsyncRwLock::new(inner))))
     }
 
-    pub async fn get_or_create_room(&self, room_id: RoomId) -> Result<RoomHandle<'a>> {
+    pub async fn get_or_create_room(&mut self, room_id: RoomId) -> Result<RoomHandle> {
         let mut inner = self.0.write().await;
         if let Some(room_handle) = inner.rooms.get(&room_id) {
             // room was already loaded into memory
@@ -67,7 +68,7 @@ impl<'a> RoomRegistry<'a> {
                 subscriptions: HashMap::new(), // TODO figure out how to populate subscriptions
                 topic: stored_room.topic.into(),
                 message_count: stored_room.message_count,
-                storage: inner.storage,
+                storage: inner.storage.clone(),
             };
             let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
             inner.rooms.insert(room_id, room_handle.clone());
@@ -84,7 +85,7 @@ impl<'a> RoomRegistry<'a> {
                 subscriptions: HashMap::new(),
                 topic: topic.into(),
                 message_count: 0,
-                storage: inner.storage,
+                storage: inner.storage.clone(),
             };
             let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
             inner.rooms.insert(room_id, room_handle.clone());
@@ -93,7 +94,7 @@ impl<'a> RoomRegistry<'a> {
         }
     }
 
-    pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle<'a>> {
+    pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle> {
         let inner = self.0.read().await;
         let res = inner.rooms.get(room_id);
         res.map(|r| r.clone())
@@ -113,15 +114,15 @@ impl<'a> RoomRegistry<'a> {
     }
 }
 
-struct RoomRegistryInner<'a> {
-    rooms: HashMap<RoomId, RoomHandle<'a>>,
+struct RoomRegistryInner {
+    rooms: HashMap<RoomId, RoomHandle>,
     metric_active_rooms: IntGauge,
-    storage: &'a Storage,
+    storage: Storage,
 }
 
 #[derive(Clone)]
-pub struct RoomHandle<'a>(Arc<AsyncRwLock<Room<'a>>>);
-impl<'a> RoomHandle<'a> {
+pub struct RoomHandle(Arc<AsyncRwLock<Room>>);
+impl RoomHandle {
     pub async fn subscribe(&self, player_id: PlayerId, player_handle: PlayerHandle) {
         let mut lock = self.0.write().await;
         lock.add_subscriber(player_id, player_handle).await;
@@ -139,10 +140,7 @@ impl<'a> RoomHandle<'a> {
 
     pub async fn send_message(&self, player_id: PlayerId, body: Str) {
         let mut lock = self.0.write().await;
-        let res =  lock.send_message(player_id, body).await;
-        if let Err(err) = res {
-            log::warn!("Failed to send message: {err:?}");
-        }
+        lock.send_message(player_id, body).await;
     }
 
     pub async fn get_room_info(&self) -> RoomInfo {
@@ -169,15 +167,15 @@ impl<'a> RoomHandle<'a> {
     }
 }
 
-struct Room<'a> {
+struct Room {
     storage_id: u32,
     room_id: RoomId,
     subscriptions: HashMap<PlayerId, PlayerHandle>,
     message_count: u32,
     topic: Str,
-    storage: &'a Storage,
+    storage: Storage,
 }
-impl<'a> Room<'a> {
+impl Room {
     async fn add_subscriber(&mut self, player_id: PlayerId, player_handle: PlayerHandle) {
         tracing::info!("Adding a subscriber to room");
         self.subscriptions.insert(player_id.clone(), player_handle);
@@ -202,7 +200,6 @@ impl<'a> Room<'a> {
     }
 
     async fn broadcast_update(&self, update: Updates, except: &PlayerId) {
-        tracing::debug!("Broadcasting an update to {} subs", self.subscriptions.len());
         for (player_id, sub) in &self.subscriptions {
             if player_id == except {
                 continue;
@@ -213,7 +210,7 @@ impl<'a> Room<'a> {
     }
 }
 
-#[derive(Serialize, Debug)]
+#[derive(Serialize)]
 pub struct RoomInfo {
     pub id: RoomId,
     pub members: Vec<PlayerId>,

src/main.rs

@@ -53,26 +53,29 @@ async fn main() -> Result<()> {
     } = config;
     let mut metrics = MetricsRegistry::new();
     let storage = Storage::open(storage_config).await?;
-    let rooms = RoomRegistry::new(&mut metrics, &storage)?;
-    let players = PlayerRegistry::empty(&rooms, &metrics)?;
+    let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
+    let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?;
+    let telemetry_terminator =
+        util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone()).await?;
 
-    // unsafe: outer future is never dropped, scope is joined on `scope.collect`
-    let mut scope = unsafe { Scope::create() };
-    let telemetry_terminator = util::telemetry::launch(telemetry_config, &metrics, &rooms, &mut scope).await?;
-    let irc = projections::irc::launch(&irc_config, &players, &rooms, &metrics, &storage, &mut scope).await?;
-    let xmpp = projections::xmpp::launch(xmpp_config, &players, &rooms, &metrics, &mut scope).await?;
+    unsafe { async_scoped::TokioScope::scope_and_collect(|s| {
+        s.spawn(async {
+            let irc = projections::irc::launch(&irc_config, &players, &rooms, &metrics, &storage).await?;
+        });
+    }) }.await;
+    
+    let xmpp = projections::xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?;
     tracing::info!("Started");
 
     sleep.await;
 
     tracing::info!("Begin shutdown");
-    let _ = xmpp.send(());
-    let _ = irc.send(());
-    let _ = telemetry_terminator.send(());
-    let _ = scope.collect().await;
-    drop(scope);
-
+    xmpp.terminate().await?;
+    irc.terminate().await?;
+    telemetry_terminator.terminate().await?;
     players.shutdown_all().await?;
+    drop(players);
+    drop(rooms);
     storage.close().await?;
     tracing::info!("Shutdown complete");
     Ok(())

src/prelude.rs

@@ -23,5 +23,3 @@ macro_rules! ffail {
 }
 
 pub(crate) use ffail;
-
-pub type Scope<'a> = async_scoped::Scope<'a, (), async_scoped::Tokio>;

src/projections/irc/mod.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::net::SocketAddr;
 
-use futures_util::FutureExt;
+use futures_util::future::join_all;
 use prometheus::{IntCounter, IntGauge, Registry as MetricsRegistry};
 use serde::Deserialize;
 use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter};
@@ -16,6 +16,7 @@ use crate::prelude::*;
 use crate::protos::irc::client::{client_message, ClientMessage};
 use crate::protos::irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
 use crate::protos::irc::{Chan, Recipient};
+use crate::util::Terminator;
 
 #[cfg(test)]
 mod test;
@@ -42,10 +43,10 @@ async fn handle_socket(
     config: ServerConfig,
     mut stream: TcpStream,
     socket_addr: &SocketAddr,
-    players: &PlayerRegistry<'_>,
-    rooms: &RoomRegistry<'_>,
+    players: PlayerRegistry,
+    rooms: RoomRegistry,
     termination: Deferred<()>, // TODO use it to stop the connection gracefully
-    storage: &Storage,
+    mut storage: Storage,
 ) -> Result<()> {
     let (reader, writer) = stream.split();
     let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
@@ -65,7 +66,7 @@ async fn handle_socket(
     writer.flush().await?;
 
     let registered_user: Result<RegisteredUser> =
-        handle_registration(&mut reader, &mut writer, &storage).await;
+        handle_registration(&mut reader, &mut writer, &mut storage).await;
 
     match registered_user {
         Ok(user) => {
@@ -82,7 +83,7 @@ async fn handle_socket(
 async fn handle_registration<'a>(
     reader: &mut BufReader<ReadHalf<'a>>,
     writer: &mut BufWriter<WriteHalf<'a>>,
-    storage: &Storage,
+    storage: &mut Storage,
 ) -> Result<RegisteredUser> {
     let mut buffer = vec![];
 
@@ -175,8 +176,8 @@ async fn handle_registration<'a>(
 
 async fn handle_registered_socket<'a>(
     config: ServerConfig,
-    players: &PlayerRegistry<'_>,
-    rooms: &RoomRegistry<'_>,
+    mut players: PlayerRegistry,
+    rooms: RoomRegistry,
     reader: &mut BufReader<ReadHalf<'a>>,
     writer: &mut BufWriter<WriteHalf<'a>>,
     user: RegisteredUser,
@@ -300,7 +301,7 @@ async fn handle_update(
     user: &RegisteredUser,
     player_id: &PlayerId,
     writer: &mut (impl AsyncWrite + Unpin),
-    rooms: &RoomRegistry<'_>,
+    rooms: &RoomRegistry,
     update: Updates,
 ) -> Result<()> {
     log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
@@ -397,7 +398,7 @@ async fn handle_incoming_message(
     buffer: &str,
     config: &ServerConfig,
     user: &RegisteredUser,
-    rooms: &RoomRegistry<'_>,
+    rooms: &RoomRegistry,
     user_handle: &mut PlayerConnection,
     writer: &mut (impl AsyncWrite + Unpin),
 ) -> Result<HandleResult> {
@@ -688,14 +689,13 @@ async fn produce_on_join_cmd_messages(
     Ok(())
 }
 
-pub async fn launch<'a>(
-    config: &'a ServerConfig,
-    players: &'a PlayerRegistry<'_>,
-    rooms: &'a RoomRegistry<'_>,
-    metrics: &'a MetricsRegistry,
-    storage: &'a Storage,
-    scope: &mut Scope<'a>,
-) -> Result<Promise<()>> {
+pub async fn launch(
+    config: &ServerConfig,
+    players: &PlayerRegistry,
+    rooms: &RoomRegistry,
+    metrics: &MetricsRegistry,
+    storage: &Storage,
+) -> Result<Terminator> {
     log::info!("Starting IRC projection");
     let (stopped_tx, mut stopped_rx) = channel(32);
     let current_connections =
@@ -709,12 +709,10 @@ pub async fn launch<'a>(
 
     let listener = TcpListener::bind(config.listen_on).await?;
 
-    let (signal, mut rx) = oneshot();
-    let future = async move {
+    let terminator = Terminator::spawn(|mut rx| async move {
         // TODO probably should separate logic for accepting new connection and storing them
         // into two tasks so that they don't block each other
         let mut actors = HashMap::new();
-        let mut scope = unsafe { Scope::create() };
         loop {
             select! {
                 biased;
@@ -737,22 +735,23 @@ pub async fn launch<'a>(
                                 continue;
                             }
 
-
-                            let current_connections_clone = current_connections.clone();
-                            let stopped_tx = stopped_tx.clone();
-
-                            let (a, rx) = oneshot();
-                            let future = async move {
-                                match handle_socket(config, stream, &socket_addr, players, rooms, rx, storage).await {
-                                    Ok(_) => log::info!("Connection terminated"),
-                                    Err(err) => log::warn!("Connection failed: {err}"),
+                            let terminator = Terminator::spawn(|termination| {
+                                let players = players.clone();
+                                let rooms = rooms.clone();
+                                let current_connections_clone = current_connections.clone();
+                                let stopped_tx = stopped_tx.clone();
+                                let storage = storage.clone();
+                                async move {
+                                    match handle_socket(config, stream, &socket_addr, players, rooms, termination, storage).await {
+                                        Ok(_) => log::info!("Connection terminated"),
+                                        Err(err) => log::warn!("Connection failed: {err}"),
+                                    }
+                                    current_connections_clone.dec();
+                                    stopped_tx.send(socket_addr).await?;
+                                    Ok(())
                                 }
-                                current_connections_clone.dec();
-                                stopped_tx.send(socket_addr).await?;
-                                Ok(())
-                            };
-                            scope.spawn(future.map(|_: Result<()>| ()));
-                            actors.insert(socket_addr, a);
+                            });
+                            actors.insert(socket_addr, terminator);
                         },
                         Err(err) => log::warn!("Failed to accept new connection: {err}"),
                     }
@@ -761,20 +760,19 @@ pub async fn launch<'a>(
         }
 
         log::info!("Stopping IRC projection");
-        for (socket_addr, terminator) in actors {
-            match terminator.send(()) {
-                Ok(_) => log::debug!("Stopping IRC connection at {socket_addr}"),
-                Err(_) => log::debug!("IRC connection at {socket_addr} already stopped")
+        join_all(actors.into_iter().map(|(socket_addr, terminator)| async move {
+            log::debug!("Stopping IRC connection at {socket_addr}");
+            match terminator.terminate().await {
+                Ok(_) => log::debug!("Stopped IRC connection at {socket_addr}"),
+                Err(err) => {
+                    log::warn!("IRC connection to {socket_addr} finished with error: {err}")
+                }
             }
-        }
-        let _ = scope.collect().await;
-        drop(scope);
+        })).await;
         log::info!("Stopped IRC projection");
         Ok(())
-    };
-
-    scope.spawn(future.map(|_: Result<()>| ()));
+    });
 
     log::info!("Started IRC projection");
-    Ok(signal)
+    Ok(terminator)
 }

src/projections/xmpp/mod.rs

@@ -7,7 +7,7 @@ use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use futures_util::FutureExt;
+use futures_util::future::join_all;
 use prometheus::Registry as MetricsRegistry;
 use quick_xml::events::{BytesDecl, Event};
 use quick_xml::{NsReader, Writer};
@@ -30,6 +30,7 @@ use crate::protos::xmpp::roster::RosterQuery;
 use crate::protos::xmpp::session::Session;
 use crate::protos::xmpp::stream::*;
 use crate::util::xml::{Continuation, FromXml, Parser, ToXml};
+use crate::util::Terminator;
 
 use self::proto::{ClientPacket, IqClientBody};
 
@@ -52,13 +53,12 @@ struct Authenticated {
     xmpp_muc_name: Resource,
 }
 
-pub async fn launch<'a>(
+pub async fn launch(
     config: ServerConfig,
-    players: &'a PlayerRegistry<'_>,
-    rooms: &'a RoomRegistry<'_>,
-    metrics: &'a MetricsRegistry,
-    scope: &mut Scope<'a>,
-) -> Result<Promise<()>> {
+    players: PlayerRegistry,
+    rooms: RoomRegistry,
+    metrics: MetricsRegistry,
+) -> Result<Terminator> {
     log::info!("Starting XMPP projection");
 
     let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?;
@@ -75,16 +75,13 @@ pub async fn launch<'a>(
     });
 
     let listener = TcpListener::bind(config.listen_on).await?;
-
-    let (signal, mut rx) = oneshot();
-    let future = async move {
+    let terminator = Terminator::spawn(|mut termination| async move {
         let (stopped_tx, mut stopped_rx) = channel(32);
         let mut actors = HashMap::new();
-        let mut scope = unsafe { Scope::create() };
         loop {
             select! {
                 biased;
-                _ = &mut rx => break,
+                _ = &mut termination => break,
                 stopped = stopped_rx.recv() => match stopped {
                     Some(stopped) => { let _ = actors.remove(&stopped); },
                     None => unreachable!(),
@@ -98,20 +95,22 @@ pub async fn launch<'a>(
                                 // TODO kill the older connection and restart it
                                 continue;
                             }
-                            let stopped_tx = stopped_tx.clone();
-                            let loaded_config = loaded_config.clone();
-                            let (a, rx) = oneshot();
-                            let future = async move {
-                                    match handle_socket(loaded_config, stream, &socket_addr, players, rooms, rx).await {
+                            let players = players.clone();
+                            let rooms = rooms.clone();
+                            let terminator = Terminator::spawn(|termination| {
+                                let stopped_tx = stopped_tx.clone();
+                                let loaded_config = loaded_config.clone();
+                                async move {
+                                    match handle_socket(loaded_config, stream, &socket_addr, players, rooms, termination).await {
                                         Ok(_) => log::info!("Connection terminated"),
                                         Err(err) => log::warn!("Connection failed: {err}"),
                                     }
                                     stopped_tx.send(socket_addr).await?;
                                     Ok(())
-                            };
-                            scope.spawn(future.map(|_: Result<()>| ()));
+                                }
+                            });
 
-                            actors.insert(socket_addr, a);
+                            actors.insert(socket_addr, terminator);
                         },
                         Err(err) => log::warn!("Failed to accept new connection: {err}"),
                     }
@@ -119,28 +118,35 @@ pub async fn launch<'a>(
             }
         }
         log::info!("Stopping XMPP projection");
-        for (socket_addr, terminator) in actors {
-            match terminator.send(()) {
-                Ok(_) => log::debug!("Stopping XMPP connection at {socket_addr}"),
-                Err(_) => log::debug!("XMPP connection at {socket_addr} already stopped")
-            }
-        }
-        let _ = scope.collect().await;
-        drop(scope);
+        join_all(
+            actors
+                .into_iter()
+                .map(|(socket_addr, terminator)| async move {
+                    log::debug!("Stopping XMPP connection at {socket_addr}");
+                    match terminator.terminate().await {
+                        Ok(_) => log::debug!("Stopped XMPP connection at {socket_addr}"),
+                        Err(err) => {
+                            log::warn!(
+                                "XMPP connection to {socket_addr} finished with error: {err}"
+                            )
+                        }
+                    }
+                }),
+        )
+        .await;
         log::info!("Stopped XMPP projection");
         Ok(())
-    };
-    scope.spawn(future.map(|_: Result<()>| ()));
+    });
     log::info!("Started XMPP projection");
-    Ok(signal)
+    Ok(terminator)
 }
 
 async fn handle_socket(
     config: Arc<LoadedConfig>,
     mut stream: TcpStream,
     socket_addr: &SocketAddr,
-    players: &PlayerRegistry<'_>,
-    rooms: &RoomRegistry<'_>,
+    mut players: PlayerRegistry,
+    rooms: RoomRegistry,
     termination: Deferred<()>, // TODO use it to stop the connection gracefully
 ) -> Result<()> {
     log::debug!("Received an XMPP connection from {socket_addr}");
@@ -265,7 +271,7 @@ async fn socket_final(
     reader_buf: &mut Vec<u8>,
     authenticated: &Authenticated,
     user_handle: &mut PlayerConnection,
-    rooms: &RoomRegistry<'_>,
+    rooms: &RoomRegistry,
 ) -> Result<()> {
     read_xml_header(xml_reader, reader_buf).await?;
     let _ = ClientStreamStart::parse(xml_reader, reader_buf).await?;
@@ -375,7 +381,7 @@ async fn handle_packet(
     packet: ClientPacket,
     user: &Authenticated,
     user_handle: &mut PlayerConnection,
-    rooms: &RoomRegistry<'_>,
+    rooms: &RoomRegistry,
 ) -> Result<bool> {
     Ok(match packet {
         proto::ClientPacket::Iq(iq) => {
@@ -472,7 +478,7 @@ async fn handle_packet(
     })
 }
 
-async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms: &RoomRegistry<'_>) {
+async fn handle_iq(output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>, rooms: &RoomRegistry) {
     match iq.body {
         proto::IqClientBody::Bind(b) => {
             let req = Iq {
@@ -584,7 +590,7 @@ fn disco_info(to: Option<&str>, req: &InfoQuery) -> InfoQuery {
     }
 }
 
-async fn disco_items(to: Option<&str>, req: &ItemQuery, rooms: &RoomRegistry<'_>) -> ItemQuery {
+async fn disco_items(to: Option<&str>, req: &ItemQuery, rooms: &RoomRegistry) -> ItemQuery {
     let item = match to {
         Some("localhost") => {
             vec![Item {

src/util/telemetry.rs

@@ -15,6 +15,7 @@ use crate::core::room::RoomRegistry;
 use crate::prelude::*;
 
 use crate::util::http::*;
+use crate::util::Terminator;
 
 type BoxBody = http_body_util::combinators::BoxBody<Bytes, Infallible>;
 type HttpResult<T> = std::result::Result<T, Infallible>;
@@ -24,48 +25,51 @@ pub struct ServerConfig {
     pub listen_on: SocketAddr,
 }
 
-pub async fn launch<'a>(
+pub async fn launch(
     config: ServerConfig,
-    metrics: &'a MetricsRegistry,
-    rooms: &'a RoomRegistry<'_>,
-    scope: &mut Scope<'a>,
-) -> Result<Promise<()>> {
+    metrics: MetricsRegistry,
+    rooms: RoomRegistry,
+) -> Result<Terminator> {
     log::info!("Starting the telemetry service");
     let listener = TcpListener::bind(config.listen_on).await?;
     log::debug!("Listener started");
+    let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, rooms, rx.map(|_| ())));
+    Ok(terminator)
+}
 
-    let (signal, mut rx) = oneshot();
-
-    let future = async move {
-        let mut scope = unsafe { Scope::create() };
-        loop {
-            select! {
-                biased;
-                _ = &mut rx => break,
-                result = listener.accept() => {
-                    let (stream, _) = result?;
-                    scope.spawn(async move {
-                        let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(metrics, rooms, r)));
-                        if let Err(err) = server.await {
-                            tracing::error!("Error serving connection: {:?}", err);
-                        }
-                    });
-                },
-            }
+async fn main_loop(
+    listener: TcpListener,
+    metrics: MetricsRegistry,
+    rooms: RoomRegistry,
+    termination: impl Future<Output = ()>,
+) -> Result<()> {
+    pin!(termination);
+    loop {
+        select! {
+            biased;
+            _ = &mut termination => break,
+            result = listener.accept() => {
+                let (stream, _) = result?;
+                let metrics = metrics.clone();
+                let rooms = rooms.clone();
+                tokio::task::spawn(async move {
+                    let registry = metrics.clone();
+                    let rooms = rooms.clone();
+                    let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), rooms.clone(), r)));
+                    if let Err(err) = server.await {
+                        tracing::error!("Error serving connection: {:?}", err);
+                    }
+                });
+            },
         }
-        let _ = scope.collect().await;
-        drop(scope);
-        log::info!("Terminating the telemetry service");
-        Ok(())
-    };
-    scope.spawn(future.map(|_: Result<()>| ()));
-    
-    Ok(signal)
+    }
+    log::info!("Terminating the telemetry service");
+    Ok(())
 }
 
 async fn route(
-    registry: &MetricsRegistry,
-    rooms: &RoomRegistry<'_>,
+    registry: MetricsRegistry,
+    rooms: RoomRegistry,
     request: Request<hyper::body::Incoming>,
 ) -> std::result::Result<Response<BoxBody>, Infallible> {
     match (request.method(), request.uri().path()) {
@@ -75,7 +79,7 @@ async fn route(
     }
 }
 
-fn endpoint_metrics(registry: &MetricsRegistry) -> HttpResult<Response<Full<Bytes>>> {
+fn endpoint_metrics(registry: MetricsRegistry) -> HttpResult<Response<Full<Bytes>>> {
     let mf = registry.gather();
     let mut buffer = vec![];
     TextEncoder
@@ -84,7 +88,7 @@ fn endpoint_metrics(registry: &MetricsRegistry) -> HttpResult<Response<Full<Byte
     Ok(Response::new(Full::new(Bytes::from(buffer))))
 }
 
-async fn endpoint_rooms(rooms: &RoomRegistry<'_>) -> HttpResult<Response<Full<Bytes>>> {
+async fn endpoint_rooms(rooms: RoomRegistry) -> HttpResult<Response<Full<Bytes>>> {
     let room_list = rooms.get_all_rooms().await;
     let mut buffer = vec![];
     serde_json::to_writer(&mut buffer, &room_list).expect("unexpected fail when writing to vec");