mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-03 07:23:33 +00:00
refactor(redis): split the connection out of RedisCache
This commit is contained in:
Devin AI
parent
38fa821a61
commit
08276193a7
14 changed files with 770 additions and 531 deletions
|
|
@ -20,7 +20,7 @@ metadata:
|
|||
- New coordination state gets a store in `litellm/redis/stores/<keyspace>.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
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
{
|
||||
"<lambda>",
|
||||
"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:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
41
litellm/redis/connection.py
Normal file
41
litellm/redis/connection.py
Normal file
|
|
@ -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,
|
||||
)
|
||||
542
litellm/redis/health.py
Normal file
542
litellm/redis/health.py
Normal file
|
|
@ -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(
|
||||
{
|
||||
"<lambda>",
|
||||
"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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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] = {}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
65
tests/unit/redis/test_connection.py
Normal file
65
tests/unit/redis/test_connection.py
Normal file
|
|
@ -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"]
|
||||
Loading…
Add table
Reference in a new issue