diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 812b435ae5f..2c36995c4f8 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -197,10 +197,13 @@ class RedisCircuitBreaker: self._timeout_streak_started_at: float | None = None self._opened_at: float | None = None self._state = self.CLOSED + self._generation = 0 _breaker_metrics().record_state_change(None, self._state) - def is_half_open(self) -> bool: - return self._state == self.HALF_OPEN + @property + def generation(self) -> int: + """Counts state transitions, so a call can tell whether the breaker moved while it ran.""" + return self._generation def is_open(self) -> bool: """Returns True if Redis calls should be skipped.""" @@ -270,6 +273,7 @@ class RedisCircuitBreaker: _breaker_metrics().record_transition(state) _breaker_metrics().record_state_change(self._state, state) self._state = state + self._generation += 1 _RedisCallResult = TypeVar("_RedisCallResult") @@ -412,27 +416,28 @@ def log_redis_failure( @dataclass(frozen=True, slots=True) class _BreakerAdmission: swallowed_before: int - is_probe: bool + generation: int def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> _BreakerAdmission: """Reject the call if the breaker is open, else record what its success may later prove.""" if breaker.is_open(): raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open — skipping {name}") - return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), is_probe=breaker.is_half_open()) + return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), generation=breaker.generation) def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission) -> None: - """Record success only when nothing failed while the call ran and the call may vouch for Redis. + """Record success only when nothing failed while the call ran and the breaker has not moved since. Several Redis methods catch their own connection errors and return a default, so a - method that returned is not on its own proof of a healthy Redis. While the breaker is - half open only the designated recovery probe may close it: a call admitted before the - breaker opened that finishes late says nothing about whether Redis recovered. + method that returned is not on its own proof of a healthy Redis. A success also vouches + only for the breaker state that admitted the call: a call admitted before the breaker + opened, or a probe admitted before a later failure reopened it, finishes knowing nothing + about whether Redis has recovered since, so only the current probe may close the breaker. """ if _swallowed_redis_failures.get() != admission.swallowed_before: return - if breaker.is_half_open() and not admission.is_probe: + if breaker.generation != admission.generation: return breaker.record_success() diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 9df9c44a30f..bcae33b976e 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1152,3 +1152,53 @@ async def test_stale_success_during_the_recovery_probe_leaves_the_breaker_to_the assert breaker._state == breaker.CLOSED assert breaker.is_open() is False + + +@pytest.mark.asyncio +async def test_a_probe_overtaken_by_a_later_outage_leaves_the_breaker_to_the_new_probe(): + """A probe still in flight when a late failure reopens the breaker must not close it for the next probe. + + Once the breaker has reopened, only the probe admitted after that outage has reached + Redis, so the older probe's success no longer says anything about whether Redis recovered. + """ + from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker + + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) + old_probe_admitted = asyncio.Event() + old_probe_release = asyncio.Event() + new_probe_admitted = asyncio.Event() + new_probe_release = asyncio.Event() + + async def old_probe_call() -> str: + old_probe_admitted.set() + await old_probe_release.wait() + return "old probe" + + async def new_probe_call() -> str: + new_probe_admitted.set() + await new_probe_release.wait() + return "new probe" + + for _ in range(3): + breaker.record_failure() + breaker._opened_at = time.time() - 9999 + old_probe = asyncio.ensure_future(_run_under_circuit_breaker(breaker, "op", old_probe_call)) + await old_probe_admitted.wait() + assert breaker._state == breaker.HALF_OPEN + + breaker.record_failure() + assert breaker._state == breaker.OPEN + breaker._opened_at = time.time() - 9999 + new_probe = asyncio.ensure_future(_run_under_circuit_breaker(breaker, "op", new_probe_call)) + await new_probe_admitted.wait() + assert breaker._state == breaker.HALF_OPEN + + old_probe_release.set() + assert await old_probe == "old probe" + + assert breaker._state == breaker.HALF_OPEN, "the overtaken probe must not close the breaker for the new probe" + assert breaker.is_open() is True + + new_probe_release.set() + assert await new_probe == "new probe" + assert breaker._state == breaker.CLOSED