fix(redis-client): match redis-py command and failure behavior

litellm-rust/Cargo.lock

@@ -4418,6 +4418,7 @@ dependencies = [
  "litellm-secret-value",
  "redis",
  "rstest",
+ "socket2 0.6.5",
  "thiserror 2.0.19",
  "tokio",
 ]

litellm-rust/crates/gateway/src/redis.rs

@@ -7,7 +7,7 @@ use litellm_config::{
 };
 use litellm_redis_client::{
     ClusterEndpoint, ClusterNode, DEFAULT_TIMEOUT, Endpoint, HostEndpoint, Namespace, RedisClient,
-    RedisConfig,
+    RedisConfig, Timeouts,
 };
 
 use crate::Error;
@@ -178,8 +178,9 @@ fn client(params: Params, breaker: Option<BreakerConfig>) -> Result<RedisClient,
     Ok(RedisClient::new(RedisConfig {
         endpoint,
         namespace: params.namespace.map(Namespace::new).unwrap_or_default(),
-        timeout: params.timeout.unwrap_or(DEFAULT_TIMEOUT),
+        timeouts: Timeouts::new(params.timeout.unwrap_or(DEFAULT_TIMEOUT), None),
         breaker: breaker.map(|config| Breaker::new(config, Box::new(log_transition))),
+        events: None,
     })?)
 }
 

litellm-rust/crates/redis-client/AGENTS.md

@@ -1,12 +1,12 @@
-- The one async Redis client for the gateway and the Rust cache crates. It owns connecting, timeouts, error classes and the key namespace
+- The one async Redis client for the gateway and the Rust cache crates. `cache-redis` (the response cache behind `python-bridge/src/cache/native`) still opens its own r2d2 pool from a URL, without TLS options, cloud auth, Sentinel or cluster settings, until that keyspace ports onto this client. It owns connecting, timeouts, error classes and the key namespace
 - It knows nothing about domains, so key builders, TTL policy, value formats and fallback decisions stay in the crate that owns the state. Every write takes its TTL from the caller, and there is no default
 - Async only. Build one `RedisClient` per Redis instance at the process boundary and clone it. Never add a pool, a sync API, `spawn_blocking`, or a client built inside a route
-- Set every redis-rs timeout and retry explicitly. Its defaults (500ms response, 1s connect, 6 reconnect retries) are not the Python proxy's (5s socket timeout from `REDIS_SOCKET_TIMEOUT`)
+- Every connection sets TCP_NODELAY, like redis-py's `socket.setsockopt(IPPROTO_TCP, TCP_NODELAY, 1)`. Set every redis-rs timeout and retry explicitly, citing the redis-py 5.3.1 or `litellm/` source next to each constant. Its defaults (500ms response, 1s connect, 6 reconnect retries) are not the Python proxy's. `Timeouts` takes `socket_timeout` and `socket_connect_timeout`, and the connect timeout falls back to the socket timeout like redis-py. Standalone reconnects on the next command with no retry of the failed one, matching redis-py's `Retry(NoBackoff(), 0)`. Cluster keeps redis-py's 3 `cluster_error_retry_attempts` and waits a flat 250ms between them, which is redis-py's CLUSTERDOWN sleep. redis-py's jittered per-node reconnect backoff has no redis-rs equivalent. The retry set also differs, redis-rs retries LOADING and READONLY internally, while redis-py 5.3.1 raises them. Do not claim exact retry parity
 - Resolving config from `config.yaml` / `REDIS_*` env the way `litellm/_redis.py` does belongs to the consumer that reads that config, not here. `RedisConfig` takes already-resolved values
 
 ## Errors
 - `Transport`, `Server`, `Decode`, each wrapping the redis-rs error as `#[source]`. `ConnectError` is separate because only `connect` returns it
-- Classify in `From<RedisError> for Error` by `ErrorKind`. LOADING, CLUSTERDOWN, MASTERDOWN, TRYAGAIN and READONLY are typed server replies, but they mean the node can't serve right now, so they are `Transport`. Codes redis-rs doesn't type (WRONGTYPE) come back as `ErrorKind::Extension` and are `Server`. Each code gets its own rstest case built from a parsed RESP error
+- Classify with `Error::classify(error, deployment)`, never a bare `From<RedisError>`, because redis-py (5.3.1) decides by topology. `Transport` is what Python's `_redis_health_error_types` counts: LOADING everywhere, CLUSTERDOWN and MASTERDOWN only on cluster (only `ClusterParser` maps them), plus NOAUTH, WRONGPASS and the `ERR` messages redis-py maps to `ConnectionError`, matched exactly like `parse_error`. TRYAGAIN is `Server`, since the cluster client already retried it. Each code gets its own rstest case built from a parsed RESP error
 - A `Server` error never drops the connection
 - `Transport` includes timeouts, where the command may still have applied. Callers of counters and locks treat it as an unknown outcome
 - Never swallow an error or turn it into a miss. The caller picks the fallback
@@ -17,7 +17,6 @@
 - Every public call (each command, `eval`, one `run` of a pipeline) takes one admission through `guarded` / `admit` in `client.rs`. Never call the connection around it
 - `Transport` counts as a failure (`Timeout` when `is_timeout()`, else `Hard`). `Server` and `Decode` count as nothing, since Redis answered and the request was bad. A probe that gets one still reopens, through the dropped-probe rule in `circuit-breaker`. A pipeline counts one failure if any op came back `Transport`, else one success
 - An open breaker returns `Error::BreakerOpen` without touching the socket
-- Python doesn't count MASTERDOWN, TRYAGAIN or READONLY as health failures. Here they are `Transport` and do count
 
 ## Namespace
 - Applied once, here, to every key. A key that already starts with `{namespace}:` is left alone, matching `RedisCache.check_and_fix_namespace`
@@ -27,17 +26,22 @@
 - One typed method per command, built with `redis::cmd`. Empty batches (MGET, DEL) return without a round trip, since Redis rejects an empty MGET
 - On cluster, redis-rs splits MGET and DEL by slot and returns MGET values in input order. Other multi-key commands are not split, so a new one needs a cross-slot test
 - TTLs are `Duration` and are sent in milliseconds (`PX`, `PEXPIRE`, `PTTL`)
+- `incr_by`, `incr_by_float` and `rpush` take an `Expiry` and apply it in the same script as the write, so a key never outlives a lost follow-up `EXPIRE`. `KeepExisting` sets the TTL only on a key without one, like Python's `_incrbyfloat_with_ttl`, and `Refresh` always resets it
+- Atomic writes require scripting permission (`EVAL`, `EVALSHA`, `SCRIPT LOAD`) and permission for the underlying write, `PTTL`, `PEXPIRE` and `GET` for integer increments. There is no non-atomic fallback for restricted ACLs. Ordinary reads and SET do not require scripting
+- Integer increments read their exact decimal result inside the script before expiry, without converting through Lua's double. The public method and pipeline return an exact integer
 
 ## Scripts and pipelines
 - A `Script` declares every key it touches in KEYS. On cluster an undeclared key works or fails depending on which node owns its slot, so it can pass tests and break in production
-- `eval` goes through redis-rs `invoke_async`, which loads the script on NOSCRIPT. Inside a pipeline nothing reloads, so a NOSCRIPT op reruns alone with `invoke_async`
+- `eval` goes through redis-rs `invoke_async`, which loads the script on NOSCRIPT. Inside a pipeline a script is sent as `EVAL` with its source, so it can never miss the script cache and run out of order with the ops around it
 - A pipeline is never atomic. Every op gets its own `Result`, from `.ignore_errors()`, so one rejected op leaves the others alone. A transport failure fails every op of that round trip
-- redis-rs rejects an async cluster pipeline whose ops span slots. `Pipeline` groups ops by the slot of their first key, runs the groups concurrently and returns results in input order
+- redis-rs rejects an async cluster pipeline whose ops span slots. `Pipeline` groups ops by the slot of their first key and runs the slots concurrently, returning results in input order. Within a slot, runs of idempotent ops (GET, SET, DEL, EXPIRE) share a request and every script op gets its own, sent in order. redis-rs retries a cluster request whole, so a script op batched with others would be applied twice when only a neighbour failed
 - Pipeline ops take one key each. A multi-key op would span slots inside one group
 - Command builders live in `commands.rs` and are shared by `RedisClient` and `Pipeline`, so a command is written once
 
 ## Tests
 - Against real Redis from `litellm-redis-testing`, never a mocked wire or asserted command bytes
+- Tests that only need the public API live in `tests/<module>.rs`, with client builders in `tests/support`. Inline `#[cfg(test)]` modules are only for private items (`error.rs`, `namespace.rs`)
+- Observe connections from the server side (`RedisDeployment::clients`, `cli_on_every_node`), never by reaching into `RedisClient`'s connection
 - Every command test runs on each topology through `#[values(Topology::Standalone, Topology::Cluster)]`. A test is single-topology only when the behavior exists on one (a `db` number has no cluster meaning)
 - Assert values and Redis state. Outages come from the container (`restart`, `freeze` / `thaw`), not `DEBUG`, which the official image disables
 - Use a short client timeout in outage tests so they stay fast. Never sleep to wait for a state; poll with a deadline

litellm-rust/crates/redis-client/Cargo.toml

@@ -10,9 +10,10 @@ litellm-circuit-breaker.workspace = true
 litellm-secret-value.workspace = true
 futures-util.workspace = true
 redis = { workspace = true, features = ["cluster-async", "connection-manager", "tokio-rustls-comp"] }
+socket2 = "0.6.5"
 thiserror.workspace = true
+tokio = { workspace = true, features = ["sync", "time"] }
 
 [dev-dependencies]
 litellm-redis-testing.workspace = true
 rstest.workspace = true
-tokio.workspace = true

litellm-rust/crates/redis-client/src/client.rs

