diff --git a/litellm/_redis.py b/litellm/_redis.py index ed9f3580162..4a12fe03e56 100644 --- a/litellm/_redis.py +++ b/litellm/_redis.py @@ -478,6 +478,20 @@ def _get_redis_client_logic(**env_overrides): return redis_kwargs +def _apply_redis_cluster_reconnect_defaults(cluster_kwargs: dict) -> None: + """Fail-fast reconnect defaults so a dropped cluster node cannot stall callers. + + redis-py rediscovers CLUSTER SLOTS on a dead connection. Without a connect + timeout, health check, and keepalive that rediscovery blocks for the OS TCP + timeout (tens of seconds) — the 2026-08-13 staging e2e 60s ALB hang. Explicit + kwargs from config still win. + """ + cluster_kwargs.setdefault("health_check_interval", REDIS_CLUSTER_HEALTH_CHECK_INTERVAL) + cluster_kwargs.setdefault("socket_keepalive", True) + cluster_kwargs.setdefault("socket_timeout", REDIS_SOCKET_TIMEOUT) + cluster_kwargs.setdefault("socket_connect_timeout", cluster_kwargs["socket_timeout"]) + + def init_redis_cluster(redis_kwargs) -> redis.RedisCluster: _redis_cluster_nodes_in_env: Final[str | None] = get_secret("REDIS_CLUSTER_NODES") if _redis_cluster_nodes_in_env is not None: @@ -503,6 +517,7 @@ def init_redis_cluster(redis_kwargs) -> redis.RedisCluster: new_startup_nodes.append(ClusterNode(**item)) cluster_kwargs.pop("startup_nodes", None) + _apply_redis_cluster_reconnect_defaults(cluster_kwargs) return redis.RedisCluster(startup_nodes=new_startup_nodes, **cluster_kwargs) @@ -630,8 +645,7 @@ def get_redis_async_client( # by a cluster restart (e.g. ElastiCache Serverless maintenance) is revalidated and # reconnected before reuse instead of stalling in re-initialization; an explicit value # from config still wins. - cluster_kwargs.setdefault("health_check_interval", REDIS_CLUSTER_HEALTH_CHECK_INTERVAL) - cluster_kwargs.setdefault("socket_keepalive", True) + _apply_redis_cluster_reconnect_defaults(cluster_kwargs) # Create async RedisCluster with IAM token as password if available cluster_client: Final = async_redis.RedisCluster( diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 5fedfc5bcce..28316cc391a 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -184,14 +184,28 @@ 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. + 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 BusyLoadingError, ClusterDownError, RedisClusterException from redis.exceptions import ConnectionError as RedisConnectionError from redis.exceptions import TimeoutError as RedisTimeoutError - return (RedisConnectionError, RedisTimeoutError, BusyLoadingError, ClusterDownError, OSError, asyncio.TimeoutError) + return ( + RedisConnectionError, + RedisTimeoutError, + BusyLoadingError, + ClusterDownError, + RedisClusterException, + OSError, + asyncio.TimeoutError, + ) def _is_redis_health_failure(exc: BaseException) -> bool: diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 59200719197..0160aac2225 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -582,6 +582,7 @@ 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"), ], @@ -608,3 +609,42 @@ async def test_only_connectivity_failures_open_the_breaker(error, opens_breaker) await _run_under_circuit_breaker(breaker, "op", failing_call) assert breaker.is_open() is opens_breaker + + +@pytest.mark.asyncio +async def test_cluster_connect_timeout_swallowed_set_opens_breaker(): + """Staging 2026-08-13: redis-py wraps connect timeouts as RedisClusterException. + + ``async_set_cache`` swallows the error so a dead cluster degrades instead of + 500ing. The breaker must still open — RedisClusterException subclasses + Exception, not TimeoutError — otherwise every later request retries CLUSTER + SLOTS and the worker hangs for the connect timeout (the 60s ALB read timeout). + """ + from redis.exceptions import RedisClusterException + + from litellm.caching.redis_cache import ( + RedisCircuitBreaker, + _is_redis_health_failure, + _record_swallowed_redis_failure, + _run_under_circuit_breaker, + ) + + production = 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 + + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) + + async def swallowed_cluster_timeout(): + _record_swallowed_redis_failure(breaker, production) + return None + + for _ in range(breaker.failure_threshold): + await _run_under_circuit_breaker(breaker, "async_set_cache", swallowed_cluster_timeout) + + assert breaker.is_open() is True + + with pytest.raises(Exception, match="circuit breaker is open"): + await _run_under_circuit_breaker(breaker, "async_set_cache", swallowed_cluster_timeout) diff --git a/tests/test_litellm/test_redis.py b/tests/test_litellm/test_redis.py index 14c242f1096..328d0264ece 100644 --- a/tests/test_litellm/test_redis.py +++ b/tests/test_litellm/test_redis.py @@ -12,7 +12,7 @@ from litellm._redis import ( get_redis_connection_pool, get_redis_url_from_environment, ) -from litellm.constants import REDIS_CLUSTER_HEALTH_CHECK_INTERVAL +from litellm.constants import REDIS_CLUSTER_HEALTH_CHECK_INTERVAL, REDIS_SOCKET_TIMEOUT from litellm._redis_credential_provider import ( GCPIAMCredentialProvider, _token_cache, @@ -195,6 +195,25 @@ def test_async_cluster_sets_reconnect_defaults(mock_cluster_cls): assert call_kwargs["health_check_interval"] == REDIS_CLUSTER_HEALTH_CHECK_INTERVAL assert call_kwargs["health_check_interval"] > 0 assert call_kwargs["socket_keepalive"] is True + assert call_kwargs["socket_timeout"] == REDIS_SOCKET_TIMEOUT + assert call_kwargs["socket_connect_timeout"] == REDIS_SOCKET_TIMEOUT + + +@patch("litellm._redis.redis.RedisCluster") +def test_sync_cluster_sets_reconnect_defaults(mock_cluster_cls): + """Sync RedisCluster must get the same fail-fast reconnect defaults as async. + + init_redis_cluster used to skip them, so a dropped ElastiCache Serverless + connection stalled the (often event-loop) caller in CLUSTER SLOTS rediscovery. + """ + get_redis_client(startup_nodes=[{"host": "cluster-node", "port": 6379}]) + + mock_cluster_cls.assert_called_once() + call_kwargs = mock_cluster_cls.call_args[1] + assert call_kwargs["health_check_interval"] == REDIS_CLUSTER_HEALTH_CHECK_INTERVAL + assert call_kwargs["socket_keepalive"] is True + assert call_kwargs["socket_timeout"] == REDIS_SOCKET_TIMEOUT + assert call_kwargs["socket_connect_timeout"] == REDIS_SOCKET_TIMEOUT @patch("litellm._redis.async_redis.RedisCluster")