chore: bump version to 0.7.7

Cargo.lock

@@ -3452,7 +3452,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-bonfire"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "async-channel 2.3.1",
  "async-std",
@@ -3482,7 +3482,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-config"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "async-std",
  "cached",
@@ -3498,7 +3498,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-database"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "async-lock",
  "async-recursion",
@@ -3544,7 +3544,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-delta"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "async-channel 1.6.1",
  "async-std",
@@ -3590,7 +3590,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-models"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "indexmap",
  "iso8601-timestamp 0.2.11",
@@ -3607,7 +3607,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-permissions"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "async-std",
  "async-trait",
@@ -3622,7 +3622,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-presence"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "async-std",
  "log",
@@ -3633,7 +3633,7 @@ dependencies = [
 
 [[package]]
 name = "revolt-result"
-version = "0.7.6"
+version = "0.7.7"
 dependencies = [
  "revolt_okapi",
  "revolt_rocket_okapi",

crates/bonfire/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-bonfire"
-version = "0.7.6"
+version = "0.7.7"
 license = "AGPL-3.0-or-later"
 edition = "2021"
 
@@ -41,7 +41,7 @@ revolt-result = { path = "../core/result" }
 revolt-models = { path = "../core/models" }
 revolt-config = { path = "../core/config" }
 revolt-database = { path = "../core/database" }
-revolt-permissions = { version = "0.7.6", path = "../core/permissions" }
+revolt-permissions = { version = "0.7.7", path = "../core/permissions" }
 revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
 
 # redis

crates/bonfire/src/websocket.rs

@@ -8,7 +8,7 @@ use fred::{
 };
 use futures::{
     channel::oneshot,
-    pin_mut, select,
+    join, pin_mut, select,
     stream::{SplitSink, SplitStream},
     FutureExt, SinkExt, StreamExt, TryStreamExt,
 };
@@ -22,6 +22,7 @@ use revolt_presence::{create_session, delete_session};
 use async_std::{
     net::TcpStream,
     sync::{Mutex, RwLock},
+    task::spawn,
 };
 use revolt_result::create_error;
 use sentry::Level;
@@ -127,34 +128,44 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     }
 
     {
+        // Setup channels and mutexes
         let write = Mutex::new(write);
         let subscribed = state.subscribed.clone();
         let active_servers = state.active_servers.clone();
-        let (signal_s, signal_r) = async_channel::unbounded();
+        let (topic_signal_s, topic_signal_r) = async_channel::unbounded();
+
+        // TODO: this needs to be rewritten
+        // Create channels through which the tasks can signal to each other they need to clean up
+        let (kill_signal_1_s, kill_signal_1_r) = async_channel::bounded(1);
+        let (kill_signal_2_s, kill_signal_2_r) = async_channel::bounded(1);
 
         // Create a PubSub connection to poll on.
-        let listener = listener(db, &mut state, addr, &config, signal_r, &write).fuse();
+        let listener = listener_with_kill_signal(
+            db,
+            &mut state,
+            addr,
+            &config,
+            topic_signal_r,
+            kill_signal_1_r,
+            &write,
+            kill_signal_2_s,
+        );
+
         // Read from WebSocket stream.
-        let worker = worker(
+        let worker = worker_with_kill_signal(
             addr,
             subscribed,
             active_servers,
             user_id.clone(),
             &config,
-            signal_s,
+            topic_signal_s,
+            kill_signal_2_r,
             read,
             &write,
-        )
-        .fuse();
-
-        // Pin both tasks.
-        pin_mut!(listener, worker);
-
-        // Wait for either disconnect or for listener to die.
-        select!(
-            () = listener => {},
-            () = worker => {}
+            kill_signal_1_s,
         );
+
+        join!(listener, worker);
     }
     // Clean up presence session.
     let last_session = delete_session(&user_id, session_id).await;
@@ -165,16 +176,40 @@ pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr)
     }
 }
 