@@ -1,13 +1,18 @@
 use std::{
     future::Future,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
 use litellm_circuit_breaker::{Admission, Breaker, Failure};
-use redis::{Cmd, FromRedisValue, IntoConnectionInfo, ToRedisArgs, ToSingleRedisArg, Value};
+use redis::{
+    Cmd, FromRedisValue, IntoConnectionInfo, RedisError, ToRedisArgs, ToSingleRedisArg, Value,
+};
 
 use crate::{
-    ConnectError, Endpoint, Error, Namespace, Pipeline, RedisConfig, Script, commands,
+    CallEvent, ConnectError, Endpoint, Error, Expiry, FailureKind, Namespace, Operation, Outcome,
+    Pipeline, RedisConfig, RedisEvents, Script,
+    commands::{self, ScriptCall},
     connection::Connection,
 };
 
@@ -16,6 +21,7 @@ pub struct RedisClient {
     pub(crate) connection: Connection,
     namespace: Namespace,
     breaker: Option<Breaker>,
+    events: Option<Arc<dyn RedisEvents>>,
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -29,33 +35,34 @@ impl RedisClient {
     pub fn new(config: RedisConfig) -> Result<Self, ConnectError> {
         let connection = match &config.endpoint {
             Endpoint::Url(url) => {
-                Connection::standalone(url.expose().into_connection_info()?, config.timeout)?
+                Connection::standalone(url.expose().into_connection_info()?, config.timeouts)?
             }
             Endpoint::Host(host) => {
-                Connection::standalone(host.connection_info()?, config.timeout)?
+                Connection::standalone(host.connection_info()?, config.timeouts)?
             }
-            Endpoint::Cluster(cluster) => Connection::cluster(cluster, config.timeout)?,
+            Endpoint::Cluster(cluster) => Connection::cluster(cluster, config.timeouts)?,
         };
         Ok(Self {
             connection,
             namespace: config.namespace,
             breaker: config.breaker,
+            events: config.events,
         })
     }
 
     pub async fn ping(&self) -> Result<(), Error> {
-        self.query(&commands::ping()).await
+        self.query("PING", &commands::ping()).await
     }
 
     pub async fn get<T: FromRedisValue>(&self, key: &str) -> Result<Option<T>, Error> {
-        self.query(&commands::get(self.key(key))).await
+        self.query("GET", &commands::get(self.key(key))).await
     }
 
     pub async fn mget<T: FromRedisValue>(&self, keys: &[&str]) -> Result<Vec<Option<T>>, Error> {
         if keys.is_empty() {
             return Ok(Vec::new());
         }
-        self.query(&commands::mget(self.keys(keys))).await
+        self.query("MGET", &commands::mget(self.keys(keys))).await
     }
 
     pub async fn set(
@@ -64,7 +71,8 @@ impl RedisClient {
         value: impl ToSingleRedisArg,
         ttl: Duration,
     ) -> Result<(), Error> {
-        self.query(&commands::set(self.key(key), value, ttl)).await
+        self.query("SET", &commands::set(self.key(key), value, ttl))
+            .await
     }
 
     pub async fn set_nx(
@@ -74,7 +82,7 @@ impl RedisClient {
         ttl: Duration,
     ) -> Result<bool, Error> {
         let reply: Option<String> = self
-            .query(&commands::set_nx(self.key(key), value, ttl))
+            .query("SET", &commands::set_nx(self.key(key), value, ttl))
             .await?;
         Ok(reply.is_some())
     }
@@ -83,15 +91,16 @@ impl RedisClient {
         if keys.is_empty() {
             return Ok(0);
         }
-        self.query(&commands::del(self.keys(keys))).await
+        self.query("DEL", &commands::del(self.keys(keys))).await
     }
 
     pub async fn expire(&self, key: &str, ttl: Duration) -> Result<bool, Error> {
-        self.query(&commands::expire(self.key(key), ttl)).await
+        self.query("PEXPIRE", &commands::expire(self.key(key), ttl))
+            .await
     }
 
     pub async fn ttl(&self, key: &str) -> Result<Ttl, Error> {
-        let reply: i64 = self.query(&commands::ttl(self.key(key))).await?;
+        let reply: i64 = self.query("PTTL", &commands::ttl(self.key(key))).await?;
         Ok(match reply {
             -2 => Ttl::Missing,
             -1 => Ttl::Persistent,
@@ -99,17 +108,27 @@ impl RedisClient {
         })
     }
 
-    pub async fn incr_by(&self, key: &str, delta: i64) -> Result<i64, Error> {
-        self.query(&commands::incr_by(self.key(key), delta)).await
-    }
-
-    pub async fn incr_by_float(&self, key: &str, delta: f64) -> Result<f64, Error> {
-        self.query(&commands::incr_by_float(self.key(key), delta))
+    pub async fn incr_by(&self, key: &str, delta: i64, expiry: Expiry) -> Result<i64, Error> {
+        self.call("INCRBY", commands::incr_by(self.key(key), delta, expiry))
             .await
     }
 
-    pub async fn rpush(&self, key: &str, values: impl ToRedisArgs) -> Result<usize, Error> {
-        self.query(&commands::rpush(self.key(key), values)).await
+    pub async fn incr_by_float(&self, key: &str, delta: f64, expiry: Expiry) -> Result<f64, Error> {
+        self.call(
+            "INCRBYFLOAT",
+            commands::incr_by_float(self.key(key), delta, expiry),
+        )
+        .await
+    }
+
+    pub async fn rpush(
+        &self,
+        key: &str,
+        values: impl ToRedisArgs,
+        expiry: Expiry,
+    ) -> Result<usize, Error> {
+        self.call("RPUSH", commands::rpush(self.key(key), values, expiry))
+            .await
     }
 
     pub async fn eval<T: FromRedisValue>(
@@ -120,11 +139,11 @@ impl RedisClient {
     ) -> Result<T, Error> {
         let keys = self.keys(keys);
         let args = args.to_redis_args();
-        self.guarded(async {
-            Ok(script
+        self.guarded(Operation::Script, async {
+            script
                 .invocation(&keys, &args)
                 .invoke_async(&mut self.connection.clone())
-                .await?)
+                .await
         })
         .await
     }
@@ -134,9 +153,18 @@ impl RedisClient {
     }
 
     pub async fn run(&self, pipeline: &Pipeline) -> Vec<Result<Value, Error>> {
+        let operation = Operation::Pipeline {
+            ops: pipeline.len(),
+        };
+        let started = Instant::now();
         let admission = match self.admit() {
             Ok(admission) => admission,
             Err(_) => {
+                self.observe(
+                    operation,
+                    started,
+                    Outcome::Failed(FailureKind::BreakerOpen),
+                );
                 return (0..pipeline.len())
                     .map(|_| Err(Error::BreakerOpen))
                     .collect();
@@ -150,6 +178,11 @@ impl RedisClient {
                 .filter(|error| matches!(error, Error::Transport(_)))
         });
         settle(admission, transport);
+        let failure = transport.or_else(|| results.iter().find_map(|result| result.as_ref().err()));
+        let outcome = failure.map_or(Outcome::Succeeded, |error| {
+            Outcome::Failed(FailureKind::of(error))
+        });
+        self.observe(operation, started, outcome);
         results
     }
 
@@ -161,18 +194,61 @@ impl RedisClient {
         keys.iter().map(|key| self.key(key)).collect()
     }
 
-    async fn query<T: FromRedisValue>(&self, command: &Cmd) -> Result<T, Error> {
-        self.guarded(async { Ok(command.query_async(&mut self.connection.clone()).await?) })
-            .await
+    async fn query<T: FromRedisValue>(
+        &self,
+        name: &'static str,
+        command: &Cmd,
+    ) -> Result<T, Error> {
+        self.guarded(Operation::Command(name), async {
+            command.query_async(&mut self.connection.clone()).await
+        })
+        .await
     }
 
-    async fn guarded<T>(&self, call: impl Future<Output = Result<T, Error>>) -> Result<T, Error> {
-        let admission = self.admit()?;
-        let result = call.await;
-        settle(admission, result.as_ref().err());
+    async fn call<T: FromRedisValue>(
+        &self,
+        name: &'static str,
+        call: ScriptCall,
+    ) -> Result<T, Error> {
+        self.guarded(Operation::Command(name), async {
+            call.script
+                .invocation(&call.keys, &call.args)
+                .invoke_async(&mut self.connection.clone())
+                .await
+        })
+        .await
+    }
+
+    async fn guarded<T>(
+        &self,
+        operation: Operation,
+        call: impl Future<Output = Result<T, RedisError>>,
+    ) -> Result<T, Error> {
+        let started = Instant::now();
+        let result = match self.admit() {
+            Ok(admission) => {
+                let result = call
+                    .await
+                    .map_err(|error| Error::classify(error, self.connection.deployment()));
+                settle(admission, result.as_ref().err());
+                result
+            }
+            Err(error) => Err(error),
+        };
+        self.observe(operation, started, Outcome::of(&result));
         result
     }
 
+    fn observe(&self, operation: Operation, started: Instant, outcome: Outcome) {
+        if let Some(events) = &self.events {
+            events.call(&CallEvent {
+                operation,
+                duration: started.elapsed(),
+                outcome,
+            });
+        }
+    }
+
     fn admit(&self) -> Result<Option<Admission>, Error> {
         self.breaker
             .as_ref()
@@ -198,455 +274,3 @@ fn settle(admission: Option<Admission>, error: Option<&Error>) {
         Some(Error::Server(_) | Error::Decode(_) | Error::BreakerOpen) => drop(admission),
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::time::{Duration, Instant};
-
-    use litellm_redis_testing::RedisServer;
-    use litellm_secret_value::SecretValue;
-    use rstest::rstest;
-
-    use litellm_circuit_breaker::{Breaker, Config, State};
-
-    use super::{RedisClient, Ttl};
-    use crate::{
-        DEFAULT_TIMEOUT, Endpoint, Error, HostEndpoint, Namespace, RedisConfig,
-        test_support::{Server, Topology, client_id},
-    };
-
-    #[rstest]
-    #[tokio::test]
-    async fn namespaces_isolate_keys_on_the_server(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let a = server.client("a", DEFAULT_TIMEOUT).await;
-        let b = server.client("b", DEFAULT_TIMEOUT).await;
-        let bare = server.client("", DEFAULT_TIMEOUT).await;
-
-        a.set("x", "1", Duration::from_secs(60)).await.unwrap();
-
-        assert_eq!(a.get::<String>("x").await.unwrap().as_deref(), Some("1"));
-        assert_eq!(a.get::<String>("a:x").await.unwrap().as_deref(), Some("1"));
-        assert_eq!(
-            bare.get::<String>("a:x").await.unwrap().as_deref(),
-            Some("1")
-        );
-        assert_eq!(b.get::<String>("x").await.unwrap(), None);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_rejected_command_keeps_the_connection(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        client.rpush("list", "item").await.unwrap();
-        let before = client_id(&client, "list").await;
-
-        let rejected = client.incr_by("list", 1).await;
-
-        assert!(matches!(rejected, Err(Error::Server(_))), "{rejected:?}");
-        assert_eq!(client.rpush("list", "next").await.unwrap(), 2);
-        assert_eq!(client_id(&client, "list").await, before);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_reply_of_another_type_is_a_decode_error_and_a_missing_key_is_none(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        client
-            .set("float", "15.0", Duration::from_secs(60))
-            .await
-            .unwrap();
-
-        let decoded = client.get::<i64>("float").await;
-
-        assert!(matches!(decoded, Err(Error::Decode(_))), "{decoded:?}");
-        assert_eq!(client.get::<i64>("missing").await.unwrap(), None);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_timed_out_command_does_not_leak_its_reply_into_the_next_one(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", Duration::from_millis(500)).await;
-        client
-            .set("first", "1", Duration::from_secs(60))
-            .await
-            .unwrap();
-        client
-            .set("second", "2", Duration::from_secs(60))
-            .await
-            .unwrap();
-
-        server.freeze().await;
-        let timed_out = client.get::<String>("first").await;
-        server.thaw().await;
-
-        assert!(
-            matches!(timed_out, Err(Error::Transport(_))),
-            "{timed_out:?}"
-        );
-        assert_eq!(
-            client.get::<String>("second").await.unwrap().as_deref(),
-            Some("2")
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn the_same_client_recovers_after_redis_restarts(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", Duration::from_millis(500)).await;
-
-        server.restart().await;
-
-        let deadline = Instant::now() + Duration::from_secs(30);
-        let recovered = loop {
-            match client.incr_by("after_restart", 1).await {
-                Ok(value) => break Some(value),
-                Err(_) if Instant::now() < deadline => {
-                    tokio::time::sleep(Duration::from_millis(200)).await
-                }
-                Err(_) => break None,
-            }
-        };
-        assert_eq!(recovered, Some(1));
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_client_created_while_redis_is_down_works_once_redis_is_up(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        server.stop().await;
-
-        let client = server.client("", Duration::from_millis(500)).await;
-        let while_down = client.ping().await;
-        server.resume().await;
-
-        assert!(
-            matches!(while_down, Err(Error::Transport(_))),
-            "{while_down:?}"
-        );
-        let deadline = Instant::now() + Duration::from_secs(30);
-        let recovered = loop {
-            match client.ping().await {
-                Ok(()) => break true,
-                Err(_) if Instant::now() < deadline => {
-                    tokio::time::sleep(Duration::from_millis(200)).await
-                }
-                Err(_) => break false,
-            }
-        };
-        assert!(recovered);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn empty_batches_succeed_without_asking_redis(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-
-        assert_eq!(
-            client.mget::<String>(&[]).await.unwrap(),
-            Vec::<Option<String>>::new()
-        );
-        assert_eq!(client.del(&[]).await.unwrap(), 0);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn mget_and_del_span_slots_in_input_order(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        client.set("a", "1", Duration::from_secs(60)).await.unwrap();
-        client.set("c", "3", Duration::from_secs(60)).await.unwrap();
-
-        let values = client.mget::<String>(&["c", "b", "a"]).await.unwrap();
-        let deleted = client.del(&["a", "b", "c"]).await.unwrap();
-
-        assert_eq!(
-            values,
-            vec![Some("3".to_owned()), None, Some("1".to_owned())]
-        );
-        assert_eq!(deleted, 2);
-        assert_eq!(
-            client.mget::<String>(&["a", "c"]).await.unwrap(),
-            vec![None, None]
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn set_nx_only_writes_a_missing_key(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-
-        let first = client
-            .set_nx("lock", "owner-1", Duration::from_secs(60))
-            .await
-            .unwrap();
-        let second = client
-            .set_nx("lock", "owner-2", Duration::from_secs(60))
-            .await
-            .unwrap();
-
-        assert!(first);
-        assert!(!second);
-        assert_eq!(
-            client.get::<String>("lock").await.unwrap().as_deref(),
-            Some("owner-1")
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn ttl_reports_missing_persistent_and_remaining_time(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        client.incr_by("persistent", 1).await.unwrap();
-        client
-            .set("expiring", "1", Duration::from_secs(60))
-            .await
-            .unwrap();
-        client.incr_by_float("expired_later", 0.5).await.unwrap();
-        assert!(
-            client
-                .expire("expired_later", Duration::from_secs(30))
-                .await
-                .unwrap()
-        );
-
-        assert_eq!(client.ttl("missing").await.unwrap(), Ttl::Missing);
-        assert_eq!(client.ttl("persistent").await.unwrap(), Ttl::Persistent);
-        assert!(matches!(
-            client.ttl("expiring").await.unwrap(),
-            Ttl::Expires(left) if left > Duration::from_secs(50) && left <= Duration::from_secs(60)
-        ));
-        assert!(matches!(
-            client.ttl("expired_later").await.unwrap(),
-            Ttl::Expires(left) if left > Duration::from_secs(20) && left <= Duration::from_secs(30)
-        ));
-        assert!(
-            !client
-                .expire("missing", Duration::from_secs(30))
-                .await
-                .unwrap()
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn incr_by_float_accumulates_fractions(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-
-        client.incr_by_float("spend", 0.25).await.unwrap();
-
-        assert_eq!(client.incr_by_float("spend", 0.5).await.unwrap(), 0.75);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn concurrent_commands_on_one_cloned_client_all_apply(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-
-        let tasks: Vec<_> = (0..100)
-            .map(|_| {
-                let client = client.clone();
-                tokio::spawn(async move {
-                    for _ in 0..10 {
-                        client.incr_by("shared", 1).await.unwrap();
-                    }
-                })
-            })
-            .collect();
-        for task in tasks {
-            task.await.unwrap();
-        }
-
-        assert_eq!(client.get::<i64>("shared").await.unwrap(), Some(1000));
-    }
-
-    fn breaker(failure_threshold: u32) -> Breaker {
-        Breaker::new(
-            Config {
-                failure_threshold,
-                recovery_timeout: Duration::from_millis(200),
-                timeout_min_duration: Duration::ZERO,
-            },
-            Box::new(|_| {}),
-        )
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn an_open_breaker_rejects_without_waiting_on_redis_and_a_probe_closes_it(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let breaker = breaker(2);
-        let client = server
-            .guarded_client("", Duration::from_millis(300), Some(breaker.clone()))
-            .await;
-
-        server.freeze().await;
-        let first = client.get::<String>("k").await;
-        let second = client.get::<String>("k").await;
-        let started = Instant::now();
-        let third = client.get::<String>("k").await;
-        let rejected_in = started.elapsed();
-        server.thaw().await;
-
-        assert!(matches!(first, Err(Error::Transport(_))), "{first:?}");
-        assert!(matches!(second, Err(Error::Transport(_))), "{second:?}");
-        assert!(matches!(third, Err(Error::BreakerOpen)), "{third:?}");
-        assert!(rejected_in < Duration::from_millis(50), "{rejected_in:?}");
-
-        let deadline = Instant::now() + Duration::from_secs(30);
-        let recovered = loop {
-            match client.get::<String>("k").await {
-                Ok(value) => break Some(value),
-                Err(_) if Instant::now() < deadline => {
-                    tokio::time::sleep(Duration::from_millis(50)).await
-                }
-                Err(_) => break None,
-            }
-        };
-        assert_eq!(recovered, Some(None));
-        assert!(
-            matches!(breaker.state(), State::Closed { .. }),
-            "{:?}",
-            breaker.state()
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn rejected_commands_never_open_the_breaker(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let breaker = breaker(2);
-        let client = server
-            .guarded_client("", DEFAULT_TIMEOUT, Some(breaker.clone()))
-            .await;
-        client.rpush("list", "item").await.unwrap();
-
-        let rejected = [
-            client.incr_by("list", 1).await,
-            client.incr_by("list", 1).await,
-            client.incr_by("list", 1).await,
-        ];
-
-        assert!(
-            matches!(rejected[2], Err(Error::Server(_))),
-            "{:?}",
-            rejected[2]
-        );
-        assert!(
-            matches!(breaker.state(), State::Closed { hard: 0, .. }),
-            "{:?}",
-            breaker.state()
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn an_open_breaker_fails_every_pipeline_op_without_sending(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server
-            .guarded_client("", Duration::from_millis(300), Some(breaker(1)))
-            .await;
-        let pipeline = client.pipeline().incr_by("a", 1).incr_by("c", 1);
-
-        server.freeze().await;
-        let failed = client.run(&pipeline).await;
-        let rejected = client.run(&pipeline).await;
-        server.thaw().await;
-
-        assert!(
-            matches!(failed[0], Err(Error::Transport(_))),
-            "{:?}",
-            failed[0]
-        );
-        assert!(
-            matches!(rejected[0], Err(Error::BreakerOpen)),
-            "{:?}",
-            rejected[0]
-        );
-        assert!(
-            matches!(rejected[1], Err(Error::BreakerOpen)),
-            "{:?}",
-            rejected[1]
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_host_endpoint_connects_to_its_database() {
-        let server = RedisServer::start().await.unwrap();
-        let port = server.url().rsplit(':').next().unwrap().parse().unwrap();
-        let connect = |endpoint| {
-            RedisClient::new(RedisConfig {
-                endpoint,
-                namespace: Namespace::default(),
-                timeout: DEFAULT_TIMEOUT,
-                breaker: None,
-            })
-        };
-        let in_db_3 = connect(Endpoint::Host(HostEndpoint {
-            host: "127.0.0.1".to_owned(),
-            port,
-            username: None,
-            password: None,
-            tls: false,
-            db: 3,
-        }))
-        .unwrap();
-        let in_db_0 = connect(Endpoint::Url(SecretValue::new(server.url().to_owned()))).unwrap();
-        let via_url = connect(Endpoint::Url(SecretValue::new(format!(
-            "{}/3",
-            server.url()
-        ))))
-        .unwrap();
-
-        in_db_3
-            .set("k", "3", Duration::from_secs(60))
-            .await
-            .unwrap();
-
-        assert_eq!(in_db_0.get::<String>("k").await.unwrap(), None);
-        assert_eq!(
-            via_url.get::<String>("k").await.unwrap().as_deref(),
-            Some("3")
-        );
-    }
-}

litellm-rust/crates/redis-client/src/commands.rs

@@ -1,7 +1,56 @@
-use std::time::Duration;
+use std::{sync::LazyLock, time::Duration};
 
 use redis::{Cmd, ToRedisArgs, ToSingleRedisArg};
 
+use crate::Script;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Expiry {
+    Refresh(Duration),
+    KeepExisting(Duration),
+}
+
+pub(crate) struct ScriptCall {
+    pub(crate) script: &'static Script,
+    pub(crate) keys: Vec<String>,
+    pub(crate) args: Vec<Vec<u8>>,
+}
+
+macro_rules! with_expiry {
+    ($write:literal) => {
+        with_expiry!($write, "value")
+    };
+    ($write:literal, $reply:literal) => {
+        concat!(
+            "local value = ",
+            $write,
+            "\nlocal reply = ",
+            $reply,
+            "\nif ARGV[1] == 'refresh' or redis.call('PTTL', KEYS[1]) == -1 then\n",
+            "  redis.call('PEXPIRE', KEYS[1], ARGV[2])\n",
+            "end\n",
+            "return reply"
+        )
+    };
+}
+
+static INCR_BY: LazyLock<Script> = LazyLock::new(|| {
+    Script::new(concat!(
+        "redis.call('GET', KEYS[1])\n",
+        with_expiry!(
+            "redis.call('INCRBY', KEYS[1], ARGV[3])",
+            "redis.call('GET', KEYS[1])"
+        )
+    ))
+});
+static INCR_BY_FLOAT: LazyLock<Script> =
+    LazyLock::new(|| Script::new(with_expiry!("redis.call('INCRBYFLOAT', KEYS[1], ARGV[3])")));
+static RPUSH: LazyLock<Script> = LazyLock::new(|| {
+    Script::new(with_expiry!(
+        "redis.call('RPUSH', KEYS[1], unpack(ARGV, 3))"
+    ))
+});
+
 pub(crate) fn ping() -> Cmd {
     redis::cmd("PING")
 }
@@ -45,16 +94,41 @@ pub(crate) fn ttl(key: String) -> Cmd {
     redis::cmd("PTTL").arg(key).to_owned()
 }
 
-pub(crate) fn incr_by(key: String, delta: i64) -> Cmd {
-    redis::cmd("INCRBY").arg(key).arg(delta).to_owned()
+pub(crate) fn incr_by(key: String, delta: i64, expiry: Expiry) -> ScriptCall {
+    with_expiry_call(&INCR_BY, key, expiry, delta)
 }
 
-pub(crate) fn incr_by_float(key: String, delta: f64) -> Cmd {
-    redis::cmd("INCRBYFLOAT").arg(key).arg(delta).to_owned()
+pub(crate) fn incr_by_float(key: String, delta: f64, expiry: Expiry) -> ScriptCall {
+    with_expiry_call(&INCR_BY_FLOAT, key, expiry, delta)
 }
 
-pub(crate) fn rpush(key: String, values: impl ToRedisArgs) -> Cmd {
-    redis::cmd("RPUSH").arg(key).arg(values).to_owned()
+pub(crate) fn rpush(key: String, values: impl ToRedisArgs, expiry: Expiry) -> ScriptCall {
+    with_expiry_call(&RPUSH, key, expiry, values)
+}
+
+fn with_expiry_call(
+    script: &'static Script,
+    key: String,
+    expiry: Expiry,
+    payload: impl ToRedisArgs,
+) -> ScriptCall {
+    let (mode, ttl) = match expiry {
+        Expiry::Refresh(ttl) => ("refresh", ttl),
+        Expiry::KeepExisting(ttl) => ("keep", ttl),
+    };
+    let args = [
+        mode.to_redis_args(),
+        millis(ttl).to_redis_args(),
+        payload.to_redis_args(),
+    ]
+    .into_iter()
+    .flatten()
+    .collect();
+    ScriptCall {
+        script,
+        keys: vec![key],
+        args,
+    }
 }
 
 fn millis(duration: Duration) -> u64 {

litellm-rust/crates/redis-client/src/config.rs

@@ -1,19 +1,42 @@
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 
 use redis::{ConnectionAddr, ConnectionInfo, IntoConnectionInfo, RedisConnectionInfo, RedisResult};
 
 use litellm_circuit_breaker::Breaker;
 use litellm_secret_value::SecretValue;
 
-use crate::Namespace;
+use crate::{Namespace, RedisEvents};
 
+// litellm/constants.py REDIS_SOCKET_TIMEOUT, which _coerce_redis_kwargs_types sets when socket_timeout is unset
 pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
 
 pub struct RedisConfig {
     pub endpoint: Endpoint,
     pub namespace: Namespace,
-    pub timeout: Duration,
+    pub timeouts: Timeouts,
     pub breaker: Option<Breaker>,
+    pub events: Option<Arc<dyn RedisEvents>>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub struct Timeouts {
+    pub response: Duration,
+    pub connect: Duration,
+}
+
+impl Timeouts {
+    pub fn new(socket_timeout: Duration, connect_timeout: Option<Duration>) -> Self {
+        Self {
+            response: socket_timeout,
+            connect: connect_timeout.unwrap_or(socket_timeout),
+        }
+    }
+}
+
+impl Default for Timeouts {
+    fn default() -> Self {
+        Self::new(DEFAULT_TIMEOUT, None)
+    }
 }
 
 #[derive(Clone, Debug, PartialEq)]

litellm-rust/crates/redis-client/src/connection.rs

@@ -1,15 +1,20 @@
-use std::time::Duration;
-
 use redis::{
     Cmd, ConnectionInfo, Pipeline, RedisFuture, RedisResult, Value,
     aio::{ConnectionLike, ConnectionManager, ConnectionManagerConfig},
     cluster::{ClusterClientBuilder, ClusterConfig},
     cluster_async::ClusterConnection,
+    io::tcp::TcpSettings,
 };
 
-use crate::ClusterEndpoint;
+use crate::{ClusterEndpoint, Timeouts, error::Deployment};
 
-const RETRIES: u32 = 3;
+// redis-py 5.3.1 redis/asyncio/connection.py: without retry options a connection gets Retry(NoBackoff(), 0)
+const RECONNECT_RETRIES: usize = 0;
+// redis-py 5.3.1 redis/asyncio/cluster.py: cluster_error_retry_attempts=3
+const CLUSTER_RETRIES: u32 = 3;
+// redis-py 5.3.1 redis/asyncio/cluster.py sleeps a flat 0.25s before retrying after CLUSTERDOWN.
+// redis-rs draws the wait from [min, max), so a 1ms window pins it to 250ms
+const CLUSTER_RETRY_WAIT_MS: u64 = 250;
 
 #[derive(Clone)]
 pub(crate) enum Connection {
@@ -18,20 +23,36 @@ pub(crate) enum Connection {
 }
 
 impl Connection {
-    pub(crate) fn standalone(info: ConnectionInfo, timeout: Duration) -> RedisResult<Self> {
+    pub(crate) fn deployment(&self) -> Deployment {
+        match self {
+            Self::Standalone(_) => Deployment::Standalone,
+            Self::Cluster(_) => Deployment::Cluster,
+        }
+    }
+
+    pub(crate) fn standalone(info: ConnectionInfo, timeouts: Timeouts) -> RedisResult<Self> {
         let config = ConnectionManagerConfig::new()
-            .set_connection_timeout(Some(timeout))
-            .set_response_timeout(Some(timeout))
-            .set_number_of_retries(RETRIES as usize);
+            .set_connection_timeout(Some(timeouts.connect))
+            .set_response_timeout(Some(timeouts.response))
+            .set_number_of_retries(RECONNECT_RETRIES);
+        let info = info.set_tcp_settings(TcpSettings::default().set_nodelay(true));
         let manager = ConnectionManager::new_lazy_with_config(redis::Client::open(info)?, config)?;
         Ok(Self::Standalone(manager))
     }
 
-    pub(crate) fn cluster(endpoint: &ClusterEndpoint, timeout: Duration) -> RedisResult<Self> {
+    pub(crate) fn cluster(endpoint: &ClusterEndpoint, timeouts: Timeouts) -> RedisResult<Self> {
         let builder = ClusterClientBuilder::new(endpoint.addrs())
-            .connection_timeout(timeout)
-            .response_timeout(timeout)
-            .retries(RETRIES);
+            .tcp_settings(
+                TcpSettings::default()
+                    .set_nodelay(true)
+                    .set_keepalive(socket2::TcpKeepalive::new()),
+            )
+            .connection_timeout(timeouts.connect)
+            .response_timeout(timeouts.response)
+            .retries(CLUSTER_RETRIES)
+            .retry_wait_formula(0, 2)
+            .min_retry_wait(CLUSTER_RETRY_WAIT_MS)
+            .max_retry_wait(CLUSTER_RETRY_WAIT_MS + 1);
         let builder = match &endpoint.username {
             Some(username) => builder.username(username.expose()),
             None => builder,
@@ -40,9 +61,9 @@ impl Connection {
             Some(password) => builder.password(password.expose()),
             None => builder,
         };
-        let config = ClusterConfig::new()
-            .set_connection_timeout(timeout)
-            .set_response_timeout(timeout);
+        // The builder's response timeout bounds the whole request, but with_config overwrites each node's
+        // response timeout with this one even when unset, so it has to be repeated here
+        let config = ClusterConfig::new().set_response_timeout(timeouts.response);
         Ok(Self::Cluster(
             builder
                 .build()?

litellm-rust/crates/redis-client/src/error.rs

@@ -16,29 +16,62 @@ pub enum Error {
 #[error("connecting to Redis")]
 pub struct ConnectError(#[from] RedisError);
 
-impl From<RedisError> for Error {
-    fn from(error: RedisError) -> Self {
-        match error.kind() {
-            ErrorKind::Server(
-                ServerErrorKind::BusyLoading
-                | ServerErrorKind::ClusterDown
-                | ServerErrorKind::MasterDown
-                | ServerErrorKind::TryAgain
-                | ServerErrorKind::ReadOnly,
-            ) => Self::Transport(error),
-            ErrorKind::Server(_) | ErrorKind::Extension => Self::Server(error),
-            ErrorKind::Parse | ErrorKind::UnexpectedReturnType => Self::Decode(error),
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Deployment {
+    Standalone,
+    Cluster,
+}
+
+// redis-py 5.3.1 redis/_parsers/base.py EXCEPTION_CLASSES["ERR"] entries that subclass ConnectionError
+const CONNECTION_ERR_MESSAGES: &[&str] = &[
+    "max number of clients reached",
+    "invalid password",
+    "Client sent AUTH, but no password is set",
+    "AUTH <password> called without any password configured for the default user. Are you sure your configuration is correct?",
+];
+
+impl Error {
+    pub(crate) fn classify(error: RedisError, deployment: Deployment) -> Self {
+        match (error.kind(), deployment) {
+            (ErrorKind::Parse | ErrorKind::UnexpectedReturnType, _) => Self::Decode(error),
+            (ErrorKind::Server(kind), _) if is_unavailable(kind, deployment) => {
+                Self::Transport(error)
+            }
+            (ErrorKind::Server(ServerErrorKind::ResponseError), _)
+                if error
+                    .detail()
+                    .is_some_and(|detail| CONNECTION_ERR_MESSAGES.contains(&detail)) =>
+            {
+                Self::Transport(error)
+            }
+            (ErrorKind::Extension, _) if matches!(error.code(), Some("NOAUTH" | "WRONGPASS")) => {
+                Self::Transport(error)
+            }
+            (ErrorKind::Server(_) | ErrorKind::Extension | ErrorKind::InvalidClientConfig, _) => {
+                Self::Server(error)
+            }
             _ => Self::Transport(error),
         }
     }
 }
 
+fn is_unavailable(kind: ServerErrorKind, deployment: Deployment) -> bool {
+    matches!(
+        (kind, deployment),
+        (ServerErrorKind::BusyLoading, _)
+            | (
+                ServerErrorKind::ClusterDown | ServerErrorKind::MasterDown,
+                Deployment::Cluster
+            )
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use redis::{RedisError, Value};
     use rstest::rstest;
 
-    use super::Error;
+    use super::{Deployment, Error};
 
     fn reply(resp: &str) -> RedisError {
         match redis::parse_redis_value(resp.as_bytes()).unwrap() {
@@ -47,23 +80,84 @@ mod tests {
         }
     }
 
+    fn is_transport(resp: &str, deployment: Deployment) -> bool {
+        match Error::classify(reply(resp), deployment) {
+            Error::Transport(_) => true,
+            Error::Server(_) => false,
+            other => panic!("{resp:?} classified as {other:?}"),
+        }
+    }
+
     #[rstest]
     #[case::wrong_type("-WRONGTYPE Operation against a key holding the wrong kind of value\r\n")]
     #[case::no_script("-NOSCRIPT No matching script\r\n")]
     #[case::cross_slot("-CROSSSLOT Keys in request don't hash to the same slot\r\n")]
     #[case::generic("-ERR value is not an integer or out of range\r\n")]
     #[case::no_perm("-NOPERM this user has no permissions to access one of the keys\r\n")]
-    fn rejections_are_server_errors(#[case] resp: &str) {
-        assert!(matches!(Error::from(reply(resp)), Error::Server(_)));
+    #[case::try_again("-TRYAGAIN Multiple keys request during rehashing of slot\r\n")]
+    #[case::out_of_memory("-OOM command not allowed when used memory > 'maxmemory'.\r\n")]
+    #[case::auth_arity("-ERR wrong number of arguments for 'auth' command\r\n")]
+    #[case::max_clients_with_suffix("-ERR max number of clients reached, try later\r\n")]
+    fn rejections_are_server_errors_on_every_deployment(
+        #[case] resp: &str,
+        #[values(Deployment::Standalone, Deployment::Cluster)] deployment: Deployment,
+    ) {
+        assert!(!is_transport(resp, deployment));
     }
 
     #[rstest]
     #[case::loading("-LOADING Redis is loading the dataset in memory\r\n")]
-    #[case::cluster_down("-CLUSTERDOWN The cluster is down\r\n")]
-    #[case::master_down("-MASTERDOWN Link with MASTER is down\r\n")]
-    #[case::try_again("-TRYAGAIN Multiple keys request during rehashing of slot\r\n")]
-    #[case::read_only("-READONLY You can't write against a read only replica.\r\n")]
-    fn unavailable_nodes_are_transport_errors(#[case] resp: &str) {
-        assert!(matches!(Error::from(reply(resp)), Error::Transport(_)));
+    #[case::no_auth("-NOAUTH Authentication required.\r\n")]
+    #[case::wrong_pass("-WRONGPASS invalid username-password pair or user is disabled.\r\n")]
+    #[case::max_clients("-ERR max number of clients reached\r\n")]
+    #[case::invalid_password("-ERR invalid password\r\n")]
+    #[case::no_password_set("-ERR Client sent AUTH, but no password is set\r\n")]
+    #[case::no_default_password(
+        "-ERR AUTH <password> called without any password configured for the default user. Are you sure your configuration is correct?\r\n"
+    )]
+    fn unavailable_nodes_are_transport_errors_on_every_deployment(
+        #[case] resp: &str,
+        #[values(Deployment::Standalone, Deployment::Cluster)] deployment: Deployment,
+    ) {
+        assert!(is_transport(resp, deployment));
+    }
+
+    #[rstest]
+    #[case::master_down_on_standalone(
+        "-MASTERDOWN Link with MASTER is down\r\n",
+        Deployment::Standalone,
+        false
+    )]
+    #[case::master_down_on_cluster(
+        "-MASTERDOWN Link with MASTER is down\r\n",
+        Deployment::Cluster,
+        true
+    )]
+    #[case::cluster_down_on_standalone(
+        "-CLUSTERDOWN The cluster is down\r\n",
+        Deployment::Standalone,
+        false
+    )]
+    #[case::cluster_down_on_cluster(
+        "-CLUSTERDOWN The cluster is down\r\n",
+        Deployment::Cluster,
+        true
+    )]
+    #[case::read_only_on_standalone(
+        "-READONLY You can't write against a read only replica.\r\n",
+        Deployment::Standalone,
+        false
+    )]
+    #[case::read_only_on_cluster(
+        "-READONLY You can't write against a read only replica.\r\n",
+        Deployment::Cluster,
+        false
+    )]
+    fn topology_decides_like_redis_py(
+        #[case] resp: &str,
+        #[case] deployment: Deployment,
+        #[case] transport: bool,
+    ) {
+        assert_eq!(is_transport(resp, deployment), transport);
     }
 }

litellm-rust/crates/redis-client/src/events.rs

@@ -0,0 +1,57 @@
+use std::time::Duration;
+
+use crate::Error;
+
+pub trait RedisEvents: Send + Sync {
+    fn call(&self, event: &CallEvent);
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Operation {
+    Command(&'static str),
+    Script,
+    Pipeline { ops: usize },
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Outcome {
+    Succeeded,
+    Failed(FailureKind),
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum FailureKind {
+    Timeout,
+    Transport,
+    Server,
+    Decode,
+    BreakerOpen,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub struct CallEvent {
+    pub operation: Operation,
+    pub duration: Duration,
+    pub outcome: Outcome,
+}
+
+impl FailureKind {
+    pub fn of(error: &Error) -> Self {
+        match error {
+            Error::Transport(error) if error.is_timeout() => Self::Timeout,
+            Error::Transport(_) => Self::Transport,
+            Error::Server(_) => Self::Server,
+            Error::Decode(_) => Self::Decode,
+            Error::BreakerOpen => Self::BreakerOpen,
+        }
+    }
+}
+
+impl Outcome {
+    pub(crate) fn of<T>(result: &Result<T, Error>) -> Self {
+        match result {
+            Ok(_) => Self::Succeeded,
+            Err(error) => Self::Failed(FailureKind::of(error)),
+        }
+    }
+}

litellm-rust/crates/redis-client/src/lib.rs

@@ -3,18 +3,19 @@ mod commands;
 mod config;
 mod connection;
 mod error;
+mod events;
 mod namespace;
 mod pipeline;
 mod script;
-#[cfg(test)]
-mod test_support;
 
 pub use client::{RedisClient, Ttl};
+pub use commands::Expiry;
 pub use config::{
-    ClusterEndpoint, ClusterNode, DEFAULT_TIMEOUT, Endpoint, HostEndpoint, RedisConfig,
+    ClusterEndpoint, ClusterNode, DEFAULT_TIMEOUT, Endpoint, HostEndpoint, RedisConfig, Timeouts,
 };
 pub use error::{ConnectError, Error};
+pub use events::{CallEvent, FailureKind, Operation, Outcome, RedisEvents};
 pub use namespace::Namespace;
 pub use pipeline::Pipeline;
-pub use redis::Value;
+pub use redis::{ErrorKind as RedisErrorKind, RedisError, Value};
 pub use script::Script;

litellm-rust/crates/redis-client/src/pipeline.rs

@@ -2,11 +2,14 @@ use std::{collections::HashMap, time::Duration};
 
 use futures_util::future::join_all;
 use redis::{
-    Cmd, ErrorKind, RedisResult, ServerErrorKind, ToRedisArgs, ToSingleRedisArg, Value,
-    cluster_routing::Slot,
+    Cmd, FromRedisValue, RedisResult, ToRedisArgs, ToSingleRedisArg, Value, cluster_routing::Slot,
 };
 
-use crate::{Error, Namespace, Script, commands, connection::Connection};
+use crate::{
+    Error, Expiry, Namespace, Script,
+    commands::{self, ScriptCall},
+    connection::Connection,
+};
 
 pub struct Pipeline {
     namespace: Namespace,
@@ -16,13 +19,8 @@ pub struct Pipeline {
 struct Op {
     command: Cmd,
     key: String,
-    script: Option<ScriptCall>,
-}
-
-struct ScriptCall {
-    script: Script,
-    keys: Vec<String>,
-    args: Vec<Vec<u8>>,
+    idempotent: bool,
+    integer_reply: bool,
 }
 
 type Settled = (usize, Result<Value, Error>);
@@ -37,59 +35,62 @@ impl Pipeline {
 
     pub fn get(self, key: &str) -> Self {
         let key = self.namespace.apply(key);
-        self.command(commands::get(key.clone()), key)
+        self.command(commands::get(key.clone()), key, true, false)
     }
 
     pub fn set(self, key: &str, value: impl ToSingleRedisArg, ttl: Duration) -> Self {
         let key = self.namespace.apply(key);
-        self.command(commands::set(key.clone(), value, ttl), key)
+        self.command(commands::set(key.clone(), value, ttl), key, true, false)
     }
 
     pub fn del(self, key: &str) -> Self {
         let key = self.namespace.apply(key);
-        self.command(commands::del(vec![key.clone()]), key)
+        self.command(commands::del(vec![key.clone()]), key, true, false)
     }
 
     pub fn expire(self, key: &str, ttl: Duration) -> Self {
         let key = self.namespace.apply(key);
-        self.command(commands::expire(key.clone(), ttl), key)
+        self.command(commands::expire(key.clone(), ttl), key, true, false)
     }
 
-    pub fn incr_by(self, key: &str, delta: i64) -> Self {
+    pub fn incr_by(self, key: &str, delta: i64, expiry: Expiry) -> Self {
         let key = self.namespace.apply(key);
-        self.command(commands::incr_by(key.clone(), delta), key)
+        self.script_call(commands::incr_by(key, delta, expiry), true)
     }
 
-    pub fn incr_by_float(self, key: &str, delta: f64) -> Self {
+    pub fn incr_by_float(self, key: &str, delta: f64, expiry: Expiry) -> Self {
         let key = self.namespace.apply(key);
-        self.command(commands::incr_by_float(key.clone(), delta), key)
+        self.script_call(commands::incr_by_float(key, delta, expiry), false)
+    }
+
+    pub fn rpush(self, key: &str, values: impl ToRedisArgs, expiry: Expiry) -> Self {
+        let key = self.namespace.apply(key);
+        self.script_call(commands::rpush(key, values, expiry), false)
     }
 
     pub fn eval(self, script: &Script, keys: &[&str], args: impl ToRedisArgs) -> Self {
         let keys: Vec<String> = keys.iter().map(|key| self.namespace.apply(key)).collect();
-        let args = args.to_redis_args();
-        let op = Op {
-            command: script.evalsha(&keys, &args),
-            key: keys.first().cloned().unwrap_or_default(),
-            script: Some(ScriptCall {
-                script: script.clone(),
-                keys,
-                args,
-            }),
-        };
-        self.push(op)
+        let key = keys.first().cloned().unwrap_or_default();
+        self.command(script.eval(&keys, &args.to_redis_args()), key, false, false)
     }
 
-    fn command(self, command: Cmd, key: String) -> Self {
-        self.push(Op {
+    fn script_call(self, call: ScriptCall, integer_reply: bool) -> Self {
+        let key = call.keys.first().cloned().unwrap_or_default();
+        self.command(
+            call.script.eval(&call.keys, &call.args),
+            key,
+            false,
+            integer_reply,
+        )
+    }
+
+    fn command(mut self, command: Cmd, key: String, idempotent: bool, integer_reply: bool) -> Self {
+        self.ops.push(Op {
             command,
             key,
-            script: None,
-        })
-    }
-
-    fn push(mut self, op: Op) -> Self {
-        self.ops.push(op);
+            idempotent,
+            integer_reply,
+        });
         self
     }
 
@@ -102,7 +103,7 @@ impl Pipeline {
         let settled = join_all(
             groups
                 .iter()
-                .map(|group| self.run_group(connection.clone(), group)),
+                .map(|batches| self.run_in_order(connection, batches)),
         )
         .await;
         let mut settled: Vec<Settled> = settled.into_iter().flatten().collect();
@@ -110,24 +111,43 @@ impl Pipeline {
         settled.into_iter().map(|(_, result)| result).collect()
     }
 
-    fn groups(&self, connection: &Connection) -> Vec<Vec<usize>> {
+    async fn run_in_order(&self, connection: &Connection, batches: &[Vec<usize>]) -> Vec<Settled> {
+        let mut settled = Vec::with_capacity(batches.iter().map(Vec::len).sum());
+        for batch in batches {
+            settled.extend(self.run_group(connection.clone(), batch).await);
+        }
+        settled
+    }
+
+    fn groups(&self, connection: &Connection) -> Vec<Vec<Vec<usize>>> {
         match connection {
             _ if self.ops.is_empty() => Vec::new(),
-            Connection::Standalone(_) => vec![(0..self.ops.len()).collect()],
+            Connection::Standalone(_) => vec![vec![(0..self.ops.len()).collect()]],
             Connection::Cluster(_) => {
                 let mut by_slot: HashMap<Slot, Vec<usize>> = HashMap::new();
-                for (index, op) in self.ops.iter().enumerate() {
+                for index in 0..self.ops.len() {
                     by_slot
-                        .entry(Slot::for_key(&op.key))
+                        .entry(Slot::for_key(&self.ops[index].key))
                         .or_default()
                         .push(index);
                 }
-                by_slot.into_values().collect()
+                by_slot
+                    .into_values()
+                    .map(|indices| self.retry_safe_batches(indices))
+                    .collect()
             }
         }
     }
 
+    fn retry_safe_batches(&self, indices: Vec<usize>) -> Vec<Vec<usize>> {
+        indices
+            .chunk_by(|&left, &right| self.ops[left].idempotent && self.ops[right].idempotent)
+            .map(<[usize]>::to_vec)
+            .collect()
+    }
+
     async fn run_group(&self, mut connection: Connection, indices: &[usize]) -> Vec<Settled> {
+        let deployment = connection.deployment();
         let mut pipeline = redis::pipe();
         pipeline.ignore_errors();
         for &index in indices {
@@ -135,158 +155,31 @@ impl Pipeline {
         }
         let replies: RedisResult<Vec<RedisResult<Value>>> =
             pipeline.query_async(&mut connection).await;
-        let replies = match replies {
-            Ok(replies) => replies,
-            Err(error) => {
-                return indices
-                    .iter()
-                    .map(|&index| (index, Err(Error::from(error.clone()))))
-                    .collect();
-            }
-        };
-        let mut settled = Vec::with_capacity(indices.len());
-        for (&index, reply) in indices.iter().zip(replies) {
-            settled.push((index, self.settle(&mut connection, index, reply).await));
-        }
-        settled
-    }
-
-    async fn settle(
-        &self,
-        connection: &mut Connection,
-        index: usize,
-        reply: RedisResult<Value>,
-    ) -> Result<Value, Error> {
-        match (reply, &self.ops[index].script) {
-            (Err(error), Some(call))
-                if error.kind() == ErrorKind::Server(ServerErrorKind::NoScript) =>
-            {
-                Ok(call
-                    .script
-                    .invocation(&call.keys, &call.args)
-                    .invoke_async(connection)
-                    .await?)
-            }
-            (reply, _) => Ok(reply?),
+        match replies {
+            Ok(replies) => indices
+                .iter()
+                .zip(replies)
+                .map(|(&index, reply)| {
+                    (
+                        index,
+                        reply
+                            .map_err(|error| Error::classify(error, deployment))
+                            .and_then(|value| {
+                                if self.ops[index].integer_reply {
+                                    i64::from_redis_value(value)
+                                        .map(Value::Int)
+                                        .map_err(|error| Error::Decode(error.into()))
+                                } else {
+                                    Ok(value)
+                                }
+                            }),
+                    )
+                })
+                .collect(),
+            Err(error) => indices
+                .iter()
+                .map(|&index| (index, Err(Error::classify(error.clone(), deployment))))
+                .collect(),
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::time::{Duration, Instant};
-
-    use redis::Value;
-    use rstest::rstest;
-
-    use crate::{
-        DEFAULT_TIMEOUT, Error, Script,
-        test_support::{Server, Topology, flush_scripts},
-    };
-
-    #[rstest]
-    #[tokio::test]
-    async fn each_op_settles_on_its_own_in_input_order(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        client
-            .set("text", "not a number", Duration::from_secs(60))
-            .await
-            .unwrap();
-        let pipeline = client
-            .pipeline()
-            .incr_by("a", 1)
-            .incr_by("text", 1)
-            .incr_by("c", 5)
-            .set("b", "2", Duration::from_secs(60))
-            .get("b")
-            .incr_by("a", 1)
-            .get("missing");
-
-        let results = client.run(&pipeline).await;
-
-        assert_eq!(results.len(), 7);
-        assert_eq!(results[0].as_ref().unwrap(), &Value::Int(1));
-        assert!(
-            matches!(results[1], Err(Error::Server(_))),
-            "{:?}",
-            results[1]
-        );
-        assert_eq!(results[2].as_ref().unwrap(), &Value::Int(5));
-        assert_eq!(results[3].as_ref().unwrap(), &Value::Okay);
-        assert_eq!(
-            results[4].as_ref().unwrap(),
-            &Value::BulkString(b"2".to_vec())
-        );
-        assert_eq!(results[5].as_ref().unwrap(), &Value::Int(2));
-        assert_eq!(results[6].as_ref().unwrap(), &Value::Nil);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_flushed_script_in_a_pipeline_reruns_alone(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("ns", DEFAULT_TIMEOUT).await;
-        let incr = Script::new("return redis.call('INCRBY', KEYS[1], ARGV[1])");
-        client.eval::<i64>(&incr, &["counter"], 1).await.unwrap();
-        flush_scripts(&client).await;
-        let pipeline = client
-            .pipeline()
-            .eval(&incr, &["counter"], 2)
-            .incr_by("other", 7);
-
-        let results = client.run(&pipeline).await;
-
-        assert_eq!(results[0].as_ref().unwrap(), &Value::Int(3));
-        assert_eq!(results[1].as_ref().unwrap(), &Value::Int(7));
-        assert_eq!(client.get::<i64>("counter").await.unwrap(), Some(3));
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn an_unreachable_server_fails_every_op_as_transport(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", Duration::from_millis(500)).await;
-        let pipeline = client.pipeline().incr_by("a", 1).incr_by("c", 1);
-
-        server.freeze().await;
-        let results = client.run(&pipeline).await;
-        server.thaw().await;
-
-        assert_eq!(results.len(), 2);
-        assert!(
-            matches!(results[0], Err(Error::Transport(_))),
-            "{:?}",
-            results[0]
-        );
-        assert!(
-            matches!(results[1], Err(Error::Transport(_))),
-            "{:?}",
-            results[1]
-        );
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn an_empty_pipeline_returns_without_asking_redis(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", Duration::from_secs(2)).await;
-
-        server.freeze().await;
-        let started = Instant::now();
-        let results = client.run(&client.pipeline()).await;
-        let elapsed = started.elapsed();
-        server.thaw().await;
-
-        assert!(results.is_empty());
-        assert!(elapsed < Duration::from_secs(1), "{elapsed:?}");
-    }
-}

litellm-rust/crates/redis-client/src/script.rs

@@ -1,16 +1,24 @@
+use std::sync::Arc;
+
 use redis::{Cmd, ScriptInvocation};
 
 #[derive(Clone)]
-pub struct Script(redis::Script);
+pub struct Script {
+    cached: redis::Script,
+    source: Arc<str>,
+}
 
 impl Script {
     pub fn new(source: &str) -> Self {
-        Self(redis::Script::new(source))
+        Self {
+            cached: redis::Script::new(source),
+            source: source.into(),
+        }
     }
 
-    pub(crate) fn evalsha(&self, keys: &[String], args: &[Vec<u8>]) -> Cmd {
-        redis::cmd("EVALSHA")
-            .arg(self.0.get_hash())
+    pub(crate) fn eval(&self, keys: &[String], args: &[Vec<u8>]) -> Cmd {
+        redis::cmd("EVAL")
+            .arg(&*self.source)
             .arg(keys.len())
             .arg(keys)
             .arg(args)
@@ -18,71 +26,8 @@ impl Script {
     }
 
     pub(crate) fn invocation(&self, keys: &[String], args: &[Vec<u8>]) -> ScriptInvocation<'_> {
-        let mut invocation = self.0.prepare_invoke();
+        let mut invocation = self.cached.prepare_invoke();
         invocation.key(keys).arg(args);
         invocation
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::time::Duration;
-
-    use rstest::rstest;
-
-    use super::Script;
-    use crate::{
-        DEFAULT_TIMEOUT, Error,
-        test_support::{Server, Topology, flush_scripts},
-    };
-
-    #[rstest]
-    #[tokio::test]
-    async fn script_keys_are_namespaced(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("a", DEFAULT_TIMEOUT).await;
-        let bare = server.client("", DEFAULT_TIMEOUT).await;
-        let read = Script::new("return redis.call('GET', KEYS[1])");
-        bare.set("a:x", "1", Duration::from_secs(60)).await.unwrap();
-
-        let value: Option<String> = client
-            .eval(&read, &["x"], Vec::<String>::new())
-            .await
-            .unwrap();
-
-        assert_eq!(value.as_deref(), Some("1"));
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_failing_script_is_a_server_error_and_keeps_the_connection(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        let incr = Script::new("return redis.call('INCRBY', KEYS[1], ARGV[1])");
-        client.rpush("list", "item").await.unwrap();
-
-        let failed = client.eval::<i64>(&incr, &["list"], 1).await;
-
-        assert!(matches!(failed, Err(Error::Server(_))), "{failed:?}");
-        assert_eq!(client.eval::<i64>(&incr, &["counter"], 3).await.unwrap(), 3);
-    }
-
-    #[rstest]
-    #[tokio::test]
-    async fn a_flushed_script_is_loaded_again(
-        #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
-    ) {
-        let server = Server::start(topology).await;
-        let client = server.client("", DEFAULT_TIMEOUT).await;
-        let incr = Script::new("return redis.call('INCRBY', KEYS[1], ARGV[1])");
-        client.eval::<i64>(&incr, &["counter"], 1).await.unwrap();
-
-        flush_scripts(&client).await;
-
-        assert_eq!(client.eval::<i64>(&incr, &["counter"], 1).await.unwrap(), 2);
-    }
-}

litellm-rust/crates/redis-client/src/test_support.rs

@@ -1,126 +0,0 @@
-use std::time::Duration;
-
-use litellm_circuit_breaker::Breaker;
-use litellm_redis_testing::{RedisCluster, RedisServer};
-use litellm_secret_value::SecretValue;
-use redis::cluster_routing::{Route, RoutingInfo, SingleNodeRoutingInfo, SlotAddr};
-
-use crate::{
-    ClusterEndpoint, ClusterNode, Endpoint, Namespace, RedisClient, RedisConfig,
-    connection::Connection,
-};
-
-#[derive(Clone, Copy, Debug)]
-pub(crate) enum Topology {
-    Standalone,
-    Cluster,
-}
-
-pub(crate) enum Server {
-    Standalone(RedisServer),
-    Cluster(RedisCluster),
-}
-
-impl Server {
-    pub(crate) async fn start(topology: Topology) -> Self {
-        match topology {
-            Topology::Standalone => Self::Standalone(RedisServer::start().await.unwrap()),
-            Topology::Cluster => Self::Cluster(RedisCluster::start().await.unwrap()),
-        }
-    }
-
-    pub(crate) fn endpoint(&self) -> Endpoint {
-        match self {
-            Self::Standalone(server) => Endpoint::Url(SecretValue::new(server.url().to_owned())),
-            Self::Cluster(cluster) => Endpoint::Cluster(ClusterEndpoint {
-                nodes: cluster
-                    .nodes()
-                    .map(|(host, port)| ClusterNode {
-                        host: host.to_owned(),
-                        port,
-                    })
-                    .collect(),
-                username: None,
-                password: None,
-                tls: false,
-            }),
-        }
-    }
-
-    pub(crate) async fn client(&self, namespace: &str, timeout: Duration) -> RedisClient {
-        self.guarded_client(namespace, timeout, None).await
-    }
-
-    pub(crate) async fn guarded_client(
-        &self,
-        namespace: &str,
-        timeout: Duration,
-        breaker: Option<Breaker>,
-    ) -> RedisClient {
-        RedisClient::new(RedisConfig {
-            endpoint: self.endpoint(),
-            namespace: Namespace::new(namespace),
-            timeout,
-            breaker,
-        })
-        .unwrap()
-    }
-
-    pub(crate) async fn restart(&self) {
-        match self {
-            Self::Standalone(server) => server.restart().await.unwrap(),
-            Self::Cluster(cluster) => cluster.restart().await.unwrap(),
-        }
-    }
-
-    pub(crate) async fn stop(&self) {
-        match self {
-            Self::Standalone(server) => server.stop().await.unwrap(),
-            Self::Cluster(cluster) => cluster.stop().await.unwrap(),
-        }
-    }
-
-    pub(crate) async fn resume(&self) {
-        match self {
-            Self::Standalone(server) => server.resume().await.unwrap(),
-            Self::Cluster(cluster) => cluster.resume().await.unwrap(),
-        }
-    }
-
-    pub(crate) async fn freeze(&self) {
-        match self {
-            Self::Standalone(server) => server.freeze().await.unwrap(),
-            Self::Cluster(cluster) => cluster.freeze().await.unwrap(),
-        }
-    }
-
-    pub(crate) async fn thaw(&self) {
-        match self {
-            Self::Standalone(server) => server.thaw().await.unwrap(),
-            Self::Cluster(cluster) => cluster.thaw().await.unwrap(),
-        }
-    }
-}
-
-pub(crate) async fn client_id(client: &RedisClient, key: &str) -> i64 {
-    let command = redis::cmd("CLIENT").arg("ID").to_owned();
-    match client.connection.clone() {
-        Connection::Standalone(mut connection) => {
-            command.query_async(&mut connection).await.unwrap()
-        }
-        Connection::Cluster(mut connection) => {
-            let node = Route::with_key(key, SlotAddr::Master);
-            let routing = RoutingInfo::SingleNode(SingleNodeRoutingInfo::SpecificNode(node));
-            redis::from_redis_value(connection.route_command(command, routing).await.unwrap())
-                .unwrap()
-        }
-    }
-}
-
-pub(crate) async fn flush_scripts(client: &RedisClient) {
-    redis::cmd("SCRIPT")
-        .arg("FLUSH")
-        .query_async::<()>(&mut client.connection.clone())
-        .await
-        .unwrap();
-}

litellm-rust/crates/redis-client/tests/client.rs

@@ -0,0 +1,761 @@
+mod support;
+
+use std::{
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+
+use litellm_redis_testing::{RedisDeployment, RedisServer, Topology, eventually};
+use rstest::rstest;
+
+use litellm_circuit_breaker::{Breaker, Config, State};
+
+use litellm_redis_client::{
+    CallEvent, DEFAULT_TIMEOUT, Endpoint, Error, Expiry, FailureKind, HostEndpoint, Namespace,
+    Operation, Outcome, RedisClient, RedisConfig, RedisEvents, Script, Timeouts, Ttl,
+};
+use litellm_secret_value::SecretValue;
+use support::{KEEP, connect, connect_guarded, connections, endpoint};
+
+async fn persist(client: &RedisClient, key: &str) {
+    let set = Script::new("return redis.call('SET', KEYS[1], '0')");
+    client
+        .eval::<()>(&set, &[key], Vec::<String>::new())
+        .await
+        .unwrap();
+}
+
+#[derive(Clone, Copy, Debug)]
+enum Counter {
+    IncrBy,
+    IncrByFloat,
+    Rpush,
+}
+
+async fn write(client: &RedisClient, counter: Counter, key: &str, expiry: Expiry) {
+    match counter {
+        Counter::IncrBy => drop(client.incr_by(key, 1, expiry).await.unwrap()),
+        Counter::IncrByFloat => drop(client.incr_by_float(key, 0.5, expiry).await.unwrap()),
+        Counter::Rpush => drop(client.rpush(key, "item", expiry).await.unwrap()),
+    }
+}
+
+async fn persist_as(client: &RedisClient, counter: Counter, key: &str) {
+    let source = match counter {
+        Counter::IncrBy | Counter::IncrByFloat => "return redis.call('SET', KEYS[1], '1')",
+        Counter::Rpush => "return redis.call('RPUSH', KEYS[1], 'first')",
+    };
+    client
+        .eval::<redis::Value>(&Script::new(source), &[key], Vec::<String>::new())
+        .await
+        .unwrap();
+}
+
+fn remaining(ttl: Ttl) -> Duration {
+    match ttl {
+        Ttl::Expires(left) => left,
+        other => panic!("expected an expiring key, got {other:?}"),
+    }
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_counter_write_expires_a_new_key_in_the_same_command(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(Counter::IncrBy, Counter::IncrByFloat, Counter::Rpush)] counter: Counter,
+    #[values(
+        Expiry::KeepExisting(Duration::from_secs(60)),
+        Expiry::Refresh(Duration::from_secs(60))
+    )]
+    expiry: Expiry,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+
+    write(&client, counter, "fresh", expiry).await;
+
+    let left = remaining(client.ttl("fresh").await.unwrap());
+    assert!(
+        left > Duration::from_secs(50) && left <= Duration::from_secs(60),
+        "{left:?}"
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn keep_existing_leaves_a_running_ttl_and_refresh_resets_it(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(Counter::IncrBy, Counter::IncrByFloat, Counter::Rpush)] counter: Counter,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    let long = Duration::from_secs(600);
+    write(
+        &client,
+        counter,
+        "kept",
+        Expiry::Refresh(Duration::from_secs(30)),
+    )
+    .await;
+    write(
+        &client,
+        counter,
+        "refreshed",
+        Expiry::Refresh(Duration::from_secs(30)),
+    )
+    .await;
+
+    write(&client, counter, "kept", Expiry::KeepExisting(long)).await;
+    write(&client, counter, "refreshed", Expiry::Refresh(long)).await;
+
+    assert!(remaining(client.ttl("kept").await.unwrap()) <= Duration::from_secs(30));
+    assert!(remaining(client.ttl("refreshed").await.unwrap()) > Duration::from_secs(500));
+}
+
+#[rstest]
+#[tokio::test]
+async fn keep_existing_expires_a_key_that_had_no_ttl(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(Counter::IncrBy, Counter::IncrByFloat, Counter::Rpush)] counter: Counter,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "ns", DEFAULT_TIMEOUT);
+    persist_as(&client, counter, "persistent").await;
+
+    write(&client, counter, "persistent", KEEP).await;
+
+    let left = remaining(client.ttl("persistent").await.unwrap());
+    assert!(left > Duration::from_secs(50), "{left:?}");
+}
+
+#[derive(Default)]
+struct Recorded(Mutex<Vec<CallEvent>>);
+
+#[rstest]
+#[tokio::test]
+async fn atomic_writes_require_scripting_permissions(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(Counter::IncrBy, Counter::IncrByFloat, Counter::Rpush)] counter: Counter,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    server
+        .cli_on_every_node(&["ACL", "SETUSER", "default", "-@scripting"])
+        .await
+        .unwrap();
+    client
+        .set("ordinary", "allowed", Duration::from_secs(60))
+        .await
+        .unwrap();
+    assert_eq!(
+        client.get::<String>("ordinary").await.unwrap().as_deref(),
+        Some("allowed")
+    );
+    let denied = match counter {
+        Counter::IncrBy => client.incr_by("atomic", 1, KEEP).await.map(drop),
+        Counter::IncrByFloat => client.incr_by_float("atomic", 0.5, KEEP).await.map(drop),
+        Counter::Rpush => client.rpush("atomic", "item", KEEP).await.map(drop),
+    };
+    assert!(matches!(denied, Err(Error::Server(_))));
+    assert_eq!(client.get::<String>("atomic").await.unwrap(), None);
+    server
+        .cli_on_every_node(&["ACL", "SETUSER", "default", "+@scripting"])
+        .await
+        .unwrap();
+    write(&client, counter, "atomic", KEEP).await;
+    assert!(remaining(client.ttl("atomic").await.unwrap()) > Duration::ZERO);
+}
+
+#[rstest]
+#[tokio::test]
+async fn an_integer_reply_acl_failure_does_not_change_the_counter(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(false, true)] pipelined: bool,
+    #[values(None, Some(5))] initial: Option<i64>,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    if let Some(value) = initial {
+        client
+            .set("atomic", value, Duration::from_secs(60))
+            .await
+            .unwrap();
+    }
+    server
+        .cli_on_every_node(&["ACL", "SETUSER", "default", "-get"])
+        .await
+        .unwrap();
+    let result = if pipelined {
+        client
+            .run(&client.pipeline().incr_by("atomic", 1, KEEP))
+            .await
+            .remove(0)
+            .map(drop)
+    } else {
+        client.incr_by("atomic", 1, KEEP).await.map(drop)
+    };
+    assert!(matches!(result, Err(Error::Server(_))));
+    server
+        .cli_on_every_node(&["ACL", "SETUSER", "default", "+get"])
+        .await
+        .unwrap();
+    assert_eq!(client.get::<i64>("atomic").await.unwrap(), initial);
+    if initial.is_some() {
+        assert!(remaining(client.ttl("atomic").await.unwrap()) > Duration::ZERO);
+    }
+}
+
+#[rstest]
+#[case::beyond_double_precision(9_007_199_254_740_992_i64, 1_i64)]
+#[case::negative_beyond_double_precision(-9_007_199_254_740_992_i64, -1_i64)]
+#[case::maximum_integer(i64::MAX - 1, 1_i64)]
+#[case::minimum_integer(i64::MIN + 1, -1_i64)]
+#[tokio::test]
+async fn integer_increments_preserve_every_bit(
+    #[case] initial: i64,
+    #[case] delta: i64,
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(false, true)] pipelined: bool,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    client
+        .set("integer", initial, Duration::from_secs(60))
+        .await
+        .unwrap();
+    let expected = initial + delta;
+    let result = if pipelined {
+        client
+            .run(&client.pipeline().incr_by("integer", delta, KEEP))
+            .await
+            .remove(0)
+            .unwrap()
+    } else {
+        litellm_redis_client::Value::Int(client.incr_by("integer", delta, KEEP).await.unwrap())
+    };
+    assert_eq!(result, litellm_redis_client::Value::Int(expected));
+    assert_eq!(client.get::<i64>("integer").await.unwrap(), Some(expected));
+}
+
+#[rstest]
+#[tokio::test]
+async fn outage_events_report_the_failure_kind(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(false, true)] timeout: bool,
+    #[values(false, true)] pipeline: bool,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let recorded = Arc::new(Recorded::default());
+    let client = RedisClient::new(RedisConfig {
+        endpoint: endpoint(&server),
+        namespace: Namespace::default(),
+        timeouts: Timeouts::new(Duration::from_millis(300), None),
+        breaker: None,
+        events: Some(recorded.clone()),
+    })
+    .unwrap();
+    client.ping().await.unwrap();
+    if timeout {
+        server.freeze().await.unwrap();
+    } else {
+        server.stop().await.unwrap();
+    }
+    let result = if pipeline {
+        client
+            .run(&client.pipeline().get("missing"))
+            .await
+            .remove(0)
+            .map(drop)
+    } else {
+        client.get::<String>("missing").await.map(drop)
+    };
+    if timeout {
+        server.thaw().await.unwrap();
+    } else {
+        server.resume().await.unwrap();
+    }
+    assert!(matches!(result, Err(Error::Transport(_))));
+    let expected = if timeout {
+        FailureKind::Timeout
+    } else {
+        FailureKind::Transport
+    };
+    assert_eq!(
+        recorded.0.lock().unwrap().last().unwrap().outcome,
+        Outcome::Failed(expected)
+    );
+}
+
+impl RedisEvents for Recorded {
+    fn call(&self, event: &CallEvent) {
+        self.0.lock().unwrap().push(*event);
+    }
+}
+
+#[rstest]
+#[tokio::test]
+async fn every_call_reports_its_operation_and_outcome(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let recorded = Arc::new(Recorded::default());
+    let client = RedisClient::new(RedisConfig {
+        endpoint: endpoint(&server),
+        namespace: Namespace::default(),
+        timeouts: Timeouts::default(),
+        breaker: None,
+        events: Some(recorded.clone()),
+    })
+    .unwrap();
+    client.rpush("list", "item", KEEP).await.unwrap();
+
+    let _ = client.incr_by("list", 1, KEEP).await;
+    let _ = client.get::<String>("missing").await;
+    let _ = client.run(&client.pipeline().get("a").get("b")).await;
+    let failures = client
+        .run(&client.pipeline().incr_by("list", 1, KEEP))
+        .await;
+    assert!(matches!(failures[0], Err(Error::Server(_))));
+
+    let seen: Vec<(Operation, Outcome)> = recorded
+        .0
+        .lock()
+        .unwrap()
+        .iter()
+        .map(|event| (event.operation, event.outcome))
+        .collect();
+    assert_eq!(
+        seen,
+        vec![
+            (Operation::Command("RPUSH"), Outcome::Succeeded),
+            (
+                Operation::Command("INCRBY"),
+                Outcome::Failed(FailureKind::Server)
+            ),
+            (Operation::Command("GET"), Outcome::Succeeded),
+            (Operation::Pipeline { ops: 2 }, Outcome::Succeeded),
+            (
+                Operation::Pipeline { ops: 1 },
+                Outcome::Failed(FailureKind::Server)
+            ),
+        ]
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn namespaces_isolate_keys_on_the_server(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let a = connect(&server, "a", DEFAULT_TIMEOUT);
+    let b = connect(&server, "b", DEFAULT_TIMEOUT);
+    let bare = connect(&server, "", DEFAULT_TIMEOUT);
+
+    a.set("x", "1", Duration::from_secs(60)).await.unwrap();
+
+    assert_eq!(a.get::<String>("x").await.unwrap().as_deref(), Some("1"));
+    assert_eq!(a.get::<String>("a:x").await.unwrap().as_deref(), Some("1"));
+    assert_eq!(
+        bare.get::<String>("a:x").await.unwrap().as_deref(),
+        Some("1")
+    );
+    assert_eq!(b.get::<String>("x").await.unwrap(), None);
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_rejected_command_keeps_the_connection(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    client.rpush("list", "item", KEEP).await.unwrap();
+    let before = connections(&server).await;
+
+    let rejected = client.incr_by("list", 1, KEEP).await;
+
+    assert!(matches!(rejected, Err(Error::Server(_))), "{rejected:?}");
+    assert_eq!(client.rpush("list", "next", KEEP).await.unwrap(), 2);
+    assert_eq!(connections(&server).await, before);
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_reply_of_another_type_is_a_decode_error_and_a_missing_key_is_none(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    client
+        .set("float", "15.0", Duration::from_secs(60))
+        .await
+        .unwrap();
+
+    let decoded = client.get::<i64>("float").await;
+
+    assert!(matches!(decoded, Err(Error::Decode(_))), "{decoded:?}");
+    assert_eq!(client.get::<i64>("missing").await.unwrap(), None);
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_timed_out_command_does_not_leak_its_reply_into_the_next_one(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", Duration::from_millis(500));
+    client
+        .set("first", "1", Duration::from_secs(60))
+        .await
+        .unwrap();
+    client
+        .set("second", "2", Duration::from_secs(60))
+        .await
+        .unwrap();
+
+    server.freeze().await.unwrap();
+    let timed_out = client.get::<String>("first").await;
+    server.thaw().await.unwrap();
+
+    assert!(
+        matches!(timed_out, Err(Error::Transport(_))),
+        "{timed_out:?}"
+    );
+    assert_eq!(
+        client.get::<String>("second").await.unwrap().as_deref(),
+        Some("2")
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn the_same_client_recovers_after_redis_restarts(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", Duration::from_millis(500));
+
+    server.restart().await.unwrap();
+
+    let recovered = eventually(async || client.incr_by("after_restart", 1, KEEP).await.ok()).await;
+    assert_eq!(recovered, 1);
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_client_created_while_redis_is_down_works_once_redis_is_up(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    server.stop().await.unwrap();
+
+    let client = connect(&server, "", Duration::from_millis(500));
+    let started = Instant::now();
+    let while_down = client.ping().await;
+    let failed_in = started.elapsed();
+    server.resume().await.unwrap();
+
+    assert!(failed_in < Duration::from_secs(2), "{failed_in:?}");
+
+    assert!(
+        matches!(while_down, Err(Error::Transport(_))),
+        "{while_down:?}"
+    );
+    let recovered = eventually(async || client.ping().await.ok()).await;
+    assert_eq!(recovered, ());
+}
+
+#[rstest]
+#[tokio::test]
+async fn empty_batches_succeed_without_asking_redis(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+
+    assert_eq!(
+        client.mget::<String>(&[]).await.unwrap(),
+        Vec::<Option<String>>::new()
+    );
+    assert_eq!(client.del(&[]).await.unwrap(), 0);
+}
+
+#[rstest]
+#[tokio::test]
+async fn mget_and_del_span_slots_in_input_order(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    client.set("a", "1", Duration::from_secs(60)).await.unwrap();
+    client.set("c", "3", Duration::from_secs(60)).await.unwrap();
+
+    let values = client.mget::<String>(&["c", "b", "a"]).await.unwrap();
+    let deleted = client.del(&["a", "b", "c"]).await.unwrap();
+
+    assert_eq!(
+        values,
+        vec![Some("3".to_owned()), None, Some("1".to_owned())]
+    );
+    assert_eq!(deleted, 2);
+    assert_eq!(
+        client.mget::<String>(&["a", "c"]).await.unwrap(),
+        vec![None, None]
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn set_nx_only_writes_a_missing_key(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+
+    let first = client
+        .set_nx("lock", "owner-1", Duration::from_secs(60))
+        .await
+        .unwrap();
+    let second = client
+        .set_nx("lock", "owner-2", Duration::from_secs(60))
+        .await
+        .unwrap();
+
+    assert!(first);
+    assert!(!second);
+    assert_eq!(
+        client.get::<String>("lock").await.unwrap().as_deref(),
+        Some("owner-1")
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn ttl_reports_missing_persistent_and_remaining_time(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    persist(&client, "persistent").await;
+    client
+        .set("expiring", "1", Duration::from_secs(60))
+        .await
+        .unwrap();
+    client
+        .incr_by_float("expired_later", 0.5, KEEP)
+        .await
+        .unwrap();
+    assert!(
+        client
+            .expire("expired_later", Duration::from_secs(30))
+            .await
+            .unwrap()
+    );
+
+    assert_eq!(client.ttl("missing").await.unwrap(), Ttl::Missing);
+    assert_eq!(client.ttl("persistent").await.unwrap(), Ttl::Persistent);
+    assert!(matches!(
+        client.ttl("expiring").await.unwrap(),
+        Ttl::Expires(left) if left > Duration::from_secs(50) && left <= Duration::from_secs(60)
+    ));
+    assert!(matches!(
+        client.ttl("expired_later").await.unwrap(),
+        Ttl::Expires(left) if left > Duration::from_secs(20) && left <= Duration::from_secs(30)
+    ));
+    assert!(
+        !client
+            .expire("missing", Duration::from_secs(30))
+            .await
+            .unwrap()
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn incr_by_float_accumulates_fractions(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+
+    client.incr_by_float("spend", 0.25, KEEP).await.unwrap();
+
+    assert_eq!(
+        client.incr_by_float("spend", 0.5, KEEP).await.unwrap(),
+        0.75
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn concurrent_commands_on_one_cloned_client_all_apply(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+
+    let tasks: Vec<_> = (0..100)
+        .map(|_| {
+            let client = client.clone();
+            tokio::spawn(async move {
+                for _ in 0..10 {
+                    client.incr_by("shared", 1, KEEP).await.unwrap();
+                }
+            })
+        })
+        .collect();
+    for task in tasks {
+        task.await.unwrap();
+    }
+
+    assert_eq!(client.get::<i64>("shared").await.unwrap(), Some(1000));
+}
+
+fn breaker(failure_threshold: u32) -> Breaker {
+    Breaker::new(
+        Config {
+            failure_threshold,
+            recovery_timeout: Duration::from_millis(200),
+            timeout_min_duration: Duration::ZERO,
+        },
+        Box::new(|_| {}),
+    )
+}
+
+#[rstest]
+#[tokio::test]
+async fn an_open_breaker_rejects_without_waiting_on_redis_and_a_probe_closes_it(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let breaker = breaker(2);
+    let client = connect_guarded(
+        &server,
+        "",
+        Duration::from_millis(300),
+        Some(breaker.clone()),
+    );
+
+    server.freeze().await.unwrap();
+    let first = client.get::<String>("k").await;
+    let second = client.get::<String>("k").await;
+    let started = Instant::now();
+    let third = client.get::<String>("k").await;
+    let rejected_in = started.elapsed();
+    server.thaw().await.unwrap();
+
+    assert!(matches!(first, Err(Error::Transport(_))), "{first:?}");
+    assert!(matches!(second, Err(Error::Transport(_))), "{second:?}");
+    assert!(matches!(third, Err(Error::BreakerOpen)), "{third:?}");
+    assert!(rejected_in < Duration::from_millis(50), "{rejected_in:?}");
+
+    let recovered = eventually(async || client.get::<String>("k").await.ok()).await;
+    assert_eq!(recovered, None);
+    assert!(
+        matches!(breaker.state(), State::Closed { .. }),
+        "{:?}",
+        breaker.state()
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn rejected_commands_never_open_the_breaker(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let breaker = breaker(2);
+    let client = connect_guarded(&server, "", DEFAULT_TIMEOUT, Some(breaker.clone()));
+    client.rpush("list", "item", KEEP).await.unwrap();
+
+    let rejected = [
+        client.incr_by("list", 1, KEEP).await,
+        client.incr_by("list", 1, KEEP).await,
+        client.incr_by("list", 1, KEEP).await,
+    ];
+
+    assert!(
+        matches!(rejected[2], Err(Error::Server(_))),
+        "{:?}",
+        rejected[2]
+    );
+    assert!(
+        matches!(breaker.state(), State::Closed { hard: 0, .. }),
+        "{:?}",
+        breaker.state()
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn an_open_breaker_fails_every_pipeline_op_without_sending(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect_guarded(&server, "", Duration::from_millis(300), Some(breaker(1)));
+    let pipeline = client
+        .pipeline()
+        .incr_by("a", 1, KEEP)
+        .incr_by("c", 1, KEEP);
+
+    server.freeze().await.unwrap();
+    let failed = client.run(&pipeline).await;
+    let rejected = client.run(&pipeline).await;
+    server.thaw().await.unwrap();
+
+    assert!(
+        matches!(failed[0], Err(Error::Transport(_))),
+        "{:?}",
+        failed[0]
+    );
+    assert!(
+        matches!(rejected[0], Err(Error::BreakerOpen)),
+        "{:?}",
+        rejected[0]
+    );
+    assert!(
+        matches!(rejected[1], Err(Error::BreakerOpen)),
+        "{:?}",
+        rejected[1]
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_host_endpoint_connects_to_its_database() {
+    let server = RedisServer::start().await.unwrap();
+    let port = server.url().rsplit(':').next().unwrap().parse().unwrap();
+    let connect = |endpoint| {
+        RedisClient::new(RedisConfig {
+            endpoint,
+            namespace: Namespace::default(),
+            timeouts: Timeouts::default(),
+            breaker: None,
+            events: None,
+        })
+    };
+    let url = |url: String| Endpoint::Url(SecretValue::new(url));
+    let in_db_3 = connect(Endpoint::Host(HostEndpoint {
+        host: "127.0.0.1".to_owned(),
+        port,
+        username: None,
+        password: None,
+        tls: false,
+        db: 3,
+    }))
+    .unwrap();
+    let in_db_0 = connect(url(server.url().to_owned())).unwrap();
+    let via_url = connect(url(format!("{}/3", server.url()))).unwrap();
+
+    in_db_3
+        .set("k", "3", Duration::from_secs(60))
+        .await
+        .unwrap();
+
+    assert_eq!(in_db_0.get::<String>("k").await.unwrap(), None);
+    assert_eq!(
+        via_url.get::<String>("k").await.unwrap().as_deref(),
+        Some("3")
+    );
+}

litellm-rust/crates/redis-client/tests/connection.rs

@@ -0,0 +1,63 @@
+mod support;
+
+use std::time::Duration;
+
+use litellm_redis_client::{Endpoint, Error, HostEndpoint};
+use litellm_redis_testing::RedisServer;
+use litellm_secret_value::SecretValue;
+use rstest::rstest;
+
+use support::connect_to;
+
+#[rstest]
+#[tokio::test]
+async fn an_out_of_range_database_is_a_server_rejection() {
+    let server = RedisServer::start().await.unwrap();
+    let port = server.url().rsplit(':').next().unwrap().parse().unwrap();
+    let client = connect_to(
+        Endpoint::Host(HostEndpoint {
+            host: "127.0.0.1".into(),
+            port,
+            username: None,
+            password: None,
+            tls: false,
+            db: 999,
+        }),
+        "",
+        Duration::from_secs(2),
+        None,
+    );
+    let result = client.ping().await;
+    assert!(matches!(result, Err(Error::Server(_))), "{result:?}");
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_rejected_password_is_a_transport_failure() {
+    let server = RedisServer::start().await.unwrap();
+    server
+        .cli(&["ACL", "SETUSER", "default", "on", ">right"])
+        .await
+        .unwrap();
+    let port = server.url().rsplit(':').next().unwrap().parse().unwrap();
+    let with_password = |password: &str| {
+        connect_to(
+            Endpoint::Host(HostEndpoint {
+                host: "127.0.0.1".to_owned(),
+                port,
+                username: None,
+                password: Some(SecretValue::new(password)),
+                tls: false,
+                db: 0,
+            }),
+            "",
+            Duration::from_secs(2),
+            None,
+        )
+    };
+    with_password("right").ping().await.unwrap();
+
+    let rejected = with_password("wrong").ping().await;
+
+    assert!(matches!(rejected, Err(Error::Transport(_))), "{rejected:?}");
+}

litellm-rust/crates/redis-client/tests/pipeline.rs

@@ -0,0 +1,242 @@
+mod support;
+
+use std::time::{Duration, Instant};
+
+use litellm_redis_client::Value;
+use litellm_redis_testing::{RedisDeployment, Topology};
+use rstest::rstest;
+
+use litellm_redis_client::{DEFAULT_TIMEOUT, Error, Script};
+use support::{KEEP, connect};
+
+async fn begin_migration(cluster: &litellm_redis_testing::RedisCluster, key: &str) {
+    let (_, first) = cluster.nodes().next().unwrap();
+    let slot = cluster
+        .cli_on_node(first, &["CLUSTER", "KEYSLOT", key])
+        .await
+        .unwrap();
+    let slot = slot.trim();
+    let number: u16 = slot.parse().unwrap();
+    let nodes = cluster
+        .cli_on_node(first, &["CLUSTER", "NODES"])
+        .await
+        .unwrap();
+    let owner = nodes.lines().find(|line| line.split_whitespace().skip(8).any(|range| {
+        let (start, end) = range.split_once('-').unwrap_or((range, range));
+        matches!((start.parse::<u16>(), end.parse::<u16>()), (Ok(start), Ok(end)) if (start..=end).contains(&number))
+    })).unwrap();
+    let target = nodes.lines().find(|line| *line != owner).unwrap();
+    let address = |line: &str| {
+        let value = line.split_whitespace().nth(1).unwrap();
+        value
+            .split_once('@')
+            .unwrap()
+            .0
+            .rsplit_once(':')
+            .unwrap()
+            .1
+            .parse::<u16>()
+            .unwrap()
+    };
+    let source_port = address(owner);
+    let target_port = address(target);
+    let source_id = owner.split_whitespace().next().unwrap();
+    let target_id = target.split_whitespace().next().unwrap();
+    assert_eq!(
+        cluster
+            .cli_on_node(source_port, &["GET", key])
+            .await
+            .unwrap()
+            .trim(),
+        "0"
+    );
+    assert_eq!(
+        cluster
+            .cli_on_node(
+                target_port,
+                &["CLUSTER", "SETSLOT", slot, "IMPORTING", source_id]
+            )
+            .await
+            .unwrap()
+            .trim(),
+        "OK"
+    );
+    assert_eq!(
+        cluster
+            .cli_on_node(
+                source_port,
+                &["CLUSTER", "SETSLOT", slot, "MIGRATING", target_id]
+            )
+            .await
+            .unwrap()
+            .trim(),
+        "OK"
+    );
+    let state = cluster
+        .cli_on_node(source_port, &["CLUSTER", "NODES"])
+        .await
+        .unwrap();
+    assert!(
+        state.contains(&format!("[{slot}->-{target_id}]")),
+        "{state}"
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_retrying_neighbor_cannot_reapply_a_cluster_counter() {
+    let cluster = litellm_redis_testing::RedisCluster::start().await.unwrap();
+    let server = RedisDeployment::Cluster(cluster);
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    let counter = "{migration}:counter";
+    client
+        .set(counter, 0, Duration::from_secs(60))
+        .await
+        .unwrap();
+    let RedisDeployment::Cluster(cluster) = &server else {
+        unreachable!()
+    };
+    begin_migration(cluster, counter).await;
+    let both_keys = Script::new("return redis.call('MGET', KEYS[1], KEYS[2])");
+
+    let results = client
+        .run(&client.pipeline().incr_by(counter, 1, KEEP).eval(
+            &both_keys,
+            &[counter, "{migration}:missing"],
+            Vec::<String>::new(),
+        ))
+        .await;
+
+    assert_eq!(client.get::<i64>(counter).await.unwrap(), Some(1));
+    assert_eq!(results[0].as_ref().unwrap(), &Value::Int(1));
+    assert!(
+        matches!(&results[1], Err(Error::Server(error)) if error.code() == Some("TRYAGAIN")),
+        "{:?}",
+        results[1]
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn each_op_settles_on_its_own_in_input_order(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    client
+        .set("text", "not a number", Duration::from_secs(60))
+        .await
+        .unwrap();
+    let pipeline = client
+        .pipeline()
+        .incr_by("a", 1, KEEP)
+        .incr_by("text", 1, KEEP)
+        .incr_by("c", 5, KEEP)
+        .set("b", "2", Duration::from_secs(60))
+        .get("b")
+        .incr_by("a", 1, KEEP)
+        .get("missing");
+
+    let results = client.run(&pipeline).await;
+
+    assert_eq!(results.len(), 7);
+    assert_eq!(results[0].as_ref().unwrap(), &Value::Int(1));
+    assert!(
+        matches!(results[1], Err(Error::Server(_))),
+        "{:?}",
+        results[1]
+    );
+    assert_eq!(results[2].as_ref().unwrap(), &Value::Int(5));
+    assert_eq!(results[3].as_ref().unwrap(), &Value::Okay);
+    assert_eq!(
+        results[4].as_ref().unwrap(),
+        &Value::BulkString(b"2".to_vec())
+    );
+    assert_eq!(results[5].as_ref().unwrap(), &Value::Int(2));
+    assert_eq!(results[6].as_ref().unwrap(), &Value::Nil);
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_script_runs_in_pipeline_order_on_first_use_and_after_a_flush(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(false, true)] flushed: bool,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "ns", DEFAULT_TIMEOUT);
+    let incr = Script::new("return redis.call('INCRBY', KEYS[1], ARGV[1])");
+    if flushed {
+        client.eval::<i64>(&incr, &["warmup"], 1).await.unwrap();
+        server
+            .cli_on_every_node(&["SCRIPT", "FLUSH"])
+            .await
+            .unwrap();
+    }
+    let pipeline = client
+        .pipeline()
+        .eval(&incr, &["counter"], 2)
+        .get("counter")
+        .incr_by("counter", 5, KEEP)
+        .get("counter");
+
+    let results = client.run(&pipeline).await;
+
+    assert_eq!(results[0].as_ref().unwrap(), &Value::Int(2));
+    assert_eq!(
+        results[1].as_ref().unwrap(),
+        &Value::BulkString(b"2".to_vec())
+    );
+    assert_eq!(results[2].as_ref().unwrap(), &Value::Int(7));
+    assert_eq!(
+        results[3].as_ref().unwrap(),
+        &Value::BulkString(b"7".to_vec())
+    );
+    assert_eq!(client.get::<i64>("counter").await.unwrap(), Some(7));
+}
+
+#[rstest]
+#[tokio::test]
+async fn an_unreachable_server_fails_every_op_as_transport(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", Duration::from_millis(500));
+    let pipeline = client
+        .pipeline()
+        .incr_by("a", 1, KEEP)
+        .incr_by("c", 1, KEEP);
+
+    server.freeze().await.unwrap();
+    let results = client.run(&pipeline).await;
+    server.thaw().await.unwrap();
+
+    assert_eq!(results.len(), 2);
+    assert!(
+        matches!(results[0], Err(Error::Transport(_))),
+        "{:?}",
+        results[0]
+    );
+    assert!(
+        matches!(results[1], Err(Error::Transport(_))),
+        "{:?}",
+        results[1]
+    );
+}
+
+#[rstest]
+#[tokio::test]
+async fn an_empty_pipeline_returns_without_asking_redis(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", Duration::from_secs(2));
+
+    server.freeze().await.unwrap();
+    let started = Instant::now();
+    let results = client.run(&client.pipeline()).await;
+    let elapsed = started.elapsed();
+    server.thaw().await.unwrap();
+
+    assert!(results.is_empty());
+    assert!(elapsed < Duration::from_secs(1), "{elapsed:?}");
+}

litellm-rust/crates/redis-client/tests/script.rs

@@ -0,0 +1,112 @@
+mod support;
+
+use std::time::Duration;
+
+use litellm_redis_testing::{RedisDeployment, Topology};
+use rstest::rstest;
+
+use litellm_redis_client::Script;
+use litellm_redis_client::{DEFAULT_TIMEOUT, Error, Ttl};
+use support::{KEEP, connect};
+
+#[rstest]
+#[tokio::test]
+async fn script_keys_are_namespaced(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "a", DEFAULT_TIMEOUT);
+    let bare = connect(&server, "", DEFAULT_TIMEOUT);
+    let read = Script::new("return redis.call('GET', KEYS[1])");
+    bare.set("a:x", "1", Duration::from_secs(60)).await.unwrap();
+
+    let value: Option<String> = client
+        .eval(&read, &["x"], Vec::<String>::new())
+        .await
+        .unwrap();
+
+    assert_eq!(value.as_deref(), Some("1"));
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_failing_script_is_a_server_error_and_keeps_the_connection(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    let incr = Script::new("return redis.call('INCRBY', KEYS[1], ARGV[1])");
+    client.rpush("list", "item", KEEP).await.unwrap();
+
+    let failed = client.eval::<i64>(&incr, &["list"], 1).await;
+
+    assert!(matches!(failed, Err(Error::Server(_))), "{failed:?}");
+    assert_eq!(client.eval::<i64>(&incr, &["counter"], 3).await.unwrap(), 3);
+}
+
+#[rstest]
+#[tokio::test]
+async fn a_flushed_script_is_loaded_again(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "", DEFAULT_TIMEOUT);
+    let incr = Script::new("return redis.call('INCRBY', KEYS[1], ARGV[1])");
+    client.eval::<i64>(&incr, &["counter"], 1).await.unwrap();
+
+    server
+        .cli_on_every_node(&["SCRIPT", "FLUSH"])
+        .await
+        .unwrap();
+
+    assert_eq!(client.eval::<i64>(&incr, &["counter"], 1).await.unwrap(), 2);
+}
+
+const INCREMENT_WITH_FLOOR: &str = "local count = redis.call('INCRBY', KEYS[1], ARGV[1]) \
+if count < 0 then count = redis.call('INCRBY', KEYS[1], -count) end \
+if redis.call('TTL', KEYS[1]) < 0 then redis.call('EXPIRE', KEYS[1], ARGV[2]) end \
+return count";
+
+#[rstest]
+#[tokio::test]
+async fn pythons_increment_with_floor_script_floors_at_zero_and_expires_in_one_call(
+    #[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
+    #[values(false, true)] pipelined: bool,
+) {
+    let server = RedisDeployment::start(topology).await.unwrap();
+    let client = connect(&server, "ns", DEFAULT_TIMEOUT);
+    let floor = Script::new(INCREMENT_WITH_FLOOR);
+    server
+        .cli_on_every_node(&["SCRIPT", "FLUSH"])
+        .await
+        .unwrap();
+
+    let count = match pipelined {
+        false => client
+            .eval::<i64>(&floor, &["slots"], (-5, 60))
+            .await
+            .unwrap(),
+        true => {
+            let results = client
+                .run(
+                    &client
+                        .pipeline()
+                        .eval(&floor, &["slots"], (-5, 60))
+                        .get("slots"),
+                )
+                .await;
+            assert_eq!(
+                results[1].as_ref().unwrap(),
+                &redis::Value::BulkString(b"0".to_vec())
+            );
+            redis::from_redis_value(results[0].as_ref().unwrap().clone()).unwrap()
+        }
+    };
+
+    assert_eq!(count, 0);
+    assert_eq!(client.get::<i64>("slots").await.unwrap(), Some(0));
+    assert!(matches!(
+        client.ttl("slots").await.unwrap(),
+        Ttl::Expires(left) if left > Duration::from_secs(50) && left <= Duration::from_secs(60)
+    ));
+}

litellm-rust/crates/redis-client/tests/support/mod.rs

@@ -0,0 +1,63 @@
+#![allow(dead_code)] // each test binary compiles this module on its own and uses a different subset
+
+use std::time::Duration;
+
+use litellm_circuit_breaker::Breaker;
+use litellm_redis_client::{
+    ClusterEndpoint, ClusterNode, Endpoint, Expiry, Namespace, RedisClient, RedisConfig, Timeouts,
+};
+use litellm_redis_testing::{ClientConnection, RedisDeployment};
+use litellm_secret_value::SecretValue;
+
+pub const KEEP: Expiry = Expiry::KeepExisting(Duration::from_secs(60));
+
+pub fn endpoint(deployment: &RedisDeployment) -> Endpoint {
+    match deployment {
+        RedisDeployment::Standalone(server) => Endpoint::Url(SecretValue::new(server.url())),
+        RedisDeployment::Cluster(cluster) => Endpoint::Cluster(ClusterEndpoint {
+            nodes: cluster
+                .nodes()
+                .map(|(host, port)| ClusterNode {
+                    host: host.to_owned(),
+                    port,
+                })
+                .collect(),
+            username: None,
+            password: None,
+            tls: false,
+        }),
+    }
+}
+
+pub fn connect(deployment: &RedisDeployment, namespace: &str, timeout: Duration) -> RedisClient {
+    connect_guarded(deployment, namespace, timeout, None)
+}
+
+pub fn connect_guarded(
+    deployment: &RedisDeployment,
+    namespace: &str,
+    timeout: Duration,
+    breaker: Option<Breaker>,
+) -> RedisClient {
+    connect_to(endpoint(deployment), namespace, timeout, breaker)
+}
+
+pub fn connect_to(
+    endpoint: Endpoint,
+    namespace: &str,
+    timeout: Duration,
+    breaker: Option<Breaker>,
+) -> RedisClient {
+    RedisClient::new(RedisConfig {
+        endpoint,
+        namespace: Namespace::new(namespace),
+        timeouts: Timeouts::new(timeout, None),
+        breaker,
+        events: None,
+    })
+    .unwrap()
+}
+
+pub async fn connections(deployment: &RedisDeployment) -> Vec<ClientConnection> {
+    deployment.clients().await.unwrap()
+}

litellm-rust/crates/redis-testing/AGENTS.md

@@ -4,3 +4,7 @@
 - `RedisCluster` runs 3 masters in one container. Each node listens on the same port inside and outside and announces 127.0.0.1, so MOVED redirects work from the test process
 - Outages come from the container (`restart`, `freeze` / `thaw`), never `DEBUG`, which the official image disables
 - No env gate that makes a test pass without Redis. A test that can't start its container fails
+- `cli` / `cli_on_every_node` run `redis-cli -e` inside the container for setup such as ACL users, so tests don't need a second client. `-e` makes an error reply fail the call, and a test whose setup changes server state also asserts that state before relying on it
+- Not `redis-test`'s server helpers, which need local `redis-server` and `openssl` binaries
+- `RedisDeployment` runs either `Topology`, so a test can take `#[values(Topology::Standalone, Topology::Cluster)]`. `clients()` returns every client connection on every node from `CLIENT LIST`, leaving out the CLI's own
+- Server fixtures only. This crate never depends on a Redis client crate, so the client crates can dev-depend on it without a cycle

litellm-rust/crates/redis-testing/Cargo.toml

@@ -8,4 +8,4 @@ repository.workspace = true
 [dependencies]
 testcontainers-modules = { workspace = true, features = ["redis"] }
 thiserror.workspace = true
-tokio = { workspace = true, features = ["io-util"] }
+tokio = { workspace = true, features = ["io-util", "time"] }

litellm-rust/crates/redis-testing/src/cluster.rs

@@ -5,7 +5,8 @@ use testcontainers_modules::testcontainers::{
 };
 
 use crate::{
-    Error,
+    ClientConnection, Error,
+    deployment::parse_client_list,
     ports::{HOST, start_on_free_ports, wait_for_pong},
     standalone::REDIS_TAG,
 };
@@ -70,6 +71,28 @@ impl RedisCluster {
         self.ports.iter().map(|&port| (HOST, port))
     }
 
+    pub async fn cli_on_every_node(&self, args: &[&str]) -> Result<(), Error> {
+        for port in self.ports {
+            crate::exec::run(&self.container, crate::exec::cli(port, args)).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn cli_on_node(&self, port: u16, args: &[&str]) -> Result<String, Error> {
+        crate::exec::run(&self.container, crate::exec::cli(port, args)).await
+    }
+
+    pub async fn clients(&self) -> Result<Vec<ClientConnection>, Error> {
+        let mut clients = Vec::new();
+        for port in self.ports {
+            let reply =
+                crate::exec::run(&self.container, crate::exec::cli(port, &["CLIENT", "LIST"]))
+                    .await?;
+            clients.extend(parse_client_list(port, &reply));
+        }
+        Ok(clients)
+    }
+
     pub async fn restart(&self) -> Result<(), Error> {
         self.stop().await?;
         self.resume().await

litellm-rust/crates/redis-testing/src/deployment.rs

@@ -0,0 +1,113 @@
+use std::time::{Duration, Instant};
+
+use crate::{Error, RedisCluster, RedisServer};
+
+const EVENTUALLY_TIMEOUT: Duration = Duration::from_secs(30);
+
+#[derive(Clone, Copy, Debug)]
+pub enum Topology {
+    Standalone,
+    Cluster,
+}
+
+pub enum RedisDeployment {
+    Standalone(RedisServer),
+    Cluster(RedisCluster),
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ClientConnection {
+    pub node: u16,
+    pub id: u64,
+    pub user: String,
+}
+
+impl RedisDeployment {
+    pub async fn start(topology: Topology) -> Result<Self, Error> {
+        Ok(match topology {
+            Topology::Standalone => Self::Standalone(RedisServer::start().await?),
+            Topology::Cluster => Self::Cluster(RedisCluster::start().await?),
+        })
+    }
+
+    pub async fn cli_on_every_node(&self, args: &[&str]) -> Result<(), Error> {
+        match self {
+            Self::Standalone(server) => server.cli(args).await.map(drop),
+            Self::Cluster(cluster) => cluster.cli_on_every_node(args).await,
+        }
+    }
+
+    pub async fn clients(&self) -> Result<Vec<ClientConnection>, Error> {
+        match self {
+            Self::Standalone(server) => server.clients().await,
+            Self::Cluster(cluster) => cluster.clients().await,
+        }
+    }
+
+    pub async fn restart(&self) -> Result<(), Error> {
+        match self {
+            Self::Standalone(server) => server.restart().await,
+            Self::Cluster(cluster) => cluster.restart().await,
+        }
+    }
+
+    pub async fn stop(&self) -> Result<(), Error> {
+        match self {
+            Self::Standalone(server) => server.stop().await,
+            Self::Cluster(cluster) => cluster.stop().await,
+        }
+    }
+
+    pub async fn resume(&self) -> Result<(), Error> {
+        match self {
+            Self::Standalone(server) => server.resume().await,
+            Self::Cluster(cluster) => cluster.resume().await,
+        }
+    }
+
+    pub async fn freeze(&self) -> Result<(), Error> {
+        match self {
+            Self::Standalone(server) => server.freeze().await,
+            Self::Cluster(cluster) => cluster.freeze().await,
+        }
+    }
+
+    pub async fn thaw(&self) -> Result<(), Error> {
+        match self {
+            Self::Standalone(server) => server.thaw().await,
+            Self::Cluster(cluster) => cluster.thaw().await,
+        }
+    }
+}
+
+pub(crate) fn parse_client_list(node: u16, reply: &str) -> Vec<ClientConnection> {
+    reply
+        .lines()
+        .filter(|line| !line.split(' ').any(|field| field == "cmd=client|list"))
+        .filter_map(|line| {
+            let field = |name: &str| {
+                line.split(' ')
+                    .find_map(|field| field.strip_prefix(name)?.strip_prefix('='))
+            };
+            Some(ClientConnection {
+                node,
+                id: field("id")?.parse().ok()?,
+                user: field("user")?.to_owned(),
+            })
+        })
+        .collect()
+}
+
+pub async fn eventually<T>(mut attempt: impl AsyncFnMut() -> Option<T>) -> T {
+    let deadline = Instant::now() + EVENTUALLY_TIMEOUT;
+    loop {
+        if let Some(value) = attempt().await {
+            return value;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "condition not met within {EVENTUALLY_TIMEOUT:?}"
+        );
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}

litellm-rust/crates/redis-testing/src/error.rs

@@ -8,4 +8,6 @@ pub enum Error {
     Unreachable { port: u16 },
     #[error("controlling the Redis container")]
     Container(#[from] testcontainers_modules::testcontainers::TestcontainersError),
+    #[error("{command} exited with {code}")]
+    Command { command: String, code: i64 },
 }

litellm-rust/crates/redis-testing/src/exec.rs

@@ -0,0 +1,31 @@
+use testcontainers_modules::testcontainers::{
+    ContainerAsync, Image,
+    core::{CmdWaitFor, ExecCommand},
+};
+
+use crate::Error;
+
+pub(crate) async fn run<I: Image>(
+    container: &ContainerAsync<I>,
+    command: Vec<String>,
+) -> Result<String, Error> {
+    let mut result = container
+        .exec(ExecCommand::new(command.clone()).with_cmd_ready_condition(CmdWaitFor::exit()))
+        .await?;
+    let stdout = String::from_utf8_lossy(&result.stdout_to_vec().await?).into_owned();
+    match result.exit_code().await? {
+        Some(0) | None => Ok(stdout),
+        Some(code) => Err(Error::Command {
+            command: command.join(" "),
+            code,
+        }),
+    }
+}
+
+pub(crate) fn cli(port: u16, args: &[&str]) -> Vec<String> {
+    ["redis-cli", "-e", "-p", &port.to_string()]
+        .into_iter()
+        .chain(args.iter().copied())
+        .map(str::to_owned)
+        .collect()
+}

litellm-rust/crates/redis-testing/src/lib.rs

@@ -1,8 +1,11 @@
 mod cluster;
+mod deployment;
 mod error;
+mod exec;
 mod ports;
 mod standalone;
 
 pub use cluster::RedisCluster;
+pub use deployment::{ClientConnection, RedisDeployment, Topology, eventually};
 pub use error::Error;
 pub use standalone::RedisServer;

litellm-rust/crates/redis-testing/src/standalone.rs

@@ -5,7 +5,8 @@ use testcontainers_modules::{
 };
 
 use crate::{
-    Error,
+    ClientConnection, Error,
+    deployment::parse_client_list,
     ports::{HOST, start_on_free_ports, wait_for_pong},
 };
 
@@ -39,6 +40,17 @@ impl RedisServer {
         &self.url
     }
 
+    pub async fn cli(&self, args: &[&str]) -> Result<String, Error> {
+        crate::exec::run(&self.container, crate::exec::cli(REDIS_PORT, args)).await
+    }
+
+    pub async fn clients(&self) -> Result<Vec<ClientConnection>, Error> {
+        Ok(parse_client_list(
+            REDIS_PORT,
+            &self.cli(&["CLIENT", "LIST"]).await?,
+        ))
+    }
+
     pub async fn restart(&self) -> Result<(), Error> {
         self.stop().await?;
         self.resume().await