From aa606d839b7e8da2dcfb78da5557f76d07efad83 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Fri, 10 Jul 2026 12:37:51 +0300 Subject: [PATCH 1/2] refactor: improve Redis error tracing with detailed context and adjust instrumentation annotations --- crates/key-value-store/src/lib.rs | 12 ++++++------ crates/key-value-store/src/redis_impl.rs | 24 +++++++++++++++--------- 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/crates/key-value-store/src/lib.rs b/crates/key-value-store/src/lib.rs index a7a75be..05d4336 100644 --- a/crates/key-value-store/src/lib.rs +++ b/crates/key-value-store/src/lib.rs @@ -145,7 +145,7 @@ impl key_value::Host for StoreImpl {} impl StoreImpl { /// Open a store by name. Return the store ID. - #[instrument(skip(self), level = "trace", ret, err)] + #[instrument(skip(self), level = "trace", ret)] pub async fn open(&mut self, name: &str) -> Result { if let Some(param) = self .allowed_stores @@ -160,7 +160,7 @@ impl StoreImpl { } /// Get a value from a store by key. - #[instrument(skip(self), level = "trace", ret, err)] + #[instrument(skip(self), level = "trace", ret)] pub async fn get(&self, store: u32, key: &str) -> Result, Error> { let Some(store) = self.stores.get(store as usize) else { return Err(Error::NoSuchStore); @@ -169,7 +169,7 @@ impl StoreImpl { } /// Get a values from a store by key. - #[instrument(skip(self), level = "trace", ret, err)] + #[instrument(skip(self), level = "trace", ret)] pub async fn zrange_by_score( &self, store: u32, @@ -183,7 +183,7 @@ impl StoreImpl { store.zrange_by_score(key, min, max).await } - #[instrument(skip(self), level = "trace", ret, err)] + #[instrument(skip(self), level = "trace", ret)] pub async fn scan(&mut self, store: u32, pattern: &str) -> Result, Error> { let Some(store) = self.stores.get(store as usize) else { return Err(Error::NoSuchStore); @@ -191,7 +191,7 @@ impl StoreImpl { store.scan(pattern).await } - #[instrument(skip(self), level = "trace", ret, err)] + #[instrument(skip(self), level = "trace", ret)] pub async fn zscan( &mut self, store: u32, @@ -205,7 +205,7 @@ impl StoreImpl { } /// Get a value from a store by key. - #[instrument(skip(self), level = "trace", ret, err)] + #[instrument(skip(self), level = "trace", ret)] pub async fn bf_exists(&self, store: u32, key: &str, item: &str) -> Result { let Some(store) = self.stores.get(store as usize) else { return Err(Error::NoSuchStore); diff --git a/crates/key-value-store/src/redis_impl.rs b/crates/key-value-store/src/redis_impl.rs index c0800eb..d20ad4e 100644 --- a/crates/key-value-store/src/redis_impl.rs +++ b/crates/key-value-store/src/redis_impl.rs @@ -16,11 +16,11 @@ impl RedisStore { /// follow-up calls land on the freshly re-established connection. pub async fn open(params: &str) -> Result { let client = ::redis::Client::open(params).map_err(|error| { - tracing::warn!(error=?error, "redis open"); + tracing::warn!(error = ?error, "kv-store: redis open"); Error::InternalError })?; let conn = ConnectionManager::new(client).await.map_err(|error| { - tracing::warn!(error=?error, "redis open"); + tracing::warn!(error = ?error, "kv-store: redis open"); Error::InternalError })?; Ok(Self { inner: conn }) @@ -31,7 +31,7 @@ impl RedisStore { impl Store for RedisStore { async fn get(&self, key: &str) -> Result, Error> { self.inner.clone().get(key).await.map_err(|error| { - tracing::warn!(cause=?error, "redis get"); + tracing::warn!(cause = ?error, key, "kv-store: redis get"); Error::InternalError }) } @@ -47,7 +47,7 @@ impl Store for RedisStore { .zrangebyscore_withscores(key, min, max) .await .map_err(|error| { - tracing::warn!(cause=?error, "redis zrangebyscore"); + tracing::warn!(cause = ?error, key, min, max, "kv-store: redis zrangebyscore"); Error::InternalError }) } @@ -55,12 +55,15 @@ impl Store for RedisStore { async fn scan(&self, pattern: &str) -> Result, Error> { let mut conn = self.inner.clone(); let mut it = conn.scan_match(pattern).await.map_err(|error| { - tracing::warn!(cause=?error, "redis scan_match"); + tracing::warn!(cause = ?error, pattern, "kv-store: redis scan_match"); Error::InternalError })?; let mut ret = vec![]; while let Some(element) = it.next_item().await { - ret.push(element.map_err(|error| Error::Other(error.to_string()))?); + ret.push(element.map_err(|error| { + tracing::warn!(cause = ?error, pattern, "kv-store: redis scan_match: item"); + Error::Other(error.to_string()) + })?); } Ok(ret) } @@ -69,12 +72,15 @@ impl Store for RedisStore { let mut conn = self.inner.clone(); let mut it: AsyncIter<(Value, f64)> = conn.zscan_match(key, pattern).await.map_err(|error| { - tracing::warn!(cause=?error, "redis zscan_match"); + tracing::warn!(cause = ?error, key, pattern, "kv-store: redis zscan_match"); Error::InternalError })?; let mut ret = vec![]; while let Some(element) = it.next_item().await { - ret.push(element.map_err(|error| Error::Other(error.to_string()))?); + ret.push(element.map_err(|error| { + tracing::warn!(cause = ?error, key, pattern, "kv-store: redis zscan_match: item"); + Error::Other(error.to_string()) + })?); } Ok(ret) } @@ -86,7 +92,7 @@ impl Store for RedisStore { .query_async(&mut self.inner.clone()) .await .map_err(|error| { - tracing::warn!(cause=?error, "redis bf_exists"); + tracing::warn!(cause = ?error, key, item, "kv-store: redis bf_exists"); Error::InternalError }) } From 6bd5c455b9e0b73dc364b9efc72558c4fe523657 Mon Sep 17 00:00:00 2001 From: Ruslan Pislari Date: Fri, 10 Jul 2026 15:34:12 +0300 Subject: [PATCH 2/2] feat: add fail-fast Redis connection management with configurable timeouts and extend error metrics --- crates/key-value-store/src/redis_impl.rs | 32 ++++++++++++++++++++---- crates/runtime/src/lib.rs | 9 +++++++ crates/runtime/src/util/metrics.rs | 3 +++ 3 files changed, 39 insertions(+), 5 deletions(-) diff --git a/crates/key-value-store/src/redis_impl.rs b/crates/key-value-store/src/redis_impl.rs index d20ad4e..b88ca7a 100644 --- a/crates/key-value-store/src/redis_impl.rs +++ b/crates/key-value-store/src/redis_impl.rs @@ -1,7 +1,27 @@ use crate::Store; use reactor::gcore::fastedge::key_value::{Error, Value}; -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::{AsyncCommands, AsyncIter}; +use std::time::Duration; + +/// Fail-fast timeouts for the KV-store Redis connection. Redis sits on the +/// request hot path, so a slow/unreachable Redis must surface as a quick error +/// rather than parking the calling task: a stalled dependency with no bound +/// lets in-flight requests pile up until the node collapses. Set explicitly so +/// a redis-crate bump can't silently change them. +const REDIS_RESPONSE_TIMEOUT: Duration = Duration::from_millis(100); +const REDIS_CONNECTION_TIMEOUT: Duration = Duration::from_millis(100); +const REDIS_NUMBER_OF_RETRIES: usize = 2; +const REDIS_MAX_RECONNECT_DELAY: Duration = Duration::from_millis(100); + +/// Build the fail-fast connection-manager config for KV-store Redis connections. +fn connection_manager_config() -> ConnectionManagerConfig { + ConnectionManagerConfig::new() + .set_response_timeout(Some(REDIS_RESPONSE_TIMEOUT)) + .set_connection_timeout(Some(REDIS_CONNECTION_TIMEOUT)) + .set_number_of_retries(REDIS_NUMBER_OF_RETRIES) + .set_max_delay(REDIS_MAX_RECONNECT_DELAY) +} #[derive(Clone)] pub struct RedisStore { @@ -19,10 +39,12 @@ impl RedisStore { tracing::warn!(error = ?error, "kv-store: redis open"); Error::InternalError })?; - let conn = ConnectionManager::new(client).await.map_err(|error| { - tracing::warn!(error = ?error, "kv-store: redis open"); - Error::InternalError - })?; + let conn = ConnectionManager::new_with_config(client, connection_manager_config()) + .await + .map_err(|error| { + tracing::warn!(error = ?error, "kv-store: redis open"); + Error::InternalError + })?; Ok(Self { inner: conn }) } } diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 3428d29..462d37b 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -60,6 +60,15 @@ pub enum AppResult { TIMEOUT, OOM, OTHER, + /// Request shed by admission control (server overloaded). + #[cfg(feature = "metrics")] + OVERLOADED, + /// Request rejected because the app is rate limited. + #[cfg(feature = "metrics")] + RATE_LIMITED, + /// Request rejected because the app is disabled or in draft. + #[cfg(feature = "metrics")] + DISABLED, } pub type InstancePre = wasmtime::component::InstancePre>; diff --git a/crates/runtime/src/util/metrics.rs b/crates/runtime/src/util/metrics.rs index 79c2956..eefa490 100644 --- a/crates/runtime/src/util/metrics.rs +++ b/crates/runtime/src/util/metrics.rs @@ -46,6 +46,9 @@ pub fn metrics(result: AppResult, label: &[&str], duration: Option, memory_ AppResult::TIMEOUT => values.push("timeout"), AppResult::OOM => values.push("oom"), AppResult::OTHER => values.push("other"), + AppResult::OVERLOADED => values.push("overloaded"), + AppResult::RATE_LIMITED => values.push("ratelimited"), + AppResult::DISABLED => values.push("disabled"), _ => {} };