Merge pull request #40764 from BerriAI/litellm_redis_pool_timeout_counts_as_timeout

fix(redis): count pool wait timeouts as breaker timeouts

litellm/caching/redis_cache.py

@@ -16,7 +16,7 @@ import inspect
 import json
 import logging
 import time
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable, Iterator, Sequence
 from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import timedelta
@@ -317,19 +317,37 @@ def _is_redis_health_failure(exc: BaseException) -> bool:
 def _redis_timeout_error_types() -> tuple[type, ...]:
     """Health failures that are timeouts rather than unambiguous connectivity errors.
 
-    ``builtins.TimeoutError`` covers ``asyncio.TimeoutError`` and ``socket.timeout``
-    (aliases since py3.11 / py3.10). ``redis.exceptions.TimeoutError`` does not subclass
-    either, so it is listed explicitly.
+    ``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.
     """
     try:
         from redis.exceptions import TimeoutError as RedisTimeoutError
     except ImportError:
-        return (TimeoutError,)
-    return (RedisTimeoutError, TimeoutError)
+        return (TimeoutError, asyncio.TimeoutError)
+    return (RedisTimeoutError, TimeoutError, asyncio.TimeoutError)
+
+
+_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:
-    return isinstance(exc, _redis_timeout_error_types())
+    """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))
 
 
 class _BreakerMetrics:

tests/test_litellm/caching/test_redis_cache.py

@@ -1205,6 +1205,66 @@ async def test_a_probe_overtaken_by_a_later_outage_leaves_the_breaker_to_the_new
     assert breaker._state == breaker.CLOSED
 
 
+@pytest.mark.asyncio
+async def test_pool_wait_timeout_is_a_timeout_failure_not_hard_connectivity():
+    """A saturated blocking pool must not open the breaker before the timeout minimum duration.
+
+    redis-py's async BlockingConnectionPool gives up waiting for a free connection by raising
+    ConnectionError("No connection available.") chained from asyncio.TimeoutError. Redis itself
+    is healthy in that case, so the failure has to be classed as a timeout and stay behind the
+    duration gate instead of being counted as a hard connectivity failure.
+    """
+    from fakeredis import FakeServer
+    from fakeredis.aioredis import FakeConnection
+    from redis.asyncio import BlockingConnectionPool, Redis
+    from redis.exceptions import ConnectionError as RedisConnectionError
+
+    from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker
+
+    pool = BlockingConnectionPool(connection_class=FakeConnection, server=FakeServer(), max_connections=1, timeout=0.01)
+    client = Redis(connection_pool=pool)
+    breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=5.0)
+
+    busy_connection = await pool.get_connection()
+    try:
+        for _ in range(breaker.failure_threshold * 2):
+            with pytest.raises(RedisConnectionError, match="No connection available"):
+                await _run_under_circuit_breaker(breaker, "op", lambda: client.get("k"))
+    finally:
+        await pool.release(busy_connection)
+
+    assert breaker.is_open() is False, "a busy pool is a timeout gated on duration, not a dead Redis"
+    assert await _run_under_circuit_breaker(breaker, "op", lambda: client.get("k")) is None
+    await client.aclose()
+
+
+def test_timeout_classification_follows_the_explicit_cause_chain_only():
+    from redis.exceptions import ConnectionError as RedisConnectionError
+
+    from litellm.caching.redis_cache import _is_redis_timeout_failure
+
+    def raise_chained_from_timeout() -> None:
+        try:
+            raise asyncio.TimeoutError()
+        except asyncio.TimeoutError as err:
+            raise RedisConnectionError("No connection available.") from err
+
+    def raise_while_handling_timeout() -> None:
+        try:
+            raise asyncio.TimeoutError()
+        except asyncio.TimeoutError:
+            raise RedisConnectionError("refused")
+
+    with pytest.raises(RedisConnectionError) as chained:
+        raise_chained_from_timeout()
+    with pytest.raises(RedisConnectionError) as contextual:
+        raise_while_handling_timeout()
+
+    assert _is_redis_timeout_failure(chained.value) is True
+    assert _is_redis_timeout_failure(contextual.value) is False
+    assert _is_redis_timeout_failure(RedisConnectionError("refused")) is False
+
+
 class _RoundTripCountingRedis:
     """Fake redis.asyncio client: one round trip per awaited command or pipeline execute."""