From abe12992e7987f90507f4cb4c04e6fdc0df0fa47 Mon Sep 17 00:00:00 2001 From: Artyom Savchenko Date: Sat, 21 Feb 2026 21:58:08 +0700 Subject: [PATCH] Take upstream (#10538) * fix: redesign and compact Signed-off-by: Leonid Kaganov * fix: comments Signed-off-by: Leonid Kaganov * fix: correctly dropping connections in timeout Signed-off-by: Leonid Kaganov * feature: loglevel in config Signed-off-by: Leonid Kaganov * fix: loglevel in config Signed-off-by: Leonid Kaganov * features lopt: direct personal messages between websockets by username Signed-off-by: Leonid Kaganov * fix collaborator security query Signed-off-by: Alexander Onnikov * Fix svelte warnings Signed-off-by: Artem Savchenko * Change log Signed-off-by: Artem Savchenko * Add workspace permissions enum Signed-off-by: Artem Savchenko * Add changelog Signed-off-by: Artem Savchenko * Permission methods Signed-off-by: Artem Savchenko * Update core and account versions Signed-off-by: Artem Savchenko * Fix predicate Signed-off-by: Artem Savchenko * Add changelog Signed-off-by: Artem Savchenko * Fix object clone Signed-off-by: Artem Savchenko * Bump and add changelog Signed-off-by: Artem Savchenko * Fix exception in getTypeOf Signed-off-by: Artem Savchenko * Disable changelog check Signed-off-by: Artem Savchenko * Fix failed to fetch errors in account client Signed-off-by: Artem Savchenko * Clean up Signed-off-by: Artem Savchenko * Support in-memory mode for hulypulse Signed-off-by: Artem Savchenko --------- Signed-off-by: Leonid Kaganov Signed-off-by: Alexander Onnikov Signed-off-by: Artem Savchenko Co-authored-by: Leonid Kaganov Co-authored-by: Alexander Onnikov --- foundations/hulypulse/Cargo.lock | 2 +- foundations/hulypulse/Cargo.toml | 7 +- foundations/hulypulse/README.md | 20 +- foundations/hulypulse/src/config.rs | 15 +- foundations/hulypulse/src/config/default.toml | 1 + foundations/hulypulse/src/db.rs | 286 ++++++++++-------- foundations/hulypulse/src/handlers_http.rs | 24 +- foundations/hulypulse/src/hub_service.rs | 6 +- foundations/hulypulse/src/main.rs | 41 ++- foundations/hulypulse/src/redis.rs | 18 +- foundations/hulypulse/tests/rest_api.rs | 7 +- 11 files changed, 233 insertions(+), 194 deletions(-) diff --git a/foundations/hulypulse/Cargo.lock b/foundations/hulypulse/Cargo.lock index 1668acc159..52644883b8 100644 --- a/foundations/hulypulse/Cargo.lock +++ b/foundations/hulypulse/Cargo.lock @@ -1175,7 +1175,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hulypulse" -version = "0.4.0" +version = "0.4.1" dependencies = [ "actix-cors", "actix-web", diff --git a/foundations/hulypulse/Cargo.toml b/foundations/hulypulse/Cargo.toml index 0cdcc1233a..50f1dafe32 100644 --- a/foundations/hulypulse/Cargo.toml +++ b/foundations/hulypulse/Cargo.toml @@ -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"] \ No newline at end of file diff --git a/foundations/hulypulse/README.md b/foundations/hulypulse/README.md index 90940eb91b..8e457bd91c 100644 --- a/foundations/hulypulse/README.md +++ b/foundations/hulypulse/README.md @@ -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: "<invalid>") - ```HULY_REDIS_MODE```: redis mode "direct" or "sentinel" (default: "direct") diff --git a/foundations/hulypulse/src/config.rs b/foundations/hulypulse/src/config.rs index 50f555c117..e5e7f07440 100644 --- a/foundations/hulypulse/src/config.rs +++ b/foundations/hulypulse/src/config.rs @@ -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::")] pub redis_urls: Vec, - #[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, - // pub backend: BackendType, pub heartbeat_timeout: u64, pub ping_timeout: u64, diff --git a/foundations/hulypulse/src/config/default.toml b/foundations/hulypulse/src/config/default.toml index 46779e7197..06551e18ea 100644 --- a/foundations/hulypulse/src/config/default.toml +++ b/foundations/hulypulse/src/config/default.toml @@ -3,6 +3,7 @@ bind_host = "0.0.0.0" token_secret = "secret" +backend = "redis" redis_urls = "redis://huly.local:6379" redis_password = "" redis_mode = "direct" diff --git a/foundations/hulypulse/src/db.rs b/foundations/hulypulse/src/db.rs index b33836d7b2..006728d6fd 100644 --- a/foundations/hulypulse/src/db.rs +++ b/foundations/hulypulse/src/db.rs @@ -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 = Result; -#[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 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(code: u16, msg: impl Into) -> DbResult { -// let msg = msg.into(); -// let full = format!("{}: {}", code, msg); -// Err(redis::RedisError::from(( -// redis::ErrorKind::ExtensionError, -// "", -// full, -// ))) -// } - pub fn error(code: u16, msg: impl Into) -> DbResult { - 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>, + }, +} + #[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>, + 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>, - ) -> 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>) -> 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 { - #[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> { - #[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> { - #[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, mode: Option, ) -> 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) -> DbResult { - #[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"); + } +} diff --git a/foundations/hulypulse/src/handlers_http.rs b/foundations/hulypulse/src/handlers_http.rs index 5efabece51..ee6a8a9368 100644 --- a/foundations/hulypulse/src/handlers_http.rs +++ b/foundations/hulypulse/src/handlers_http.rs @@ -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") } diff --git a/foundations/hulypulse/src/hub_service.rs b/foundations/hulypulse/src/hub_service.rs index 137bdf5013..0327a883ab 100644 --- a/foundations/hulypulse/src/hub_service.rs +++ b/foundations/hulypulse/src/hub_service.rs @@ -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(), diff --git a/foundations/hulypulse/src/main.rs b/foundations/hulypulse/src/main.rs index 7221b2c9f9..083d9365a9 100644 --- a/foundations/hulypulse/src/main.rs +++ b/foundations/hulypulse/src/main.rs @@ -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() diff --git a/foundations/hulypulse/src/redis.rs b/foundations/hulypulse/src/redis.rs index 37ea4bcdbd..4cb83fcac8 100644 --- a/foundations/hulypulse/src/redis.rs +++ b/foundations/hulypulse/src/redis.rs @@ -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 { +pub async fn redis_info(conn: &mut MultiplexedConnection) -> DbResult { let info: String = redis::cmd("INFO").query_async(conn).await?; let mut redis_keys: Option = 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> { +pub async fn redis_list(conn: &mut MultiplexedConnection, key: &str) -> DbResult> { 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> { +pub async fn redis_read(conn: &mut MultiplexedConnection, key: &str) -> DbResult> { deprecated_symbol_error(key)?; if key.ends_with('/') { @@ -306,7 +300,7 @@ pub async fn redis_delete( conn: &mut MultiplexedConnection, key: &str, mode: Option, -) -> RedisResult { +) -> DbResult { 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}"); diff --git a/foundations/hulypulse/tests/rest_api.rs b/foundations/hulypulse/tests/rest_api.rs index 6a9f86fa70..a072d8deac 100644 --- a/foundations/hulypulse/tests/rest_api.rs +++ b/foundations/hulypulse/tests/rest_api.rs @@ -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());