Merge 8ee19759e0 into 08f2a98250

litellm/caching/redis_cache.py

@@ -483,6 +483,67 @@ def redis_breaker_state_counts() -> Mapping[str, int]:
     return MappingProxyType(dict(python_states + Counter(native_redis_breaker_states())))
 
 
+_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
 
@@ -707,6 +768,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()
 

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.