From 08276193a70a3288fedbabc4cac6acf7013c4066 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 30 Oct 2026 22:59:41 +0000 Subject: [PATCH] refactor(redis): split the connection out of RedisCache --- .agents/skills/redis-access/SKILL.md | 6 +- litellm/caching/redis_cache.py | 625 +++++-------------------- litellm/redis/AGENTS.md | 3 +- litellm/redis/connection.py | 41 ++ litellm/redis/health.py | 542 +++++++++++++++++++++ litellm/redis/transport/batch.py | 2 +- litellm/redis/transport/breaker.py | 2 +- litellm/redis/transport/events.py | 2 +- litellm/redis/transport/redis_py.py | 2 +- tests/unit/caching/test_caching.py | 2 +- tests/unit/caching/test_dual_cache.py | 2 +- tests/unit/caching/test_redis_batch.py | 1 + tests/unit/caching/test_redis_cache.py | 6 +- tests/unit/redis/test_connection.py | 65 +++ 14 files changed, 770 insertions(+), 531 deletions(-) create mode 100644 litellm/redis/connection.py create mode 100644 litellm/redis/health.py create mode 100644 tests/unit/redis/test_connection.py diff --git a/.agents/skills/redis-access/SKILL.md b/.agents/skills/redis-access/SKILL.md index ddc0db766ca..8c4b9266717 100644 --- a/.agents/skills/redis-access/SKILL.md +++ b/.agents/skills/redis-access/SKILL.md @@ -20,7 +20,7 @@ metadata: - New coordination state gets a store in `litellm/redis/stores/.py`, not a new `RedisCache` method - A Python store is a finished state. Adding one does not require a Rust port - Callers get typed domain results from the store and keep policy, which TTL applies, fail open or closed, retries, and the `with_service_target` that labels the calls -- 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`. `litellm/caching/redis_cache.py` and `redis_cluster_cache.py` stay allowlisted while they own the breaker's exception types and the generic verbs +- 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`, except for exception classification in `litellm/redis/health.py`. `litellm/caching/redis_cache.py` and `redis_cluster_cache.py` stay allowlisted while they own the generic verbs ## The transport @@ -38,8 +38,8 @@ metadata: - Replies are undecoded bytes. `eval` returns `int | bytes | None | tuple[...]`. Decoding belongs to the store that owns the keyspace - `eval` takes script text every call and does `EVALSHA` with `EVAL` fallback, so nothing registers scripts - These rules are shared with `crates/redis-client`, whose `AGENTS.md` has the Rust side. Change both together, since Python and Rust pods share keys in one Redis -- `RedisCache.transport()` builds the transport over that cache's own client, pub/sub client, namespace, circuit breaker, `_admit_async_call`, cluster-aware `mget` and `ServiceLogging`. No new connection, no second breaker, and `RedisClusterCache`'s connection gate still applies -- Every method runs its command through two steps, in this order. `run_under_breaker` wraps the existing `_run_under_circuit_breaker` from `litellm/caching/redis_cache.py`, so admission, the generation check, the health failure classification and the timeout flag are the ones legacy traffic uses, and maps the outcome to a `RedisFailure` (`breaker_open`, `transport`, `server`) instead of raising. `RedisServiceEvents.observe` times the call, logs a failure and emits the service event +- `RedisCache.transport()` delegates to its internal `RedisConnection`, which builds the transport over that cache's own client, pub/sub client, namespace, circuit breaker, `_admit_async_call`, cluster-aware `mget` and `ServiceLogging`. No new connection, no second breaker, and `RedisClusterCache`'s connection gate still applies +- Every method runs its command through two steps, in this order. `run_under_breaker` wraps the shared `_run_under_circuit_breaker` from `litellm/redis/health.py`, so admission, the generation check, the health failure classification and the timeout flag are the ones legacy traffic uses, and maps the outcome to a `RedisFailure` (`breaker_open`, `transport`, `server`) instead of raising. `RedisServiceEvents.observe` times the call, logs a failure and emits the service event - `ping`, the admin commands and pub/sub skip the breaker, like the `RedisCache` methods they replace, so health checks still reach a Redis the breaker took out. They still go through `RedisServiceEvents.observe`, except `get_message` polling diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index def36fa5891..9f10e873096 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -12,38 +12,63 @@ import ast import asyncio import functools import hashlib -import inspect -import itertools import json import logging -import threading import time -import weakref -from collections import Counter -from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence -from contextvars import ContextVar -from dataclasses import dataclass +from collections.abc import Awaitable, Callable, Sequence from datetime import timedelta -from types import FrameType, MappingProxyType -from typing import TYPE_CHECKING, Any, Final, Protocol, TypeVar, cast +from types import MappingProxyType +from typing import TYPE_CHECKING, Any, Final, Protocol, cast from pydantic import TypeAdapter import litellm -from litellm._internal_context import current_service_caller from litellm._logging import print_verbose, verbose_logger from litellm.constants import ( DEFAULT_REDIS_MAJOR_VERSION, REDIS_CIRCUIT_BREAKER_ENABLED, REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD, REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT, - REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION, - REDIS_TIMEOUT_LOG_INTERVAL, ) 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.connection import RedisConnection +from litellm.redis.health import ( + _CALL_STACK_END_MODULES, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _GENERIC_CALLER_FRAME_NAMES, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _GENERIC_CALLER_MODULES, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _LIVE_BREAKERS, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _MAX_EXCEPTION_CAUSE_DEPTH, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + RedisCircuitBreaker, + RedisCircuitBreakerOpenError, + _BreakerAdmission, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _caller_frames, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _ends_call_stack, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _enter_circuit_breaker, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _exit_circuit_breaker, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _explicit_causes, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _get_call_stack_info, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _is_generic_caller_frame, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _is_redis_health_failure, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _log_throttling_timeouts, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _record_swallowed_redis_failure, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _redis_health_error_types, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _redis_timeout_error_types, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _redis_timeout_log_throttle, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _RedisCallResult, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _RedisTimeoutLogThrottle, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _run_under_circuit_breaker, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _run_under_circuit_breaker_sync, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + _swallowed_redis_failures, # pyright: ignore[reportPrivateUsage] # compatibility re-export of shared Redis internals + is_redis_error, + is_redis_timeout_failure, + log_redis_failure, + log_redis_failure_value, + redis_breaker_state_counts, +) +from litellm.redis.transport.batch import PipelineBatch from litellm.redis.transport.client import AsyncPubSubClient, AsyncRedisCommands -from litellm.redis.transport.contract import RedisFailure, RedisTransport +from litellm.redis.transport.contract import RedisTransport from litellm.types.caching import ( RedisPipelineIncrementOperation, RedisPipelineLpopOperation, @@ -59,8 +84,6 @@ if TYPE_CHECKING: from redis.asyncio.client import Pipeline from redis.asyncio.cluster import ClusterPipeline - from litellm._service_logger import ServiceLogging - pipeline = Pipeline cluster_pipeline = ClusterPipeline async_redis_client = Redis @@ -74,6 +97,42 @@ else: Span = Any +__all__ = ( + "_CALL_STACK_END_MODULES", + "_GENERIC_CALLER_FRAME_NAMES", + "_GENERIC_CALLER_MODULES", + "_LIVE_BREAKERS", + "_MAX_EXCEPTION_CAUSE_DEPTH", + "RedisCache", + "RedisCircuitBreaker", + "RedisCircuitBreakerOpenError", + "_BreakerAdmission", + "_RedisCallResult", + "_RedisTimeoutLogThrottle", + "_caller_frames", + "_ends_call_stack", + "_enter_circuit_breaker", + "_exit_circuit_breaker", + "_explicit_causes", + "_get_call_stack_info", + "_is_generic_caller_frame", + "_is_redis_health_failure", + "_log_throttling_timeouts", + "_record_swallowed_redis_failure", + "_redis_health_error_types", + "_redis_timeout_error_types", + "_redis_timeout_log_throttle", + "_run_under_circuit_breaker", + "_run_under_circuit_breaker_sync", + "_swallowed_redis_failures", + "is_redis_error", + "is_redis_timeout_failure", + "log_redis_failure", + "log_redis_failure_value", + "redis_breaker_state_counts", +) + + class _AsyncRedisCommands(AsyncRedisCommands, Protocol): """Async redis commands this cache issues. @@ -94,37 +153,6 @@ class _AsyncRedisCommands(AsyncRedisCommands, Protocol): def pipeline(self, transaction: bool = True) -> "Pipeline[bytes]": ... -_GENERIC_CALLER_MODULES: Final = frozenset( - { - __name__, - "litellm.caching.redis_batch", - "litellm.caching.dual_cache", - "litellm.caching.caching", - "litellm.rust_bridge.execution", - "contextlib", - } -) -_GENERIC_CALLER_FRAME_NAMES: Final = frozenset( - { - "", - "wrapper", - "_run_under_circuit_breaker", - "_run_under_circuit_breaker_sync", - "run_alone", - "_settle_alone", - "get_cache", - "set_cache", - "async_get_cache", - "async_set_cache", - "async_batch_get_cache", - "async_batch_get_cache_shared", - "async_set_cache_pipeline", - "async_increment_cache", - "async_delete_cache", - } -) -_CALL_STACK_END_MODULES: Final = ("asyncio", "concurrent", "threading") - _INCREMENT_WITH_FLOOR_LUA: Final = ( "local count = redis.call('INCRBY', KEYS[1], ARGV[1]) " "if count < 0 then count = redis.call('INCRBY', KEYS[1], -count) end " @@ -147,216 +175,6 @@ def _decoded_counts(values: Sequence[bytes | str | None]) -> tuple[int | None, . ) -def _is_generic_caller_frame(frame: FrameType) -> bool: - module: Final = frame.f_globals.get("__name__") - return ( - module in _GENERIC_CALLER_MODULES - or (isinstance(module, str) and module.startswith("litellm.redis.")) - or frame.f_code.co_name in _GENERIC_CALLER_FRAME_NAMES - ) - - -def _ends_call_stack(frame: FrameType) -> bool: - module: Final = frame.f_globals.get("__name__") - return isinstance(module, str) and module.startswith(_CALL_STACK_END_MODULES) - - -def _caller_frames(first: FrameType) -> Iterator[FrameType]: - frame: FrameType | None = first - while frame is not None and not _ends_call_stack(frame): - yield frame - frame = frame.f_back - - -def _get_call_stack_info(num_frames: int = 2) -> str: - """ - Get the function names of the nearest meaningful callers of the cache method. - - Frames that merely forward the call (this module's circuit-breaker guards, the - cache facades, the batch pipeline's retry path, generic cache verbs) are - skipped, and the walk stops at the event loop, so the chain names the litellm - code that wanted the call. When nothing but forwarding frames is found (the call - runs in a task of its own, like a batch op retried on the flush) the chain the - declaring code threaded through ``service_caller`` is reported, else ``unknown``. - - Args: - num_frames: Number of previous frames to include (default: 2) - - Returns: - A string with format "caller_function [<- grandparent_function]" - """ - try: - current_frame: Final = inspect.currentframe() - if current_frame is None: - return "unknown" - - # Skip this function and the immediate caller (which sets call_type) - f_back: Final = current_frame.f_back - if f_back is None: - return "unknown" - first: Final = f_back.f_back - if first is None: - return "unknown" - frames: Final = _caller_frames(first) - leading: Final = tuple(itertools.islice(frames, num_frames)) - leading_names: Final = tuple(frame.f_code.co_name for frame in leading if not _is_generic_caller_frame(frame)) - further_names: Final = tuple( - itertools.islice( - (frame.f_code.co_name for frame in frames if not _is_generic_caller_frame(frame)), - num_frames - len(leading_names), - ) - ) - function_names: Final = leading_names + further_names - - if function_names: - return " <- ".join(function_names) - return current_service_caller() or "unknown" - except Exception: - return "unknown" - - -class RedisCircuitBreaker: - """ - Tracks Redis health for a RedisCache instance. - - States: - CLOSED - normal, Redis is called - OPEN - Redis is down, raise immediately (no network call) - HALF_OPEN - recovery probe: allow one request through - - Transitions: - CLOSED -> OPEN after failure_threshold consecutive hard connectivity - failures, or after an unbroken run of timeout failures - (no success or hard failure in between) that reaches - failure_threshold and spans timeout_min_duration seconds - OPEN -> HALF_OPEN after recovery_timeout seconds - HALF_OPEN -> CLOSED on success - HALF_OPEN -> OPEN on failure (resets timer) - - Timeouts are accounted separately from hard connectivity failures because the async - Redis timeout includes time waiting for the worker event loop to resume: one loop - stall makes every in-flight operation time out together, which satisfies a purely - consecutive threshold instantly even though Redis is healthy. Requiring a - timeout-only streak to also span timeout_min_duration filters such bursts while a - real outage that surfaces as timeouts still opens the breaker after that duration. - """ - - CLOSED = "closed" - OPEN = "open" - HALF_OPEN = "half_open" - - def __init__( - self, - failure_threshold: int, - recovery_timeout: int, - enabled: bool = True, - timeout_min_duration: float = REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION, - service_logger: "ServiceLogging | None" = None, - ) -> None: - self._service_logger = service_logger - self.failure_threshold = failure_threshold - self.recovery_timeout = recovery_timeout - self.enabled = enabled - self.timeout_min_duration = timeout_min_duration - self._failure_count = 0 - self._hard_failure_count = 0 - self._timeout_count = 0 - self._timeout_streak_started_at: float | None = None - self._opened_at: float | None = None - self._state = self.CLOSED - self._generation = 0 - _LIVE_BREAKERS.add(self) - self._report(transitions_to=None, failure_class=None) - - @property - def state(self) -> str: - return self._state - - @property - def generation(self) -> int: - """Counts state transitions, so a call can tell whether the breaker moved while it ran.""" - return self._generation - - def is_open(self) -> bool: - """Returns True if Redis calls should be skipped.""" - if not self.enabled: - return False - if self._state == self.HALF_OPEN: - # Probe already in flight — fast-fail all concurrent requests. - # Only the one call that caused the OPEN→HALF_OPEN transition - # (which returned False) is the designated probe. - return True - if self._state == self.OPEN: - if time.time() - (self._opened_at or 0) > self.recovery_timeout: - self._set_state(self.HALF_OPEN) - return False # this caller is the designated probe - return True - return False - - def _should_open(self, now: float) -> bool: - if self._state == self.HALF_OPEN: - return True - if self._hard_failure_count >= self.failure_threshold: - return True - if self._timeout_count < self.failure_threshold: - return False - return now - (self._timeout_streak_started_at or now) >= self.timeout_min_duration - - def record_failure(self, is_timeout: bool = False) -> None: - if not self.enabled: - return - now: Final = time.time() - self._failure_count += 1 - if is_timeout: - self._timeout_count += 1 - if self._timeout_streak_started_at is None: - self._timeout_streak_started_at = now - else: - self._hard_failure_count += 1 - self._timeout_count = 0 - self._timeout_streak_started_at = None - self._opened_at = now - self._report(transitions_to=None, failure_class="timeout" if is_timeout else "connectivity") - if self._should_open(now): - if self._state != self.OPEN: - verbose_logger.warning( - "Redis circuit breaker OPENED after %d consecutive failures" - " (%d hard connectivity) — fast-failing Redis calls for %ds", - self._failure_count, - self._hard_failure_count, - self.recovery_timeout, - ) - self._set_state(self.OPEN) - - def record_success(self) -> None: - if not self.enabled or self._state == self.OPEN: - return - if self._state == self.HALF_OPEN: - verbose_logger.info("Redis circuit breaker CLOSED — Redis recovered") - self._failure_count = 0 - self._hard_failure_count = 0 - self._timeout_count = 0 - self._timeout_streak_started_at = None - self._set_state(self.CLOSED) - - def _set_state(self, state: str) -> None: - if state == self._state: - return - self._state = state - self._generation += 1 - self._report(transitions_to=state, failure_class=None) - - def _report(self, transitions_to: str | None, failure_class: str | None) -> None: - if self._service_logger is not None: - self._service_logger.redis_breaker_hook(transitions_to=transitions_to, failure_class=failure_class) - - -_RedisCallResult = TypeVar("_RedisCallResult") - - -_swallowed_redis_failures: Final[ContextVar[int]] = ContextVar("litellm_swallowed_redis_failures", default=0) - - def _opaque_kwarg_key(value: object) -> str: return f"{type(value).__name__}-{id(value)}" @@ -384,251 +202,6 @@ def _cluster_node_pubsub_client( # pyright: ignore[reportUnknownParameterType] return Redis.from_pool(pool) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # redis generics -@functools.lru_cache(maxsize=1) -def _redis_health_error_types() -> tuple[type, ...]: - """Exception types that mean the Redis backend itself is unhealthy. - - Command and data errors say nothing about connectivity: an INCR against a non-numeric - value or an undecodable cached entry is a request problem, and counting those would let - a caller trip the shared breaker on demand, dropping rate limiting to per-process - counters that spreading traffic across replicas can outrun. - - Imported lazily because this module is reachable from a base ``import litellm`` while - redis is not a base dependency. - """ - from redis.exceptions import BusyLoadingError, ClusterDownError - from redis.exceptions import ConnectionError as RedisConnectionError - from redis.exceptions import TimeoutError as RedisTimeoutError - - return (RedisConnectionError, RedisTimeoutError, BusyLoadingError, ClusterDownError, OSError, asyncio.TimeoutError) - - -def _is_redis_health_failure(exc: BaseException) -> bool: - """True when ``exc`` indicates Redis is unreachable rather than the request being bad.""" - try: - return isinstance(exc, _redis_health_error_types()) - except ImportError: - return True - - -@functools.lru_cache(maxsize=1) -def _redis_timeout_error_types() -> tuple[type, ...]: - """Health failures that are timeouts rather than unambiguous connectivity errors. - - ``builtins.TimeoutError`` covers ``socket.timeout`` (an alias since py3.10) and, from - py3.11, ``asyncio.TimeoutError``; on py3.10 ``asyncio.TimeoutError`` is still its own - class, so it is listed explicitly. ``redis.exceptions.TimeoutError`` subclasses neither. - ``MaxConnectionsError`` is the async cluster client's pool-exhausted signal (its nodes have - no blocking pool, so it is raised instantly instead of after a wait): a busy pool, not a - dead Redis, like the standalone pool's chained wait timeout. - """ - try: - from redis import exceptions as redis_exceptions - except ImportError: - return (TimeoutError, asyncio.TimeoutError) - max_connections_error: Final = cast( # cast-ok: types-redis 4.6 stubs predate MaxConnectionsError - "type[Exception]", - redis_exceptions.MaxConnectionsError, # pyright: ignore[reportAttributeAccessIssue] # types-redis 4.6 stubs predate it - ) - return (redis_exceptions.TimeoutError, TimeoutError, asyncio.TimeoutError, max_connections_error) - - -_MAX_EXCEPTION_CAUSE_DEPTH: Final = 20 - - -def _explicit_causes(exc: BaseException) -> Iterator[BaseException]: - current = exc # rebind-ok: advances one link per iteration of the bounded walk - for _ in range(_MAX_EXCEPTION_CAUSE_DEPTH): - yield current - if current.__cause__ is None: - return - current = current.__cause__ - - -def is_redis_timeout_failure(exc: BaseException) -> bool: - """True when ``exc`` or any exception it was explicitly raised ``from`` is a timeout. - - redis-py's blocking pool reports a pool wait timeout as ``ConnectionError`` chained from - ``asyncio.TimeoutError``, which is a busy pool rather than an unreachable Redis. - """ - timeout_types: Final = _redis_timeout_error_types() - return any(isinstance(link, timeout_types) for link in _explicit_causes(exc)) - - -def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseException) -> None: - """Record a Redis failure that the calling method is about to swallow. - - The marker is a ContextVar rather than a counter on the breaker because breakers are - shared by every concurrent caller. A plain shared counter cannot tell "my call failed" - from "some other in-flight call failed", so a success overlapping someone else's - failure would be discarded and a Redis that is answering would still be evicted. - asyncio gives each task its own copy of the context, so this is per-call. - """ - if not _is_redis_health_failure(exc): - return - breaker.record_failure(is_timeout=is_redis_timeout_failure(exc)) - _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1) - - -_LIVE_BREAKERS: Final[weakref.WeakSet[RedisCircuitBreaker]] = weakref.WeakSet() - - -def redis_breaker_state_counts() -> Mapping[str, int]: - """How many live Redis circuit breakers sit in each state, for the Prometheus services logger.""" - return MappingProxyType(dict(Counter(breaker.state for breaker in _LIVE_BREAKERS))) - - -class RedisCircuitBreakerOpenError(Exception): - pass - - -class _RedisTimeoutLogThrottle: - """Admits one Redis timeout log line per interval and counts the timeouts it suppressed in between.""" - - def __init__(self, interval: float, clock: Callable[[], float] = time.monotonic) -> None: - self.interval = interval - self._clock = clock - self._lock = threading.Lock() - self._last_logged_at: float | None = None - self._suppressed = 0 - - def admit(self) -> int | None: - """Return the number of timeouts suppressed since the last admitted line, or None to suppress this one.""" - with self._lock: - now: Final = self._clock() - if self._last_logged_at is not None and now - self._last_logged_at < self.interval: - self._suppressed += 1 - return None - suppressed: Final = self._suppressed - self._suppressed = 0 - self._last_logged_at = now - return suppressed - - -_redis_timeout_log_throttle: Final = _RedisTimeoutLogThrottle(REDIS_TIMEOUT_LOG_INTERVAL) - - -def log_redis_failure( - logger: logging.Logger, level: int, message: str, exc: BaseException, with_traceback: bool = False -) -> None: - if isinstance(exc, RedisCircuitBreakerOpenError): - logger.debug("%s: %s", message, exc, stacklevel=2) - return - _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, detail, stacklevel=3) - return - if suppressed == 0: - 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, - detail, - suppressed, - exc_info=exc_info, - stacklevel=3, - ) - - -@dataclass(frozen=True, slots=True) -class _BreakerAdmission: - swallowed_before: int - generation: int - - -def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> _BreakerAdmission: - """Reject the call if the breaker is open, else record what its success may later prove.""" - if breaker.is_open(): - raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open — skipping {name}") - return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), generation=breaker.generation) - - -def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission) -> None: - """Record success only when nothing failed while the call ran and the breaker has not moved since. - - Several Redis methods catch their own connection errors and return a default, so a - method that returned is not on its own proof of a healthy Redis. A success also vouches - only for the breaker state that admitted the call: a call admitted before the breaker - opened, or a probe admitted before a later failure reopened it, finishes knowing nothing - about whether Redis has recovered since, so only the current probe may close the breaker. - """ - if _swallowed_redis_failures.get() != admission.swallowed_before: - return - if breaker.generation != admission.generation: - return - breaker.record_success() - - -async def _run_under_circuit_breaker( - breaker: RedisCircuitBreaker, - name: str, - call: Callable[[], Awaitable[_RedisCallResult]], -) -> _RedisCallResult: - """Run one Redis coroutine under a circuit breaker. - - Shared by the method decorator and the Lua script executor so both feed the same - health signal. - """ - admission: Final = _enter_circuit_breaker(breaker, name) - try: - result: Final = await call() - except Exception as e: - if _is_redis_health_failure(e): - breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) - raise - _exit_circuit_breaker(breaker, admission) - return result - - -def is_redis_error(exc: BaseException) -> bool: - from redis.exceptions import RedisError - - return isinstance(exc, RedisError) - - -def _run_under_circuit_breaker_sync( - breaker: RedisCircuitBreaker, - name: str, - call: Callable[[], _RedisCallResult], -) -> _RedisCallResult: - """Run one blocking Redis call under a circuit breaker, feeding the same health signal as the async path.""" - admission: Final = _enter_circuit_breaker(breaker, name) - try: - result: Final = call() - except Exception as e: - if _is_redis_health_failure(e): - breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) - raise - _exit_circuit_breaker(breaker, admission) - return result - - def _redis_circuit_breaker_guard(method): """ Decorator for RedisCache async methods. @@ -734,6 +307,27 @@ class RedisCache(BaseCache): else: super().__init__() # defaults to 60s + self._connection: Final = self._make_connection() + + def _make_connection(self) -> RedisConnection: + from litellm.redis.transport.events import RedisServiceEvents + + return RedisConnection( + client=lambda: self._async_commands(), + pubsub_client=lambda: self._pubsub_commands(), + namespace=lambda: self.namespace, + breaker=lambda: self._circuit_breaker, + admit=self._admit_transport_call, + events=lambda: RedisServiceEvents(self.service_logger_obj), + multi_get=lambda keys: self._async_run_redis_mget_operation(list(keys)), + server_version=lambda: self._parse_redis_major_version(), + request_engine=lambda: self._request_batch_engine(), + post_call_engine=lambda: self._post_call_batch_engine(), + ) + + def _admit_transport_call(self, call: Callable[[], Awaitable[_RedisCallResult]]) -> Awaitable[_RedisCallResult]: + return self._admit_async_call(call) + def _admit_async_call(self, call: Callable[[], Awaitable[_RedisCallResult]]) -> Awaitable[_RedisCallResult]: return call() @@ -860,23 +454,18 @@ class RedisCache(BaseCache): def _pubsub_commands(self) -> AsyncPubSubClient: return self.init_pubsub_client() # pyright: ignore[reportUnknownVariableType] # redis generics - def transport(self) -> RedisTransport: - from litellm.caching.redis_batch import post_call_batch_engine, request_batch_engine - from litellm.redis.transport.events import RedisServiceEvents - from litellm.redis.transport.redis_py import RedisPyTransport + def _request_batch_engine(self) -> PipelineBatch | None: + from litellm.caching.redis_batch import request_batch_engine - return RedisPyTransport( - client=self._async_commands, - namespace=self.namespace, - breaker=self._circuit_breaker, - admit=self._admit_async_call, - events=RedisServiceEvents(self.service_logger_obj), - multi_get=lambda keys: self._async_run_redis_mget_operation(list(keys)), - server_version=self._parse_redis_major_version, - pubsub_client=self._pubsub_commands, - request_engine=lambda: request_batch_engine(self), - post_call_engine=lambda: post_call_batch_engine(self), - ) + return request_batch_engine(self) + + def _post_call_batch_engine(self) -> PipelineBatch | None: + from litellm.caching.redis_batch import post_call_batch_engine + + return post_call_batch_engine(self) + + def transport(self) -> RedisTransport: + return self._connection.transport() def check_and_fix_namespace(self, key: str) -> str: """ diff --git a/litellm/redis/AGENTS.md b/litellm/redis/AGENTS.md index 1285bdc57d5..f87b5f6f67e 100644 --- a/litellm/redis/AGENTS.md +++ b/litellm/redis/AGENTS.md @@ -7,7 +7,8 @@ For adding Redis access or moving raw Redis access here, follow [.agents/skills/ - Every write takes its TTL from the caller, `None` means no expiry, there is no default - Replies are undecoded bytes, `mget` and `eval` arrays are tuples - 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 here that imports redis-py. `transport/breaker.py` reuses `RedisCache`'s breaker admission and health classification, it never reimplements them +- `connection.py` owns transport inputs without depending on caching. `health.py` owns the shared breaker, health classification and failure logging +- `transport/redis_py.py` is the redis-py command boundary. `transport/breaker.py` maps shared breaker failures to values - `transport/events.py` is the only place REDIS service events for transport traffic are emitted. Stores and callers emit nothing and do not log Redis health - `stores/` holds one module per keyspace, and a keyspace that has a store is only reached through it. A store is a `Protocol` of domain operations plus a frozen dataclass over `RedisTransport`. It owns its keys, value codec, expiry, Lua and decoding, and callers never see a Redis command - Keys, value bytes and TTL units in Redis are a wire format shared with older pods and Rust. Never change them in place diff --git a/litellm/redis/connection.py b/litellm/redis/connection.py new file mode 100644 index 00000000000..d5da1ff3d05 --- /dev/null +++ b/litellm/redis/connection.py @@ -0,0 +1,41 @@ +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from litellm.redis.health import RedisCircuitBreaker +from litellm.redis.transport.batch import PipelineBatch +from litellm.redis.transport.client import AsyncPubSubClient, AsyncRedisCommands +from litellm.redis.transport.contract import Admit, RedisTransport + +if TYPE_CHECKING: + from litellm.redis.transport.events import RedisServiceEvents + + +@dataclass(frozen=True, slots=True) +class RedisConnection: + client: Callable[[], AsyncRedisCommands] + pubsub_client: Callable[[], AsyncPubSubClient] + namespace: Callable[[], str | None] + breaker: Callable[[], RedisCircuitBreaker] + admit: Admit + events: Callable[[], "RedisServiceEvents"] + multi_get: Callable[[Sequence[str]], Awaitable[object]] + server_version: Callable[[], int] + request_engine: Callable[[], PipelineBatch | None] + post_call_engine: Callable[[], PipelineBatch | None] + + def transport(self) -> RedisTransport: + from litellm.redis.transport.redis_py import RedisPyTransport + + return RedisPyTransport( + client=self.client, + pubsub_client=self.pubsub_client, + namespace=self.namespace(), + breaker=self.breaker(), + admit=self.admit, + events=self.events(), + multi_get=self.multi_get, + server_version=self.server_version, + request_engine=self.request_engine, + post_call_engine=self.post_call_engine, + ) diff --git a/litellm/redis/health.py b/litellm/redis/health.py new file mode 100644 index 00000000000..02cef70fa9b --- /dev/null +++ b/litellm/redis/health.py @@ -0,0 +1,542 @@ +import asyncio +import functools +import inspect +import itertools +import logging +import threading +import time +import weakref +from collections import Counter +from collections.abc import Awaitable, Callable, Iterator, Mapping +from contextvars import ContextVar +from dataclasses import dataclass +from types import FrameType, MappingProxyType +from typing import TYPE_CHECKING, Final, TypeVar, cast + +from litellm._internal_context import current_service_caller +from litellm._logging import verbose_logger +from litellm.constants import REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION, REDIS_TIMEOUT_LOG_INTERVAL +from litellm.redis.transport.contract import RedisFailure + +if TYPE_CHECKING: + from litellm._service_logger import ServiceLogging + +__all__ = ( + "RedisCircuitBreaker", + "RedisCircuitBreakerOpenError", + "_BreakerAdmission", + "_RedisTimeoutLogThrottle", + "_caller_frames", + "_ends_call_stack", + "_enter_circuit_breaker", + "_exit_circuit_breaker", + "_explicit_causes", + "_get_call_stack_info", + "_is_generic_caller_frame", + "_is_redis_health_failure", + "_log_throttling_timeouts", + "_record_swallowed_redis_failure", + "_redis_health_error_types", + "_redis_timeout_error_types", + "_run_under_circuit_breaker", + "_run_under_circuit_breaker_sync", + "is_redis_error", + "is_redis_timeout_failure", + "log_redis_failure", + "log_redis_failure_value", + "redis_breaker_state_counts", +) + + +_GENERIC_CALLER_MODULES: Final = frozenset( + { + "caching.redis_cache", + "caching.redis_batch", + "caching.dual_cache", + "caching.caching", + "rust_bridge.execution", + "contextlib", + } +) +_GENERIC_CALLER_FRAME_NAMES: Final = frozenset( + { + "", + "wrapper", + "_run_under_circuit_breaker", + "_run_under_circuit_breaker_sync", + "run_alone", + "_settle_alone", + "get_cache", + "set_cache", + "async_get_cache", + "async_set_cache", + "async_batch_get_cache", + "async_batch_get_cache_shared", + "async_set_cache_pipeline", + "async_increment_cache", + "async_delete_cache", + } +) +_CALL_STACK_END_MODULES: Final = ("asyncio", "concurrent", "threading") + + +def _is_generic_caller_frame(frame: FrameType) -> bool: + module: Final = frame.f_globals.get("__name__") + return ( + (isinstance(module, str) and module.removeprefix("litellm.") in _GENERIC_CALLER_MODULES) + or (isinstance(module, str) and module.startswith("litellm.redis.")) + or frame.f_code.co_name in _GENERIC_CALLER_FRAME_NAMES + ) + + +def _ends_call_stack(frame: FrameType) -> bool: + module: Final = frame.f_globals.get("__name__") + return isinstance(module, str) and module.startswith(_CALL_STACK_END_MODULES) + + +def _caller_frames(first: FrameType) -> Iterator[FrameType]: + frame: FrameType | None = first + while frame is not None and not _ends_call_stack(frame): + yield frame + frame = frame.f_back + + +def _get_call_stack_info(num_frames: int = 2) -> str: + """ + Get the function names of the nearest meaningful callers of the cache method. + + Frames that merely forward the call (this module's circuit-breaker guards, the + cache facades, the batch pipeline's retry path, generic cache verbs) are + skipped, and the walk stops at the event loop, so the chain names the litellm + code that wanted the call. When nothing but forwarding frames is found (the call + runs in a task of its own, like a batch op retried on the flush) the chain the + declaring code threaded through ``service_caller`` is reported, else ``unknown``. + + Args: + num_frames: Number of previous frames to include (default: 2) + + Returns: + A string with format "caller_function [<- grandparent_function]" + """ + try: + current_frame: Final = inspect.currentframe() + if current_frame is None: + return "unknown" + + # Skip this function and the immediate caller (which sets call_type) + f_back: Final = current_frame.f_back + if f_back is None: + return "unknown" + first: Final = f_back.f_back + if first is None: + return "unknown" + frames: Final = _caller_frames(first) + leading: Final = tuple(itertools.islice(frames, num_frames)) + leading_names: Final = tuple(frame.f_code.co_name for frame in leading if not _is_generic_caller_frame(frame)) + further_names: Final = tuple( + itertools.islice( + (frame.f_code.co_name for frame in frames if not _is_generic_caller_frame(frame)), + num_frames - len(leading_names), + ) + ) + function_names: Final = leading_names + further_names + + if function_names: + return " <- ".join(function_names) + return current_service_caller() or "unknown" + except Exception: + return "unknown" + + +class RedisCircuitBreaker: + """ + Tracks Redis health for a RedisCache instance. + + States: + CLOSED - normal, Redis is called + OPEN - Redis is down, raise immediately (no network call) + HALF_OPEN - recovery probe: allow one request through + + Transitions: + CLOSED -> OPEN after failure_threshold consecutive hard connectivity + failures, or after an unbroken run of timeout failures + (no success or hard failure in between) that reaches + failure_threshold and spans timeout_min_duration seconds + OPEN -> HALF_OPEN after recovery_timeout seconds + HALF_OPEN -> CLOSED on success + HALF_OPEN -> OPEN on failure (resets timer) + + Timeouts are accounted separately from hard connectivity failures because the async + Redis timeout includes time waiting for the worker event loop to resume: one loop + stall makes every in-flight operation time out together, which satisfies a purely + consecutive threshold instantly even though Redis is healthy. Requiring a + timeout-only streak to also span timeout_min_duration filters such bursts while a + real outage that surfaces as timeouts still opens the breaker after that duration. + """ + + CLOSED = "closed" + OPEN = "open" + HALF_OPEN = "half_open" + + def __init__( + self, + failure_threshold: int, + recovery_timeout: int, + enabled: bool = True, + timeout_min_duration: float = REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION, + service_logger: "ServiceLogging | None" = None, + ) -> None: + self._service_logger = service_logger + self.failure_threshold = failure_threshold + self.recovery_timeout = recovery_timeout + self.enabled = enabled + self.timeout_min_duration = timeout_min_duration + self._failure_count = 0 + self._hard_failure_count = 0 + self._timeout_count = 0 + self._timeout_streak_started_at: float | None = None + self._opened_at: float | None = None + self._state = self.CLOSED + self._generation = 0 + _LIVE_BREAKERS.add(self) + self._report(transitions_to=None, failure_class=None) + + @property + def state(self) -> str: + return self._state + + @property + def generation(self) -> int: + """Counts state transitions, so a call can tell whether the breaker moved while it ran.""" + return self._generation + + def is_open(self) -> bool: + """Returns True if Redis calls should be skipped.""" + if not self.enabled: + return False + if self._state == self.HALF_OPEN: + # Probe already in flight — fast-fail all concurrent requests. + # Only the one call that caused the OPEN→HALF_OPEN transition + # (which returned False) is the designated probe. + return True + if self._state == self.OPEN: + if time.time() - (self._opened_at or 0) > self.recovery_timeout: + self._set_state(self.HALF_OPEN) + return False # this caller is the designated probe + return True + return False + + def _should_open(self, now: float) -> bool: + if self._state == self.HALF_OPEN: + return True + if self._hard_failure_count >= self.failure_threshold: + return True + if self._timeout_count < self.failure_threshold: + return False + return now - (self._timeout_streak_started_at or now) >= self.timeout_min_duration + + def record_failure(self, is_timeout: bool = False) -> None: + if not self.enabled: + return + now: Final = time.time() + self._failure_count += 1 + if is_timeout: + self._timeout_count += 1 + if self._timeout_streak_started_at is None: + self._timeout_streak_started_at = now + else: + self._hard_failure_count += 1 + self._timeout_count = 0 + self._timeout_streak_started_at = None + self._opened_at = now + self._report(transitions_to=None, failure_class="timeout" if is_timeout else "connectivity") + if self._should_open(now): + if self._state != self.OPEN: + verbose_logger.warning( + "Redis circuit breaker OPENED after %d consecutive failures" + " (%d hard connectivity) — fast-failing Redis calls for %ds", + self._failure_count, + self._hard_failure_count, + self.recovery_timeout, + ) + self._set_state(self.OPEN) + + def record_success(self) -> None: + if not self.enabled or self._state == self.OPEN: + return + if self._state == self.HALF_OPEN: + verbose_logger.info("Redis circuit breaker CLOSED — Redis recovered") + self._failure_count = 0 + self._hard_failure_count = 0 + self._timeout_count = 0 + self._timeout_streak_started_at = None + self._set_state(self.CLOSED) + + def _set_state(self, state: str) -> None: + if state == self._state: + return + self._state = state + self._generation += 1 + self._report(transitions_to=state, failure_class=None) + + def _report(self, transitions_to: str | None, failure_class: str | None) -> None: + if self._service_logger is not None: + self._service_logger.redis_breaker_hook(transitions_to=transitions_to, failure_class=failure_class) + + +_RedisCallResult = TypeVar("_RedisCallResult") + + +_swallowed_redis_failures: Final[ContextVar[int]] = ContextVar("litellm_swallowed_redis_failures", default=0) + + +@functools.lru_cache(maxsize=1) +def _redis_health_error_types() -> tuple[type, ...]: + """Exception types that mean the Redis backend itself is unhealthy. + + Command and data errors say nothing about connectivity: an INCR against a non-numeric + value or an undecodable cached entry is a request problem, and counting those would let + a caller trip the shared breaker on demand, dropping rate limiting to per-process + counters that spreading traffic across replicas can outrun. + + Imported lazily because this module is reachable from a base ``import litellm`` while + redis is not a base dependency. + """ + from redis.exceptions import ( # noqa: TID251 # shared Redis exception classification + BusyLoadingError, + ClusterDownError, + ) + from redis.exceptions import ( # noqa: TID251 # shared Redis exception classification + ConnectionError as RedisConnectionError, + ) + from redis.exceptions import ( # noqa: TID251 # shared Redis exception classification + TimeoutError as RedisTimeoutError, + ) + + return (RedisConnectionError, RedisTimeoutError, BusyLoadingError, ClusterDownError, OSError, asyncio.TimeoutError) + + +def _is_redis_health_failure(exc: BaseException) -> bool: + """True when ``exc`` indicates Redis is unreachable rather than the request being bad.""" + try: + return isinstance(exc, _redis_health_error_types()) + except ImportError: + return True + + +@functools.lru_cache(maxsize=1) +def _redis_timeout_error_types() -> tuple[type, ...]: + """Health failures that are timeouts rather than unambiguous connectivity errors. + + ``builtins.TimeoutError`` covers ``socket.timeout`` (an alias since py3.10) and, from + py3.11, ``asyncio.TimeoutError``; on py3.10 ``asyncio.TimeoutError`` is still its own + class, so it is listed explicitly. ``redis.exceptions.TimeoutError`` subclasses neither. + ``MaxConnectionsError`` is the async cluster client's pool-exhausted signal (its nodes have + no blocking pool, so it is raised instantly instead of after a wait): a busy pool, not a + dead Redis, like the standalone pool's chained wait timeout. + """ + try: + from redis import exceptions as redis_exceptions # noqa: TID251 # shared Redis exception classification + except ImportError: + return (TimeoutError, asyncio.TimeoutError) + max_connections_error: Final = cast( # cast-ok: types-redis 4.6 stubs predate MaxConnectionsError + "type[Exception]", + redis_exceptions.MaxConnectionsError, # pyright: ignore[reportAttributeAccessIssue] # types-redis 4.6 stubs predate it + ) + return (redis_exceptions.TimeoutError, TimeoutError, asyncio.TimeoutError, max_connections_error) + + +_MAX_EXCEPTION_CAUSE_DEPTH: Final = 20 + + +def _explicit_causes(exc: BaseException) -> Iterator[BaseException]: + current = exc # rebind-ok: advances one link per iteration of the bounded walk + for _ in range(_MAX_EXCEPTION_CAUSE_DEPTH): + yield current + if current.__cause__ is None: + return + current = current.__cause__ + + +def is_redis_timeout_failure(exc: BaseException) -> bool: + """True when ``exc`` or any exception it was explicitly raised ``from`` is a timeout. + + redis-py's blocking pool reports a pool wait timeout as ``ConnectionError`` chained from + ``asyncio.TimeoutError``, which is a busy pool rather than an unreachable Redis. + """ + timeout_types: Final = _redis_timeout_error_types() + return any(isinstance(link, timeout_types) for link in _explicit_causes(exc)) + + +def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseException) -> None: + """Record a Redis failure that the calling method is about to swallow. + + The marker is a ContextVar rather than a counter on the breaker because breakers are + shared by every concurrent caller. A plain shared counter cannot tell "my call failed" + from "some other in-flight call failed", so a success overlapping someone else's + failure would be discarded and a Redis that is answering would still be evicted. + asyncio gives each task its own copy of the context, so this is per-call. + """ + if not _is_redis_health_failure(exc): + return + breaker.record_failure(is_timeout=is_redis_timeout_failure(exc)) + _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1) + + +_LIVE_BREAKERS: Final[weakref.WeakSet[RedisCircuitBreaker]] = weakref.WeakSet() + + +def redis_breaker_state_counts() -> Mapping[str, int]: + """How many live Redis circuit breakers sit in each state, for the Prometheus services logger.""" + return MappingProxyType(dict(Counter(breaker.state for breaker in _LIVE_BREAKERS))) + + +class RedisCircuitBreakerOpenError(Exception): + pass + + +class _RedisTimeoutLogThrottle: + """Admits one Redis timeout log line per interval and counts the timeouts it suppressed in between.""" + + def __init__(self, interval: float, clock: Callable[[], float] = time.monotonic) -> None: + self.interval = interval + self._clock = clock + self._lock = threading.Lock() + self._last_logged_at: float | None = None + self._suppressed = 0 + + def admit(self) -> int | None: + """Return the number of timeouts suppressed since the last admitted line, or None to suppress this one.""" + with self._lock: + now: Final = self._clock() + if self._last_logged_at is not None and now - self._last_logged_at < self.interval: + self._suppressed += 1 + return None + suppressed: Final = self._suppressed + self._suppressed = 0 + self._last_logged_at = now + return suppressed + + +_redis_timeout_log_throttle: Final = _RedisTimeoutLogThrottle(REDIS_TIMEOUT_LOG_INTERVAL) + + +def log_redis_failure( + logger: logging.Logger, level: int, message: str, exc: BaseException, with_traceback: bool = False +) -> None: + if isinstance(exc, RedisCircuitBreakerOpenError): + logger.debug("%s: %s", message, exc, stacklevel=2) + return + _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, detail, stacklevel=3) + return + if suppressed == 0: + 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, + detail, + suppressed, + exc_info=exc_info, + stacklevel=3, + ) + + +@dataclass(frozen=True, slots=True) +class _BreakerAdmission: + swallowed_before: int + generation: int + + +def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> _BreakerAdmission: + """Reject the call if the breaker is open, else record what its success may later prove.""" + if breaker.is_open(): + raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open — skipping {name}") + return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), generation=breaker.generation) + + +def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission) -> None: + """Record success only when nothing failed while the call ran and the breaker has not moved since. + + Several Redis methods catch their own connection errors and return a default, so a + method that returned is not on its own proof of a healthy Redis. A success also vouches + only for the breaker state that admitted the call: a call admitted before the breaker + opened, or a probe admitted before a later failure reopened it, finishes knowing nothing + about whether Redis has recovered since, so only the current probe may close the breaker. + """ + if _swallowed_redis_failures.get() != admission.swallowed_before: + return + if breaker.generation != admission.generation: + return + breaker.record_success() + + +async def _run_under_circuit_breaker( + breaker: RedisCircuitBreaker, + name: str, + call: Callable[[], Awaitable[_RedisCallResult]], +) -> _RedisCallResult: + """Run one Redis coroutine under a circuit breaker. + + Shared by the method decorator and the Lua script executor so both feed the same + health signal. + """ + admission: Final = _enter_circuit_breaker(breaker, name) + try: + result: Final = await call() + except Exception as e: + if _is_redis_health_failure(e): + breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) + raise + _exit_circuit_breaker(breaker, admission) + return result + + +def is_redis_error(exc: BaseException) -> bool: + from redis.exceptions import RedisError # noqa: TID251 # shared Redis exception classification + + return isinstance(exc, RedisError) + + +def _run_under_circuit_breaker_sync( + breaker: RedisCircuitBreaker, + name: str, + call: Callable[[], _RedisCallResult], +) -> _RedisCallResult: + """Run one blocking Redis call under a circuit breaker, feeding the same health signal as the async path.""" + admission: Final = _enter_circuit_breaker(breaker, name) + try: + result: Final = call() + except Exception as e: + if _is_redis_health_failure(e): + breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) + raise + _exit_circuit_breaker(breaker, admission) + return result diff --git a/litellm/redis/transport/batch.py b/litellm/redis/transport/batch.py index 7585410181c..7d7228083d0 100644 --- a/litellm/redis/transport/batch.py +++ b/litellm/redis/transport/batch.py @@ -11,7 +11,7 @@ from litellm._internal_context import ( service_target, ) from litellm._logging import verbose_logger -from litellm.caching.redis_cache import ( +from litellm.redis.health import ( RedisCircuitBreaker, _get_call_stack_info, # pyright: ignore[reportPrivateUsage] # same caller chain every RedisCache method reports _run_under_circuit_breaker, # pyright: ignore[reportPrivateUsage] # same health signal as every RedisCache method diff --git a/litellm/redis/transport/breaker.py b/litellm/redis/transport/breaker.py index daa23cade72..6fede3f9682 100644 --- a/litellm/redis/transport/breaker.py +++ b/litellm/redis/transport/breaker.py @@ -1,7 +1,7 @@ from collections.abc import Awaitable, Callable from typing import Final, TypeVar -from litellm.caching.redis_cache import ( +from litellm.redis.health import ( RedisCircuitBreaker, RedisCircuitBreakerOpenError, _is_redis_health_failure, # pyright: ignore[reportPrivateUsage] # same health classification as RedisCache diff --git a/litellm/redis/transport/events.py b/litellm/redis/transport/events.py index ab24a7796c5..aaaf3056006 100644 --- a/litellm/redis/transport/events.py +++ b/litellm/redis/transport/events.py @@ -8,7 +8,7 @@ from typing import Final, TypeVar from litellm._internal_context import current_service_target, service_target from litellm._logging import verbose_logger from litellm._service_logger import ServiceLogging -from litellm.caching.redis_cache import ( +from litellm.redis.health import ( _get_call_stack_info, # pyright: ignore[reportPrivateUsage] # same caller attribution as RedisCache events log_redis_failure, log_redis_failure_value, diff --git a/litellm/redis/transport/redis_py.py b/litellm/redis/transport/redis_py.py index 9641f5c2bca..e98e4a2986d 100644 --- a/litellm/redis/transport/redis_py.py +++ b/litellm/redis/transport/redis_py.py @@ -9,7 +9,7 @@ from typing import Final, Generic, TypeVar from pydantic import TypeAdapter, ValidationError from redis.exceptions import NoScriptError -from litellm.caching.redis_cache import RedisCircuitBreaker +from litellm.redis.health import RedisCircuitBreaker from litellm.redis.transport.batch import BatchResult, PipelineBackend, PipelineBatch from litellm.redis.transport.breaker import failure_from, run_under_breaker from litellm.redis.transport.client import AsyncPubSub, AsyncPubSubClient, AsyncRedisCommands, RedisPipeline diff --git a/tests/unit/caching/test_caching.py b/tests/unit/caching/test_caching.py index c4727cbeff0..9230fea6319 100644 --- a/tests/unit/caching/test_caching.py +++ b/tests/unit/caching/test_caching.py @@ -14,8 +14,8 @@ import pytest import respx import litellm +import litellm.redis.health as redis_cache_module from litellm import acompletion, aembedding, completion, embedding -import litellm.caching.redis_cache as redis_cache_module from litellm._internal_context import current_service_target from litellm.caching.caching import Cache, CacheMode, response_cache_phase from litellm.caching.caching_handler import _PENDING_CACHE_WRITES diff --git a/tests/unit/caching/test_dual_cache.py b/tests/unit/caching/test_dual_cache.py index 5ebc3bf7336..1cf179242af 100644 --- a/tests/unit/caching/test_dual_cache.py +++ b/tests/unit/caching/test_dual_cache.py @@ -727,8 +727,8 @@ async def test_redis_timeouts_falling_back_to_memory_log_once_per_interval(caplo """The first fallback WARNING of a timeout streak logs, the rest stay at DEBUG until the summary.""" from redis.exceptions import TimeoutError as RedisTimeoutError - from litellm.caching import redis_cache as redis_cache_module from litellm.caching.redis_cache import _RedisTimeoutLogThrottle + from litellm.redis import health as redis_cache_module clock = MagicMock(return_value=1_000.0) monkeypatch.setattr( diff --git a/tests/unit/caching/test_redis_batch.py b/tests/unit/caching/test_redis_batch.py index cd270035416..e52a64164ff 100644 --- a/tests/unit/caching/test_redis_batch.py +++ b/tests/unit/caching/test_redis_batch.py @@ -95,6 +95,7 @@ class FakeRedisCache(RedisCache): self.namespace = namespace self._circuit_breaker = RedisCircuitBreaker(failure_threshold=5, recovery_timeout=30) self.service_logger_obj = ServiceLogging() + self._connection: Final = self._make_connection() self.default_ttl = None self.alone: list[tuple[str, Any]] = [] self.store: dict[str, Any] = {} diff --git a/tests/unit/caching/test_redis_cache.py b/tests/unit/caching/test_redis_cache.py index 2bdc7b9549b..71f7c8b65c4 100644 --- a/tests/unit/caching/test_redis_cache.py +++ b/tests/unit/caching/test_redis_cache.py @@ -683,12 +683,12 @@ def test_call_stack_info_skips_guard_frames_when_deployed_without_sources(monkey check comparing those two paths stops skipping and the service log then names the guard machinery instead of the real caller. """ - from litellm.caching import redis_cache as redis_cache_module from litellm.caching.redis_cache import ( RedisCircuitBreaker, _get_call_stack_info, _redis_circuit_breaker_guard_sync, ) + from litellm.redis import health as redis_cache_module monkeypatch.setattr(redis_cache_module, "__file__", redis_cache_module.__file__ + "c") @@ -1197,8 +1197,8 @@ def test_timeouts_during_a_blip_log_once_per_interval_not_once_per_call(sync_bat from redis.exceptions import TimeoutError as RedisTimeoutError - from litellm.caching import redis_cache as redis_cache_module from litellm.caching.redis_cache import _RedisTimeoutLogThrottle + from litellm.redis import health as redis_cache_module clock = MagicMock(return_value=1_000.0) monkeypatch.setattr( @@ -1271,8 +1271,8 @@ async def test_write_path_timeouts_inside_the_interval_stay_at_debug(call_method from redis.exceptions import TimeoutError as RedisTimeoutError - from litellm.caching import redis_cache as redis_cache_module from litellm.caching.redis_cache import _RedisTimeoutLogThrottle + from litellm.redis import health as redis_cache_module clock = MagicMock(return_value=1_000.0) throttle = _RedisTimeoutLogThrottle(interval=5.0, clock=clock) diff --git a/tests/unit/redis/test_connection.py b/tests/unit/redis/test_connection.py new file mode 100644 index 00000000000..00481d6728e --- /dev/null +++ b/tests/unit/redis/test_connection.py @@ -0,0 +1,65 @@ +from collections.abc import Awaitable, Callable +from typing import Final, TypeVar + +import fakeredis +import pytest + +from litellm._service_logger import ServiceLogging +from litellm.redis.connection import RedisConnection +from litellm.redis.health import RedisCircuitBreaker +from litellm.redis.transport.contract import RedisFailure +from litellm.redis.transport.events import RedisServiceEvents +from tests._support.fake_redis import fake_redis_cache + +_T: Final = TypeVar("_T") + + +def _admit(call: Callable[[], Awaitable[_T]]) -> Awaitable[_T]: + return call() + + +@pytest.mark.asyncio +async def test_connection_without_cache_namespaces_keys_and_records_failures( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + breaker: Final = RedisCircuitBreaker(failure_threshold=1, recovery_timeout=3600) + events: Final = RedisServiceEvents(ServiceLogging()) + connection: Final = RedisConnection( + client=lambda: raw, + pubsub_client=lambda: raw, + namespace=lambda: "connection", + breaker=lambda: breaker, + admit=_admit, + events=lambda: events, + multi_get=lambda keys: raw.mget(keys), + server_version=lambda: 7, + request_engine=lambda: None, + post_call_engine=lambda: None, + ) + transport: Final = connection.transport() + assert await transport.get("missing") is None + assert await transport.set("key", b"value", ttl=None) is True + assert await raw.get("connection:key") == b"value" + assert await raw.get("key") is None + assert await transport.mget(("key", "missing")) == (b"value", None) + server.connected = False + failed: Final = await transport.get("key") + assert isinstance(failed, RedisFailure) and failed.kind == "transport" + assert breaker.state == RedisCircuitBreaker.OPEN + refused: Final = await transport.get("key") + assert isinstance(refused, RedisFailure) and refused.kind == "breaker_open" + + +@pytest.mark.asyncio +async def test_cache_connection_reads_namespace_at_transport_construction( + server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis +) -> None: + cache: Final = fake_redis_cache(server, namespace="before") + first: Final = cache.transport() + assert await first.set("key", b"first", ttl=None) is True + cache.namespace = "after" + second: Final = cache.transport() + assert await second.set("key", b"second", ttl=None) is True + assert await first.get("key") == b"first" + assert await second.get("key") == b"second" + assert await raw.mget(("before:key", "after:key")) == [b"first", b"second"]