mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-04 02:31:27 +00:00
fix(redis): count pool wait timeouts as breaker timeouts
redis-py's blocking pool reports a saturated pool as ConnectionError chained from asyncio.TimeoutError. The circuit breaker classified that as a hard connectivity failure and opened at once while Redis was healthy. Follow the explicit cause chain so it counts as a timeout and stays behind the timeout_min_duration gate
This commit is contained in:
parent
7419a536ad
commit
54e247998e
2 changed files with 69 additions and 1 deletions
|
|
@ -321,7 +321,15 @@ def _redis_timeout_error_types() -> tuple[type, ...]:
|
|||
|
||||
|
||||
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.
|
||||
"""
|
||||
if isinstance(exc, _redis_timeout_error_types()):
|
||||
return True
|
||||
cause: Final = exc.__cause__
|
||||
return cause is not None and _is_redis_timeout_failure(cause)
|
||||
|
||||
|
||||
class _BreakerMetrics:
|
||||
|
|
|
|||
|
|
@ -1030,3 +1030,63 @@ def test_sync_guard_counts_a_timeout_as_a_timeout():
|
|||
_run_under_circuit_breaker_sync(breaker, "op", timing_out_call)
|
||||
|
||||
assert breaker.is_open() is False
|
||||
|
||||
|
||||
@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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue