diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index da8f2690649..941887b249c 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -4381,6 +4381,7 @@ dependencies = [ "criterion", "fancy-regex 0.19.2", "futures-util", + "indexmap 2.14.0", "litellm-auth", "litellm-auth-aws", "litellm-cache", @@ -5773,6 +5774,7 @@ version = "0.29.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4688ddedf473e32662b9b067670129a8afb8c18e351482c70d62ba4a88171e8b" dependencies = [ + "indexmap 2.14.0", "libc", "once_cell", "portable-atomic", diff --git a/litellm-rust/crates/python-bridge/Cargo.toml b/litellm-rust/crates/python-bridge/Cargo.toml index ff8f540c806..6ecfebf0096 100644 --- a/litellm-rust/crates/python-bridge/Cargo.toml +++ b/litellm-rust/crates/python-bridge/Cargo.toml @@ -60,7 +60,8 @@ litellm-router-core.workspace = true litellm-host-python.workspace = true litellm-model-catalog.workspace = true litellm-token-counter = { path = "../token-counter", default-features = false } -pyo3.workspace = true +indexmap = "2.14.0" +pyo3 = { workspace = true, features = ["indexmap"] } prost.workspace = true pyo3-async-runtimes.workspace = true reqwest.workspace = true diff --git a/litellm-rust/crates/python-bridge/src/lib.rs b/litellm-rust/crates/python-bridge/src/lib.rs index 0a9efbdcac3..dc127e41129 100644 --- a/litellm-rust/crates/python-bridge/src/lib.rs +++ b/litellm-rust/crates/python-bridge/src/lib.rs @@ -42,9 +42,24 @@ mod _native { #[pymodule_export] use crate::redis::client::native_redis_breaker_states; #[pymodule_export] + use crate::redis::store::auth_cache::{NativeAuthCache, PyLocalSizes}; + #[pymodule_export] + use crate::redis::store::rate_limits::{ + NativeRateLimits, PyAtomicApplied, PyAtomicOverLimit, PyGaugeOverLimit, PyGaugesAcquired, + PyGuardedReply, PyWindowCount, + }; + #[pymodule_export] use crate::redis::store::request::{NativeRequestRedis, NativeSend}; #[pymodule_export] - use crate::redis::store::{NativePending, NativeRedisStoreError}; + use crate::redis::store::router_usage::NativeRouterUsage; + #[pymodule_export] + use crate::redis::store::sensitive_route_pins::NativeSensitiveRoutePins; + #[pymodule_export] + use crate::redis::store::spend_counters::{NativeSpendCounters, PySpendRead}; + #[pymodule_export] + use crate::redis::store::{ + NativePending, NativeRedisStoreError, router_cooldowns::NativeRouterCooldowns, + }; #[pymodule_export] use crate::router::{RouterCall, RouterCoreHandle}; #[pymodule_export] @@ -123,8 +138,22 @@ mod tests { "NativeRedisClient", "NativeRequestRedis", "NativeSend", + "NativeAuthCache", + "NativeRateLimits", + "WindowCount", + "AtomicApplied", + "AtomicOverLimit", + "GuardedReply", + "GaugesAcquired", + "GaugeOverLimit", "NativePending", + "NativeRouterCooldowns", + "NativeRouterUsage", + "NativeSensitiveRoutePins", "NativeRedisStoreError", + "NativeSpendCounters", + "SpendRead", + "LocalSizes", "native_redis_breaker_states", "NativeClickHouseSpendConfig", "NativeClickHouseSpendStorage", diff --git a/litellm-rust/crates/python-bridge/src/python_settings.rs b/litellm-rust/crates/python-bridge/src/python_settings.rs index eacec6859b5..0152fe8a5b9 100644 --- a/litellm-rust/crates/python-bridge/src/python_settings.rs +++ b/litellm-rust/crates/python-bridge/src/python_settings.rs @@ -13,6 +13,8 @@ pub(crate) enum PythonSettings { UrlPolicy, #[strum(serialize = "provider_defaults")] ProviderDefaults, + #[strum(serialize = "router_tiers")] + RouterTiers, #[strum(serialize = "secret_manager")] SecretManager, #[strum(serialize = "secret_manager_binding")] diff --git a/litellm-rust/crates/python-bridge/src/redis/client/mod.rs b/litellm-rust/crates/python-bridge/src/redis/client/mod.rs index c47e9aec671..5b0e0ec5d94 100644 --- a/litellm-rust/crates/python-bridge/src/redis/client/mod.rs +++ b/litellm-rust/crates/python-bridge/src/redis/client/mod.rs @@ -18,7 +18,7 @@ use pyo3::{ exceptions::{PyOverflowError, PyTypeError, PyValueError}, prelude::*, sync::MutexExt, - types::{PyBool, PyDict, PyFloat, PyInt, PyList, PyString, PyTuple}, + types::{PyBool, PyDict, PyFloat, PyInt, PyList, PyMapping, PyString, PyTuple}, }; use self::credentials::PythonCredentialSource; @@ -40,16 +40,18 @@ impl<'a, 'py> FromPyObject<'a, 'py> for RedisSpec { fn extract(kwargs: Borrowed<'a, 'py, PyAny>) -> PyResult { let py = kwargs.py(); + let source = PyDict::new(py); + let mapping = kwargs.cast::()?; + source.update(&mapping)?; let plain = PyDict::new(py); let mut host_objects = Vec::new(); - for (name, value) in kwargs.cast::()?.iter() { + for (name, value) in source.iter() { match value.is_none() || is_plain(&value) { true => plain.set_item(&name, value)?, false => host_objects.push(name.extract::()?), } } - let credential_provider = kwargs - .cast::()? + let credential_provider = source .get_item("credential_provider")? .filter(|provider| !provider.is_none() && !is_plain(provider)) .map(Bound::unbind); @@ -65,27 +67,44 @@ impl<'a, 'py> FromPyObject<'a, 'py> for RedisSpec { } } -/// `litellm.constants`' `REDIS_CIRCUIT_BREAKER_*`, which Python reads from the environment once. -pub(crate) struct RedisBreaker(BreakerConfig); +/// `litellm.redis.store_settings.RedisBreakerSettings`, which `RedisCache` fills from `litellm.constants`. +#[derive(FromPyObject)] +pub(crate) struct RedisBreaker { + #[pyo3(attribute)] + failure_threshold: u32, + #[pyo3(attribute)] + recovery_timeout: f64, + #[pyo3(attribute)] + timeout_min_duration: f64, +} -impl<'a, 'py> FromPyObject<'a, 'py> for RedisBreaker { - type Error = PyErr; - - fn extract(breaker: Borrowed<'a, 'py, PyAny>) -> PyResult { - let (failure_threshold, recovery_timeout, timeout_min_duration): (u32, f64, f64) = - breaker.extract()?; +impl RedisBreaker { + fn config(&self) -> PyResult { let seconds = |seconds: f64| { Duration::try_from_secs_f64(seconds) - .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string())) + .map_err(|error| PyValueError::new_err(error.to_string())) }; - Ok(Self(BreakerConfig { - failure_threshold, - recovery_timeout: seconds(recovery_timeout)?, - timeout_min_duration: seconds(timeout_min_duration)?, - })) + Ok(BreakerConfig { + failure_threshold: self.failure_threshold, + recovery_timeout: seconds(self.recovery_timeout)?, + timeout_min_duration: seconds(self.timeout_min_duration)?, + }) } } +/// `litellm.redis.store_settings.RedisStoreSettings`, which a `RedisCache` captures when it is built. +#[derive(FromPyObject)] +pub(crate) struct StoreSettings { + #[pyo3(attribute)] + redis_kwargs: RedisSpec, + #[pyo3(attribute)] + namespace: Option, + #[pyo3(attribute)] + default_ttl: f64, + #[pyo3(attribute)] + breaker: Option, +} + fn is_plain(value: &Bound<'_, PyAny>) -> bool { value.is_instance_of::() || value.is_instance_of::() @@ -119,7 +138,7 @@ fn shared_client( py: Python<'_>, spec: RedisSpec, namespace: Option, - breaker: Option, + breaker: Option, ) -> PyResult> { let environment: SharedEnvironment = Arc::new(ProcessEnvironment); let RedisSpec { @@ -176,7 +195,7 @@ fn shared_client( } let connected = { let _runtime = runtime.enter(); - litellm_redis_connect::connect(resolved.clone(), host, breaker.map(|breaker| breaker.0)) + litellm_redis_connect::connect(resolved.clone(), host, breaker) }; let client = match connected { Ok(client) => client, @@ -205,15 +224,15 @@ pub(crate) struct NativeRedisClient { impl NativeRedisClient { /// `None` when redis-py would fail only once a command connects, which leaves every store given it on memory. #[staticmethod] - #[pyo3(signature = (redis_kwargs, *, default_ttl_seconds, namespace = None, breaker = None))] - fn connect( - py: Python<'_>, - redis_kwargs: RedisSpec, - default_ttl_seconds: f64, - namespace: Option, - breaker: Option, - ) -> PyResult> { - let default_ttl = number::seconds(default_ttl_seconds).map_err(seconds_error)?; + fn connect(py: Python<'_>, settings: StoreSettings) -> PyResult> { + let StoreSettings { + redis_kwargs, + namespace, + default_ttl, + breaker, + } = settings; + let default_ttl = number::seconds(default_ttl).map_err(seconds_error)?; + let breaker = breaker.as_ref().map(RedisBreaker::config).transpose()?; Ok( shared_client(py, redis_kwargs, namespace, breaker)?.map(|client| Self { client, diff --git a/litellm-rust/crates/python-bridge/src/redis/mod.rs b/litellm-rust/crates/python-bridge/src/redis/mod.rs index ad91f8349ac..0a54822bcd7 100644 --- a/litellm-rust/crates/python-bridge/src/redis/mod.rs +++ b/litellm-rust/crates/python-bridge/src/redis/mod.rs @@ -1,11 +1,3 @@ -#[expect( - dead_code, - reason = "the native stores that use these helpers land in the next PR" -)] pub(crate) mod client; mod events; -#[expect( - dead_code, - reason = "the native stores that use these helpers land in the next PR" -)] pub(crate) mod store; diff --git a/litellm-rust/crates/python-bridge/src/redis/store/auth_cache.rs b/litellm-rust/crates/python-bridge/src/redis/store/auth_cache.rs new file mode 100644 index 00000000000..8d7788b136e --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/auth_cache.rs @@ -0,0 +1,265 @@ +use std::sync::Arc; + +use litellm_python_compat::Value; +use litellm_redis_store::auth_cache::{AuthCache, Settings}; +use pyo3::prelude::*; + +use super::{ + NativePending, PyNumber, kind_error, pending, redis_async, redis_sync, remote, + request::{NativeRequestRedis, NativeSend}, + ttl, + value::PyValue, +}; +use crate::redis::client::NativeRedisClient; + +fn values(values: Vec>) -> Vec> { + values.into_iter().map(|value| value.map(PyValue)).collect() +} + +/// `UserApiKeyCache` with `enable_redis_auth_cache` on. The sync methods back `DualCache`'s sync +/// API and block with the GIL released, like redis-py's sync client. +#[pyclass(frozen, name = "NativeAuthCache")] +pub(crate) struct NativeAuthCache(Arc); + +#[pymethods] +impl NativeAuthCache { + #[new] + #[pyo3(signature = (client, *, memory_entries, memory_ttl_seconds = None))] + fn new( + client: Option>, + memory_entries: usize, + memory_ttl_seconds: Option, + ) -> PyResult { + Ok(Self(Arc::new(AuthCache::new( + client.map(|client| remote(&client)), + Settings { + memory_entries, + memory_ttl: ttl(memory_ttl_seconds)?, + }, + )))) + } + + #[getter] + fn has_redis(&self) -> bool { + self.0.has_redis() + } + + /// `request` answers a key its pre-call reads already found absent without another `GET`. + #[pyo3(signature = (key, request = None))] + fn get<'py>( + &self, + py: Python<'py>, + key: String, + request: Option<&NativeRequestRedis>, + ) -> PyResult> { + let store = self.0.clone(); + let request = request.map(|request| request.0.clone()); + redis_async(py, "async_get_cache", kind_error, async move { + store + .get(request.as_ref(), &key) + .await + .map(|value| value.map(PyValue)) + }) + } + + fn get_sync(&self, py: Python<'_>, key: String) -> PyResult> { + let store = self.0.clone(); + redis_sync(py, "get_cache", kind_error, async move { + store.get(None, &key).await.map(|value| value.map(PyValue)) + }) + } + + /// `update_cache_ttl` and `update_in_memory_max_size`: both partitions keep what they hold. + #[pyo3(signature = (*, memory_entries, memory_ttl_seconds = None))] + fn reconfigure(&self, memory_entries: usize, memory_ttl_seconds: Option) -> PyResult<()> { + self.0.reconfigure(Settings { + memory_entries, + memory_ttl: ttl(memory_ttl_seconds)?, + }); + Ok(()) + } + + #[pyo3(signature = (key, value, ttl_seconds = None))] + fn set<'py>( + &self, + py: Python<'py>, + key: String, + value: PyValue, + ttl_seconds: Option, + ) -> PyResult> { + let store = self.0.clone(); + let ttl = ttl(ttl_seconds)?; + redis_async(py, "async_set_cache", kind_error, async move { + store.set(&key, value.0, ttl).await + }) + } + + #[pyo3(signature = (key, value, ttl_seconds = None))] + fn set_sync( + &self, + py: Python<'_>, + key: String, + value: PyValue, + ttl_seconds: Option, + ) -> PyResult<()> { + let store = self.0.clone(); + let ttl = ttl(ttl_seconds)?; + redis_sync(py, "set_cache", kind_error, async move { + store.set_str(&key, value.0, ttl).await + }) + } + + /// Memory now and the `SET` on the batch `send` picks. + #[pyo3(signature = (key, value, ttl_seconds, send))] + fn declare_set( + &self, + key: String, + value: PyValue, + ttl_seconds: Option, + send: &NativeSend, + ) -> PyResult { + let written = self + .0 + .declare_set(send.send(), &key, value.0, ttl(ttl_seconds)?) + .map_err(kind_error)?; + Ok(pending( + "async_set_cache", + kind_error, + written, + std::convert::identity, + )) + } + + fn declare_delete(&self, key: String, send: &NativeSend) -> PyResult { + let deleted = self + .0 + .declare_delete(send.send(), &key) + .map_err(kind_error)?; + Ok(pending( + "async_delete_cache", + kind_error, + deleted, + std::convert::identity, + )) + } + + #[pyo3(signature = (entries, ttl_seconds = None))] + fn set_many<'py>( + &self, + py: Python<'py>, + entries: Vec<(String, PyValue)>, + ttl_seconds: Option, + ) -> PyResult> { + let store = self.0.clone(); + let entries = entries + .into_iter() + .map(|(key, value)| (key, value.0)) + .collect(); + let ttl = ttl(ttl_seconds)?; + redis_async(py, "async_set_cache_pipeline", kind_error, async move { + store.set_many(entries, ttl).await + }) + } + + /// On a request batch it returns before anything is sent, and `Now` is a pipeline of its own. + fn write_back<'py>( + &self, + py: Python<'py>, + entries: Vec<(String, PyValue, Option)>, + send: &NativeSend, + ) -> PyResult> { + let entries = entries + .into_iter() + .map(|(key, value, seconds)| Ok((key, value.0, ttl(seconds)?))) + .collect::>()?; + let written = self + .0 + .write_back(send.send(), entries) + .map_err(kind_error)?; + redis_async( + py, + "async_set_cache_pipeline_with_ttls", + kind_error, + written, + ) + } + + fn delete<'py>(&self, py: Python<'py>, keys: Vec) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_delete_cache", kind_error, async move { + store.delete(&keys).await + }) + } + + fn delete_sync(&self, py: Python<'_>, keys: Vec) -> PyResult<()> { + let store = self.0.clone(); + redis_sync(py, "delete_cache", kind_error, async move { + store.delete(&keys).await + }) + } + + /// One `MGET` of keys the caller found missing in memory, on the batch `send` picks; `None` + /// without Redis. + fn get_many(&self, keys: Vec, send: &NativeSend) -> Option { + let read = self.0.get_many(send.send(), &keys)?; + Some(pending("async_batch_get_cache", kind_error, read, values)) + } + + fn increment_if_present<'py>( + &self, + py: Python<'py>, + key: String, + amount: PyNumber, + ) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_increment_if_present", kind_error, async move { + store.increment_if_present(&key, amount.0).await + }) + } + + fn get_local(&self, key: &str) -> PyResult> { + Ok(self.0.get_local(key).map_err(kind_error)?.map(PyValue)) + } + + #[pyo3(signature = (key, value, ttl_seconds = None))] + fn set_local(&self, key: &str, value: PyValue, ttl_seconds: Option) -> PyResult<()> { + self.0 + .set_local(key, value.0, ttl(ttl_seconds)?) + .map_err(kind_error) + } + + fn delete_local(&self, key: &str) -> PyResult<()> { + self.0.delete_local(key).map_err(kind_error) + } + + fn flush_local(&self) -> PyResult<()> { + self.0.flush_local().map_err(kind_error) + } + + fn len_local(&self) -> PyResult { + let (objects, key_objects) = self.0.len_local().map_err(kind_error)?; + Ok(PyLocalSizes { + objects, + key_objects, + }) + } +} + +/// Entries the management-object and the user-key partitions hold in memory. +#[pyclass(frozen, eq, get_all, name = "LocalSizes")] +#[derive(PartialEq)] +pub(crate) struct PyLocalSizes { + objects: usize, + key_objects: usize, +} + +#[pymethods] +impl PyLocalSizes { + #[new] + fn new(objects: usize, key_objects: usize) -> Self { + Self { + objects, + key_objects, + } + } +} diff --git a/litellm-rust/crates/python-bridge/src/redis/store/mod.rs b/litellm-rust/crates/python-bridge/src/redis/store/mod.rs index 2c48bf2c689..d9c3aabb814 100644 --- a/litellm-rust/crates/python-bridge/src/redis/store/mod.rs +++ b/litellm-rust/crates/python-bridge/src/redis/store/mod.rs @@ -1,4 +1,11 @@ +pub(crate) mod auth_cache; +mod policy; +pub(crate) mod rate_limits; pub(crate) mod request; +pub(crate) mod router_cooldowns; +pub(crate) mod router_usage; +pub(crate) mod sensitive_route_pins; +pub(crate) mod spend_counters; mod value; use value::PyValue; @@ -159,6 +166,7 @@ where } /// A Python `int` or `float` across the bridge, keeping which one it is. +#[derive(Clone, Copy, PartialEq)] pub(crate) struct PyNumber(Number); impl<'a, 'py> FromPyObject<'a, 'py> for PyNumber { @@ -188,6 +196,21 @@ impl<'py> IntoPyObject<'py> for PyNumber { } } +/// A store enum read from its wire spelling, so the bridge names no variant. +pub(crate) struct Spelled(pub(crate) T); + +impl<'a, 'py, T: std::str::FromStr> FromPyObject<'a, 'py> for Spelled { + type Error = PyErr; + + fn extract(value: Borrowed<'a, 'py, PyAny>) -> PyResult { + let spelling: String = value.extract()?; + spelling + .parse() + .map(Self) + .map_err(|_| PyValueError::new_err(format!("unknown spelling {spelling:?}"))) + } +} + /// A store type whose Python side is plain data, converted through its `FromValue` and `IntoValue` /// so the bridge never names a field. A shape the store refuses is `ValueError`. pub(crate) struct PyData(pub(crate) T); diff --git a/litellm-rust/crates/python-bridge/src/redis/store/policy.rs b/litellm-rust/crates/python-bridge/src/redis/store/policy.rs new file mode 100644 index 00000000000..ffdf7b97a2e --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/policy.rs @@ -0,0 +1,151 @@ +use std::time::Duration; + +use litellm_redis_store::TierPolicy; +use pyo3::prelude::*; + +use crate::{ + coercion::{Field, FieldSpec, ProjectionError}, + python_settings::{PythonSettings, Snapshot}, +}; + +const COOLDOWN_REFRESH: FieldSpec = + FieldSpec::new("cooldown_redis_read_interval", seconds); +const USAGE_REFRESH: FieldSpec = FieldSpec::new("usage_redis_read_interval", seconds); +const TRACKED_READS: FieldSpec = FieldSpec::new("max_redis_batch_cache_size", count); +const MEMORY_TTL: FieldSpec> = + FieldSpec::new("default_in_memory_ttl", optional_seconds); + +/// The tier values this Python process runs with, from its environment and `litellm` globals. +#[derive(Debug)] +pub(crate) struct RouterTierPolicies { + pub(crate) cooldowns: TierPolicy, + pub(crate) usage: TierPolicy, +} + +pub(crate) fn router_tier_policies(py: Python<'_>) -> PyResult { + Ok(project(&PythonSettings::RouterTiers.read(py)?)?) +} + +fn project(snapshot: &Snapshot<'_>) -> Result { + let tracked_reads = snapshot.read(&TRACKED_READS)?; + let default_in_memory_ttl = snapshot.read(&MEMORY_TTL)?; + Ok(RouterTierPolicies { + cooldowns: TierPolicy { + refresh_after: snapshot.read(&COOLDOWN_REFRESH)?, + tracked_reads, + default_in_memory_ttl, + }, + usage: TierPolicy { + refresh_after: snapshot.read(&USAGE_REFRESH)?, + tracked_reads, + default_in_memory_ttl, + }, + }) +} + +fn seconds(field: &Field<'_>) -> Result { + match field.value().extract::() { + Ok(seconds) if seconds.is_finite() && seconds >= 0.0 => { + Ok(Duration::from_secs_f64(seconds)) + } + _ => Err(field.invalid("a non-negative number of seconds")), + } +} + +fn optional_seconds(field: &Field<'_>) -> Result, ProjectionError> { + match field.value().is_none() { + true => Ok(None), + false => seconds(field).map(Some), + } +} + +fn count(field: &Field<'_>) -> Result { + field + .value() + .extract::() + .map_err(|_| field.invalid("a non-negative integer")) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use pyo3::{prelude::*, types::PyDict}; + + use super::project; + use crate::python_settings::PythonSettings; + + fn projected(py: Python<'_>, fields: &str) -> Result { + let locals = PyDict::new(py); + let source = std::ffi::CString::new(format!( + "import types\nsettings = types.SimpleNamespace({fields})" + )) + .unwrap(); + py.run(&source, Some(&locals), Some(&locals)).unwrap(); + let snapshot = + PythonSettings::RouterTiers.snapshot(locals.get_item("settings").unwrap().unwrap()); + project(&snapshot).map_err(|error| PyErr::from(error).to_string()) + } + + #[test] + fn the_host_values_are_each_familys_refresh_and_tracked_reads() { + Python::initialize(); + Python::attach(|py| { + let policies = projected( + py, + "cooldown_redis_read_interval=2.5, usage_redis_read_interval=30, max_redis_batch_cache_size=42, default_in_memory_ttl=None", + ) + .unwrap(); + assert_eq!( + policies.cooldowns.refresh_after, + Duration::from_millis(2500) + ); + assert_eq!(policies.usage.refresh_after, Duration::from_secs(30)); + assert_eq!(policies.cooldowns.tracked_reads, 42); + assert_eq!(policies.usage.tracked_reads, 42); + assert_eq!(policies.cooldowns.default_in_memory_ttl, None); + }); + } + + #[test] + fn a_set_default_in_memory_ttl_reaches_both_familys_backfill() { + Python::initialize(); + Python::attach(|py| { + let policies = projected( + py, + "cooldown_redis_read_interval=1, usage_redis_read_interval=10, max_redis_batch_cache_size=1000, default_in_memory_ttl=7.5", + ) + .unwrap(); + assert_eq!( + policies.cooldowns.default_in_memory_ttl, + Some(Duration::from_millis(7500)) + ); + assert_eq!( + policies.usage.default_in_memory_ttl, + Some(Duration::from_millis(7500)) + ); + }); + } + + #[test] + fn a_value_python_would_not_run_with_is_a_configuration_error() { + Python::initialize(); + Python::attach(|py| { + let negative = projected( + py, + "cooldown_redis_read_interval=-1, usage_redis_read_interval=10, max_redis_batch_cache_size=1000, default_in_memory_ttl=None", + ) + .unwrap_err(); + let text = projected( + py, + "cooldown_redis_read_interval=1, usage_redis_read_interval=10, max_redis_batch_cache_size='many', default_in_memory_ttl=None", + ) + .unwrap_err(); + assert!( + negative.contains("cooldown_redis_read_interval"), + "{negative}" + ); + assert!(text.contains("max_redis_batch_cache_size"), "{text}"); + }); + } +} diff --git a/litellm-rust/crates/python-bridge/src/redis/store/rate_limits.rs b/litellm-rust/crates/python-bridge/src/redis/store/rate_limits.rs new file mode 100644 index 00000000000..bf6ddfbeb52 --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/rate_limits.rs @@ -0,0 +1,591 @@ +use std::{sync::Arc, time::Duration}; + +use indexmap::IndexMap; +use litellm_redis_store::{ + Pending, + rate_limits::{ + AcquireReply, Applied, AtomicCounter, AtomicReply, Counter, Entry, Gauge, GuardedIncrement, + GuardedReply, Increment, Kind, Local, RateLimits, Scope, WindowCount, + }, +}; +use pyo3::{exceptions::PyValueError, prelude::*}; + +use super::{ + NativePending, PyNumber, Spelled, kind_error, redis_async, remote, + request::{NativeRequestRedis, NativeSend}, + ttl, +}; +use crate::redis::client::NativeRedisClient; + +/// The v3 rate limiter's keyspace. Memory reads and writes are synchronous like `InMemoryCache`; +/// every Redis call is a future. +#[pyclass(frozen, name = "NativeRateLimits")] +pub(crate) struct NativeRateLimits(Arc); + +/// An `int` or `float` counter, a `str` window start, or a `{slot_id: acquired_at}` registry. +#[derive(FromPyObject)] +enum PyLocal { + Number(PyNumber), + Text(String), + Slots(IndexMap), +} + +impl From for Local { + fn from(local: PyLocal) -> Self { + match local { + PyLocal::Number(number) => Self::Number(number.0), + PyLocal::Text(text) => Self::Text(text), + PyLocal::Slots(slots) => Self::Slots(slots.into_iter().collect()), + } + } +} + +/// What the memory tier holds, back as the Python value it was written as. +struct LocalValue(Local); + +impl<'py> IntoPyObject<'py> for LocalValue { + type Target = PyAny; + type Output = Bound<'py, PyAny>; + type Error = PyErr; + + fn into_pyobject(self, py: Python<'py>) -> PyResult { + Ok(match self.0 { + Local::Number(number) => PyNumber(number).into_pyobject(py)?, + Local::Text(text) => text.into_pyobject(py)?.into_any(), + Local::Slots(slots) => slots + .into_iter() + .collect::>() + .into_pyobject(py)? + .into_any(), + }) + } +} + +/// `litellm.redis.stores.rate_limits.RateLimitScope`. +#[derive(FromPyObject)] +struct PyScope { + #[pyo3(attribute)] + descriptor_key: String, + #[pyo3(attribute)] + descriptor_value: String, +} + +impl From for Scope { + fn from(scope: PyScope) -> Self { + Self { + descriptor: scope.descriptor_key, + value: scope.descriptor_value, + } + } +} + +/// `RateLimitCounter`. +#[derive(FromPyObject)] +struct PyCounter { + #[pyo3(attribute)] + scope: PyScope, + #[pyo3(attribute)] + kind: Spelled, +} + +impl From for Counter { + fn from(counter: PyCounter) -> Self { + Self { + scope: counter.scope.into(), + kind: counter.kind.0, + } + } +} + +/// `RateLimitWindow`, `RateLimitCounter` or `RateLimitGauge`, told apart by the key suffix each names. +#[derive(FromPyObject)] +struct PyEntry { + #[pyo3(attribute)] + suffix: String, + #[pyo3(attribute)] + scope: PyScope, +} + +impl TryFrom for Entry { + type Error = PyErr; + + fn try_from(entry: PyEntry) -> PyResult { + Self::new(&entry.suffix, entry.scope.into()) + .map_err(|_| PyValueError::new_err(format!("unknown key suffix {:?}", entry.suffix))) + } +} + +/// `CounterIncrement`. +#[derive(FromPyObject)] +struct PyIncrement { + #[pyo3(attribute)] + counter: PyCounter, + #[pyo3(attribute)] + amount: PyNumber, + #[pyo3(attribute)] + ttl: Option, +} + +impl From for Increment { + fn from(increment: PyIncrement) -> Self { + Self { + counter: increment.counter.into(), + amount: increment.amount.0, + ttl: increment.ttl.map(Duration::from_secs), + } + } +} + +/// `AtomicCounter`. +#[derive(FromPyObject)] +struct PyAtomicCounter { + #[pyo3(attribute)] + counter: PyCounter, + #[pyo3(attribute)] + limit: i64, + #[pyo3(attribute)] + increment: i64, + #[pyo3(attribute)] + ttl: i64, + #[pyo3(attribute)] + window_size: i64, +} + +impl From for AtomicCounter { + fn from(counter: PyAtomicCounter) -> Self { + Self { + counter: counter.counter.into(), + limit: counter.limit, + increment: counter.increment, + ttl_seconds: counter.ttl, + window_size_seconds: counter.window_size, + } + } +} + +fn all>(records: Vec

) -> Vec { + records.into_iter().map(T::from).collect() +} + +#[pyclass(frozen, eq, get_all, name = "WindowCount")] +#[derive(PartialEq)] +pub(crate) struct PyWindowCount { + window_start: String, + count: i64, +} + +#[pymethods] +impl PyWindowCount { + #[new] + fn new(window_start: String, count: i64) -> Self { + Self { + window_start, + count, + } + } +} + +/// Per counter, the count after the increment and the start of the window it landed in. +#[pyclass(frozen, eq, get_all, name = "AtomicApplied")] +#[derive(PartialEq)] +pub(crate) struct PyAtomicApplied { + counters: Vec<(i64, i64)>, +} + +#[pymethods] +impl PyAtomicApplied { + #[new] + fn new(counters: Vec<(i64, i64)>) -> Self { + Self { counters } + } +} + +#[pyclass(frozen, eq, get_all, name = "AtomicOverLimit")] +#[derive(PartialEq)] +pub(crate) struct PyAtomicOverLimit { + index: usize, + current: i64, + limit: i64, +} + +#[pymethods] +impl PyAtomicOverLimit { + #[new] + fn new(index: usize, current: i64, limit: i64) -> Self { + Self { + index, + current, + limit, + } + } +} + +#[derive(IntoPyObject)] +enum PyAtomicReply { + Applied(PyAtomicApplied), + OverLimit(PyAtomicOverLimit), +} + +#[pyclass(frozen, eq, get_all, name = "GuardedReply")] +#[derive(PartialEq)] +pub(crate) struct PyGuardedReply { + applied: bool, + count: i64, +} + +#[pymethods] +impl PyGuardedReply { + #[new] + fn new(applied: bool, count: i64) -> Self { + Self { applied, count } + } +} + +#[pyclass(frozen, eq, get_all, name = "GaugesAcquired")] +#[derive(PartialEq)] +pub(crate) struct PyGaugesAcquired { + in_flight: Vec, +} + +#[pymethods] +impl PyGaugesAcquired { + #[new] + fn new(in_flight: Vec) -> Self { + Self { in_flight } + } +} + +#[pyclass(frozen, eq, get_all, name = "GaugeOverLimit")] +#[derive(PartialEq)] +pub(crate) struct PyGaugeOverLimit { + index: usize, + in_flight: i64, + limit: i64, +} + +#[pymethods] +impl PyGaugeOverLimit { + #[new] + fn new(index: usize, in_flight: i64, limit: i64) -> Self { + Self { + index, + in_flight, + limit, + } + } +} + +#[derive(IntoPyObject)] +enum PyAcquireReply { + Acquired(PyGaugesAcquired), + OverLimit(PyGaugeOverLimit), +} + +fn window_counts(counts: Vec) -> Vec { + counts + .into_iter() + .map( + |WindowCount { + window_start, + count, + }| PyWindowCount { + window_start, + count, + }, + ) + .collect() +} + +fn atomic(reply: AtomicReply) -> PyAtomicReply { + match reply { + AtomicReply::Applied(applied) => PyAtomicReply::Applied(PyAtomicApplied { + counters: applied + .into_iter() + .map( + |Applied { + count, + window_start, + }| (count, window_start), + ) + .collect(), + }), + AtomicReply::OverLimit { + index, + current, + limit, + } => PyAtomicReply::OverLimit(PyAtomicOverLimit { + index, + current, + limit, + }), + } +} + +fn guarded(GuardedReply { applied, count }: GuardedReply) -> PyGuardedReply { + PyGuardedReply { applied, count } +} + +fn acquire(reply: AcquireReply) -> PyAcquireReply { + match reply { + AcquireReply::Acquired(in_flight) => { + PyAcquireReply::Acquired(PyGaugesAcquired { in_flight }) + } + AcquireReply::OverLimit { + index, + in_flight, + limit, + } => PyAcquireReply::OverLimit(PyGaugeOverLimit { + index, + in_flight, + limit, + }), + } +} + +/// One script call declared on the batch the caller's `NativeSend` picked; `None` without Redis. +fn script( + call_type: &'static str, + declared: Option>, + reply: fn(T) -> R, +) -> Option +where + T: Send + 'static, + R: for<'a> IntoPyObject<'a> + Send + 'static, +{ + Some(super::pending(call_type, kind_error, declared?, reply)) +} + +#[pymethods] +impl NativeRateLimits { + #[new] + #[pyo3(signature = (client, backfill_ttl_seconds = None))] + fn new( + client: Option>, + backfill_ttl_seconds: Option, + ) -> PyResult { + Ok(Self(Arc::new(RateLimits::new( + client.map(|client| remote(&client)), + ttl(backfill_ttl_seconds)?, + )))) + } + + #[getter] + fn has_redis(&self) -> bool { + self.0.has_redis() + } + + fn slot_groups(&self, scopes: Vec, force: bool) -> Vec> { + self.0.slot_groups(&all(scopes), force) + } + + fn get_local(&self, entry: PyEntry) -> PyResult> { + Ok(self + .0 + .get_local(&entry.try_into()?) + .map_err(kind_error)? + .map(LocalValue)) + } + + #[pyo3(signature = (entry, value, ttl_seconds))] + fn set_local(&self, entry: PyEntry, value: PyLocal, ttl_seconds: Option) -> PyResult<()> { + self.0 + .set_local(&entry.try_into()?, value.into(), ttl(ttl_seconds)?) + .map_err(kind_error) + } + + /// `request` answers a key its pre-call reads already found absent without another `GET`. + #[pyo3(signature = (entry, *, backfill_ttl_seconds = None, request = None))] + fn get<'py>( + &self, + py: Python<'py>, + entry: PyEntry, + backfill_ttl_seconds: Option, + request: Option<&NativeRequestRedis>, + ) -> PyResult> { + let store = self.0.clone(); + let entry = Entry::try_from(entry)?; + let backfill_ttl = ttl(backfill_ttl_seconds)?; + let request = request.map(|request| request.0.clone()); + redis_async(py, "async_get_cache", kind_error, async move { + store + .get(request.as_ref(), &entry, backfill_ttl) + .await + .map(|local| local.map(LocalValue)) + }) + } + + fn sliding_window( + &self, + counters: Vec, + now: i64, + window_size: i64, + send: &NativeSend, + ) -> Option { + script( + "batch_rate_limiter_script", + self.0 + .sliding_window(send.send(), &all(counters), now, window_size), + window_counts, + ) + } + + fn read_counters(&self, counters: Vec, send: &NativeSend) -> Option { + script( + "batch_counter_read_script", + self.0.read_counters(send.send(), &all(counters)), + std::convert::identity, + ) + } + + fn check_and_increment( + &self, + counters: Vec, + send: &NativeSend, + ) -> Option { + script( + "check_and_increment_by_n_script", + self.0.check_and_increment(send.send(), &all(counters)), + atomic, + ) + } + + fn guarded_increment( + &self, + counter: PyCounter, + expected_window_start: String, + amount: PyNumber, + ttl_seconds: Option, + send: &NativeSend, + ) -> Option { + let increment = GuardedIncrement { + counter: counter.into(), + expected_window_start, + amount: amount.0, + ttl: ttl_seconds.map(Duration::from_secs), + }; + script( + "window_guarded_token_increment_script", + self.0.guarded_increment(send.send(), &increment), + guarded, + ) + } + + fn parallel_acquire( + &self, + gauges: Vec<(PyScope, i64)>, + slot_ttl_seconds: u64, + slot_id: &str, + send: &NativeSend, + ) -> Option { + let gauges: Vec = gauges + .into_iter() + .map(|(scope, limit)| Gauge { + scope: scope.into(), + limit, + }) + .collect(); + script( + "parallel_acquire_script", + self.0.parallel_acquire( + send.send(), + &gauges, + Duration::from_secs(slot_ttl_seconds), + slot_id, + ), + acquire, + ) + } + + fn parallel_release( + &self, + gauges: Vec, + slot_id: &str, + send: &NativeSend, + ) -> Option { + script( + "parallel_release_script", + self.0.parallel_release(send.send(), &all(gauges), slot_id), + std::convert::identity, + ) + } + + fn parallel_count( + &self, + gauges: Vec, + slot_ttl_seconds: u64, + send: &NativeSend, + ) -> Option { + script( + "parallel_count_script", + self.0.parallel_count( + send.send(), + &all(gauges), + Duration::from_secs(slot_ttl_seconds), + ), + std::convert::identity, + ) + } + + fn token_increment( + &self, + increments: Vec, + send: &NativeSend, + ) -> Option { + script( + "token_increment_script", + self.0.token_increment(send.send(), &all(increments)), + std::convert::identity, + ) + } + + /// `False` without Redis. + fn token_increment_post_call( + &self, + request: &NativeRequestRedis, + increments: Vec, + ) -> bool { + self.0 + .token_increment_post_call(&request.0, all(increments)) + } + + /// `False` without Redis. + fn parallel_release_post_call( + &self, + request: &NativeRequestRedis, + gauges: Vec, + slot_id: &str, + ) -> bool { + self.0 + .parallel_release_post_call(&request.0, &all(gauges), slot_id) + } + + fn increment<'py>( + &self, + py: Python<'py>, + increment: PyIncrement, + ) -> PyResult> { + let store = self.0.clone(); + let increment = Increment::from(increment); + redis_async(py, "async_increment", kind_error, async move { + Ok(store.increment(&increment).await.map(PyNumber)) + }) + } + + fn increment_many<'py>( + &self, + py: Python<'py>, + increments: Vec, + ) -> PyResult> { + let store = self.0.clone(); + let increments: Vec = all(increments); + redis_async(py, "async_increment_pipeline", kind_error, async move { + store.increment_many(&increments).await; + Ok(()) + }) + } + + fn increment_many_post_call(&self, request: &NativeRequestRedis, increments: Vec) { + self.0 + .increment_many_post_call(&request.0, &all(increments)); + } +} diff --git a/litellm-rust/crates/python-bridge/src/redis/store/router_cooldowns.rs b/litellm-rust/crates/python-bridge/src/redis/store/router_cooldowns.rs new file mode 100644 index 00000000000..f9f2364da19 --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/router_cooldowns.rs @@ -0,0 +1,183 @@ +use std::sync::Arc; + +use litellm_redis_store::{ + Send, + router_cooldowns::{Cooldown, RouterCooldowns}, +}; +use pyo3::{ + prelude::*, + types::{PyDict, PyTuple}, +}; + +use super::{ + NativePending, PyData, PyNumber, duration, kind_error, pending, policy::router_tier_policies, + redis_async, redis_sync, remote, request::NativeSend, +}; +use crate::redis::client::NativeRedisClient; + +/// One router's cooldown store. The router's sync completion path calls `put`, `active_sync`, +/// `minimum` and `increment_allowed_fails`, so those block with the GIL released like redis-py; the +/// `async_` twins return awaitables for async callers. +#[pyclass(frozen, name = "NativeRouterCooldowns")] +pub(crate) struct NativeRouterCooldowns(Arc); + +/// `[(model_id, cooldown), ...]` the way `RedisCache` hands the router its batch read. +struct PyActive(Vec<(String, Cooldown)>); + +impl<'py> IntoPyObject<'py> for PyActive { + type Target = PyAny; + type Output = Bound<'py, PyAny>; + type Error = PyErr; + + fn into_pyobject(self, py: Python<'py>) -> PyResult { + active_to_py(py, self.0)? + .into_pyobject(py) + .map(Bound::into_any) + } +} + +fn active_to_py( + py: Python<'_>, + active: Vec<(String, Cooldown)>, +) -> PyResult>> { + active + .into_iter() + .map(|(model_id, cooldown)| { + PyTuple::new( + py, + [ + model_id.into_pyobject(py)?.into_any(), + PyData(cooldown).into_pyobject(py)?, + ], + ) + }) + .collect() +} + +#[pymethods] +impl NativeRouterCooldowns { + #[new] + fn new(py: Python<'_>, client: Option>) -> PyResult { + let remote = client.map(|client| remote(&client)); + let policy = router_tier_policies(py)?.cooldowns; + Ok(Self(Arc::new(RouterCooldowns::new(remote, policy)))) + } + + /// Adds the Redis tier and keeps memory; `false` when one was already attached. + fn attach(&self, client: PyRef<'_, NativeRedisClient>) -> bool { + self.0.attach(remote(&client)) + } + + #[getter] + fn has_redis(&self) -> bool { + self.0.has_redis() + } + + fn put( + &self, + py: Python<'_>, + model_id: String, + cooldown: PyData, + ttl_seconds: f64, + ) -> PyResult<()> { + let store = self.0.clone(); + let ttl = duration(ttl_seconds)?; + redis_sync(py, "set_cache", kind_error, async move { + store.put_str(&model_id, &cooldown.0, ttl).await + }) + } + + fn increment_allowed_fails( + &self, + py: Python<'_>, + model_id: String, + suffix: Option, + ttl_seconds: f64, + ) -> PyResult { + let store = self.0.clone(); + let ttl = duration(ttl_seconds)?; + redis_sync(py, "increment_cache", kind_error, async move { + store + .increment_allowed_fails_int(&model_id, suffix.as_deref(), ttl) + .await + }) + } + + fn async_put<'py>( + &self, + py: Python<'py>, + model_id: String, + cooldown: PyData, + ttl_seconds: f64, + ) -> PyResult> { + let store = self.0.clone(); + let ttl = duration(ttl_seconds)?; + redis_async(py, "async_set_cache", kind_error, async move { + store.put(&model_id, &cooldown.0, ttl).await + }) + } + + fn async_increment_allowed_fails<'py>( + &self, + py: Python<'py>, + model_id: String, + suffix: Option, + ttl_seconds: f64, + ) -> PyResult> { + let store = self.0.clone(); + let ttl = duration(ttl_seconds)?; + redis_async(py, "async_increment", kind_error, async move { + store + .increment_allowed_fails(&model_id, suffix.as_deref(), ttl) + .await + }) + } + + /// The deployments still cooling down, read on the batch `send` picks. + fn active(&self, model_ids: Vec, send: &NativeSend) -> PyResult { + let active = self.0.active(send.send(), &model_ids).map_err(kind_error)?; + Ok(pending( + "async_batch_get_cache", + kind_error, + active, + PyActive, + )) + } + + /// `active` on its own with the GIL released, for the router's sync completion path. + fn active_sync<'py>( + &self, + py: Python<'py>, + model_ids: Vec, + ) -> PyResult>> { + let active = self.0.active(Send::Now, &model_ids).map_err(kind_error)?; + let active = redis_sync(py, "batch_get_cache", kind_error, active)?; + active_to_py(py, active) + } + + fn minimum(&self, py: Python<'_>, model_ids: Vec) -> PyResult> { + let store = self.0.clone(); + redis_sync(py, "batch_get_cache", kind_error, async move { + store + .minimum(&model_ids) + .await + .map(|minimum| minimum.map(PyNumber)) + }) + } + + fn delete_local(&self, model_id: &str) -> PyResult<()> { + self.0.delete_local(model_id).map_err(kind_error) + } + + fn flush_local_counters(&self) -> PyResult<()> { + self.0.flush_local_counters().map_err(kind_error) + } + + fn snapshot<'py>(&self, py: Python<'py>) -> PyResult> { + let snapshot = PyDict::new(py); + for (key, cooldown) in self.0.snapshot().map_err(kind_error)? { + snapshot.set_item(key, PyData(cooldown))?; + } + Ok(snapshot) + } +} diff --git a/litellm-rust/crates/python-bridge/src/redis/store/router_usage.rs b/litellm-rust/crates/python-bridge/src/redis/store/router_usage.rs new file mode 100644 index 00000000000..bad4085cfe6 --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/router_usage.rs @@ -0,0 +1,184 @@ +use std::{sync::Arc, time::Duration}; + +use litellm_redis_store::router_usage::{Counter, Kind, RouterUsage}; +use pyo3::{prelude::*, types::PyDict}; + +use super::{ + PyNumber, Spelled, kind_error, policy::router_tier_policies, redis_async, redis_sync, remote, +}; +use crate::redis::client::NativeRedisClient; + +/// `litellm.redis.stores.router_usage.UsageCounter`. +#[derive(FromPyObject)] +struct PyCounter { + #[pyo3(attribute)] + deployment_id: String, + #[pyo3(attribute)] + model: Option, + #[pyo3(attribute)] + minute: String, + #[pyo3(attribute)] + kind: Spelled, +} + +impl From for Counter { + fn from(counter: PyCounter) -> Self { + Self { + deployment_id: counter.deployment_id, + model: counter.model, + minute: counter.minute, + kind: counter.kind.0, + } + } +} + +fn counters(counters: Vec) -> Vec { + counters.into_iter().map(Counter::from).collect() +} + +/// One router's global TPM/RPM counters. The `_sync` methods block with the GIL released, for the +/// Rust router's sync calls, which have no event loop to await on. +#[pyclass(frozen, name = "NativeRouterUsage")] +pub(crate) struct NativeRouterUsage(Arc); + +#[pymethods] +impl NativeRouterUsage { + #[new] + fn new(py: Python<'_>, client: Option>) -> PyResult { + let remote = client.map(|client| remote(&client)); + let policy = router_tier_policies(py)?.usage; + Ok(Self(Arc::new(RouterUsage::new(remote, policy)))) + } + + /// Adds the Redis tier and keeps memory; `false` when one was already attached. + fn attach(&self, client: PyRef<'_, NativeRedisClient>) -> bool { + self.0.attach(remote(&client)) + } + + #[getter] + fn has_redis(&self) -> bool { + self.0.has_redis() + } + + fn increment<'py>( + &self, + py: Python<'py>, + counter: PyCounter, + value: PyNumber, + ttl_seconds: u64, + ) -> PyResult> { + let store = self.0.clone(); + let counter = Counter::from(counter); + redis_async(py, "async_increment", kind_error, async move { + Ok(store + .increment(&counter, value.0, Duration::from_secs(ttl_seconds)) + .await + .map(PyNumber)) + }) + } + + /// `_increment_deployment`'s `async_increment_pipeline`, always on its own round trip. + fn increment_deployment<'py>( + &self, + py: Python<'py>, + deployment_id: String, + model: Option, + minute: String, + tokens: PyNumber, + requests: i64, + ) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_increment_pipeline", kind_error, async move { + store + .increment_deployment( + &deployment_id, + model.as_deref(), + &minute, + tokens.0, + requests, + ) + .await + }) + } + + fn get_many<'py>( + &self, + py: Python<'py>, + counters: Vec, + ) -> PyResult> { + let store = self.0.clone(); + let counters = self::counters(counters); + redis_async(py, "async_batch_get_cache", kind_error, async move { + let read = store.get_many(&counters).await?; + Ok(read + .into_iter() + .map(|count| count.map(PyNumber)) + .collect::>()) + }) + } + + fn increment_sync( + &self, + py: Python<'_>, + counter: PyCounter, + value: PyNumber, + ttl_seconds: u64, + ) -> PyResult> { + let store = self.0.clone(); + let counter = Counter::from(counter); + redis_sync(py, "increment_cache", kind_error, async move { + Ok(store + .increment(&counter, value.0, Duration::from_secs(ttl_seconds)) + .await + .map(PyNumber)) + }) + } + + fn increment_deployment_sync( + &self, + py: Python<'_>, + deployment_id: String, + model: Option, + minute: String, + tokens: PyNumber, + requests: i64, + ) -> PyResult<()> { + let store = self.0.clone(); + redis_sync(py, "increment_pipeline", kind_error, async move { + store + .increment_deployment( + &deployment_id, + model.as_deref(), + &minute, + tokens.0, + requests, + ) + .await + }) + } + + fn get_many_sync( + &self, + py: Python<'_>, + counters: Vec, + ) -> PyResult>> { + let store = self.0.clone(); + let counters = self::counters(counters); + redis_sync(py, "batch_get_cache", kind_error, async move { + let read = store.get_many(&counters).await?; + Ok(read.into_iter().map(|count| count.map(PyNumber)).collect()) + }) + } + + fn flush_local_counters(&self) -> PyResult<()> { + self.0.flush_local_counters().map_err(kind_error) + } + + fn snapshot<'py>(&self, py: Python<'py>) -> PyResult> { + let snapshot = PyDict::new(py); + for (key, count) in self.0.snapshot().map_err(kind_error)? { + snapshot.set_item(key, PyNumber(count))?; + } + Ok(snapshot) + } +} diff --git a/litellm-rust/crates/python-bridge/src/redis/store/sensitive_route_pins.rs b/litellm-rust/crates/python-bridge/src/redis/store/sensitive_route_pins.rs new file mode 100644 index 00000000000..8b1b82945fb --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/sensitive_route_pins.rs @@ -0,0 +1,49 @@ +use std::sync::Arc; + +use litellm_redis_store::sensitive_route_pins::SensitiveRoutePins; +use pyo3::prelude::*; + +use super::{duration, kind_error, redis_async, remote}; +use crate::redis::client::NativeRedisClient; + +#[pyclass(frozen, name = "NativeSensitiveRoutePins")] +pub(crate) struct NativeSensitiveRoutePins(Arc); + +#[pymethods] +impl NativeSensitiveRoutePins { + #[new] + fn new(client: Option>) -> Self { + Self(Arc::new(SensitiveRoutePins::new( + client.map(|client| remote(&client)), + ))) + } + + fn get<'py>( + &self, + py: Python<'py>, + tenant: String, + session_id: String, + default_ttl_seconds: f64, + ) -> PyResult> { + let store = self.0.clone(); + let default_ttl = duration(default_ttl_seconds)?; + redis_async(py, "async_get_cache", kind_error, async move { + store.get(&tenant, &session_id, default_ttl).await + }) + } + + fn set<'py>( + &self, + py: Python<'py>, + tenant: String, + session_id: String, + model: String, + ttl_seconds: f64, + ) -> PyResult> { + let store = self.0.clone(); + let ttl = duration(ttl_seconds)?; + redis_async(py, "async_set_cache", kind_error, async move { + store.set(&tenant, &session_id, &model, ttl).await + }) + } +} diff --git a/litellm-rust/crates/python-bridge/src/redis/store/spend_counters.rs b/litellm-rust/crates/python-bridge/src/redis/store/spend_counters.rs new file mode 100644 index 00000000000..58b77222f5e --- /dev/null +++ b/litellm-rust/crates/python-bridge/src/redis/store/spend_counters.rs @@ -0,0 +1,256 @@ +use std::sync::Arc; + +use litellm_redis_store::{ + Number, + spend_counters::{LostSeed, Reading, SpendCounter, SpendCounters}, +}; +use pyo3::{prelude::*, types::PyDict}; + +use super::{ + NativePending, PyData, PyNumber, kind_error, pending, redis_async, remote, + request::{NativeRequestRedis, NativeSend}, + ttl, unwrap_all, +}; +use crate::redis::client::NativeRedisClient; + +/// A counter's value and whether Redis answered it, a clean miss included. +#[pyclass(frozen, eq, get_all, name = "SpendRead")] +#[derive(PartialEq)] +pub(crate) struct PySpendRead { + value: Option, + from_redis: bool, +} + +#[pymethods] +impl PySpendRead { + #[new] + fn new(value: Option, from_redis: bool) -> Self { + Self { value, from_redis } + } +} + +fn increments(increments: Vec<(PyData, PyNumber)>) -> Vec<(SpendCounter, Number)> { + increments + .into_iter() + .map(|(counter, amount)| (counter.0, amount.0)) + .collect() +} + +fn numbers(values: Option>>) -> Option>> { + values.map(|values| { + values + .into_iter() + .map(|value| value.map(PyNumber)) + .collect() + }) +} + +/// The proxy's spend counters, Redis first with this worker's memory copy behind it. +#[pyclass(frozen, name = "NativeSpendCounters")] +pub(crate) struct NativeSpendCounters(Arc); + +#[pymethods] +impl NativeSpendCounters { + #[new] + fn new(client: Option>) -> Self { + Self(Arc::new(SpendCounters::new( + client.map(|client| remote(&client)), + ))) + } + + /// Adds the Redis tier and keeps memory; `false` when one was already attached. + fn attach(&self, client: PyRef<'_, NativeRedisClient>) -> bool { + self.0.attach(remote(&client)) + } + + #[getter] + fn has_redis(&self) -> bool { + self.0.has_redis() + } + + #[getter] + fn redis_ttl_seconds(&self) -> Option { + self.0.redis_ttl().map(|ttl| ttl.as_secs_f64()) + } + + /// `("redis" | "memory", value)`: where the answer came from, a Redis miss included. + fn read<'py>( + &self, + py: Python<'py>, + counter: PyData, + ) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_get_cache", kind_error, async move { + let reading = store.read(&counter.0).await?; + Ok(PySpendRead { + value: reading.value().map(PyNumber), + from_redis: matches!(reading, Reading::Redis(_)), + }) + }) + } + + /// One `MGET` on the batch `send` picks; `None` without Redis. It resolves to `None` when the + /// read failed, so each caller reads on its own. + fn read_many( + &self, + counters: Vec>, + send: &NativeSend, + ) -> Option { + let read = self.0.read_many(send.send(), &unwrap_all(counters))?; + Some(pending("async_batch_get_cache", kind_error, read, numbers)) + } + + fn increment<'py>( + &self, + py: Python<'py>, + counter: PyData, + amount: PyNumber, + ) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_increment", kind_error, async move { + store.increment(&counter.0, amount.0).await.map(PyNumber) + }) + } + + fn increment_many<'py>( + &self, + py: Python<'py>, + increments: Vec<(PyData, PyNumber)>, + ) -> PyResult> { + let store = self.0.clone(); + let increments = self::increments(increments); + redis_async(py, "async_increment_pipeline", kind_error, async move { + let totals = store.increment_many(&increments).await?; + Ok(totals.into_iter().map(PyNumber).collect::>()) + }) + } + + /// Declares the increments on the request's post-call batch; `False` when there is no Redis. + fn increment_many_post_call( + &self, + request: &NativeRequestRedis, + increments: Vec<(PyData, PyNumber)>, + ) -> bool { + self.0 + .increment_many_post_call(&request.0, &self::increments(increments)) + } + + fn set_max<'py>( + &self, + py: Python<'py>, + counter: PyData, + spend: PyNumber, + ) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_set_max", kind_error, async move { + store + .set_max(&counter.0, spend.0) + .await + .map(|stored| stored.map(PyNumber)) + }) + } + + #[pyo3(signature = (counter, spend, *, increment_if_lost))] + fn seed<'py>( + &self, + py: Python<'py>, + counter: PyData, + spend: PyNumber, + increment_if_lost: bool, + ) -> PyResult> { + let store = self.0.clone(); + let lost = if increment_if_lost { + LostSeed::Increment + } else { + LostSeed::Answer + }; + redis_async(py, "async_set_cache", kind_error, async move { + store.seed(&counter.0, spend.0, lost).await.map(PyNumber) + }) + } + + #[pyo3(signature = (counter, spend, ttl_seconds = None))] + fn set<'py>( + &self, + py: Python<'py>, + counter: PyData, + spend: PyNumber, + ttl_seconds: Option, + ) -> PyResult> { + let store = self.0.clone(); + let ttl = ttl(ttl_seconds)?; + redis_async(py, "async_set_cache", kind_error, async move { + store.set(&counter.0, spend.0, ttl).await + }) + } + + fn delete<'py>( + &self, + py: Python<'py>, + counters: Vec>, + ) -> PyResult> { + let store = self.0.clone(); + let counters = unwrap_all(counters); + redis_async(py, "async_delete_cache", kind_error, async move { + store.delete(&counters).await + }) + } + + fn refresh_ttl<'py>( + &self, + py: Python<'py>, + counter: PyData, + ) -> PyResult> { + let store = self.0.clone(); + redis_async(py, "async_refresh_ttl", kind_error, async move { + store.refresh_ttl(&counter.0).await + }) + } + + fn get_local(&self, counter: PyData) -> PyResult> { + Ok(self + .0 + .get_local(&counter.0) + .map_err(kind_error)? + .map(PyNumber)) + } + + #[pyo3(signature = (counter, spend, ttl_seconds = None))] + fn set_local( + &self, + counter: PyData, + spend: PyNumber, + ttl_seconds: Option, + ) -> PyResult<()> { + self.0 + .set_local(&counter.0, spend.0, ttl(ttl_seconds)?) + .map_err(kind_error) + } + + fn delete_local(&self, counter: PyData) -> PyResult<()> { + self.0.delete_local(&counter.0).map_err(kind_error) + } + + fn increment_local( + &self, + counter: PyData, + amount: PyNumber, + ) -> PyResult { + self.0 + .increment_local(&counter.0, amount.0) + .map(PyNumber) + .map_err(kind_error) + } + + fn flush_local(&self) -> PyResult<()> { + self.0.flush_local().map_err(kind_error) + } + + fn snapshot<'py>(&self, py: Python<'py>) -> PyResult> { + let snapshot = PyDict::new(py); + for (key, spend) in self.0.snapshot().map_err(kind_error)? { + snapshot.set_item(key, PyNumber(spend))?; + } + Ok(snapshot) + } +} diff --git a/litellm-rust/crates/redis-store/AGENTS.md b/litellm-rust/crates/redis-store/AGENTS.md index 9aad49d1d71..cdd2bd7ca64 100644 --- a/litellm-rust/crates/redis-store/AGENTS.md +++ b/litellm-rust/crates/redis-store/AGENTS.md @@ -1,11 +1,11 @@ -- One folder per Redis family (`router_cooldowns/`, `router_usage/`, `spend_counters/`, `auth_cache/`, `rate_limits/`), each with `mod.rs` (the store and its ops), `keys.rs` (key types and spellings) and `types.rs` (Python-facing values with their `FromValue` / `IntoValue`). It is the only place that family's keys, value codec and scripts are written, and it takes domain ids, never keys. Shared pieces live once at the crate root: `Error` and `ErrorKind` (`error.rs`), `Memory` and `Clock` (`memory.rs`), `Pending` (`request.rs`), `Remote` and its attach-once `RemoteSlot`, which knows nothing of memory (`remote.rs`), `DualCache` (`dual_cache/`) and the Python codecs (`codec.rs`, `value.rs`) +- One folder per Redis family (`router_cooldowns/`, `router_usage/`, `spend_counters/`, `auth_cache/`, `rate_limits/`), each with `mod.rs` (the store and its ops), `keys.rs` (key types and spellings) and `types.rs` (the family's values, with the `FromValue` / `IntoValue` codecs for what it stores in Redis). It is the only place that family's keys, value codec and scripts are written, and it takes domain ids, never keys. Shared pieces live once at the crate root: `Error` and `ErrorKind` (`error.rs`), `Memory` and `Clock` (`memory.rs`), `Pending` (`request.rs`), `Remote` and its attach-once `RemoteSlot`, which knows nothing of memory (`remote.rs`), `DualCache` (`dual_cache/`) and the Python codecs (`codec.rs`, `value.rs`) - An op declared on a `Send` returns `Pending`, a lazy future built when the op is declared. An op that only means something on Redis (an `MGET` the caller falls back from, a script) returns `None` without one, and an op that replaced a `DualCache` call answers from memory. A post-call method returns `false` when there was no Redis to declare on - A store that sends a pipeline of its own uses `RedisBatch::pipelined`, never `Pipeline` and raw `Value`s - Python and Rust pods share these keys during a rolling deploy, so a module writes the bytes `main`'s Python writes and reads everything `main` reads. Python formats go through `litellm-python-compat` (`repr`, `literal_eval`, `json`), never a hand-rolled codec - A family holds one `DualCache` per `DualCache` its Python held on `main`, and each `DualCache` method is the Python method it names, so a family op that was a `DualCache` call is one call here. Where `main`'s Python reached past `DualCache` to `in_memory_cache` or `redis_cache`, the family does the same through `memory()` and `redis()`, which keeps that reach visible. Caches a family's Python attached to one `RedisCache` share one `Arc`. Only a family that batch-reads builds its cache with `DualCache::batch_reading` from a `TierPolicy` the host passes in, and the Python host fills it from what `main` reads from the environment or `litellm` globals (`litellm/rust_bridge/host/settings.py`, `router_tiers`). Callers pass a `Remote` (or none), never a cache. Redis-only families are plain functions over `&RedisClient` - Every family works with no Redis, answering from memory. `attach` adds the Redis tier at most once and keeps memory, as `DualCache.attach_redis_cache` does, so a host builds the store at startup and never rebuilds it to connect - Where `main` swallows a Redis failure and serves memory, the family does the same and logs. It never turns a failure into a miss that `main` would not -- Key suffixes and wire spellings derive from strum on the family's enums, and a type whose Python side is plain data implements `FromValue` / `IntoValue` over `litellm_python_compat::Value`, so the bridge never names a field or a spelling +- Key suffixes and wire spellings derive from strum on the family's enums. A value the family stores in Redis implements `FromValue` / `IntoValue` over `litellm_python_compat::Value`. An argument or reply is a typed record, never a positional tuple: a frozen Python dataclass read in `python-bridge` by a `#[derive(FromPyObject)]` struct with `#[pyo3(attribute)]` fields and converted into the family's type with `From`, and a reply returned as a frozen `#[pyclass]`. The bridge reads an enum through its strum spelling, never a hand-written match - Wire contracts are hand-written cases with expected values, inlined as the same literals in this crate's test and in `tests/unit/redis/stores/`, checked here against real Redis and there against Python's own codec. Change both sides together - `rate_limits.rs` keeps the v3 limiter's Lua byte for byte (`rate_limits/lua.rs`) so pods on either stack share one script SHA; the admission decisions stay in the Python limiter - `RequestRedis` is one request's batches: a pre-call and a post-call `RedisBatch` per client, the post one flushed by `flush_post` or its deadline. Each op exists once and takes a `Send` (`Now`, `PreCall`, `PostCall`) picked by the caller, and the store calls `send.batch(&client)`. A store never encodes a phase in a method name. A post-call method exists only where a follow-up runs after the write settles (`token_increment_post_call`), as a thin wrapper over the shared op. `Send::Now` is an alone batch, so a script call outside a request goes out alone, like a redis-py `Script` outside a pipeline, so `NOSCRIPT` costs `SCRIPT LOAD` and one more `EVALSHA` the way it does on main. Only what main's Python puts on its request pipeline rides these batches. An op main sends on its own round trip, like a deployment's tpm and rpm increment, stays on its own round trip here diff --git a/litellm-rust/crates/redis-store/src/auth_cache/keys.rs b/litellm-rust/crates/redis-store/src/auth_cache/keys.rs new file mode 100644 index 00000000000..feb87a40b00 --- /dev/null +++ b/litellm-rust/crates/redis-store/src/auth_cache/keys.rs @@ -0,0 +1,8 @@ +/// `is_user_key_cache_key`: user-key objects are cached under a bare `hash_token` digest, every +/// other object under a prefixed key. +pub fn is_key_object(key: &str) -> bool { + key.len() == 64 + && key + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} diff --git a/litellm-rust/crates/redis-store/src/auth_cache/mod.rs b/litellm-rust/crates/redis-store/src/auth_cache/mod.rs new file mode 100644 index 00000000000..b5a83b5aec6 --- /dev/null +++ b/litellm-rust/crates/redis-store/src/auth_cache/mod.rs @@ -0,0 +1,258 @@ +mod keys; +mod types; + +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; + +use litellm_python_compat::{Value, json}; +use litellm_redis_client::{Declared, RedisBatch, commands}; + +pub use self::keys::is_key_object; +pub use self::types::Settings; +use crate::{ + Error, Number, Pending, RequestRedis, Send, + dual_cache::DualCache, + memory::{Clock, lock, system_clock}, + remote::{Remote, RemoteSlot}, + request::ready, +}; + +/// `RedisCache.async_increment_if_present`, byte for byte, sent as `EVAL` like it. +const INCREMENT_IF_PRESENT: &str = "if redis.call('EXISTS', KEYS[1]) == 0 then return false end return redis.call('INCRBYFLOAT', KEYS[1], ARGV[1])"; + +/// The proxy's auth object cache with `enable_redis_auth_cache` on: `UserApiKeyCache`'s two +/// `DualCache` partitions, so churn in management objects cannot evict user-key objects, in front of +/// one Redis shared by every pod. Keys are the logical cache keys `UserApiKeyCache` callers already +/// use; values are what `json.loads` returns, written as `json.dumps`, like `RedisCache.async_set_cache`. +pub struct AuthCache { + key_objects: DualCache, + objects: DualCache, + redis: Arc, + settings: Mutex, +} + +impl AuthCache { + pub fn new(remote: Option, settings: Settings) -> Self { + Self::with_clock(remote, settings, system_clock()) + } + + pub fn with_clock(remote: Option, settings: Settings, now: Clock) -> Self { + let redis = Arc::new(RemoteSlot::new(remote)); + let partition = || DualCache::new(redis.clone(), settings.memory(), now.clone()); + Self { + key_objects: partition(), + objects: partition(), + redis, + settings: Mutex::new(settings), + } + } + + pub fn has_redis(&self) -> bool { + self.redis.get().is_some() + } + + pub fn settings(&self) -> Settings { + *lock(&self.settings) + } + + /// `UserApiKeyCache.update_cache_ttl` and `update_in_memory_max_size` on both partitions, which + /// keep what they hold. + pub fn reconfigure(&self, settings: Settings) { + *lock(&self.settings) = settings; + self.key_objects.reconfigure(settings.memory()); + self.objects.reconfigure(settings.memory()); + } + + pub fn get_local(&self, key: &str) -> Result, Error> { + Ok(self.partition(key).memory().get(key)?) + } + + /// A memory write with `ttl`, or `default_in_memory_ttl` when it names none. + pub fn set_local(&self, key: &str, value: Value, ttl: Option) -> Result<(), Error> { + let partition = self.partition(key); + Ok(partition + .memory() + .set(key, value, ttl.or(partition.default_in_memory_ttl()))?) + } + + pub fn delete_local(&self, key: &str) -> Result<(), Error> { + Ok(self.partition(key).memory().delete(key)?) + } + + pub fn flush_local(&self) -> Result<(), Error> { + self.key_objects.memory().flush()?; + Ok(self.objects.memory().flush()?) + } + + /// Entries held by the management-object and the user-key partitions. + pub fn len_local(&self) -> Result<(usize, usize), Error> { + Ok(( + self.objects.memory().len()?, + self.key_objects.memory().len()?, + )) + } + + /// `DualCache.async_get_cache`. A key `request` already read as missing from Redis is not read + /// again. + pub async fn get( + &self, + request: Option<&RequestRedis>, + key: &str, + ) -> Result, Error> { + self.partition(key).get(request, key, None).await + } + + /// `DualCache.async_set_cache`, writing `json.dumps(value)`. + pub async fn set(&self, key: &str, value: Value, ttl: Option) -> Result<(), Error> { + self.partition(key).set(key, value, ttl).await + } + + /// `DualCache.set_cache`, whose `RedisCache.set_cache` writes `str(value)`. + pub async fn set_str( + &self, + key: &str, + value: Value, + ttl: Option, + ) -> Result<(), Error> { + self.partition(key).set_str(key, value, ttl).await + } + + /// `DualCache.async_set_cache_pre_call`: memory now, the `SET` on the batch `send` picks. + pub fn declare_set( + &self, + send: Send<'_>, + key: &str, + value: Value, + ttl: Option, + ) -> Result, Error> { + self.partition(key).declare_set(send, key, value, ttl) + } + + /// `DualCache.async_delete_cache_pre_call`, memory now and the `DEL` on the batch `send` picks. + pub fn declare_delete(&self, send: Send<'_>, key: &str) -> Result, Error> { + self.partition(key).declare_delete(send, key) + } + + /// `UserApiKeyCache.async_set_cache_pipeline`: the key-object entries in one pipeline, then + /// the rest in another, each with one TTL. + pub async fn set_many( + &self, + entries: Vec<(String, Value)>, + ttl: Option, + ) -> Result<(), Error> { + let (key_objects, others): (Vec<_>, Vec<_>) = + entries.into_iter().partition(|(key, _)| is_key_object(key)); + self.key_objects.set_pipeline(key_objects, ttl).await?; + self.objects.set_pipeline(others, ttl).await + } + + /// The auth prefetch write-back: memory first, an entry without a TTL taking + /// `default_in_memory_ttl`, then `RedisCache.async_set_cache_pipeline_with_ttls`, which encodes + /// every value before it sends anything and gives an entry without a TTL the connection's + /// default. On a request batch the `SET`s are declared and nothing is left to wait for. + /// [`Send::Now`] is one pipeline of its own rather than one round trip per entry. + pub fn write_back( + &self, + send: Send<'_>, + entries: Vec<(String, Value, Option)>, + ) -> Result, Error> { + let encoded = self + .redis + .get() + .map(|_| { + entries + .iter() + .map(|(_, value, _)| json::dumps(value)) + .collect::, _>>() + }) + .transpose(); + let writes = entries + .into_iter() + .map(|(key, value, ttl)| { + self.set_local(&key, value, ttl)?; + Ok((key, ttl)) + }) + .collect::, Error>>()?; + let (Some(remote), Some(encoded)) = (self.redis.get(), encoded?) else { + return Ok(ready(())); + }; + let sets = writes.iter().zip(encoded).map(|((key, ttl), value)| { + commands::set(key, value, ttl.unwrap_or(remote.default_ttl)) + }); + let Send::Now = send else { + let batch = send.batch(&remote.client); + sets.for_each(|set| drop(batch.add(set))); + return Ok(ready(())); + }; + let pipeline = RedisBatch::pipelined(remote.client.clone()); + let written: Vec<_> = sets.map(|set| pipeline.add(set)).collect(); + Ok(Box::pin(async move { + Declared::all(written) + .await + .into_iter() + .try_for_each(|reply| Ok(reply?)) + })) + } + + /// `UserApiKeyCache.async_delete_cache_keys`: the key-object partition and the rest as two + /// concurrent `DualCache.async_delete_cache_keys`. Both run even when one fails. + pub async fn delete(&self, keys: &[String]) -> Result<(), Error> { + let (key_objects, others): (Vec, Vec) = + keys.iter().cloned().partition(|key| is_key_object(key)); + let (key_objects, others) = tokio::join!( + self.key_objects.delete_keys(&key_objects), + self.objects.delete_keys(&others) + ); + key_objects.and(others) + } + + /// `RedisCache.async_batch_get_cache` for keys the caller found missing in memory, one `MGET` on + /// the batch `send` picks, noting the keys Redis lacks as the request's misses. `None` without + /// Redis. Any failed read or undecodable value fails the whole read, as a failed `MGET` does. + pub fn get_many(&self, send: Send<'_>, keys: &[String]) -> Option>>> { + let client = self.redis.client()?; + let misses = send.misses(client); + let read = send.batch(client).add(commands::mget::>( + &keys.iter().map(String::as_str).collect::>(), + )); + let keys = keys.to_vec(); + Some(Box::pin(async move { + let raw = read.resolve().await?; + if let Some(misses) = misses { + misses.note( + keys.into_iter() + .zip(&raw) + .filter(|(_, raw)| raw.is_none()) + .map(|(key, _)| key), + ); + } + raw.into_iter().map(|raw| Ok(types::decode(raw)?)).collect() + })) + } + + /// `RedisCache.async_increment_if_present`: `INCRBYFLOAT` only an entry Redis still holds. + /// `None` without Redis or for an absent entry. + pub async fn increment_if_present( + &self, + key: &str, + amount: Number, + ) -> Result, Error> { + let Some(client) = self.redis.client() else { + return Ok(None); + }; + let amount = litellm_redis_client::Amount::from(amount); + Ok(client + .send(commands::eval_source(INCREMENT_IF_PRESENT, &[key], amount)) + .await?) + } + + fn partition(&self, key: &str) -> &DualCache { + if is_key_object(key) { + &self.key_objects + } else { + &self.objects + } + } +} diff --git a/litellm-rust/crates/redis-store/src/auth_cache/types.rs b/litellm-rust/crates/redis-store/src/auth_cache/types.rs new file mode 100644 index 00000000000..e933ac19af9 --- /dev/null +++ b/litellm-rust/crates/redis-store/src/auth_cache/types.rs @@ -0,0 +1,35 @@ +use std::time::Duration; + +use litellm_python_compat::Value; + +use crate::{DecodeError, codec::python_value, dual_cache::MemorySettings}; + +/// `UserApiKeyCache`'s memory configuration: each partition's size and `default_in_memory_ttl`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Settings { + pub memory_entries: usize, + pub memory_ttl: Option, +} + +impl Default for Settings { + /// `InMemoryCache`'s 200 entries. + fn default() -> Self { + Self { + memory_entries: 200, + memory_ttl: None, + } + } +} + +impl Settings { + pub(super) fn memory(self) -> MemorySettings { + MemorySettings { + entries: Some(self.memory_entries), + ttl: self.memory_ttl, + } + } +} + +pub(super) fn decode(raw: Option>) -> Result, DecodeError> { + raw.as_deref().map(python_value).transpose() +} diff --git a/litellm-rust/crates/redis-store/src/lib.rs b/litellm-rust/crates/redis-store/src/lib.rs index b7ee6e3e35e..69600aac2a6 100644 --- a/litellm-rust/crates/redis-store/src/lib.rs +++ b/litellm-rust/crates/redis-store/src/lib.rs @@ -1,23 +1,17 @@ -#![expect( - dead_code, - reason = "the store families built on these tiers and codecs land in the next PR" -)] -#![cfg_attr( - not(test), - expect( - unused_imports, - reason = "the store families built on these tiers and codecs land in the next PR" - ) -)] - +pub mod auth_cache; mod codec; mod dual_cache; mod error; mod failure_log; mod memory; mod number; +pub mod rate_limits; mod remote; mod request; +pub mod router_cooldowns; +pub mod router_usage; +pub mod sensitive_route_pins; +pub mod spend_counters; mod value; pub use codec::DecodeError; diff --git a/litellm-rust/crates/redis-store/src/rate_limits/keys.rs b/litellm-rust/crates/redis-store/src/rate_limits/keys.rs new file mode 100644 index 00000000000..7f5714a19ad --- /dev/null +++ b/litellm-rust/crates/redis-store/src/rate_limits/keys.rs @@ -0,0 +1,74 @@ +/// A rate-limit descriptor, `{descriptor:value}`, the hash tag every key of its window shares. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct Scope { + pub descriptor: String, + pub value: String, +} + +impl Scope { + pub(super) fn key(&self, suffix: &str) -> String { + format!("{{{}:{}}}:{suffix}", self.descriptor, self.value) + } + + pub(super) fn window_key(&self) -> String { + self.key(WINDOW) + } + + pub(super) fn gauge_key(&self) -> String { + self.key(GAUGE) + } +} + +const WINDOW: &str = "window"; +const GAUGE: &str = "max_parallel_requests"; + +/// A counter's key suffix, which is also how the Python limiter names it. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, strum::EnumString, strum::IntoStaticStr)] +pub enum Kind { + #[strum(serialize = "requests")] + Requests, + #[strum(serialize = "tokens")] + Tokens, +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct Counter { + pub scope: Scope, + pub kind: Kind, +} + +impl Counter { + pub(super) fn key(&self) -> String { + self.scope.key(self.kind.into()) + } +} + +/// One value of the limiter's keyspace, as both tiers address it. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum Entry { + Window(Scope), + Counter(Counter), + Gauge(Scope), +} + +impl Entry { + /// The entry the Python limiter names by its key suffix: `window`, `max_parallel_requests`, or a + /// counter's `Kind`. + pub fn new(suffix: &str, scope: Scope) -> Result { + match suffix { + WINDOW => Ok(Self::Window(scope)), + GAUGE => Ok(Self::Gauge(scope)), + kind => kind + .parse() + .map(|kind| Self::Counter(Counter { scope, kind })), + } + } + + pub(super) fn key(&self) -> String { + match self { + Self::Window(scope) => scope.window_key(), + Self::Counter(counter) => counter.key(), + Self::Gauge(scope) => scope.gauge_key(), + } + } +} diff --git a/litellm-rust/crates/redis-store/src/rate_limits/lua.rs b/litellm-rust/crates/redis-store/src/rate_limits/lua.rs new file mode 100644 index 00000000000..96451212d35 --- /dev/null +++ b/litellm-rust/crates/redis-store/src/rate_limits/lua.rs @@ -0,0 +1,260 @@ +//! The limiter's Lua, byte for byte as `main`'s Python registers it, so both share one script SHA. + +pub(super) const BATCH_RATE_LIMITER: &str = r#" +local results = {} +local now = tonumber(ARGV[1]) +local window_size = tonumber(ARGV[2]) + +-- Process each window/counter pair +for i = 1, #KEYS, 2 do + local window_key = KEYS[i] + local counter_key = KEYS[i + 1] + local increment_value = 1 + + -- Check if window exists and is valid + local window_start = redis.call('GET', window_key) + if not window_start or (now - tonumber(window_start)) >= window_size then + -- Reset window and counter + local prefix = string.sub(window_key, 1, -(#':window') - 1) + redis.call('DEL', prefix .. ':requests', prefix .. ':tokens') + redis.call('SET', window_key, tostring(now)) + redis.call('SET', counter_key, increment_value) + redis.call('EXPIRE', window_key, window_size) + redis.call('EXPIRE', counter_key, window_size) + table.insert(results, tostring(now)) -- window_start + table.insert(results, increment_value) -- counter + else + local counter = redis.call('INCR', counter_key) + -- This happens when window_key exists but counter_key doesn't (e.g., tokens key + -- created after requests key when both share the same window_key) + local current_ttl = redis.call('TTL', counter_key) + if current_ttl == -1 then + redis.call('EXPIRE', counter_key, window_size) + end + table.insert(results, window_start) -- window_start + table.insert(results, counter) -- counter + end +end + +return results +"#; + +pub(super) const BATCH_COUNTER_READ: &str = r#"return redis.call('MGET', unpack(KEYS))"#; + +pub(super) const CHECK_AND_INCREMENT_BY_N: &str = r#" +-- Atomic check-and-increment-by-N across one or more descriptors. +-- All-or-nothing: if any descriptor would exceed its limit, no counter is +-- modified. +-- +-- Uses Redis server time (`redis.call('TIME')`) instead of a client-supplied +-- timestamp so that window resets are deterministic across replicas with +-- skewed wall-clocks. This prevents a clock-skew-induced reopening of the +-- TOCTOU window across multi-replica deployments. +-- +-- KEYS layout: pairs of (window_key, counter_key), one pair per descriptor. +-- ARGV layout: per-descriptor 4-tuple, starting at ARGV[1]: +-- ARGV[(i-1)*4 + 1] = limit +-- ARGV[(i-1)*4 + 2] = increment +-- ARGV[(i-1)*4 + 3] = ttl_seconds (counter TTL when window resets) +-- ARGV[(i-1)*4 + 4] = window_size_seconds (sliding-window length) +-- +-- Return on success: +-- { 0, new_counter_1, window_start_1, new_counter_2, window_start_2, ... } +-- Return on over-limit: { 1, descriptor_index, current_counter, limit } +local time_reply = redis.call('TIME') +local now = tonumber(time_reply[1]) +local descriptor_count = #KEYS / 2 +local reset_windows = {} + +-- Pass 1: read state, validate. Abort without writing if any over limit. +local descriptor_state = {} +for i = 1, descriptor_count do + local window_key = KEYS[(i - 1) * 2 + 1] + local counter_key = KEYS[(i - 1) * 2 + 2] + local arg_base = (i - 1) * 4 + 1 + local limit = tonumber(ARGV[arg_base]) + local increment = tonumber(ARGV[arg_base + 1]) + local window_size = tonumber(ARGV[arg_base + 3]) + + local window_start = redis.call('GET', window_key) + local window_expired = (not window_start) or + ((now - tonumber(window_start)) >= window_size) + + local current_counter + if window_expired then + current_counter = 0 + else + current_counter = tonumber(redis.call('GET', counter_key) or 0) + end + + local blocked + if increment > 0 then + blocked = current_counter + increment > limit + else + blocked = current_counter >= limit + end + if blocked then + return { 1, i, current_counter, limit } + end + + descriptor_state[i] = { window_expired, current_counter, window_start } +end + +-- Pass 2: all checks passed. Apply increments. +local results = { 0 } +for i = 1, descriptor_count do + local window_key = KEYS[(i - 1) * 2 + 1] + local counter_key = KEYS[(i - 1) * 2 + 2] + local arg_base = (i - 1) * 4 + 1 + local increment = tonumber(ARGV[arg_base + 1]) + local ttl = tonumber(ARGV[arg_base + 2]) + local window_size = tonumber(ARGV[arg_base + 3]) + + local window_expired = descriptor_state[i][1] + local active_window_start + + if window_expired then + active_window_start = now + if not reset_windows[window_key] then + local prefix = string.sub(window_key, 1, -(#':window') - 1) + redis.call('DEL', prefix .. ':requests', prefix .. ':tokens') + reset_windows[window_key] = true + end + redis.call('SET', window_key, tostring(now)) + redis.call('SET', counter_key, increment) + redis.call('EXPIRE', window_key, window_size) + if ttl > 0 then + redis.call('EXPIRE', counter_key, ttl) + end + table.insert(results, increment) + else + active_window_start = tonumber(descriptor_state[i][3]) + local new_counter = redis.call('INCRBY', counter_key, increment) + local current_ttl = redis.call('TTL', counter_key) + if current_ttl == -1 and ttl > 0 then + redis.call('EXPIRE', counter_key, ttl) + end + table.insert(results, new_counter) + end + table.insert(results, active_window_start) +end + +return results +"#; + +pub(super) const WINDOW_GUARDED_TOKEN_INCREMENT: &str = r#" +local results = {} +for i = 1, #KEYS, 2 do + local window_key = KEYS[i] + local counter_key = KEYS[i + 1] + local arg_base = ((i - 1) / 2) * 3 + 1 + local expected_window_start = ARGV[arg_base] + local increment = tonumber(ARGV[arg_base + 1]) + local ttl = tonumber(ARGV[arg_base + 2]) + local active_window_start = redis.call('GET', window_key) + + if active_window_start and active_window_start == expected_window_start then + local new_counter = redis.call('INCRBY', counter_key, increment) + local current_ttl = redis.call('TTL', counter_key) + if current_ttl == -1 and ttl > 0 then + redis.call('EXPIRE', counter_key, ttl) + end + table.insert(results, 1) + table.insert(results, new_counter) + else + table.insert(results, 0) + table.insert(results, tonumber(redis.call('GET', counter_key) or 0)) + end +end +return results +"#; + +pub(super) const PARALLEL_ACQUIRE: &str = r#" +-- Atomic check-and-acquire for the max_parallel_requests concurrency gauge. +-- Each gauge key is a sorted set of per-request slot ids scored by acquire +-- time (Redis server clock). In-flight requests are counted by ZCARD after +-- pruning slots older than the slot TTL, so unlike the windowed RPM/TPM +-- counters the gauge is never reset while requests are in flight, a +-- rejected request never occupies a slot, and a slot leaked by a crashed +-- worker self-heals after the slot TTL even under continuous traffic. +-- +-- KEYS: one gauge zset key per descriptor. +-- ARGV: per-key triples (limit, slot_ttl_seconds, slot_id). +-- Success: { 0, in_flight_1, ... }. Over-limit: { 1, key_index, in_flight, limit }. +local time_reply = redis.call('TIME') +local now = tonumber(time_reply[1]) +for i = 1, #KEYS do + local limit = tonumber(ARGV[(i - 1) * 3 + 1]) + local slot_ttl = tonumber(ARGV[(i - 1) * 3 + 2]) + redis.call('ZREMRANGEBYSCORE', KEYS[i], '-inf', now - slot_ttl) + local in_flight = redis.call('ZCARD', KEYS[i]) + if in_flight + 1 > limit then + return { 1, i, in_flight, limit } + end +end +local results = { 0 } +for i = 1, #KEYS do + local slot_ttl = tonumber(ARGV[(i - 1) * 3 + 2]) + local slot_id = ARGV[(i - 1) * 3 + 3] + redis.call('ZADD', KEYS[i], now, slot_id) + redis.call('EXPIRE', KEYS[i], slot_ttl) + table.insert(results, redis.call('ZCARD', KEYS[i])) +end +return results +"#; + +pub(super) const PARALLEL_RELEASE: &str = r#" +-- Release one slot per gauge key by removing this request's slot id. +-- ZREM of an absent member (or key) is a no-op, so a release without a +-- matching acquire (proxy-side rejection, double-fired callback, slot +-- already expired) can never free a slot owned by another request. +-- KEYS: gauge zset keys. ARGV: per-key slot_id. +-- Returns the remaining in-flight count per key. +local results = {} +for i = 1, #KEYS do + redis.call('ZREM', KEYS[i], ARGV[i]) + table.insert(results, redis.call('ZCARD', KEYS[i])) +end +return results +"#; + +pub(super) const PARALLEL_COUNT: &str = r#" +-- Read the current in-flight count per gauge key (prunes expired slots +-- first so leaked slots do not inflate the reading). +-- KEYS: gauge zset keys. ARGV: per-key slot_ttl_seconds. +local time_reply = redis.call('TIME') +local now = tonumber(time_reply[1]) +local results = {} +for i = 1, #KEYS do + redis.call('ZREMRANGEBYSCORE', KEYS[i], '-inf', now - tonumber(ARGV[i])) + table.insert(results, redis.call('ZCARD', KEYS[i])) +end +return results +"#; + +pub(super) const TOKEN_INCREMENT: &str = r#" +local results = {} + +-- Process each key/increment_value/ttl triplet +for i = 1, #KEYS do + local key = KEYS[i] + local increment_value = tonumber(ARGV[i * 2 - 1]) + local ttl_seconds = tonumber(ARGV[i * 2]) + + -- Increment the value + local new_value = redis.call('INCRBYFLOAT', key, increment_value) + + -- Handle TTL: only set expire if ttl_seconds > 0 and key has no current TTL + -- ttl_seconds can be 0 (no TTL) or positive (set TTL) + if ttl_seconds and ttl_seconds > 0 then + local current_ttl = redis.call('TTL', key) + if current_ttl == -1 then + redis.call('EXPIRE', key, ttl_seconds) + end + end + + table.insert(results, new_value) +end + +return results +"#; diff --git a/litellm-rust/crates/redis-store/src/rate_limits/mod.rs b/litellm-rust/crates/redis-store/src/rate_limits/mod.rs new file mode 100644 index 00000000000..71b67d2ecc0 --- /dev/null +++ b/litellm-rust/crates/redis-store/src/rate_limits/mod.rs @@ -0,0 +1,304 @@ +mod keys; +mod lua; +mod scripts; +mod types; + +use std::{sync::Arc, time::Duration}; + +use litellm_redis_client::{Declared, RedisBatch, RedisClient, Slot}; + +pub use self::keys::{Counter, Entry, Kind, Scope}; +pub use self::types::{ + AcquireReply, Applied, AtomicCounter, AtomicReply, Gauge, GuardedIncrement, GuardedReply, + Increment, Local, WindowCount, +}; +use crate::{ + Error, Number, Pending, RequestRedis, Send, + dual_cache::{DualCache, MemorySettings, PipelineIncrement}, + memory::{Clock, system_clock}, + remote::{Remote, RemoteSlot}, + request::pending, +}; + +/// The v3 rate limiter's keyspace: its memory tier, its Lua scripts and the plain counter writes +/// `DualCache` made for it on `main`. The admission logic stays with the caller. A script needs +/// Redis, so without one it declares nothing and answers `None`. +pub struct RateLimits { + cache: DualCache, +} + +impl RateLimits { + /// `backfill_ttl` is the `default_in_memory_ttl` of the `DualCache` this tier replaces. An + /// increment without a TTL takes the remote's default. + pub fn new(remote: Option, backfill_ttl: Option) -> Self { + Self::with_clock(remote, backfill_ttl, system_clock()) + } + + pub fn with_clock(remote: Option, backfill_ttl: Option, now: Clock) -> Self { + Self { + cache: DualCache::new( + Arc::new(RemoteSlot::new(remote)), + MemorySettings { + entries: None, + ttl: backfill_ttl, + }, + now, + ), + } + } + + pub fn has_redis(&self) -> bool { + self.cache.redis().get().is_some() + } + + /// `_group_keys_by_hash_tag`: indices of `scopes` sharing a cluster slot, in first-seen order, + /// or one group when the client is not a cluster and grouping is not forced. + pub fn slot_groups(&self, scopes: &[Scope], force: bool) -> Vec> { + let by_slot = force + || self + .cache + .redis() + .client() + .is_some_and(RedisClient::is_cluster); + if !by_slot { + return vec![(0..scopes.len()).collect()]; + } + scopes + .iter() + .enumerate() + .fold( + Vec::<(Slot, Vec)>::new(), + |mut groups, (index, scope)| { + let slot = Slot::for_key(scope.window_key()); + match groups.iter_mut().find(|(seen, _)| *seen == slot) { + Some((_, members)) => members.push(index), + None => groups.push((slot, vec![index])), + } + groups + }, + ) + .into_iter() + .map(|(_, members)| members) + .collect() + } + + pub fn get_local(&self, entry: &Entry) -> Result, Error> { + Ok(self.cache.memory().get(&entry.key())?) + } + + pub fn set_local( + &self, + entry: &Entry, + value: Local, + ttl: Option, + ) -> Result<(), Error> { + Ok(self.cache.memory().set(&entry.key(), value, ttl)?) + } + + /// `DualCache.async_get_cache`, backfilling memory for `backfill_ttl` or the store's default. + /// A key `request` already read as missing from Redis is not read again. + pub async fn get( + &self, + request: Option<&RequestRedis>, + entry: &Entry, + backfill_ttl: Option, + ) -> Result, Error> { + self.cache.get(request, &entry.key(), backfill_ttl).await + } + + pub fn sliding_window( + &self, + send: Send<'_>, + counters: &[Counter], + now: i64, + window_size: i64, + ) -> Option>> { + Some(pending(scripts::sliding_window( + &self.batch(send)?, + counters, + now, + window_size, + ))) + } + + pub fn read_counters( + &self, + send: Send<'_>, + counters: &[Counter], + ) -> Option>>> { + Some(pending(scripts::read_counters( + &self.batch(send)?, + counters, + ))) + } + + pub fn check_and_increment( + &self, + send: Send<'_>, + counters: &[AtomicCounter], + ) -> Option> { + Some(pending(scripts::check_and_increment( + &self.batch(send)?, + counters, + ))) + } + + pub fn guarded_increment( + &self, + send: Send<'_>, + increment: &GuardedIncrement, + ) -> Option> { + Some(pending(scripts::guarded_increment( + &self.batch(send)?, + increment, + ))) + } + + pub fn parallel_acquire( + &self, + send: Send<'_>, + gauges: &[Gauge], + slot_ttl: Duration, + slot_id: &str, + ) -> Option> { + Some(pending(scripts::parallel_acquire( + &self.batch(send)?, + gauges, + slot_ttl, + slot_id, + ))) + } + + pub fn parallel_release( + &self, + send: Send<'_>, + gauges: &[Scope], + slot_id: &str, + ) -> Option>> { + Some(pending(self.declare_release(send, gauges, slot_id)?)) + } + + pub fn parallel_count( + &self, + send: Send<'_>, + gauges: &[Scope], + slot_ttl: Duration, + ) -> Option>> { + Some(pending(scripts::parallel_count( + &self.batch(send)?, + gauges, + slot_ttl, + ))) + } + + pub fn token_increment( + &self, + send: Send<'_>, + increments: &[Increment], + ) -> Option>> { + Some(pending(self.declare_token_increment(send, increments)?)) + } + + /// `TOKEN_INCREMENT_SCRIPT` on the request's post-call batch. A failed script falls back to + /// `increment_many` for these increments once the batch has gone out. `false` without Redis. + pub fn token_increment_post_call( + self: &Arc, + request: &RequestRedis, + increments: Vec, + ) -> bool { + let Some(declared) = self.declare_token_increment(Send::PostCall(request), &increments) + else { + return false; + }; + let store = Arc::clone(self); + request.spawn(async move { + let Err(error) = declared.settled().await else { + return; + }; + crate::failure_log::redis_failure!( + &error, + "TTL preservation failed, falling back to regular pipeline" + ); + store.increment_many(&increments).await; + }); + true + } + + /// `PARALLEL_RELEASE_SCRIPT` on the request's post-call batch; the caller has already freed + /// the slot in memory. `false` without Redis. + pub fn parallel_release_post_call( + &self, + request: &RequestRedis, + gauges: &[Scope], + slot_id: &str, + ) -> bool { + let Some(declared) = self.declare_release(Send::PostCall(request), gauges, slot_id) else { + return false; + }; + request.spawn(async move { + if let Err(error) = declared.settled().await { + crate::failure_log::redis_failure!( + &error, + "parallel_release_script failed, the slot stays released in memory only" + ); + } + }); + true + } + + /// `DualCache.async_increment_cache`. A Redis failure answers with the memory count, and an + /// entry memory cannot add to with `None`. + pub async fn increment(&self, increment: &Increment) -> Option { + self.cache + .increment(&increment.counter.key(), increment.amount, increment.ttl) + .await + } + + /// `DualCache.async_increment_cache_pipeline`, whose failures only log. + pub async fn increment_many(&self, increments: &[Increment]) { + self.cache.increment_pipeline(&pipeline(increments)).await; + } + + /// `DualCache.async_increment_cache_pipeline_post_call`: memory now, the Redis increments on + /// the request's post-call batch. + pub fn increment_many_post_call(&self, request: &RequestRedis, increments: &[Increment]) { + self.cache + .increment_pipeline_post_call(request, &pipeline(increments)); + } + + fn declare_release( + &self, + send: Send<'_>, + gauges: &[Scope], + slot_id: &str, + ) -> Option>> { + Some(scripts::parallel_release( + &self.batch(send)?, + gauges, + slot_id, + )) + } + + fn declare_token_increment( + &self, + send: Send<'_>, + increments: &[Increment], + ) -> Option>> { + Some(scripts::token_increment(&self.batch(send)?, increments)) + } + + fn batch(&self, send: Send<'_>) -> Option { + self.cache.redis().batch(send) + } +} + +fn pipeline(increments: &[Increment]) -> Vec { + increments + .iter() + .map(|increment| PipelineIncrement { + key: increment.counter.key(), + amount: increment.amount, + ttl: increment.ttl, + }) + .collect() +} diff --git a/litellm-rust/crates/redis-store/src/rate_limits/scripts.rs b/litellm-rust/crates/redis-store/src/rate_limits/scripts.rs new file mode 100644 index 00000000000..04bdad38268 --- /dev/null +++ b/litellm-rust/crates/redis-store/src/rate_limits/scripts.rs @@ -0,0 +1,303 @@ +use std::{sync::LazyLock, time::Duration}; + +use litellm_python_compat::repr::float_repr; +use litellm_redis_client::{ + Command, Declared, Error, RedisBatch, Script, Value, + commands::{self, unexpected_reply}, +}; + +use super::{ + keys::{Counter, Scope}, + lua, + types::{ + AcquireReply, Applied, AtomicCounter, AtomicReply, Gauge, GuardedIncrement, GuardedReply, + Increment, WindowCount, + }, +}; +use crate::Number; + +static BATCH_RATE_LIMITER: LazyLock