Merge pull request #627 from nextcloud/redis-pubsub-ping

send keepalive pings over redis pubsub connection

Author: icewind1991

Cargo.lock

@@ -8,7 +8,7 @@ edition = "2021"
 rust-version = "1.81.0"
 
 [dependencies]
-redis = { version = "0.30.0", default-features = false, features = ["tokio-comp", "aio", "cluster", "cluster-async", "keep-alive", "tls-rustls", "tokio-rustls-comp", "tls-rustls-webpki-roots", "tls-rustls-insecure"] }
+redis = { version = "0.31.0", default-features = false, features = ["tokio-comp", "aio", "cluster", "cluster-async", "keep-alive", "tls-rustls", "tokio-rustls-comp", "tls-rustls-webpki-roots", "tls-rustls-insecure"] }
 serde = { version = "1.0.217", features = ["derive"] }
 serde_json = "1.0.139"
 thiserror = "2.0.11"
@@ -24,7 +24,7 @@ miette = { version = "7.4.0", features = ["fancy"] }
 smallvec = { version = "1.13.2", features = ["serde"] }
 reqwest = { version = "0.12.12", default-features = false, features = ["rustls-tls", "rustls-tls-native-roots", "json"] }
 warp-real-ip = "0.2.0"
-parse-display = "0.9.1"
+parse-display = "0.10.0"
 rand = { version = "0.8.5", features = ["small_rng"] }
 ahash = "0.8.11"
 flexi_logger = { version = "0.29.8", features = ["colors"] }

Cargo.toml

@@ -6,6 +6,7 @@
 use crate::metrics::METRICS;
 use crate::{Redis, Result, UserId};
 use parse_display::Display;
+use redis::aio::PubSubSink;
 use redis::Msg;
 use serde::Deserialize;
 use serde_json::Value;
@@ -153,7 +154,10 @@ impl TryFrom<Msg> for Event {
 
 pub async fn subscribe(
     client: &Redis,
-) -> Result<impl Stream<Item = Result<Event, MessageDecodeError>>> {
+) -> Result<(
+    PubSubSink,
+    impl Stream<Item = Result<Event, MessageDecodeError>>,
+)> {
     let mut pubsub = client.pubsub().await?;
     let channels = [
         "notify_storage_update",
@@ -172,8 +176,12 @@ pub async fn subscribe(
         pubsub.subscribe(*channel).await?;
     }
 
-    Ok(pubsub.into_on_message().map(|event| {
-        METRICS.add_event();
-        Event::try_from(event)
-    }))
+    let (sink, stream) = pubsub.split();
+    Ok((
+        sink,
+        stream.map(|event| {
+            METRICS.add_event();
+            Event::try_from(event)
+        }),
+    ))
 }

src/event.rs

@@ -418,7 +418,7 @@ pub async fn listen_loop(app: Arc<App>, cancel: oneshot::Receiver<()>) {
 }
 
 pub async fn listen(app: Arc<App>) -> Result<()> {
-    let mut event_stream = event::subscribe(&app.redis).await?;
+    let (mut pubsub_sink, mut event_stream) = event::subscribe(&app.redis).await?;
 
     let handle = move |event: Event| {
         // todo: any way to do this without cloning the arc every event (scoped?)
@@ -428,6 +428,13 @@ pub async fn listen(app: Arc<App>) -> Result<()> {
         }
     };
 
+    let ping_handle = tokio::spawn(async move {
+        loop {
+            sleep(Duration::from_secs(15)).await;
+            let _ = pubsub_sink.ping::<()>().await;
+        }
+    });
+
     while let Some(event) = event_stream.next().await {
         match event {
             Ok(event) => {
@@ -440,5 +447,7 @@ pub async fn listen(app: Arc<App>) -> Result<()> {
             Err(e) => log::warn!("{e:#}"),
         }
     }
+
+    ping_handle.abort();
     Ok(())
 }