Take upstream (#10538)

* fix: redesign and compact Signed-off-by: Leonid Kaganov <lleo@lleo.me> * fix: comments Signed-off-by: Leonid Kaganov <lleo@lleo.me> * fix: correctly dropping connections in timeout Signed-off-by: Leonid Kaganov <lleo@lleo.me> * feature: loglevel in config Signed-off-by: Leonid Kaganov <lleo@lleo.me> * fix: loglevel in config Signed-off-by: Leonid Kaganov <lleo@lleo.me> * features lopt: direct personal messages between websockets by username Signed-off-by: Leonid Kaganov <lleo@lleo.me> * fix collaborator security query Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com> * Fix svelte warnings Signed-off-by: Artem Savchenko <armisav@gmail.com> * Change log Signed-off-by: Artem Savchenko <armisav@gmail.com> * Add workspace permissions enum Signed-off-by: Artem Savchenko <armisav@gmail.com> * Add changelog Signed-off-by: Artem Savchenko <armisav@gmail.com> * Permission methods Signed-off-by: Artem Savchenko <armisav@gmail.com> * Update core and account versions Signed-off-by: Artem Savchenko <armisav@gmail.com> * Fix predicate Signed-off-by: Artem Savchenko <armisav@gmail.com> * Add changelog Signed-off-by: Artem Savchenko <armisav@gmail.com> * Fix object clone Signed-off-by: Artem Savchenko <armisav@gmail.com> * Bump and add changelog Signed-off-by: Artem Savchenko <armisav@gmail.com> * Fix exception in getTypeOf Signed-off-by: Artem Savchenko <armisav@gmail.com> * Disable changelog check Signed-off-by: Artem Savchenko <armisav@gmail.com> * Fix failed to fetch errors in account client Signed-off-by: Artem Savchenko <armisav@gmail.com> * Clean up Signed-off-by: Artem Savchenko <armisav@gmail.com> * Support in-memory mode for hulypulse Signed-off-by: Artem Savchenko <armisav@gmail.com> --------- Signed-off-by: Leonid Kaganov <lleo@lleo.me> Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com> Signed-off-by: Artem Savchenko <armisav@gmail.com> Co-authored-by: Leonid Kaganov <lleo@lleo.me> Co-authored-by: Alexander Onnikov <Alexander.Onnikov@xored.com>
2026-08-17 18:05:42 +02:00 · 2026-02-21 21:58:08 +07:00
parent a46dd77f9e
commit abe12992e7
11 changed files with 233 additions and 194 deletions
@@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"

 [[package]]
 name = "hulypulse"
-version = "0.4.0"
+version = "0.4.1"
 dependencies = [
 "actix-cors",
 "actix-web",
@@ -1,6 +1,6 @@
 [package]
 name = "hulypulse"
-version = "0.4.0"
+version = "0.4.1"
 edition = "2024"

 [dependencies]
@@ -30,7 +30,7 @@ hulyrs = { git = "https://github.com/hcengineering/hulyrs.git", features = [ "ac
 secrecy = { version = "0.10.3", optional = true }

 #redis
-redis = { version = "=0.32.5", features = ["aio", "tokio-comp", "sentinel"], optional = true }
+redis = { version = "=0.32.5", features = ["aio", "tokio-comp", "sentinel"] }

 [[bin]]
 name = "hulypulse"
@@ -43,7 +43,6 @@ tokio-tungstenite = { version = "0.21", default-features = false, features = [
 ] }

 [features]
-default = ["db-redis","auth"] # lopt
+default = ["auth"] # lopt
 auth = ["regorus", "uuid", "hulyrs", "secrecy"]
 lopt = []
-db-redis = ["redis"]
@@ -190,15 +190,14 @@ Size of data is limited to some reasonable size
    - `{"message":"Del","key":"00000000-0000-0000-0000-000000000001/foo/bar"}`

 ## Special options in config/default.toml
-    - ```memory_mode = true``` Use native memory storage instead Redis
+    - ```backend = "memory"``` Use native memory storage instead Redis
    - ```max_size = 100``` Max value size in bytes

 ## Special cargo build options
-    - "db-redis" (default) - use Redis (Memory instead)
    - "auth" (default) - use huly-authorization
-    Disable both:
+    Disable auth:
        cargo build --no-default-features
-    Enable one:
+    Enable auth:
        cargo build --no-default-features --features "auth"

 ## Running
@@ -207,7 +206,17 @@ Pre-build docker images is available at: hardcoreeng/service_hulypulse:{tag}.

 You can use the following command to run the image locally:
 ```bash
-docker run -p 8095:8095 -it --rm hardcoreeng/service_hulypulse:{tag}"
+docker run -p 8095:8095 -it --rm hardcoreeng/service_hulypulse:{tag}
+```
+
+Run from source using Redis:
+```bash
+HULY_REDIS_URLS=redis://huly.local:6379 cargo run
+```
+
+Run from source in in-memory mode:
+```bash
+HULY_BACKEND=memory cargo run
 ```

 If you want to run the service as a part of local huly development environment use the following command:
@@ -228,6 +237,7 @@ The following environment variables are used to configure hulypulse:
   - ```HULY_BIND_HOST```: host to bind the server to (default: 0.0.0.0)
   - ```HULY_BIND_PORT```: port to bind the server to (default: 8094)
   - ```HULY_TOKEN_SECRET```: secret used to sign JWT tokens (default: secret)
+   - ```HULY_BACKEND```: storage backend "redis" or "memory" (default: "redis")
   - ```HULY_REDIS_URLS```: redis connection string (default: redis://huly.local:6379)
   - ```HULY_REDIS_PASSWORD```: redis password (default: "&lt;invalid&gt;")
   - ```HULY_REDIS_MODE```: redis mode "direct" or "sentinel" (default: "direct")
@@ -19,12 +19,9 @@ use std::{path::Path, sync::LazyLock};
 use secrecy::SecretString;

 use serde::Deserialize;
-#[cfg(feature = "db-redis")]
 use serde_with::StringWithSeparator;
-#[cfg(feature = "db-redis")]
 use serde_with::formats::CommaSeparator;
 use serde_with::serde_as;
-#[cfg(feature = "db-redis")]
 use url::Url;

 use config::FileFormat;
@@ -43,6 +40,10 @@ pub enum BackendType {
    Redis,
 }

+fn default_backend() -> BackendType {
+    BackendType::Redis
+}
+
 #[serde_as]
 #[derive(Deserialize, Debug)]
 pub struct Config {
@@ -52,20 +53,18 @@ pub struct Config {
    #[cfg(feature = "auth")]
    pub token_secret: SecretString,

-    #[cfg(feature = "db-redis")]
+    #[serde(default = "default_backend")]
+    pub backend: BackendType,
+
    #[serde_as(as = "StringWithSeparator::<CommaSeparator, url::Url>")]
    pub redis_urls: Vec<Url>,
-    #[cfg(feature = "db-redis")]
    pub redis_password: String,
-    #[cfg(feature = "db-redis")]
    pub redis_mode: RedisMode,
-    #[cfg(feature = "db-redis")]
    pub redis_service: String,

    pub max_ttl: usize,
    pub max_size: Option<usize>,

-    // pub backend: BackendType,
    pub heartbeat_timeout: u64,
    pub ping_timeout: u64,

@@ -3,6 +3,7 @@ bind_host = "0.0.0.0"

 token_secret = "secret"

+backend = "redis"
 redis_urls = "redis://huly.local:6379"
 redis_password = "<invalid>"
 redis_mode = "direct"
@@ -1,43 +1,38 @@
-#[cfg(not(feature = "db-redis"))]
 use std::sync::Arc;

-#[cfg(not(feature = "db-redis"))]
 use crate::hub_service::{HubState, RedisEvent, RedisEventAction, broadcast_event};
-
-#[cfg(not(feature = "db-redis"))]
 use crate::memory::{
    MemoryBackend, memory_delete, memory_info, memory_list, memory_read, memory_save,
 };
-
-#[cfg(feature = "db-redis")]
 use crate::redis::{redis_delete, redis_info, redis_list, redis_read, redis_save};
+use redis::aio::MultiplexedConnection;
+use serde::Serialize;
+use tokio::sync::RwLock;

-#[cfg(feature = "db-redis")]
-use ::redis::aio::MultiplexedConnection;
-
-#[cfg(feature = "db-redis")]
-pub type DbError = redis::RedisError;
-
-#[cfg(not(feature = "db-redis"))]
 #[derive(Debug)]
-pub struct DbError(pub String);
+pub enum DbError {
+    Redis(redis::RedisError),
+    Message(String),
+}

 pub type DbResult<T> = Result<T, DbError>;

-#[cfg(not(feature = "db-redis"))]
 impl std::fmt::Display for DbError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.0)
+        match self {
+            Self::Redis(err) => write!(f, "{err}"),
+            Self::Message(msg) => write!(f, "{msg}"),
+        }
    }
 }

-#[cfg(not(feature = "db-redis"))]
 impl std::error::Error for DbError {}

-#[cfg(not(feature = "db-redis"))]
-use tokio::sync::RwLock;
-
-use serde::Serialize;
+impl From<redis::RedisError> for DbError {
+    fn from(value: redis::RedisError) -> Self {
+        Self::Redis(value)
+    }
+}

 #[derive(Debug, Serialize)]
 pub struct DbArray {
@@ -61,33 +56,8 @@ pub enum SaveMode {
    Equal(String), // only if md5 matches provided
 }

-/// return Error
-// pub fn error<T>(code: u16, msg: impl Into<String>) -> DbResult<T> {
-//     let msg = msg.into();
-//     let full = format!("{}: {}", code, msg);
-//     Err(redis::RedisError::from((
-//         redis::ErrorKind::ExtensionError,
-//         "",
-//         full,
-//     )))
-// }
-
 pub fn error<T>(code: u16, msg: impl Into<String>) -> DbResult<T> {
-    let msg = format!("{}: {}", code, msg.into());
-
-    #[cfg(feature = "db-redis")]
-    {
-        return Err(redis::RedisError::from((
-            redis::ErrorKind::ExtensionError,
-            "",
-            msg,
-        )));
-    }
-
-    #[cfg(not(feature = "db-redis"))]
-    {
-        return Err(DbError(msg));
-    }
+    Err(DbError::Message(format!("{}: {}", code, msg.into())))
 }

 /// Check for redis-deprecated symbols
@@ -108,59 +78,67 @@ pub fn deprecated_symbol_error(s: &str) -> DbResult<()> {
    }
 }

+#[derive(Clone)]
+enum DbBackend {
+    Redis(MultiplexedConnection),
+    Memory {
+        db: MemoryBackend,
+        hub: Arc<RwLock<HubState>>,
+    },
+}
+
 #[derive(Clone)]
 pub struct Db {
-    #[cfg(feature = "db-redis")]
-    db: MultiplexedConnection,
-    #[cfg(not(feature = "db-redis"))]
-    db: MemoryBackend,
-    #[cfg(not(feature = "db-redis"))]
-    hub: Arc<RwLock<HubState>>,
+    backend: DbBackend,
 }

 impl Db {
-    pub fn new_db(
-        #[cfg(not(feature = "db-redis"))] db: MemoryBackend,
-        #[cfg(feature = "db-redis")] db: MultiplexedConnection,
-        #[cfg(not(feature = "db-redis"))] hub: Arc<RwLock<HubState>>,
-    ) -> Self {
+    pub fn new_redis(db: MultiplexedConnection) -> Self {
        Self {
-            db,
-            #[cfg(not(feature = "db-redis"))]
-            hub,
+            backend: DbBackend::Redis(db),
+        }
+    }
+
+    pub fn new_memory(db: MemoryBackend, hub: Arc<RwLock<HubState>>) -> Self {
+        Self {
+            backend: DbBackend::Memory { db, hub },
+        }
+    }
+
+    pub fn mode(&self) -> &'static str {
+        match &self.backend {
+            DbBackend::Redis(_) => "redis",
+            DbBackend::Memory { .. } => "memory",
        }
    }

    pub async fn info(&self) -> DbResult<String> {
-        #[cfg(not(feature = "db-redis"))]
-        return memory_info(&self.db).await;
-
-        #[cfg(feature = "db-redis")]
-        {
-            let mut c = self.db.clone();
-            redis_info(&mut c).await
+        match &self.backend {
+            DbBackend::Memory { db, .. } => memory_info(db).await,
+            DbBackend::Redis(conn) => {
+                let mut c = conn.clone();
+                redis_info(&mut c).await
+            }
        }
    }

    pub async fn list(&self, key: &str) -> DbResult<Vec<DbArray>> {
-        #[cfg(not(feature = "db-redis"))]
-        return memory_list(&self.db, key).await;
-
-        #[cfg(feature = "db-redis")]
-        {
-            let mut c = self.db.clone();
-            redis_list(&mut c, key).await
+        match &self.backend {
+            DbBackend::Memory { db, .. } => memory_list(db, key).await,
+            DbBackend::Redis(conn) => {
+                let mut c = conn.clone();
+                redis_list(&mut c, key).await
+            }
        }
    }

    pub async fn read(&self, key: &str) -> DbResult<Option<DbArray>> {
-        #[cfg(not(feature = "db-redis"))]
-        return memory_read(&self.db, key).await;
-
-        #[cfg(feature = "db-redis")]
-        {
-            let mut c = self.db.clone();
-            redis_read(&mut c, key).await
+        match &self.backend {
+            DbBackend::Memory { db, .. } => memory_read(db, key).await,
+            DbBackend::Redis(conn) => {
+                let mut c = conn.clone();
+                redis_read(&mut c, key).await
+            }
        }
    }

@@ -171,54 +149,112 @@ impl Db {
        ttl: Option<Ttl>,
        mode: Option<SaveMode>,
    ) -> DbResult<()> {
-        #[cfg(not(feature = "db-redis"))]
-        {
-            memory_save(&self.db, key, value.as_ref(), ttl, mode).await?;
-            // Send events
-            let value_str = std::str::from_utf8(value.as_ref())
-                .ok()
-                .map(|s| s.to_string());
-            broadcast_event(
-                &self.hub,
-                RedisEvent {
-                    message: RedisEventAction::Set,
-                    key: key.to_string(),
-                },
-                value_str,
-            )
-            .await;
-            return Ok(());
-        }
-
-        #[cfg(feature = "db-redis")]
-        {
-            let mut c = self.db.clone();
-            redis_save(&mut c, key, value.as_ref(), ttl, mode).await
+        match &self.backend {
+            DbBackend::Memory { db, hub } => {
+                memory_save(db, key, value.as_ref(), ttl, mode).await?;
+                let value_str = std::str::from_utf8(value.as_ref())
+                    .ok()
+                    .map(|s| s.to_string());
+                broadcast_event(
+                    hub,
+                    RedisEvent {
+                        message: RedisEventAction::Set,
+                        key: key.to_string(),
+                    },
+                    value_str,
+                )
+                .await;
+                Ok(())
+            }
+            DbBackend::Redis(conn) => {
+                let mut c = conn.clone();
+                redis_save(&mut c, key, value.as_ref(), ttl, mode).await
+            }
        }
    }

    pub async fn delete(&self, key: &str, mode: Option<SaveMode>) -> DbResult<bool> {
-        #[cfg(not(feature = "db-redis"))]
-        {
-            let deleted = memory_delete(&self.db, key, mode).await?;
-            if deleted {
-                broadcast_event(
-                    &self.hub,
-                    RedisEvent {
-                        message: RedisEventAction::Del,
-                        key: key.to_string(),
-                    },
-                    None,
-                )
-                .await;
+        match &self.backend {
+            DbBackend::Memory { db, hub } => {
+                let deleted = memory_delete(db, key, mode).await?;
+                if deleted {
+                    broadcast_event(
+                        hub,
+                        RedisEvent {
+                            message: RedisEventAction::Del,
+                            key: key.to_string(),
+                        },
+                        None,
+                    )
+                    .await;
+                }
+                Ok(deleted)
+            }
+            DbBackend::Redis(conn) => {
+                let mut c = conn.clone();
+                redis_delete(&mut c, key, mode).await
            }
-            return Ok(deleted);
-        }
-
-        #[cfg(feature = "db-redis")]
-        {
-            let mut c = self.db.clone();
-            redis_delete(&mut c, key, mode).await
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::hub_service::HubState;
+    use crate::memory::MemoryBackend;
+    use std::sync::Arc;
+    use tokio::sync::RwLock;
+
+    fn memory_db() -> Db {
+        let hub = Arc::new(RwLock::new(HubState::default()));
+        let backend = MemoryBackend::new();
+        Db::new_memory(backend, hub)
+    }
+
+    #[tokio::test]
+    async fn memory_db_mode_and_crud_work() {
+        let db = memory_db();
+        assert_eq!(db.mode(), "memory");
+
+        db.save("workspace/tests/key1", b"hello", Some(Ttl::Sec(60)), None)
+            .await
+            .expect("save should succeed");
+
+        let item = db
+            .read("workspace/tests/key1")
+            .await
+            .expect("read should succeed")
+            .expect("key should exist");
+        assert_eq!(item.data, "hello");
+
+        let list = db
+            .list("workspace/tests/")
+            .await
+            .expect("list should succeed");
+        assert_eq!(list.len(), 1);
+        assert_eq!(list[0].key, "workspace/tests/key1");
+
+        let deleted = db
+            .delete("workspace/tests/key1", None)
+            .await
+            .expect("delete should succeed");
+        assert!(deleted);
+        assert!(
+            db.read("workspace/tests/key1")
+                .await
+                .expect("read should succeed")
+                .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn memory_db_status_reports_memory_backend() {
+        let hub = Arc::new(RwLock::new(HubState::default()));
+        let db = Db::new_memory(MemoryBackend::new(), hub.clone());
+
+        let info = hub.read().await.info_json(&db).await;
+        assert_eq!(info["backend"], "memory");
+        assert_eq!(info["status"], "OK");
+    }
+}
@@ -35,17 +35,19 @@ use crate::workspace_owner::test_rego_http;
 pub fn map_redis_error(err: impl std::fmt::Display) -> Error {
    let msg = err.to_string();

-    if let Some(detail) = msg.split(" - ExtensionError: ").nth(1) {
-        if let Some((code, text)) = detail.split_once(": ") {
-            let text = format!("{} {}", code, text);
-            return match code {
-                "400" => actix_web::error::ErrorBadRequest(text),
-                "404" => actix_web::error::ErrorNotFound(text),
-                "412" => actix_web::error::ErrorPreconditionFailed(text),
-                "500" => actix_web::error::ErrorInternalServerError(text),
-                _ => actix_web::error::ErrorInternalServerError("unexpected error"),
-            };
-        }
+    let detail = msg
+        .split(" - ExtensionError: ")
+        .nth(1)
+        .unwrap_or(msg.as_str());
+    if let Some((code, text)) = detail.split_once(": ") {
+        let text = format!("{} {}", code, text);
+        return match code {
+            "400" => actix_web::error::ErrorBadRequest(text),
+            "404" => actix_web::error::ErrorNotFound(text),
+            "412" => actix_web::error::ErrorPreconditionFailed(text),
+            "500" => actix_web::error::ErrorInternalServerError(text),
+            _ => actix_web::error::ErrorInternalServerError("unexpected error"),
+        };
    }
    actix_web::error::ErrorInternalServerError("internal error")
 }
@@ -23,7 +23,7 @@ use tokio::sync::RwLock;

 use serde_json::{Value, json};

-use crate::{BACKEND, db::Db};
+use crate::db::Db;

 fn subscription_matches(sub_key: &str, key: &str) -> bool {
    if sub_key == key {
@@ -54,10 +54,8 @@ pub fn new_session_id() -> SessionId {
 pub enum RedisEventAction {
    Set,
    Del,
-    #[cfg(feature = "db-redis")]
    Unlink,
    Expired,
-    #[cfg(feature = "db-redis")]
    Other(String),
 }

@@ -171,7 +169,7 @@ impl HubState {
        let info = db.info().await.unwrap_or_else(|_| "error".to_string());
        json!({
            "memory_info": info,
-            "backend": BACKEND,
+            "backend": db.mode(),
            "websockets": self.sessions.len(),
            "subscriptions": self.subs.len(),
            "heartbeats": self.heartbeats.len(),
@@ -19,6 +19,8 @@ use actix_web::{
    middleware::{self},
    web::{self},
 };
+use std::sync::Arc;
+use tokio::sync::RwLock;

 #[cfg(feature = "auth")]
 use actix_web::{
@@ -45,7 +47,7 @@ mod config;
 mod handlers_http;
 mod handlers_ws;

-#[cfg(feature = "db-redis")]
+mod memory;
 mod redis;

 #[cfg(feature = "auth")]
@@ -57,20 +59,12 @@ use hub_service::HubState;
 use config::CONFIG;

 mod db;
+use crate::config::BackendType;
 use crate::db::Db;
-
-#[cfg(not(feature = "db-redis"))]
-mod memory;
-#[cfg(not(feature = "db-redis"))]
 use crate::memory::MemoryBackend;

 use crate::hub_service::check_heartbeat;

-#[cfg(feature = "db-redis")]
-pub const BACKEND: &str = "REDIS";
-#[cfg(not(feature = "db-redis"))]
-pub const BACKEND: &str = "MEMORY";
-
 fn initialize_tracing() {
    use tracing_subscriber::{filter::targets::Targets, prelude::*};

@@ -147,9 +141,8 @@ async fn main() -> anyhow::Result<()> {
    // starting heartbeat checker
    check_heartbeat(hub_state.clone());

-    let db_backend = {
-        #[cfg(feature = "db-redis")]
-        {
+    let db_backend = match &CONFIG.backend {
+        BackendType::Redis => {
            let redis_client = redis::client().await?;
            let db_connection = redis_client
                .get_multiplexed_async_connection()
@@ -166,19 +159,24 @@ async fn main() -> anyhow::Result<()> {
                    );
                    e
                })?;
-            tokio::spawn(crate::redis::receiver(redis_client, hub_state.clone()));
-            Db::new_db(db_connection)
+            tokio::spawn({
+                let hub_state = hub_state.clone();
+                async move {
+                    if let Err(err) = crate::redis::receiver(redis_client, hub_state).await {
+                        tracing::error!("Redis receiver stopped: {err}");
+                    }
+                }
+            });
+            Db::new_redis(db_connection)
        }
-
-        #[cfg(not(feature = "db-redis"))]
-        {
+        BackendType::Memory => {
            let db_connection = MemoryBackend::new();
            db_connection.spawn_ticker(hub_state.clone());
-            Db::new_db(db_connection, hub_state.clone())
+            Db::new_memory(db_connection, hub_state.clone())
        }
    };

-    tracing::info!("DB mode: {}", BACKEND);
+    tracing::info!("DB mode: {}", db_backend.mode());

    let socket = std::net::SocketAddr::new(CONFIG.bind_host.as_str().parse()?, CONFIG.bind_port);

@@ -192,9 +190,6 @@ async fn main() -> anyhow::Result<()> {
    );
    tracing::info!("Status: {}/status", &url);

-    use std::sync::Arc;
-    use tokio::sync::RwLock;
-
    let server = HttpServer::new(move || {
        let cors = Cors::default()
            .allow_any_origin()
@@ -30,7 +30,7 @@ use crate::{
 };

 use redis::{
-    Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, RedisResult, ToRedisArgs,
+    Client, ConnectionInfo, ProtocolVersion, RedisConnectionInfo, ToRedisArgs,
    aio::MultiplexedConnection,
 };
 // use serde::Serialize;
@@ -61,7 +61,7 @@ pub async fn push_event(
 }

 /// redis_info(&connection)
-pub async fn redis_info(conn: &mut MultiplexedConnection) -> redis::RedisResult<String> {
+pub async fn redis_info(conn: &mut MultiplexedConnection) -> DbResult<String> {
    let info: String = redis::cmd("INFO").query_async(conn).await?;

    let mut redis_keys: Option<usize> = None;
@@ -91,10 +91,7 @@ pub async fn redis_info(conn: &mut MultiplexedConnection) -> redis::RedisResult<
 }

 /// redis_list(&connection,prefix)
-pub async fn redis_list(
-    conn: &mut MultiplexedConnection,
-    key: &str,
-) -> redis::RedisResult<Vec<DbArray>> {
+pub async fn redis_list(conn: &mut MultiplexedConnection, key: &str) -> DbResult<Vec<DbArray>> {
    deprecated_symbol_error(key)?;
    if !key.ends_with('/') {
        return error(412, "Key must end with slash");
@@ -146,10 +143,7 @@ pub async fn redis_list(
 }

 /// redis_read(&connection,key)
-pub async fn redis_read(
-    conn: &mut MultiplexedConnection,
-    key: &str,
-) -> redis::RedisResult<Option<DbArray>> {
+pub async fn redis_read(conn: &mut MultiplexedConnection, key: &str) -> DbResult<Option<DbArray>> {
    deprecated_symbol_error(key)?;

    if key.ends_with('/') {
@@ -306,7 +300,7 @@ pub async fn redis_delete(
    conn: &mut MultiplexedConnection,
    key: &str,
    mode: Option<SaveMode>,
-) -> RedisResult<bool> {
+) -> DbResult<bool> {
    deprecated_symbol_error(key)?;

    if key.ends_with('/') {
@@ -433,7 +427,7 @@ pub async fn receiver(
    while let Some(message) = messages.next().await {
        match RedisEvent::try_from(message) {
            Ok(ev) => {
-                push_event(&hub_state, &mut redis, ev); // .await;
+                push_event(&hub_state, &mut redis, ev).await;
            }
            Err(e) => {
                warn!("invalid redis message: {e}");
@@ -31,7 +31,12 @@ async fn status(base: &str, client: &reqwest::Client) -> () {
    let text = resp.text().await.unwrap();
    let json: Value = serde_json::from_str(&text).unwrap();

-    assert_eq!(json["backend"], "memory");
+    let backend = json["backend"].as_str().unwrap_or_default();
+    if let Ok(expected_backend) = env::var("TEST_BACKEND") {
+        assert_eq!(backend, expected_backend);
+    } else {
+        assert!(backend == "memory" || backend == "redis");
+    }
    assert_eq!(json["status"], "OK");
    assert!(json.get("memory_info").is_some());
    assert!(json.get("websockets").is_some());