diff --git a/.agents/skills/redis-access/SKILL.md b/.agents/skills/redis-access/SKILL.md new file mode 100644 index 00000000000..7f77801623d --- /dev/null +++ b/.agents/skills/redis-access/SKILL.md @@ -0,0 +1,17 @@ +--- +name: redis-access +description: Add Redis operations or move raw redis-py clients and scripts behind RedisTransport in litellm/redis, without porting them to Rust or expanding migration scope +metadata: + internal: true +--- + +# Redis access + +Python reaches Redis through `RedisTransport` (`litellm/redis/transport/contract.py`). Read `references/transport.md` for adding a command, the failure kinds, the namespace and TTL rules, and moving raw `init_async_client` / `async_register_script` / `.redis_client` access onto the transport + +Rules that always apply + +- New code never calls `init_async_client`, `init_pubsub_client`, `async_register_script` or `.redis_client`, and never imports redis-py outside `litellm/redis/transport/redis_py.py` +- Keys and values in Redis are a wire format shared with older pods and Rust. A move keeps them byte for byte +- A `transport` failure on a write is an unknown outcome, never "not applied" +- Do only what was asked. Confirm before any Rust work diff --git a/.agents/skills/redis-access/references/transport.md b/.agents/skills/redis-access/references/transport.md new file mode 100644 index 00000000000..6d98a0c4135 --- /dev/null +++ b/.agents/skills/redis-access/references/transport.md @@ -0,0 +1,76 @@ +# RedisTransport + +How Python code reaches Redis. Read this before adding a command, or moving raw redis-py access off `RedisCache` + +## Why it exists + +Today commands reach Redis three ways + +- Cache methods on `RedisCache` and `RedisBatch` (get, set, incr, lists, pipelines), which apply the namespace and the circuit breaker +- Raw access, where callers take `init_async_client()`, `init_pubsub_client()`, `async_register_script()` or `.redis_client` and run commands, Lua and pub/sub themselves + - Scripts from `async_register_script` keep the namespace and breaker + - Raw clients skip both, so every caller namespaces keys and channels by hand, and some forget +- Vector search (`RedisSemanticCache` through RedisVL, `ValkeySemanticCache` and the Valkey vector store through `Redis.from_url`), which builds its own clients on separate connections + +`RedisTransport` replaces the raw path with one command-shaped seam, so the backend behind it can later swap from redis-py to `litellm-rust/crates/redis-client` without touching callers + +## The contract + +Defined in `litellm/redis/transport/contract.py` + +- Commands are frozen dataclasses, `Get(key)`, `Set(key, value, ttl, nx=False)` and `Delete(key)`. `Arg` is `str | bytes | int | float` +- The transport has one method, `execute(command)`, overloaded per command for its reply type (`bytes | None`, `bool`, `int`) +- Every command returns its value or a `RedisFailure`. Nothing raises for a Redis problem +- `RedisFailure.kind` + - `transport` is a timeout or lost connection. The command may still have applied, so a write that fails this way has an unknown outcome. `timeout=True` marks timeouts + - `server` means Redis answered with an error (WRONGTYPE). The connection is fine and the breaker does not count it + - `decode` means the reply was not the shape the method or store expected + - `breaker_open` means the breaker refused the call without touching the socket +- The namespace is applied once, inside the transport. A key that already starts with `{namespace}:` is left alone, matching `RedisCache.check_and_fix_namespace` and `crates/redis-client`. Callers and stores pass logical keys +- Every write takes its TTL from the caller. `ttl=None` means no expiry, there is no default +- Replies are undecoded bytes. Decoding belongs to whoever owns the keyspace + +These rules are shared with `crates/redis-client` (its `AGENTS.md` has the Rust side). If you change one, change both, since Python and Rust pods share keys in one Redis + +## The redis-py backend + +- `litellm/redis/transport/redis_py.py` holds `RedisPyTransport`, the only module that touches redis-py's client +- `RedisCache.transport()` builds one over that cache's own async client, namespace and circuit breaker. No new connection, no second breaker +- The guard is `BreakerGuard` in `redis_cache.py`. It runs the call under the cache's breaker and maps exceptions with `_redis_failure` + - `ResponseError` / `DataError` that are not health failures become `server` + - health failures and other `RedisError`s become `transport` + - anything else re-raises, since it is a bug, not a Redis failure +- `BreakerGuard` wraps `ReportingGuard`, which owns observability for every command + - one REDIS service success or failure event per command, with the command name as `call_type`, sent to the cache's `service_logger_obj`. A `breaker_open` refusal sends none + - the event and the log line carry the caller's `with_service_target`, so callers label their commands and emit nothing themselves + - one WARNING per failure through the timeout log throttle, `breaker_open` at DEBUG +- Callers don't log Redis health. Log only what a failure means for the feature, and only when it adds something the transport line doesn't say + +## Adding a command + +- Add a frozen dataclass for the command to `contract.py`, add it to `Command`, and add an `execute` overload for its reply type. Keep it command-shaped, never redis-py shaped (no `**kwargs`, no client options), so the native client fits behind it unchanged +- Add a `match` arm in every backend's `execute`, and add the redis-py method to the `AsyncRedisCommands` Protocol there and `_AsyncRedisCommands` in `redis_cache.py` +- Namespace every key argument with `_key`, take TTLs as `timedelta`, send them as milliseconds +- Validate the reply shape and return `decode` for anything unexpected +- Test it in `tests/unit/redis/transport/test_redis_py.py` against fakeredis: the namespace is applied once, TTL lands, a rejected command is `server`, an outage is `transport`. Service events and logging come from the guard, so a new command needs no test of its own for them +- Multi-key commands must keep their keys in one slot, or they break on cluster depending on slot ownership + +## Moving raw access onto the transport + +- Replace the raw client or script handle with transport calls, and delete hand-written namespacing and script registration +- Keep the caller's outcome per failure. If it failed closed on any exception, it fails closed on any `RedisFailure` +- Don't treat a `transport` failure on a write as "not applied" + +## Not on the transport yet + +- Pub/sub (config and auth-cache invalidation), admin commands, pipelines and `RedisBatch` stay on `RedisCache` until the transport grows them +- Vector and semantic search keep their own clients. The plan gives them native index handles on `crates/redis-client`, not `RedisTransport` + +## Where it's going + +- A `NativeRedisTransport` (a PyO3 adapter over one `RedisClient` per resolved target) goes behind the same Protocol and takes the same `Guard`. The backend is picked at construction, with redis-py kept for config the native client doesn't support yet (Sentinel, cloud auth, TLS cert options) +- Settings for both stacks resolve through `crates/config` (`RedisSettings.resolve`), so Python and Rust connect to the same target with the same namespace +- During the swap the Python `RedisCircuitBreaker` still wraps the transport. The breaker and the service events have to be in Rust before any store runs natively, because native stores call `crates/redis-client` directly +- Once stores and admin/pub-sub have replacements, redis-py and redisvl are dropped or become optional extras + +Don't build any of this unless asked diff --git a/litellm/_service_logger.py b/litellm/_service_logger.py index 060b4fb878e..75e88b880ac 100644 --- a/litellm/_service_logger.py +++ b/litellm/_service_logger.py @@ -1,5 +1,5 @@ import asyncio -from collections.abc import Callable, Coroutine +from collections.abc import Callable, Coroutine, Mapping from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any, Final, Protocol @@ -224,15 +224,16 @@ class ServiceLogging(CustomLogger): parent_otel_span: Span | None = None, start_time: datetime | float | None = None, end_time: datetime | float | None = None, - event_metadata: dict | None = None, + event_metadata: Mapping[str, object] | None = None, caller: str | None = None, - ): + ) -> None: """ - For counting if the redis, postgres call is successful """ if self.mock_testing: self.mock_testing_async_success_hook += 1 + metadata: Final = None if event_metadata is None else dict(event_metadata) payload: Final = ServiceLoggerPayload( is_error=False, error=None, @@ -241,7 +242,7 @@ class ServiceLogging(CustomLogger): call_type=call_type, caller=caller, target=current_service_target() if service == ServiceTypes.REDIS else None, - event_metadata=event_metadata, + event_metadata=metadata, ) # OTel loggers already fired this event. ``service_callback`` can hold more @@ -262,7 +263,7 @@ class ServiceLogging(CustomLogger): parent_otel_span=parent_otel_span, start_time=start_time, end_time=end_time, - event_metadata=event_metadata, + event_metadata=metadata, ) else: _otel_logger_to_use = self._resolve_otel_service_logger(callback) @@ -278,7 +279,7 @@ class ServiceLogging(CustomLogger): parent_otel_span=parent_otel_span, start_time=start_time, end_time=end_time, - event_metadata=event_metadata, + event_metadata=metadata, ) async def init_prometheus_services_logger_if_none(self): @@ -325,9 +326,9 @@ class ServiceLogging(CustomLogger): parent_otel_span: Span | None = None, start_time: datetime | float | None = None, end_time: float | datetime | None = None, - event_metadata: dict | None = None, + event_metadata: Mapping[str, object] | None = None, caller: str | None = None, - ): + ) -> None: """ - For counting if the redis, postgres call is unsuccessful """ @@ -340,6 +341,7 @@ class ServiceLogging(CustomLogger): elif isinstance(error, str): error_message = error + metadata: Final = None if event_metadata is None else dict(event_metadata) payload: Final = ServiceLoggerPayload( is_error=True, error=error_message, @@ -348,7 +350,7 @@ class ServiceLogging(CustomLogger): call_type=call_type, caller=caller, target=current_service_target() if service == ServiceTypes.REDIS else None, - event_metadata=event_metadata, + event_metadata=metadata, ) # Dedupe OTel loggers per event — see ``async_service_success_hook`` for why @@ -369,7 +371,7 @@ class ServiceLogging(CustomLogger): parent_otel_span=parent_otel_span, start_time=start_time, end_time=end_time, - event_metadata=event_metadata, + event_metadata=metadata, ) else: _otel_logger_to_use = self._resolve_otel_service_logger(callback) @@ -387,7 +389,7 @@ class ServiceLogging(CustomLogger): parent_otel_span=parent_otel_span, start_time=start_time, end_time=end_time, - event_metadata=event_metadata, + event_metadata=metadata, ) async def async_post_call_failure_hook( diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index afe6f8a1eab..587505e1011 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -30,7 +30,7 @@ from typing import TYPE_CHECKING, Any, Final, Protocol, TypeVar, cast from pydantic import TypeAdapter import litellm -from litellm._internal_context import current_service_caller +from litellm._internal_context import current_service_caller, current_service_target from litellm._logging import print_verbose, verbose_logger from litellm.constants import ( DEFAULT_REDIS_MAJOR_VERSION, @@ -42,6 +42,7 @@ from litellm.constants import ( ) from litellm.litellm_core_utils.core_helpers import get_parent_otel_span_from_kwargs from litellm.litellm_core_utils.coroutine_checker import coroutine_checker +from litellm.redis.transport.contract import RedisFailure, RedisTransport from litellm.types.caching import ( RedisPipelineIncrementOperation, RedisPipelineLpopOperation, @@ -81,6 +82,10 @@ class _AsyncRedisCommands(Protocol): def ping(self) -> Awaitable[bool]: ... + def get(self, name: str) -> Awaitable[object]: ... + + def set(self, name: str, value: str | bytes | float, *, nx: bool, px: int | None) -> Awaitable[object]: ... + def delete(self, *names: str) -> Awaitable[int]: ... def ttl(self, name: str) -> Awaitable[int]: ... @@ -103,6 +108,7 @@ _GENERIC_CALLER_MODULES: Final = frozenset( "litellm.caching.dual_cache", "litellm.caching.caching", "litellm.rust_bridge.execution", + "litellm.redis.transport.redis_py", "contextlib", } ) @@ -512,25 +518,45 @@ def log_redis_failure( if isinstance(exc, RedisCircuitBreakerOpenError): logger.debug("%s: %s", message, exc, stacklevel=2) return - exc_info: Final = exc if with_traceback else None - if not is_redis_timeout_failure(exc): - logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2) + _log_throttling_timeouts( + logger, level, message, exc, timeout=is_redis_timeout_failure(exc), exc_info=exc if with_traceback else None + ) + + +def _log_redis_failure_value(logger: logging.Logger, level: int, message: str, failure: RedisFailure) -> None: + if failure.kind == "breaker_open": + logger.debug("%s: %s", message, failure.message, stacklevel=2) + return + _log_throttling_timeouts(logger, level, message, failure.message, timeout=failure.timeout, exc_info=None) + + +def _log_throttling_timeouts( + logger: logging.Logger, + level: int, + message: str, + detail: object, + *, + timeout: bool, + exc_info: BaseException | None, +) -> None: + if not timeout: + logger.log(level, "%s: %s", message, detail, exc_info=exc_info, stacklevel=3) return suppressed: Final = _redis_timeout_log_throttle.admit() if suppressed is None: - logger.debug("%s: %s", message, exc, stacklevel=2) + logger.debug("%s: %s", message, detail, stacklevel=3) return if suppressed == 0: - logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2) + logger.log(level, "%s: %s", message, detail, exc_info=exc_info, stacklevel=3) return logger.log( level, "%s: %s (%d more Redis timeouts since the previous Redis timeout line were logged at DEBUG)", message, - exc, + detail, suppressed, exc_info=exc_info, - stacklevel=2, + stacklevel=3, ) @@ -584,6 +610,87 @@ async def _run_under_circuit_breaker( return result +def _redis_failure(exc: Exception) -> RedisFailure | None: + from redis.exceptions import DataError, RedisError, ResponseError + + if isinstance(exc, (ResponseError, DataError)) and not _is_redis_health_failure(exc): + return RedisFailure(kind="server", message=str(exc)) + if _is_redis_health_failure(exc) or isinstance(exc, RedisError): + return RedisFailure(kind="transport", message=str(exc), timeout=is_redis_timeout_failure(exc)) + return None + + +class _Admit(Protocol): + def __call__(self, call: Callable[[], Awaitable[_RedisCallResult]], /) -> Awaitable[_RedisCallResult]: ... + + +def _admit_directly(call: Callable[[], Awaitable[_RedisCallResult]]) -> Awaitable[_RedisCallResult]: + return call() + + +@dataclass(frozen=True, slots=True) +class ReportingGuard: + events: "ServiceLogging" + + async def __call__(self, name: str, call: Callable[[], Awaitable[_RedisCallResult]], /) -> _RedisCallResult: + caller: Final = _get_call_stack_info() + start_time: Final = time.time() + result: Final = await call() + end_time: Final = time.time() + if not isinstance(result, RedisFailure): + asyncio.create_task( + self.events.async_service_success_hook( + service=ServiceTypes.REDIS, + call_type=name, + duration=end_time - start_time, + start_time=start_time, + end_time=end_time, + caller=caller, + ) + ) + return result + target: Final = current_service_target() + _log_redis_failure_value( + verbose_logger, logging.WARNING, f"Redis {name} failed" + (f" for {target}" if target else ""), result + ) + if result.kind != "breaker_open": + asyncio.create_task( + self.events.async_service_failure_hook( + service=ServiceTypes.REDIS, + duration=end_time - start_time, + error=result.message, + call_type=name, + start_time=start_time, + end_time=end_time, + caller=caller, + ) + ) + return result + + +@dataclass(frozen=True, slots=True) +class BreakerGuard: + breaker: RedisCircuitBreaker + report: ReportingGuard + admit: _Admit = _admit_directly + + async def __call__( + self, name: str, call: Callable[[], Awaitable[_RedisCallResult]], / + ) -> _RedisCallResult | RedisFailure: + return await self.report(name, lambda: self._run(name, call)) + + async def _run(self, name: str, call: Callable[[], Awaitable[_RedisCallResult]]) -> _RedisCallResult | RedisFailure: + try: + return await _run_under_circuit_breaker(self.breaker, name, lambda: self.admit(call)) + except RedisCircuitBreakerOpenError as e: + return RedisFailure(kind="breaker_open", message=str(e)) + except Exception as e: + failure: Final = _redis_failure(e) + if failure is None: + raise + return failure + + def _run_under_circuit_breaker_sync( breaker: RedisCircuitBreaker, name: str, @@ -829,6 +936,15 @@ class RedisCache(BaseCache): def _async_commands(self) -> _AsyncRedisCommands: return self.init_async_client() + def transport(self) -> RedisTransport: + from litellm.redis.transport.redis_py import RedisPyTransport + + return RedisPyTransport( + client=self._async_commands, + namespace=self.namespace, + guard=BreakerGuard(self._circuit_breaker, ReportingGuard(self.service_logger_obj), self._admit_async_call), + ) + def check_and_fix_namespace(self, key: str) -> str: """ Make sure each key starts with the given namespace diff --git a/litellm/redis/AGENTS.md b/litellm/redis/AGENTS.md new file mode 100644 index 00000000000..7f035e4b385 --- /dev/null +++ b/litellm/redis/AGENTS.md @@ -0,0 +1,11 @@ +For adding Redis access or moving raw Redis access here, follow [.agents/skills/redis-access/SKILL.md](../../.agents/skills/redis-access/SKILL.md) + +- `transport/contract.py` is the command-shaped seam. Keep it one command dataclass per Redis command the Python code uses, run through `execute`, never redis-py shaped, so `litellm-rust/crates/redis-client` can sit behind it unchanged +- Rules shared with `crates/redis-client`, so the two implementations agree + - The namespace is applied here, once. A key that already starts with `{namespace}:` is left alone + - Every write takes its TTL from the caller, `None` means no expiry, there is no default + - Replies are undecoded bytes + - Failures are values. `transport` covers timeouts and connection loss, where the command may still have applied. `server` means Redis answered with an error and the connection is fine +- `transport/redis_py.py` is the only module that touches redis-py's client. The breaker and the exception mapping come from `RedisCache` through the injected guard +- Every backend runs each command through a `Guard` that emits one REDIS service event and logs Redis health failures, so swapping backends never drops metrics. Callers don't log Redis health +- `litellm/caching` imports `litellm/redis`, never the reverse diff --git a/litellm/redis/__init__.py b/litellm/redis/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/redis/transport/__init__.py b/litellm/redis/transport/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/redis/transport/contract.py b/litellm/redis/transport/contract.py new file mode 100644 index 00000000000..dc9818417c0 --- /dev/null +++ b/litellm/redis/transport/contract.py @@ -0,0 +1,56 @@ +import math +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import timedelta +from typing import Final, Literal, Protocol, TypeAlias, TypeVar, overload + +Arg: TypeAlias = str | bytes | int | float + +_T: Final = TypeVar("_T") + + +@dataclass(frozen=True, slots=True) +class RedisFailure: + kind: Literal["transport", "server", "decode", "breaker_open"] + message: str + timeout: bool = False + + +@dataclass(frozen=True, slots=True) +class Get: + key: str + + +@dataclass(frozen=True, slots=True) +class Set: + key: str + value: Arg + ttl: timedelta | None + nx: bool = False + + +@dataclass(frozen=True, slots=True) +class Delete: + key: str + + +Command: TypeAlias = Get | Set | Delete + + +class Guard(Protocol): + def __call__(self, name: str, call: Callable[[], Awaitable[_T]], /) -> Awaitable[_T | RedisFailure]: ... + + +def ttl_ms(ttl: timedelta) -> int: + return max(1, math.ceil(ttl / timedelta(milliseconds=1))) + + +class RedisTransport(Protocol): + @overload + async def execute(self, command: Get) -> bytes | None | RedisFailure: ... + + @overload + async def execute(self, command: Set) -> bool | RedisFailure: ... + + @overload + async def execute(self, command: Delete) -> int | RedisFailure: ... diff --git a/litellm/redis/transport/redis_py.py b/litellm/redis/transport/redis_py.py new file mode 100644 index 00000000000..0985634ee90 --- /dev/null +++ b/litellm/redis/transport/redis_py.py @@ -0,0 +1,59 @@ +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Final, Protocol, assert_never, overload + +from litellm.redis.transport.contract import Arg, Command, Delete, Get, Guard, RedisFailure, Set, ttl_ms + + +class AsyncRedisCommands(Protocol): + def get(self, name: str) -> Awaitable[object]: ... + + def set(self, name: str, value: Arg, *, nx: bool, px: int | None) -> Awaitable[object]: ... + + def delete(self, *names: str) -> Awaitable[int]: ... + + +def _decode_failure(value: object) -> RedisFailure: + return RedisFailure(kind="decode", message=f"unexpected Redis reply of type {type(value).__name__}") + + +def _bytes_reply(reply: object) -> bytes | None | RedisFailure: + if reply is None or isinstance(reply, (bytes, RedisFailure)): + return reply + if isinstance(reply, str): + return reply.encode() + return _decode_failure(reply) + + +@dataclass(frozen=True, slots=True) +class RedisPyTransport: + client: Callable[[], AsyncRedisCommands] + namespace: str | None + guard: Guard + + def _key(self, key: str) -> str: + if not self.namespace or key.startswith(f"{self.namespace}:"): + return key + return f"{self.namespace}:{key}" + + @overload + async def execute(self, command: Get) -> bytes | None | RedisFailure: ... + + @overload + async def execute(self, command: Set) -> bool | RedisFailure: ... + + @overload + async def execute(self, command: Delete) -> int | RedisFailure: ... + + async def execute(self, command: Command) -> bytes | None | bool | int | RedisFailure: + match command: + case Get(key=key): + return _bytes_reply(await self.guard("get", lambda: self.client().get(self._key(key)))) + case Set(key=key, value=value, ttl=ttl, nx=nx): + px: Final = None if ttl is None else ttl_ms(ttl) + written: Final = await self.guard("set", lambda: self.client().set(self._key(key), value, nx=nx, px=px)) + return written if isinstance(written, RedisFailure) else written is not None and written is not False + case Delete(key=key): + return await self.guard("delete", lambda: self.client().delete(self._key(key))) + case _: + assert_never(command) diff --git a/tests/_support/fake_redis.py b/tests/_support/fake_redis.py new file mode 100644 index 00000000000..d6f4b5e3791 --- /dev/null +++ b/tests/_support/fake_redis.py @@ -0,0 +1,43 @@ +import fakeredis + +import litellm +from litellm._service_logger import ServiceLogging +from litellm.caching.redis_cache import ( + BreakerGuard, + RedisCache, + RedisCircuitBreaker, + ReportingGuard, +) +from litellm.redis.transport.redis_py import RedisPyTransport + + +def fake_redis_cache( + server: fakeredis.FakeServer, namespace: str | None = None, service_logger: ServiceLogging | None = None +) -> RedisCache: + cache = RedisCache( + host="127.0.0.1", + port=1, + namespace=namespace, + socket_timeout=0.1, + service_logger_obj=service_logger or ServiceLogging(), + ) + litellm.in_memory_llm_clients_cache.set_cache( + key=cache._get_async_client_cache_key(), # pyright: ignore[reportPrivateUsage] # seeds the per-loop client slot RedisCache reads + value=fakeredis.FakeAsyncRedis(server=server), + ) + return cache + + +def fake_redis_transport( + client: fakeredis.FakeAsyncRedis, + *, + failure_threshold: int, + events: ServiceLogging | None = None, + namespace: str | None = None, +) -> RedisPyTransport: + breaker = RedisCircuitBreaker(failure_threshold=failure_threshold, recovery_timeout=3600) + return RedisPyTransport( + client=lambda: client, + namespace=namespace, + guard=BreakerGuard(breaker, ReportingGuard(events or ServiceLogging())), + ) diff --git a/tests/unit/redis/__init__.py b/tests/unit/redis/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/redis/conftest.py b/tests/unit/redis/conftest.py new file mode 100644 index 00000000000..d101deb7e2d --- /dev/null +++ b/tests/unit/redis/conftest.py @@ -0,0 +1,12 @@ +import fakeredis +import pytest + + +@pytest.fixture +def server() -> fakeredis.FakeServer: + return fakeredis.FakeServer() + + +@pytest.fixture +def raw(server: fakeredis.FakeServer) -> fakeredis.FakeAsyncRedis: + return fakeredis.FakeAsyncRedis(server=server) diff --git a/tests/unit/redis/transport/__init__.py b/tests/unit/redis/transport/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/redis/transport/test_contract.py b/tests/unit/redis/transport/test_contract.py new file mode 100644 index 00000000000..300c538a3a2 --- /dev/null +++ b/tests/unit/redis/transport/test_contract.py @@ -0,0 +1,17 @@ +from datetime import timedelta + +import pytest + +from litellm.redis.transport.contract import ttl_ms + + +@pytest.mark.parametrize( + ("ttl", "expected"), + [ + pytest.param(timedelta(seconds=1.5), 1500, id="whole-ms"), + pytest.param(timedelta(microseconds=1500), 2, id="rounds-up"), + pytest.param(timedelta(microseconds=1), 1, id="sub-ms-is-not-zero"), + ], +) +def test_ttl_ms_never_shortens_the_ttl_or_sends_zero(ttl: timedelta, expected: int) -> None: + assert ttl_ms(ttl) == expected diff --git a/tests/unit/redis/transport/test_redis_py.py b/tests/unit/redis/transport/test_redis_py.py new file mode 100644 index 00000000000..ca336903662 --- /dev/null +++ b/tests/unit/redis/transport/test_redis_py.py @@ -0,0 +1,237 @@ +import asyncio +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Final, Literal + +import fakeredis +import pytest + +from litellm._internal_context import current_service_target, service_target +from litellm._service_logger import ServiceLogging +from litellm.redis.transport.contract import Delete, Get, RedisFailure, Set +from litellm.types.services import ServiceTypes +from tests._support.fake_redis import fake_redis_cache, fake_redis_transport + + +@pytest.mark.asyncio +async def test_every_command_prefixes_the_namespace_once( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + transport = fake_redis_cache(server, namespace="ns").transport() + + assert await transport.execute(Set("k", "v", ttl=None)) is True + assert await transport.execute(Set("ns:gone", "v", ttl=None)) is True + + assert sorted(await raw.keys("*")) == [b"ns:gone", b"ns:k"] + assert await transport.execute(Get("k")) == await transport.execute(Get("ns:k")) == b"v" + assert await transport.execute(Delete("gone")) == 1 + assert await raw.keys("*") == [b"ns:k"] + + +@pytest.mark.asyncio +async def test_set_nx_refuses_a_live_key_and_applies_the_ttl( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + transport = fake_redis_cache(server).transport() + + assert await transport.execute(Set("k", "first", ttl=timedelta(seconds=30), nx=True)) is True + assert await transport.execute(Set("k", "second", ttl=timedelta(seconds=30), nx=True)) is False + + assert await raw.get("k") == b"first" + assert 0 < await raw.pttl("k") <= 30_000 + + +@pytest.mark.asyncio +async def test_set_without_ttl_persists(server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis) -> None: + transport = fake_redis_cache(server).transport() + + assert await transport.execute(Set("k", "v", ttl=None)) is True + + assert await raw.pttl("k") == -1 + + +@pytest.mark.asyncio +async def test_delete_reports_how_many_keys_it_removed( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + transport = fake_redis_cache(server).transport() + await raw.set("k", "v") + + assert await transport.execute(Delete("k")) == 1 + assert await transport.execute(Delete("k")) == 0 + assert await raw.exists("k") == 0 + + +@pytest.mark.asyncio +async def test_a_rejected_command_is_a_server_failure_and_keeps_the_connection( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + transport = fake_redis_cache(server).transport() + await raw.rpush("list", "x") + + rejected = await transport.execute(Get("list")) + + assert isinstance(rejected, RedisFailure) + assert rejected.kind == "server" + assert await transport.execute(Set("k", "v", ttl=None)) is True + + +@pytest.mark.asyncio +async def test_an_unreachable_server_is_a_transport_failure(server: fakeredis.FakeServer) -> None: + transport = fake_redis_cache(server).transport() + server.connected = False + + failure = await transport.execute(Get("k")) + + assert isinstance(failure, RedisFailure) + assert failure.kind == "transport" + + +@pytest.mark.asyncio +async def test_a_transport_failure_opens_the_breaker_and_it_answers_without_redis( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + transport = fake_redis_transport(raw, failure_threshold=1) + server.connected = False + + tripped = await transport.execute(Get("k")) + server.connected = True + refused = await transport.execute(Get("k")) + + assert isinstance(tripped, RedisFailure) + assert tripped.kind == "transport" + assert isinstance(refused, RedisFailure) + assert refused.kind == "breaker_open" + + +@pytest.mark.asyncio +async def test_replies_stay_bytes_on_a_client_that_decodes_responses(server: fakeredis.FakeServer) -> None: + transport = fake_redis_transport( + fakeredis.FakeAsyncRedis(server=server, decode_responses=True), failure_threshold=1 + ) + + assert await transport.execute(Set("k", "v", ttl=None)) is True + assert await transport.execute(Get("k")) == b"v" + + +async def _settle() -> None: + pending: Final = {task for task in asyncio.all_tasks() if task is not asyncio.current_task()} + if pending: + await asyncio.gather(*pending) + await _settle() + + +@dataclass(frozen=True, slots=True) +class _Event: + outcome: Literal["success", "failure"] + call_type: str + target: str | None + failed_with: str | None = None + + +class _RecordingEvents(ServiceLogging): + def __init__(self) -> None: + super().__init__() + self.seen: tuple[_Event, ...] = () + + async def async_service_success_hook( + self, + service: ServiceTypes, + call_type: str, + duration: float, + parent_otel_span: object | None = None, + start_time: datetime | float | None = None, + end_time: datetime | float | None = None, + event_metadata: object | None = None, + caller: str | None = None, + ) -> None: + assert service == ServiceTypes.REDIS and 0 <= duration + self.seen = (*self.seen, _Event("success", call_type, current_service_target())) + + async def async_service_failure_hook( + self, + service: ServiceTypes, + duration: float, + error: str | Exception, + call_type: str, + parent_otel_span: object | None = None, + start_time: datetime | float | None = None, + end_time: float | datetime | None = None, + event_metadata: object | None = None, + caller: str | None = None, + ) -> None: + assert service == ServiceTypes.REDIS and 0 <= duration + self.seen = (*self.seen, _Event("failure", call_type, current_service_target(), str(error))) + + +@pytest.mark.asyncio +async def test_every_command_reports_one_service_event_under_the_callers_target( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + events = _RecordingEvents() + transport = fake_redis_transport(raw, failure_threshold=2, events=events) + + with service_target("response_polling"): + await transport.execute(Set("k", "v", ttl=None)) + await transport.execute(Get("k")) + server.connected = False + failed = await transport.execute(Delete("k")) + await asyncio.sleep(0) + + assert isinstance(failed, RedisFailure) + assert events.seen == ( + _Event("success", "set", "response_polling"), + _Event("success", "get", "response_polling"), + _Event("failure", "delete", "response_polling", failed.message), + ) + + +@pytest.mark.asyncio +async def test_a_call_the_breaker_refuses_reports_no_service_event( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + events = _RecordingEvents() + transport = fake_redis_transport(raw, failure_threshold=1, events=events) + server.connected = False + + await transport.execute(Get("k")) + refused = await transport.execute(Get("k")) + await asyncio.sleep(0) + + assert isinstance(refused, RedisFailure) and refused.kind == "breaker_open" + assert tuple(event.outcome for event in events.seen) == ("failure",) + + +@pytest.mark.asyncio +async def test_the_cache_transport_reports_to_the_caches_service_logger(server: fakeredis.FakeServer) -> None: + service_logger = ServiceLogging(mock_testing=True) + transport = fake_redis_cache(server, service_logger=service_logger).transport() + await _settle() + before = (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook) + + await transport.execute(Get("k")) + server.connected = False + await transport.execute(Get("k")) + await _settle() + + after = (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook) + assert (after[0] - before[0], after[1] - before[1]) == (1, 1) + + +@pytest.mark.asyncio +async def test_a_failure_is_logged_once_with_the_command_and_target( + server: fakeredis.FakeServer, caplog: pytest.LogCaptureFixture +) -> None: + transport = fake_redis_cache(server).transport() + await _settle() + server.connected = False + caplog.clear() + + with caplog.at_level(logging.WARNING), service_target("response_polling"): + failure = await transport.execute(Get("k")) + + assert isinstance(failure, RedisFailure) + assert [record.getMessage() for record in caplog.records] == [ + f"Redis get failed for response_polling: {failure.message}" + ]