Merge pull request #41501 from BerriAI/litellm_1789598099-cache-redis

feat(rust): scaffold Redis cache crate

litellm-rust/Cargo.lock

@@ -70,6 +70,12 @@ dependencies = [
  "rustversion",
 ]
 
+[[package]]
+name = "arcstr"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d"
+
 [[package]]
 name = "async-compression"
 version = "0.4.46"
@@ -1837,6 +1843,12 @@ version = "0.2.186"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
+
 [[package]]
 name = "litellm-auth"
 version = "0.1.0"
@@ -1915,6 +1927,17 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "litellm-cache-redis"
+version = "0.1.0"
+dependencies = [
+ "litellm-cache",
+ "redis",
+ "redis-test",
+ "serde_json",
+ "tokio",
+]
+
 [[package]]
 name = "litellm-core"
 version = "0.1.0"
@@ -2140,6 +2163,16 @@ dependencies = [
  "minimal-lexical",
 ]
 
+[[package]]
+name = "num-bigint"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0"
+dependencies = [
+ "num-integer",
+ "num-traits",
+]
+
 [[package]]
 name = "num-conv"
 version = "0.2.2"
@@ -2656,6 +2689,36 @@ dependencies = [
  "crossbeam-utils",
 ]
 
+[[package]]
+name = "redis"
+version = "1.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acbc41a996f7652b2ddd9dfd98cc4ff602cfd742ae35382f07f608405ab50ed"
+dependencies = [
+ "arcstr",
+ "combine",
+ "itoa",
+ "num-bigint",
+ "percent-encoding",
+ "ryu",
+ "sha1_smol",
+ "socket2 0.6.5",
+ "url",
+ "xxhash-rust",
+]
+
+[[package]]
+name = "redis-test"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "804d36862e4323b69f96440cbb13c9894fc90176abdeaf91264e21d5d77f6aca"
+dependencies = [
+ "rand 0.9.5",
+ "redis",
+ "socket2 0.6.5",
+ "tempfile",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.18"
@@ -2846,6 +2909,19 @@ dependencies = [
  "semver",
 ]
 
+[[package]]
+name = "rustix"
+version = "1.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d"
+dependencies = [
+ "bitflags",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "rustls"
 version = "0.21.12"
@@ -3096,6 +3172,12 @@ dependencies = [
  "digest 0.10.7",
 ]
 
+[[package]]
+name = "sha1_smol"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d"
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -3299,6 +3381,19 @@ version = "0.13.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca"
 
+[[package]]
+name = "tempfile"
+version = "3.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
+dependencies = [
+ "fastrand",
+ "getrandom 0.4.3",
+ "once_cell",
+ "rustix",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "thiserror"
 version = "1.0.69"
@@ -4182,6 +4277,12 @@ version = "0.13.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4"
 
+[[package]]
+name = "xxhash-rust"
+version = "0.8.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6"
+
 [[package]]
 name = "yoke"
 version = "0.8.3"

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

@@ -0,0 +1,15 @@
+[package]
+name = "litellm-cache-redis"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+litellm-cache.workspace = true
+redis = "1.7.0"
+serde_json.workspace = true
+tokio.workspace = true
+
+[dev-dependencies]
+redis-test = "1.0.4"

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

@@ -0,0 +1,315 @@
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use litellm_cache::{
+    BaseCache, CacheConnectionResult, CacheConnectionStatus, CacheEntry, CacheFuture, CacheKwargs,
+    Error,
+};
+use redis::Commands;
+
+const DEFAULT_TTL: Duration = Duration::from_secs(600);
+const KEY_PREFIX: &str = "litellm-cache:";
+
+pub struct RedisCache<C = redis::Connection> {
+    connection: Arc<Mutex<C>>,
+    default_ttl: Duration,
+}
+
+impl RedisCache<redis::Connection> {
+    pub fn new(url: &str, default_ttl: Option<Duration>) -> Result<Self, Error> {
+        let client = redis::Client::open(url).map_err(|_| Error::Unavailable)?;
+        let connection = client.get_connection().map_err(|_| Error::Unavailable)?;
+        Ok(Self::with_connection(connection, default_ttl))
+    }
+}
+
+impl<C> RedisCache<C>
+where
+    C: redis::ConnectionLike + Send + 'static,
+{
+    fn with_connection(connection: C, default_ttl: Option<Duration>) -> Self {
+        Self {
+            connection: Arc::new(Mutex::new(connection)),
+            default_ttl: default_ttl.unwrap_or(DEFAULT_TTL),
+        }
+    }
+
+    fn connection(&self) -> Result<MutexGuard<'_, C>, Error> {
+        self.connection.lock().map_err(|_| Error::Unavailable)
+    }
+
+    fn namespaced_key(key: &str) -> String {
+        format!("{KEY_PREFIX}{key}")
+    }
+
+    fn namespaced_pattern() -> &'static str {
+        const PATTERN: &str = "litellm-cache:*";
+        PATTERN
+    }
+
+    fn encode(value: &CacheEntry) -> Result<Vec<u8>, Error> {
+        serde_json::to_vec(value).map_err(|_| Error::InvalidEntry)
+    }
+
+    fn decode(value: Vec<u8>) -> Result<CacheEntry, Error> {
+        serde_json::from_slice(&value).map_err(|_| Error::InvalidEntry)
+    }
+
+    fn ttl_seconds(ttl: Duration) -> u64 {
+        ttl.as_secs()
+            .saturating_add(u64::from(ttl.subsec_nanos() > 0))
+            .max(1)
+    }
+
+    fn run_blocking<T, F>(connection: Arc<Mutex<C>>, operation: F) -> CacheFuture<'static, T>
+    where
+        T: Send + 'static,
+        F: FnOnce(&mut C) -> Result<T, Error> + Send + 'static,
+    {
+        Box::pin(async move {
+            tokio::task::spawn_blocking(move || {
+                let mut connection = connection.lock().map_err(|_| Error::Unavailable)?;
+                operation(&mut connection)
+            })
+            .await
+            .map_err(|_| Error::Unavailable)?
+        })
+    }
+}
+
+impl<C> BaseCache for RedisCache<C>
+where
+    C: redis::ConnectionLike + Send + 'static,
+{
+    type Value = CacheEntry;
+
+    fn default_ttl(&self) -> Duration {
+        self.default_ttl
+    }
+
+    fn set_cache(&self, key: &str, value: Self::Value, kwargs: CacheKwargs) -> Result<(), Error> {
+        let payload = Self::encode(&value)?;
+        let ttl = Self::ttl_seconds(self.get_ttl(&kwargs));
+        self.connection()?
+            .set_ex::<_, _, ()>(Self::namespaced_key(key), payload, ttl)
+            .map_err(|_| Error::Unavailable)
+    }
+
+    fn get_cache(&self, key: &str, _: &CacheKwargs) -> Result<Option<Self::Value>, Error> {
+        self.connection()?
+            .get::<_, Option<Vec<u8>>>(Self::namespaced_key(key))
+            .map_err(|_| Error::Unavailable)?
+            .map(Self::decode)
+            .transpose()
+    }
+
+    fn delete_cache(&self, key: &str) -> Result<(), Error> {
+        self.connection()?
+            .del::<_, ()>(Self::namespaced_key(key))
+            .map_err(|_| Error::Unavailable)
+    }
+
+    fn flush_cache(&self) -> Result<(), Error> {
+        let mut connection = self.connection()?;
+        let keys = connection
+            .scan_match(Self::namespaced_pattern())
+            .map_err(|_| Error::Unavailable)?
+            .collect::<redis::RedisResult<Vec<String>>>()
+            .map_err(|_| Error::Unavailable)?;
+        if keys.is_empty() {
+            return Ok(());
+        }
+        connection
+            .del::<_, usize>(keys)
+            .map(|_| ())
+            .map_err(|_| Error::Unavailable)
+    }
+
+    fn async_set_cache<'a>(
+        &'a self,
+        key: &'a str,
+        value: Self::Value,
+        kwargs: CacheKwargs,
+    ) -> CacheFuture<'a, ()> {
+        let payload = Self::encode(&value);
+        let key = Self::namespaced_key(key);
+        let ttl = Self::ttl_seconds(self.get_ttl(&kwargs));
+        Self::run_blocking(Arc::clone(&self.connection), move |connection| {
+            connection
+                .set_ex::<_, _, ()>(key, payload?, ttl)
+                .map_err(|_| Error::Unavailable)
+        })
+    }
+
+    fn async_get_cache<'a>(
+        &'a self,
+        key: &'a str,
+        _: &'a CacheKwargs,
+    ) -> CacheFuture<'a, Option<Self::Value>> {
+        let key = Self::namespaced_key(key);
+        Box::pin(async move {
+            Self::run_blocking(Arc::clone(&self.connection), move |connection| {
+                connection
+                    .get::<_, Option<Vec<u8>>>(key)
+                    .map_err(|_| Error::Unavailable)
+            })
+            .await?
+            .map(Self::decode)
+            .transpose()
+        })
+    }
+
+    fn async_set_cache_pipeline<'a>(
+        &'a self,
+        cache_list: Vec<(String, Self::Value)>,
+        kwargs: CacheKwargs,
+    ) -> CacheFuture<'a, ()> {
+        let entries = cache_list
+            .into_iter()
+            .map(|(key, value)| {
+                Self::encode(&value).map(|payload| (Self::namespaced_key(&key), payload))
+            })
+            .collect::<Result<Vec<_>, _>>();
+        let ttl = Self::ttl_seconds(self.get_ttl(&kwargs));
+        Self::run_blocking(Arc::clone(&self.connection), move |connection| {
+            for (key, payload) in entries? {
+                connection
+                    .set_ex::<_, _, ()>(key, payload, ttl)
+                    .map_err(|_| Error::Unavailable)?;
+            }
+            Ok(())
+        })
+    }
+
+    fn async_delete_cache<'a>(&'a self, key: &'a str) -> CacheFuture<'a, ()> {
+        let key = Self::namespaced_key(key);
+        Self::run_blocking(Arc::clone(&self.connection), move |connection| {
+            connection.del::<_, ()>(key).map_err(|_| Error::Unavailable)
+        })
+    }
+
+    fn disconnect(&self) -> CacheFuture<'_, ()> {
+        Box::pin(async { Ok(()) })
+    }
+
+    fn test_connection(&self) -> CacheFuture<'_, CacheConnectionResult> {
+        Box::pin(async move {
+            Self::run_blocking(Arc::clone(&self.connection), |connection| {
+                redis::cmd("PING")
+                    .query::<String>(connection)
+                    .map_err(|_| Error::Unavailable)
+            })
+            .await?;
+            Ok(CacheConnectionResult {
+                status: CacheConnectionStatus::Success,
+                message: "Redis cache connection test successful".into(),
+                error: None,
+            })
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::RedisCache;
+    use litellm_cache::{BaseCache, CacheEntry, CacheKwargs};
+    use redis_test::{MockCmd, MockRedisConnection};
+    use serde_json::json;
+    use std::time::Duration;
+
+    fn entry() -> CacheEntry {
+        CacheEntry {
+            timestamp: 123.0,
+            response: json!({"choices": [{"text": "cached"}]}),
+        }
+    }
+
+    #[test]
+    fn cache_entries_round_trip_through_json() {
+        let entry = entry();
+        let encoded = RedisCache::<redis::Connection>::encode(&entry).unwrap();
+        assert_eq!(
+            RedisCache::<redis::Connection>::decode(encoded).unwrap(),
+            entry
+        );
+    }
+
+    #[test]
+    fn invalid_json_is_rejected() {
+        assert!(RedisCache::<redis::Connection>::decode(b"not json".to_vec()).is_err());
+    }
+
+    #[test]
+    fn ttl_seconds_rounds_up_and_keeps_expiration_positive() {
+        assert_eq!(
+            RedisCache::<redis::Connection>::ttl_seconds(Duration::ZERO),
+            1
+        );
+        assert_eq!(
+            RedisCache::<redis::Connection>::ttl_seconds(Duration::from_millis(1500)),
+            2
+        );
+        assert_eq!(
+            RedisCache::<redis::Connection>::ttl_seconds(Duration::from_secs(15)),
+            15
+        );
+    }
+
+    #[test]
+    fn redis_commands_round_trip_entries_and_delete_only_namespaced_keys() {
+        let value = entry();
+        let payload = RedisCache::<redis::Connection>::encode(&value).unwrap();
+        let connection = MockRedisConnection::new([
+            MockCmd::new(
+                redis::cmd("SETEX")
+                    .arg("litellm-cache:key")
+                    .arg(600)
+                    .arg(payload.clone()),
+                Ok("OK"),
+            ),
+            MockCmd::new(redis::cmd("GET").arg("litellm-cache:key"), Ok(payload)),
+            MockCmd::new(redis::cmd("DEL").arg("litellm-cache:key"), Ok(1u32)),
+        ])
+        .assert_all_commands_consumed();
+        let cache = RedisCache::with_connection(connection, None);
+
+        cache
+            .set_cache("key", value.clone(), CacheKwargs::default())
+            .unwrap();
+        assert_eq!(
+            cache.get_cache("key", &CacheKwargs::default()).unwrap(),
+            Some(value)
+        );
+        cache.delete_cache("key").unwrap();
+    }
+
+    #[test]
+    fn flush_scans_and_deletes_only_cache_keys() {
+        let connection = MockRedisConnection::new([
+            MockCmd::new(
+                redis::cmd("SCAN")
+                    .cursor_arg(0)
+                    .arg("MATCH")
+                    .arg("litellm-cache:*"),
+                Ok(redis_test::redis_value!(["0", ["litellm-cache:key"]])),
+            ),
+            MockCmd::new(redis::cmd("DEL").arg("litellm-cache:key"), Ok(1u32)),
+        ])
+        .assert_all_commands_consumed();
+        let cache = RedisCache::with_connection(connection, None);
+
+        cache.flush_cache().unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_connection_runs_ping_off_executor() {
+        let connection = MockRedisConnection::new([MockCmd::new(redis::cmd("PING"), Ok("PONG"))])
+            .assert_all_commands_consumed();
+        let cache = RedisCache::with_connection(connection, None);
+
+        assert_eq!(
+            cache.test_connection().await.unwrap().status,
+            litellm_cache::CacheConnectionStatus::Success
+        );
+    }
+}

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

@@ -0,0 +1,3 @@
+mod cache;
+
+pub use cache::RedisCache;

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

@@ -0,0 +1,6 @@
+use litellm_cache_redis::RedisCache;
+
+#[test]
+fn constructor_rejects_invalid_urls() {
+    assert!(RedisCache::new("not a redis url", None).is_err());
+}