diff --git a/litellm/_redis.py b/litellm/_redis.py index fe5c5cdabe9f..9e3b247f5773 100644 --- a/litellm/_redis.py +++ b/litellm/_redis.py @@ -61,23 +61,51 @@ def _get_redis_kwargs(): return available_args -def _get_redis_url_kwargs(client=None): - if client is None: - client = redis.Redis.from_url - arg_spec = inspect.getfullargspec(redis.Redis.from_url) +def _init_arg_names(cls: type) -> frozenset[str]: + """Every ``__init__`` parameter accepted anywhere in a class's MRO. - # Only allow primitive arguments - exclude_args = { - "self", - "connection_pool", - "retry", - } + Keyword-only parameters are included, and the MRO is walked because redis-py splits a + connection's parameters between ``AbstractConnection`` and its concrete subclasses. + """ + return frozenset( + name + for klass in inspect.getmro(cls) + if klass is not object + for spec in (inspect.getfullargspec(klass.__init__),) + for name in spec.args + spec.kwonlyargs + ) - include_args = ["url"] - available_args = [x for x in arg_spec.args if x not in exclude_args] + include_args +def _get_redis_url_kwargs(client: Optional[type] = None) -> tuple[str, ...]: + """Connection kwargs that redis-py forwards from ``from_url`` down to the connection. - return available_args + ``from_url`` is declared as ``(cls, url, **kwargs)``, so introspecting it yields no + connection kwargs at all. What it really does is hand its kwargs to the connection + class, so that class's signature is the allowlist. + + Taking the client's signature instead would be wrong in both directions: it omits + nothing useful, but it admits client-only parameters such as + ``single_connection_client`` and ``auto_close_connection_pool``, plus the ``ssl_*`` + family that only ``SSLConnection`` accepts. Those reach ``AbstractConnection`` and + raise ``TypeError`` the first time a connection is created. TLS on a url config is + selected by the ``rediss://`` scheme, which picks ``SSLConnection`` on its own. + """ + if client is None: + client = redis.Redis + connection_cls = async_redis.Connection if client is async_redis.Redis else redis.Connection + + exclude_args = frozenset( + { + "self", + "connection_pool", + "retry", + } + ) + + # Only allow primitive arguments + include_args = ("url", "max_connections") + + return tuple(x for x in _init_arg_names(connection_cls) if x not in exclude_args) + include_args def _get_redis_cluster_kwargs(client=None): @@ -614,7 +642,7 @@ def get_redis_async_client( if "url" in redis_kwargs and redis_kwargs["url"] is not None: if connection_pool is not None: return async_redis.Redis(connection_pool=connection_pool) - args = _get_redis_url_kwargs(client=async_redis.Redis.from_url) + args = _get_redis_url_kwargs(client=async_redis.Redis) url_kwargs = {} for arg in redis_kwargs: if arg in args: @@ -662,10 +690,10 @@ def get_redis_connection_pool( return None if "url" in redis_kwargs and redis_kwargs["url"] is not None: - pool_kwargs = { - "timeout": REDIS_CONNECTION_POOL_TIMEOUT, - "url": redis_kwargs["url"], - } + allowed_args = _get_redis_url_kwargs(client=async_redis.Redis) + pool_kwargs = {k: v for k, v in redis_kwargs.items() if k in allowed_args and k != "max_connections"} + pool_kwargs["timeout"] = REDIS_CONNECTION_POOL_TIMEOUT + pool_kwargs["url"] = redis_kwargs["url"] if "max_connections" in redis_kwargs: try: pool_kwargs["max_connections"] = int(redis_kwargs["max_connections"]) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index dd1c152a4214..9e0f022262b8 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -17,7 +17,8 @@ import time from collections.abc import Awaitable, Callable, Sequence from datetime import timedelta -from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Union, cast +from contextvars import ContextVar +from typing import TYPE_CHECKING, Any, List, Optional, Tuple, TypeVar, Union, cast import litellm from litellm._logging import print_verbose, verbose_logger @@ -168,24 +169,97 @@ def record_success(self) -> None: self._state = self.CLOSED +_RedisCallResult = TypeVar("_RedisCallResult") + + +_swallowed_redis_failures: 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 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 + + +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() + _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1) + + +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. Success is recorded only when nothing failed while ``call`` ran, + because several Redis methods catch their own connection errors and return a default. + """ + if breaker.is_open(): + raise Exception(f"Redis circuit breaker is open — skipping {name}") + swallowed_before = _swallowed_redis_failures.get() + try: + result = await call() + except Exception as e: + if _is_redis_health_failure(e): + breaker.record_failure() + raise + if _swallowed_redis_failures.get() == swallowed_before: + breaker.record_success() + return result + + def _redis_circuit_breaker_guard(method): # type: ignore """ Decorator for RedisCache async methods. Checks the circuit breaker before each call; records success/failure after. Does not apply to ping/disconnect/test_connection (health/teardown must always run). + + A returning method is not proof of a healthy Redis: several methods catch their own + connection errors and return a default so callers degrade rather than fail. Counting + those as successes reset the failure streak on every request, so the breaker could + never open and Redis was never taken out of the pool. Success is therefore recorded + only when no failure was registered while the method ran. """ @functools.wraps(method) async def wrapper(self, *args, **kwargs): # type: ignore - if self._circuit_breaker.is_open(): - raise Exception(f"Redis circuit breaker is open — skipping {method.__name__}") - try: - result = await method(self, *args, **kwargs) - self._circuit_breaker.record_success() - return result - except Exception: - self._circuit_breaker.record_failure() - raise + return await _run_under_circuit_breaker( + self._circuit_breaker, method.__name__, lambda: method(self, *args, **kwargs) + ) return wrapper @@ -551,13 +625,16 @@ def async_register_script(self, script: str) -> Callable[..., Awaitable[Any]]: ) async def run_script(keys: Sequence[str], args: Sequence[Any], client: Any = None) -> Any: - executor: Optional[Callable[..., Awaitable[Any]]] = litellm.in_memory_llm_clients_cache.get_cache( - key=script_cache_key - ) - if executor is None: - executor = self._register_script_for_current_loop(script) - litellm.in_memory_llm_clients_cache.set_cache(key=script_cache_key, value=executor) - return await executor(keys=keys, args=args, client=client) + async def execute() -> object: + executor: Optional[Callable[..., Awaitable[Any]]] = litellm.in_memory_llm_clients_cache.get_cache( + key=script_cache_key + ) + if executor is None: + executor = self._register_script_for_current_loop(script) + litellm.in_memory_llm_clients_cache.set_cache(key=script_cache_key, value=executor) + return await executor(keys=keys, args=args, client=client) + + return await _run_under_circuit_breaker(self._circuit_breaker, "run_script", execute) return run_script @@ -674,6 +751,7 @@ async def async_set_cache(self, key, value, **kwargs): str(e), value, ) + _record_swallowed_redis_failure(self._circuit_breaker, e) async def _pipeline_helper( self, @@ -758,6 +836,7 @@ async def async_set_cache_pipeline(self, cache_list: List[Tuple[Any, Any]], ttl: str(e), cache_value, ) + _record_swallowed_redis_failure(self._circuit_breaker, e) async def _set_cache_sadd_helper( self, @@ -842,6 +921,7 @@ async def async_set_cache_sadd(self, key, value: List, ttl: Optional[float], **k str(e), value, ) + _record_swallowed_redis_failure(self._circuit_breaker, e) @_redis_circuit_breaker_guard async def batch_cache_write(self, key, value, **kwargs): @@ -1106,6 +1186,7 @@ async def async_get_cache(self, key, parent_otel_span: Optional[Span] = None, ** ) ) print_verbose(f"litellm.caching.caching: async get() - Got exception from REDIS: {str(e)}") + _record_swallowed_redis_failure(self._circuit_breaker, e) @_redis_circuit_breaker_guard async def async_batch_get_cache( @@ -1177,6 +1258,7 @@ async def async_batch_get_cache( ) ) verbose_logger.error(f"Error occurred in async batch get cache - {str(e)}") + _record_swallowed_redis_failure(self._circuit_breaker, e) return key_value_dict def sync_ping(self) -> bool: @@ -1432,6 +1514,7 @@ async def async_get_ttl(self, key: str) -> Optional[int]: return ttl except Exception as e: verbose_logger.debug(f"Redis TTL Error: {e}") + _record_swallowed_redis_failure(self._circuit_breaker, e) return None @_redis_circuit_breaker_guard diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index a2e18a626387..59200719197a 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -59,9 +59,24 @@ def test_delete_cache_applies_namespace(namespace, monkeypatch, redis_no_ping): @pytest.mark.asyncio -async def test_redis_client_init_with_socket_timeout(monkeypatch, redis_no_ping): - monkeypatch.setenv("REDIS_HOST", "my-fake-host") - redis_cache = RedisCache(socket_timeout=1.0) +@pytest.mark.parametrize( + "redis_config", + [ + pytest.param({"host": "my-fake-host"}, id="host_port"), + pytest.param({"url": "redis://my-fake-host:6379"}, id="url"), + ], +) +async def test_redis_client_init_with_socket_timeout(monkeypatch, redis_no_ping, redis_config): + """socket_timeout has to reach the connection however Redis was configured. + + A url config used to drop every connection kwarg, so redis-py was left with + socket_timeout (and socket_connect_timeout, which falls back to it) unset. A + Redis host that drops packets instead of refusing them then blocks each caller + indefinitely, and the circuit breaker never trips because no call ever returns. + """ + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.delenv("REDIS_HOST", raising=False) + redis_cache = RedisCache(socket_timeout=1.0, **redis_config) assert redis_cache.redis_kwargs["socket_timeout"] == 1.0 client = redis_cache.init_async_client() assert client is not None @@ -428,3 +443,168 @@ def test_delete_cache_namespaces_key(namespace, expected, monkeypatch, redis_no_ redis_cache.redis_client = mock_client redis_cache.delete_cache(key="k") mock_client.delete.assert_called_once_with(expected) + + +def _closed_port() -> int: + """A port with nothing listening, so Redis calls fail fast and deterministically.""" + import socket + + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "call_method", + [ + pytest.param(lambda c: c.async_get_cache("lit4930"), id="async_get_cache"), + pytest.param(lambda c: c.async_batch_get_cache(["lit4930"]), id="async_batch_get_cache"), + pytest.param(lambda c: c.async_set_cache("lit4930", "v"), id="async_set_cache"), + pytest.param(lambda c: c.async_get_ttl("lit4930"), id="async_get_ttl"), + ], +) +async def test_circuit_breaker_opens_when_method_swallows_redis_failure(redis_no_ping, call_method): + """A guarded method that swallows its own Redis error must still count as a failure. + + These methods catch connection errors and return a default so callers degrade instead + of failing, which is correct. But that returns cleanly through the circuit breaker + guard, and counting it as a success reset the failure streak on every call, so the + breaker could never open. An unreachable Redis then stayed in the pool and every + request kept paying the full socket timeout on it. + """ + from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD + + cache = RedisCache(host="127.0.0.1", port=_closed_port(), socket_timeout=0.5) + + for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD): + await call_method(cache) + + with pytest.raises(Exception, match="circuit breaker is open"): + await call_method(cache) + + +@pytest.mark.asyncio +async def test_circuit_breaker_success_still_resets_the_failure_streak(redis_no_ping): + """A reachable Redis must keep the breaker closed, however many earlier calls failed. + + The guard now records success only when nothing failed while the method ran, so this + pins the other half of that contract: a call that genuinely reaches Redis has to clear + the streak, or a healthy Redis would eventually be evicted from the pool. + """ + from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD + + cache = RedisCache(host="127.0.0.1", port=_closed_port(), socket_timeout=0.5) + + for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD - 1): + await cache.async_get_cache("lit4930") + assert cache._circuit_breaker.is_open() is False + + reachable_redis = AsyncMock() + reachable_redis.get.return_value = None + with patch.object(cache, "init_async_client", return_value=reachable_redis): + await cache.async_get_cache("lit4930") + + for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD - 1): + await cache.async_get_cache("lit4930") + + assert cache._circuit_breaker.is_open() is False, "one success must clear the streak" + + +@pytest.mark.asyncio +async def test_circuit_breaker_covers_lua_script_execution(redis_no_ping): + """Lua script execution must feed the breaker like every other Redis call. + + The v3 rate limiter issues all of its Redis traffic through async_register_script, so + leaving that path unguarded meant the coordination calls during an outage never + counted toward taking Redis out of the pool and kept paying a full socket timeout + each, which is the traffic the outage hurts most. + """ + from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD + + cache = RedisCache(host="127.0.0.1", port=_closed_port(), socket_timeout=0.5) + run_script = cache.async_register_script("return 1") + + for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD): + with pytest.raises(Exception): + await run_script(keys=["lit4930"], args=[1]) + + with pytest.raises(Exception, match="circuit breaker is open"): + await run_script(keys=["lit4930"], args=[1]) + + +@pytest.mark.asyncio +async def test_concurrent_success_is_not_cancelled_by_another_calls_failure(): + """One caller's failure must not discard a different caller's success. + + A breaker is shared by every concurrent caller, so tracking "did this call fail" on the + breaker itself cannot tell my failure from someone else's. A Redis that is still + answering would then be evicted from the pool by unrelated in-flight failures, which is + the opposite of the outage this guard exists to handle. + """ + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.caching.redis_cache import ( + RedisCircuitBreaker, + _record_swallowed_redis_failure, + _run_under_circuit_breaker, + ) + + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) + + # The failure has to land after both calls are already in flight, which is the only + # ordering where a shared counter confuses the two. Failing before the healthy call + # starts would leave its snapshot correct and prove nothing. + async def swallows_a_failure(): + await asyncio.sleep(0.02) + _record_swallowed_redis_failure(breaker, RedisConnectionError("redis unreachable")) + return None + + async def succeeds_while_the_other_fails(): + await asyncio.sleep(0.05) + return "ok" + + rounds = breaker.failure_threshold + 1 + for _ in range(rounds): + await asyncio.gather( + _run_under_circuit_breaker(breaker, "failing", swallows_a_failure), + _run_under_circuit_breaker(breaker, "healthy", succeeds_while_the_other_fails), + ) + + assert breaker._failure_count < breaker.failure_threshold, "the healthy call must clear the streak" + assert breaker.is_open() is False, "a Redis answering every round must stay in the pool" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error, opens_breaker", + [ + pytest.param("ConnectionError", True, id="connection_refused_is_unhealthy"), + pytest.param("TimeoutError", True, id="timeout_is_unhealthy"), + pytest.param("BusyLoadingError", True, id="loading_is_unhealthy"), + pytest.param("ResponseError", False, id="wrong_type_command_is_not"), + pytest.param("DataError", False, id="bad_data_is_not"), + ], +) +async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker): + """Command and data errors must not count against Redis health. + + They say nothing about connectivity, and a caller able to provoke them (an INCR against + a non-numeric value, say) could otherwise trip the shared breaker on demand and drop + rate limiting to per-process counters, which spreading traffic across replicas outruns. + """ + import redis.exceptions + + from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker + + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) + raised = getattr(redis.exceptions, error)("boom") + + async def failing_call(): + raise raised + + for _ in range(breaker.failure_threshold + 1): + with pytest.raises(Exception): + await _run_under_circuit_breaker(breaker, "op", failing_call) + + assert breaker.is_open() is opens_breaker diff --git a/tests/test_litellm/test_redis.py b/tests/test_litellm/test_redis.py index 0818237655dd..e0fa800723df 100644 --- a/tests/test_litellm/test_redis.py +++ b/tests/test_litellm/test_redis.py @@ -737,3 +737,111 @@ def test_connection_pool_env_redis_ssl_false_uses_plain_connection(monkeypatch): assert pool is not None assert pool.connection_class is async_redis.Connection assert "ssl" not in pool.connection_kwargs + + +@pytest.mark.parametrize( + "redis_config", + [ + pytest.param({"host": "redis-host", "port": 6379}, id="host_port"), + pytest.param({"url": "redis://redis-host:6379"}, id="url"), + ], +) +def test_connection_pool_keeps_socket_timeout(redis_config, monkeypatch): + """The async pool must carry socket_timeout however Redis was configured. + + The url branch used to rebuild pool kwargs from scratch as {timeout, url, + max_connections}, dropping socket_timeout. redis-py then leaves both + socket_timeout and socket_connect_timeout (which falls back to it) unset, so a + Redis host that drops packets rather than refusing them blocks every caller + indefinitely instead of failing fast. + """ + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.delenv("REDIS_HOST", raising=False) + monkeypatch.delenv("REDIS_CLUSTER_NODES", raising=False) + + pool = get_redis_connection_pool(socket_timeout=5.0, **redis_config) + + assert pool is not None + assert pool.connection_kwargs.get("socket_timeout") == 5.0 + + +@pytest.mark.parametrize( + "redis_config", + [ + pytest.param({"host": "redis-host", "port": 6379}, id="host_port"), + pytest.param({"url": "redis://redis-host:6379"}, id="url"), + ], +) +def test_sync_client_keeps_socket_timeout(redis_config, monkeypatch): + """The sync client is built during RedisCache.__init__ and blocks the caller. + + Without socket_timeout it stalls for the OS TCP timeout against an unreachable + host, so merely constructing the cache stops the process. + """ + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.delenv("REDIS_HOST", raising=False) + monkeypatch.delenv("REDIS_CLUSTER_NODES", raising=False) + + client = get_redis_client(socket_timeout=5.0, **redis_config) + + assert client.connection_pool.connection_kwargs.get("socket_timeout") == 5.0 + + +@pytest.mark.parametrize( + "redis_config", + [ + pytest.param({"host": "redis-host", "port": 6379}, id="host_port"), + pytest.param({"url": "redis://redis-host:6379"}, id="url"), + ], +) +def test_async_client_keeps_socket_timeout(redis_config, monkeypatch): + """Same invariant for the async client built without an injected pool.""" + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.delenv("REDIS_HOST", raising=False) + monkeypatch.delenv("REDIS_CLUSTER_NODES", raising=False) + + client = get_redis_async_client(socket_timeout=5.0, **redis_config) + + assert client.connection_pool.connection_kwargs.get("socket_timeout") == 5.0 + + +def test_url_config_does_not_forward_ssl_kwarg(monkeypatch): + """ssl stays consumed rather than forwarded on the url path. + + TLS is selected by the rediss:// scheme there; handing ssl=True to a redis:// + url yields a plain Connection that rejects the kwarg when it first connects. + """ + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.delenv("REDIS_HOST", raising=False) + monkeypatch.delenv("REDIS_CLUSTER_NODES", raising=False) + + client = get_redis_client(url="redis://redis-host:6379", ssl=True) + + assert "ssl" not in client.connection_pool.connection_kwargs + + +@pytest.mark.parametrize( + "client_only_kwarg", + [ + pytest.param({"single_connection_client": True}, id="single_connection_client"), + pytest.param({"auto_close_connection_pool": True}, id="auto_close_connection_pool"), + pytest.param({"ssl_ca_certs": "/tmp/ca.pem"}, id="ssl_ca_certs"), + pytest.param({"ssl": True}, id="ssl"), + ], +) +def test_url_config_drops_kwargs_the_connection_cannot_accept(client_only_kwarg, monkeypatch): + """Only kwargs the connection accepts may be forwarded on the url path. + + from_url hands its kwargs down to the connection class, so client-level settings and + the SSLConnection-only ssl_* family raise TypeError the first time a connection is + created. TLS on a url config comes from the rediss:// scheme instead. + """ + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.delenv("REDIS_HOST", raising=False) + monkeypatch.delenv("REDIS_CLUSTER_NODES", raising=False) + + pool = get_redis_connection_pool(url="redis://redis-host:6379", socket_timeout=5.0, **client_only_kwarg) + + assert pool is not None + pool.make_connection() + assert pool.connection_kwargs.get("socket_timeout") == 5.0