diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index afe6f8a1eab..e3244d17e32 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -476,6 +476,67 @@ def redis_breaker_state_counts() -> Mapping[str, int]: return MappingProxyType(dict(Counter(breaker.state for breaker in _LIVE_BREAKERS))) +_REDIS_BREAKER_PROBE_INTERVAL_SECONDS: Final = 5.0 + +_LIVE_REDIS_CACHES: Final = weakref.WeakSet() + +_breaker_prober_lock: Final = threading.Lock() +_breaker_prober_thread: threading.Thread | None = None + + +def _probe_open_breaker(cache: "RedisCache") -> bool: + breaker: Final = cache._circuit_breaker # pyright: ignore[reportPrivateUsage] # the prober is RedisCache's own machinery + if breaker.state != RedisCircuitBreaker.OPEN: + return False + ping: Final = getattr(cache.redis_client, "ping", None) + if ping is None or coroutine_checker.is_async_callable(cache.redis_client): + return False + if breaker.is_open(): + return False + try: + result: Final = ping() + except Exception as e: + breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) + return True + if inspect.isawaitable(result): + result.close() # pyright: ignore[reportAttributeAccessIssue] # an awaitable ping result here is a coroutine + breaker.record_failure() + return True + if not result: + breaker.record_failure() + return True + breaker.record_success() + return True + + +def _probe_open_breakers_once() -> None: + for cache in tuple(_LIVE_REDIS_CACHES): + try: + _probe_open_breaker(cache) + except Exception: + verbose_logger.debug("Redis circuit breaker background probe failed", exc_info=True) + + +def _breaker_prober_loop() -> None: + while True: + time.sleep(_REDIS_BREAKER_PROBE_INTERVAL_SECONDS) + _probe_open_breakers_once() + + +def _register_redis_cache_for_breaker_probing(cache: "RedisCache") -> None: + global _breaker_prober_thread + _LIVE_REDIS_CACHES.add(cache) + if not cache._circuit_breaker.enabled: # pyright: ignore[reportPrivateUsage] # same registration helper + return + with _breaker_prober_lock: + if _breaker_prober_thread is not None and _breaker_prober_thread.is_alive(): + return + _breaker_prober_thread = threading.Thread( + target=_breaker_prober_loop, name="litellm-redis-breaker-prober", daemon=True + ) + _breaker_prober_thread.start() + + class RedisCircuitBreakerOpenError(Exception): pass @@ -698,6 +759,7 @@ class RedisCache(BaseCache): enabled=REDIS_CIRCUIT_BREAKER_ENABLED, service_logger=self.service_logger_obj, ) + _register_redis_cache_for_breaker_probing(self) self._setup_health_pings() diff --git a/tests/unit/caching/test_redis_cache.py b/tests/unit/caching/test_redis_cache.py index 2bdc7b9549b..ba96a9c6572 100644 --- a/tests/unit/caching/test_redis_cache.py +++ b/tests/unit/caching/test_redis_cache.py @@ -1092,6 +1092,86 @@ def test_recovery_probe_still_closes_the_breaker(): assert breaker.is_open() is False +def _open_the_breaker(cache) -> None: + cache._circuit_breaker.failure_threshold = 3 + for _ in range(3): + cache._circuit_breaker.record_failure() + assert cache._circuit_breaker._state == cache._circuit_breaker.OPEN + cache._circuit_breaker._opened_at = time.time() - 9999 + + +def test_background_probe_closes_an_open_breaker_on_an_idle_cache(sync_batch_redis_cache): + from litellm.caching.redis_cache import _probe_open_breaker + + sync_batch_redis_cache.redis_client.ping = MagicMock(return_value=True) + _open_the_breaker(sync_batch_redis_cache) + + assert _probe_open_breaker(sync_batch_redis_cache) is True + + assert sync_batch_redis_cache.redis_client.ping.call_count == 1 + assert sync_batch_redis_cache._circuit_breaker._state == sync_batch_redis_cache._circuit_breaker.CLOSED + assert sync_batch_redis_cache._circuit_breaker.is_open() is False + + +def test_background_probe_failure_reopens_the_breaker_and_restarts_the_timer(sync_batch_redis_cache): + from litellm.caching.redis_cache import _probe_open_breaker + + sync_batch_redis_cache.redis_client.ping = MagicMock(side_effect=OSError("redis still down")) + _open_the_breaker(sync_batch_redis_cache) + aged_opened_at: Final = sync_batch_redis_cache._circuit_breaker._opened_at + + assert _probe_open_breaker(sync_batch_redis_cache) is True + + breaker: Final = sync_batch_redis_cache._circuit_breaker + assert breaker._state == breaker.OPEN + assert breaker._opened_at is not None and aged_opened_at is not None and breaker._opened_at > aged_opened_at + assert _probe_open_breaker(sync_batch_redis_cache) is False + assert sync_batch_redis_cache.redis_client.ping.call_count == 1 + + +def test_background_probe_waits_for_the_recovery_timeout(sync_batch_redis_cache): + from litellm.caching.redis_cache import _probe_open_breaker + + sync_batch_redis_cache.redis_client.ping = MagicMock(return_value=True) + sync_batch_redis_cache._circuit_breaker.failure_threshold = 3 + for _ in range(3): + sync_batch_redis_cache._circuit_breaker.record_failure() + assert sync_batch_redis_cache._circuit_breaker._state == sync_batch_redis_cache._circuit_breaker.OPEN + + assert _probe_open_breaker(sync_batch_redis_cache) is False + + assert sync_batch_redis_cache.redis_client.ping.call_count == 0 + assert sync_batch_redis_cache._circuit_breaker._state == sync_batch_redis_cache._circuit_breaker.OPEN + + +def test_background_probe_yields_the_probe_slot_to_a_real_call(sync_batch_redis_cache): + from litellm.caching.redis_cache import _probe_open_breaker + + sync_batch_redis_cache.redis_client.ping = MagicMock(return_value=True) + _open_the_breaker(sync_batch_redis_cache) + breaker: Final = sync_batch_redis_cache._circuit_breaker + assert breaker.is_open() is False + assert breaker._state == breaker.HALF_OPEN + + assert _probe_open_breaker(sync_batch_redis_cache) is False + + assert sync_batch_redis_cache.redis_client.ping.call_count == 0 + assert breaker._state == breaker.HALF_OPEN + breaker.record_success() + assert breaker._state == breaker.CLOSED + + +def test_a_background_probe_pass_reaches_registered_caches(sync_batch_redis_cache): + from litellm.caching.redis_cache import _probe_open_breakers_once + + sync_batch_redis_cache.redis_client.ping = MagicMock(return_value=True) + _open_the_breaker(sync_batch_redis_cache) + + _probe_open_breakers_once() + + assert sync_batch_redis_cache._circuit_breaker._state == sync_batch_redis_cache._circuit_breaker.CLOSED + + @pytest.mark.asyncio async def test_stale_success_during_the_recovery_probe_leaves_the_breaker_to_the_probe(): """A call admitted before the trip that finishes while HALF_OPEN must not close the breaker.