+#[allow(clippy::too_many_arguments)]
+async fn listener_with_kill_signal(
+    db: &'static Database,
+    state: &mut State,
+    addr: SocketAddr,
+    config: &ProtocolConfiguration,
+    topic_signal_r: async_channel::Receiver<()>,
+    kill_signal_r: async_channel::Receiver<()>,
+    write: &Mutex<WsWriter>,
+    kill_signal_s: async_channel::Sender<()>,
+) {
+    listener(
+        db,
+        state,
+        addr,
+        config,
+        topic_signal_r,
+        kill_signal_r,
+        write,
+    )
+    .await;
+    kill_signal_s.send(()).await.ok();
+}
+
 async fn listener(
     db: &'static Database,
     state: &mut State,
     addr: SocketAddr,
     config: &ProtocolConfiguration,
-    signal_r: async_channel::Receiver<()>,
+    topic_signal_r: async_channel::Receiver<()>,
+    kill_signal_r: async_channel::Receiver<()>,
     write: &Mutex<WsWriter>,
 ) {
     let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
-
     let subscriber = match fred::types::Builder::from_config(redis_config).build_subscriber_client()
     {
         Ok(subscriber) => subscriber,
@@ -191,15 +226,26 @@ async fn listener(
         return;
     };
 
+    // Handle Redis connection dropping
+    let (clean_up_s, clean_up_r) = async_channel::bounded(1);
+    let clean_up_s = Arc::new(Mutex::new(clean_up_s));
+    subscriber.on_error(move |_| {
+        let clean_up_s = clean_up_s.clone();
+        spawn(async move {
+            clean_up_s.lock().await.send(()).await.ok();
+        });
+        Ok(())
+    });
+
     let mut message_rx = subscriber.message_rx();
-    loop {
+    'out: loop {
         // Check for state changes for subscriptions.
         match state.apply_state().await {
             SubscriptionStateChange::Reset => {
                 if let Err(err) = subscriber.unsubscribe_all().await {
                     error!("Unsubscribe all failed: {err:?}");
                     sentry::capture_error(&err);
-                    return;
+                    break 'out;
                 }
 
                 let subscribed = state.subscribed.read().await;
@@ -207,7 +253,7 @@ async fn listener(
                     if let Err(err) = subscriber.subscribe(id).await {
                         error!("Subscribe failed: {err:?}");
                         sentry::capture_error(&err);
-                        return;
+                        break 'out;
                     }
                 }
 
@@ -222,7 +268,7 @@ async fn listener(
                     if let Err(err) = subscriber.unsubscribe(id).await {
                         error!("Unsubscribe failed: {err:?}");
                         sentry::capture_error(&err);
-                        return;
+                        break 'out;
                     }
                 }
 
@@ -233,7 +279,7 @@ async fn listener(
                     if let Err(err) = subscriber.subscribe(id).await {
                         error!("Subscribe failed: {err:?}");
                         sentry::capture_error(&err);
-                        return;
+                        break 'out;
                     }
                 }
             }
@@ -241,11 +287,19 @@ async fn listener(
         }
 
         let t1 = message_rx.recv().fuse();
-        let t2 = signal_r.recv().fuse();
+        let t2 = topic_signal_r.recv().fuse();
+        let t3 = kill_signal_r.recv().fuse();
+        let t4 = clean_up_r.recv().fuse();
 
-        pin_mut!(t1, t2);
+        pin_mut!(t1, t2, t3, t4);
 
         select! {
+            _ = t4 => {
+                break 'out;
+            },
+            _ = t3 => {
+                break 'out;
+            },
             _ = t2 => {},
             message = t1 => {
                 // Handle incoming events.
@@ -254,7 +308,7 @@ async fn listener(
                     Err(e) => {
                         error!("Error while consuming pub/sub messages: {e:?}");
                         sentry::capture_error(&e);
-                        return;
+                        break 'out;
                     }
                 };
 
@@ -285,7 +339,7 @@ async fn listener(
 
                     error!("{}", err);
                     sentry::capture_message(&err, Level::Error);
-                    return;
+                    break 'out;
                 };
 
                 if let EventV1::Auth(auth) = &event {
@@ -321,16 +375,49 @@ async fn listener(
                         sentry::capture_message(&err, Level::Warning);
                     }
 
-                    return;
+                    break 'out;
                 }
 
                 if let EventV1::Logout = event {
                     info!("User {addr:?} received log out event!");
-                    return;
+                    break 'out;
                 }
             }
         }
     }
+
+    if let Err(err) = subscriber.quit().await {
+        error!("{}", err);
+        sentry::capture_error(&err);
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn worker_with_kill_signal(
+    addr: SocketAddr,
+    subscribed: Arc<RwLock<HashSet<String>>>,
+    active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
+    user_id: String,
+    config: &ProtocolConfiguration,
+    topic_signal_s: async_channel::Sender<()>,
+    kill_signal_r: async_channel::Receiver<()>,
+    read: WsReader,
+    write: &Mutex<WsWriter>,
+    kill_signal_s: async_channel::Sender<()>,
+) {
+    worker(
+        addr,
+        subscribed,
+        active_servers,
+        user_id,
+        config,
+        topic_signal_s,
+        kill_signal_r,
+        read,
+        write,
+    )
+    .await;
+    kill_signal_s.send(()).await.ok();
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -340,12 +427,22 @@ async fn worker(
     active_servers: Arc<Mutex<lru_time_cache::LruCache<String, ()>>>,
     user_id: String,
     config: &ProtocolConfiguration,
-    signal_s: async_channel::Sender<()>,
+    topic_signal_s: async_channel::Sender<()>,
+    kill_signal_r: async_channel::Receiver<()>,
     mut read: WsReader,
     write: &Mutex<WsWriter>,
 ) {
     loop {
-        let result = read.try_next().await;
+        let t1 = read.try_next().fuse();
+        let t2 = kill_signal_r.recv().fuse();
+
+        pin_mut!(t1, t2);
+
+        select! {
+            _ = t2 => {
+                return;
+            },
+            result = t1 => {
                 let msg = match result {
                     Ok(Some(msg)) => msg,
                     Ok(None) => {
@@ -401,7 +498,7 @@ async fn worker(
 
                         if !has_item {
                             // Poke the listener to adjust subscriptions
-                    signal_s.send(()).await.ok();
+                            topic_signal_s.send(()).await.ok();
                         }
                     }
                     ClientMessage::Ping { data, responded } => {
@@ -418,3 +515,5 @@ async fn worker(
                 }
             }
         }
+    }
+}

crates/core/config/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-config"
-version = "0.7.6"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/database/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-database"
-version = "0.7.6"
+version = "0.7.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -23,13 +23,13 @@ default = ["mongodb", "async-std-runtime", "tasks"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.7.6", path = "../config" }
-revolt-result = { version = "0.7.6", path = "../result" }
-revolt-models = { version = "0.7.6", path = "../models", features = [
+revolt-config = { version = "0.7.7", path = "../config" }
+revolt-result = { version = "0.7.7", path = "../result" }
+revolt-models = { version = "0.7.7", path = "../models", features = [
     "validator",
 ] }
-revolt-presence = { version = "0.7.6", path = "../presence" }
-revolt-permissions = { version = "0.7.6", path = "../permissions", features = [
+revolt-presence = { version = "0.7.7", path = "../presence" }
+revolt-permissions = { version = "0.7.7", path = "../permissions", features = [
     "serde",
     "bson",
 ] }

crates/core/models/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-models"
-version = "0.7.6"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -19,8 +19,8 @@ default = ["serde", "partials", "rocket"]
 
 [dependencies]
 # Core
-revolt-config = { version = "0.7.6", path = "../config" }
-revolt-permissions = { version = "0.7.6", path = "../permissions" }
+revolt-config = { version = "0.7.7", path = "../config" }
+revolt-permissions = { version = "0.7.7", path = "../permissions" }
 
 # Utility
 regex = "1"

crates/core/permissions/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-permissions"
-version = "0.7.6"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]
@@ -21,7 +21,7 @@ async-std = { version = "1.8.0", features = ["attributes"] }
 
 [dependencies]
 # Core
-revolt-result = { version = "0.7.6", path = "../result" }
+revolt-result = { version = "0.7.7", path = "../result" }
 
 # Utility
 auto_ops = "0.3.0"

crates/core/presence/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-presence"
-version = "0.7.6"
+version = "0.7.7"
 edition = "2021"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/core/result/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-result"
-version = "0.7.6"
+version = "0.7.7"
 edition = "2021"
 license = "MIT"
 authors = ["Paul Makles <me@insrt.uk>"]

crates/delta/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt-delta"
-version = "0.7.6"
+version = "0.7.7"
 license = "AGPL-3.0-or-later"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"