From 7ee0b70d58dd34b3919cc834198aa5e47eb2e562 Mon Sep 17 00:00:00 2001 From: mubashir1osmani Date: Thu, 13 Aug 2026 12:32:12 -0700 Subject: [PATCH] fix(redis): do not treat RedisClusterException subclasses as cluster-down SlotNotCoveredError, CrossSlotTransactionError, and InvalidPipelineStack share that base but are command/slot/pipeline errors on a live cluster. Matching isinstance(RedisClusterException) opened the breaker and dropped shared cache and rate limits to per-process state. Detect the 2026-08-13 hang by walking __cause__ (redis-py raises RedisClusterException from TimeoutError) and the unreachable-node message when the cause is stripped. --- litellm/caching/redis_cache.py | 35 +++++++--- .../test_litellm/caching/test_redis_cache.py | 66 +++++++++++++++++-- 2 files changed, 88 insertions(+), 13 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 28316cc391a..d6b416af923 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -184,16 +184,20 @@ def _redis_health_error_types() -> tuple[type, ...]: a caller trip the shared breaker on demand, dropping rate limiting to per-process counters that spreading traffic across replicas can outrun. - RedisClusterException is the wrapper redis-py raises when CLUSTER SLOTS / startup - nodes are unreachable (``Redis Cluster cannot be connected ... Timeout connecting - to server``). It subclasses ``Exception`` directly, not TimeoutError/ConnectionError, - so omitting it leaves every cache write retrying a dead cluster and blocking the - request for the connect timeout — the 2026-08-13 staging e2e hang. + RedisClusterException is *not* in this tuple. redis-py uses that base for + command, slot-layout, and pipeline errors (SlotNotCoveredError, + CrossSlotTransactionError, InvalidPipelineStack, "db must be 0", …) against + a cluster that is still answering. Matching the base would open the breaker + and drop shared cache / rate-limit state to per-process counters. + + The CLUSTER SLOTS / startup-node hang is detected in ``_is_redis_health_failure`` + by walking ``__cause__`` (redis-py does ``raise RedisClusterException(...) from + TimeoutError``) and by the unreachable-node message when the cause was lost. 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, RedisClusterException + from redis.exceptions import BusyLoadingError, ClusterDownError from redis.exceptions import ConnectionError as RedisConnectionError from redis.exceptions import TimeoutError as RedisTimeoutError @@ -202,16 +206,31 @@ def _redis_health_error_types() -> tuple[type, ...]: RedisTimeoutError, BusyLoadingError, ClusterDownError, - RedisClusterException, OSError, asyncio.TimeoutError, ) +# redis-py NodesManager.initialize: raise RedisClusterException( +# "Redis Cluster cannot be connected. Please provide at least one reachable node: …" +# ) from exception +_CLUSTER_UNREACHABLE_MARKER: Final = "redis cluster cannot be connected" + + 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()) + types: Final = _redis_health_error_types() + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + if isinstance(current, types): + return True + if _CLUSTER_UNREACHABLE_MARKER in str(current).lower(): + return True + seen.add(id(current)) + current = current.__cause__ if current.__cause__ is not None else current.__context__ + return False except ImportError: return True diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 0160aac2225..d48c889f75a 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -582,9 +582,12 @@ async def test_concurrent_success_is_not_cancelled_by_another_calls_failure(): 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("RedisClusterException", True, id="cluster_unreachable_is_unhealthy"), pytest.param("ResponseError", False, id="wrong_type_command_is_not"), pytest.param("DataError", False, id="bad_data_is_not"), + pytest.param("SlotNotCoveredError", False, id="slot_layout_is_not_unhealthy"), + pytest.param("CrossSlotTransactionError", False, id="cross_slot_is_not_unhealthy"), + pytest.param("InvalidPipelineStack", False, id="pipeline_stack_is_not_unhealthy"), + pytest.param("RedisClusterException", False, id="generic_cluster_command_error_is_not"), ], ) async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker): @@ -599,7 +602,10 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker) 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") + error_cls = getattr(redis.exceptions, error, None) + if error_cls is None: + pytest.skip(f"{error} is not in this redis-py") + raised = error_cls("boom") async def failing_call(): raise raised @@ -611,6 +617,41 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker) assert breaker.is_open() is opens_breaker +@pytest.mark.asyncio +async def test_redis_cluster_exception_subclasses_do_not_open_the_breaker(): + """Slot/pipeline/command errors share the RedisClusterException base. + + Matching that base (isinstance) would let a covered-slot miss or a + cross-slot pipeline trip the shared breaker and degrade rate limits to + per-process counters while Redis is still answering. + """ + from redis.exceptions import RedisClusterException + + from litellm.caching.redis_cache import ( + RedisCircuitBreaker, + _is_redis_health_failure, + _run_under_circuit_breaker, + ) + + class SlotLayoutError(RedisClusterException): + pass + + class PipelineStackError(RedisClusterException): + pass + + for raised in (SlotLayoutError("slot"), PipelineStackError("pipeline")): + assert _is_redis_health_failure(raised) is False + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) + + async def failing_call(exc=raised): + raise exc + + for _ in range(breaker.failure_threshold + 1): + with pytest.raises(RedisClusterException): + await _run_under_circuit_breaker(breaker, "op", failing_call) + assert breaker.is_open() is False + + @pytest.mark.asyncio async def test_cluster_connect_timeout_swallowed_set_opens_breaker(): """Staging 2026-08-13: redis-py wraps connect timeouts as RedisClusterException. @@ -629,16 +670,31 @@ async def test_cluster_connect_timeout_swallowed_set_opens_breaker(): _run_under_circuit_breaker, ) - production = RedisClusterException( + from redis.exceptions import TimeoutError as RedisTimeoutError + + timeout = RedisTimeoutError("Timeout connecting to server") + try: + raise RedisClusterException( + "Redis Cluster cannot be connected. Please provide at least one reachable node: " + "Timeout connecting to server" + ) from timeout + except RedisClusterException as production: + wrapped = production + + # Cause lost (message only) still counts — redis-py stringifies the timeout + # into the RedisClusterException text even if __cause__ is stripped. + message_only = RedisClusterException( "Redis Cluster cannot be connected. Please provide at least one reachable node: " "Timeout connecting to server" ) - assert _is_redis_health_failure(production) is True + assert _is_redis_health_failure(wrapped) is True + assert _is_redis_health_failure(message_only) is True + assert _is_redis_health_failure(RedisClusterException("Argument 'db' must be 0")) is False breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) async def swallowed_cluster_timeout(): - _record_swallowed_redis_failure(breaker, production) + _record_swallowed_redis_failure(breaker, wrapped) return None for _ in range(breaker.failure_threshold):