prevent tokio runtime queue overload under high load

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<u32, Error> {
         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<Option<Value>, 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,15 +183,15 @@ 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<Vec<String>, Error> {
         let Some(store) = self.stores.get(store as usize) else {
             return Err(Error::NoSuchStore);
         };
         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<bool, Error> {
         let Some(store) = self.stores.get(store as usize) else {
             return Err(Error::NoSuchStore);

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 {
@@ -16,13 +36,15 @@ impl RedisStore {
     /// follow-up calls land on the freshly re-established connection.
     pub async fn open(params: &str) -> Result<Self, Error> {
         let client = ::redis::Client::open(params).map_err(|error| {
-            tracing::warn!(error=?error, "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
         })?;
+        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 })
     }
 }
@@ -31,7 +53,7 @@ impl RedisStore {
 impl Store for RedisStore {
     async fn get(&self, key: &str) -> Result<Option<Value>, 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,20 +69,23 @@ 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
             })
     }
 
     async fn scan(&self, pattern: &str) -> Result<Vec<String>, 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 +94,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 +114,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
             })
     }

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<T> = wasmtime::component::InstancePre<Data<T>>;

crates/runtime/src/util/metrics.rs

@@ -46,6 +46,9 @@ pub fn metrics(result: AppResult, label: &[&str], duration: Option<u64>, 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"),
             _ => {}
         };