fix(caching): let only the recovery probe close a half-open Redis breaker

A call admitted before the breaker opened could finish while the breaker was
HALF_OPEN and close it before the designated probe reported, so Redis traffic
resumed on a stale answer. The admission now records whether the call is the
probe and only the probe's success closes a half-open breaker.

The cron job lock manager also logged an error every cycle the open breaker
refused its Redis call, one line per job per pod. That refusal is now a debug
line like every other guarded call, while real Redis errors still log at error

litellm/caching/redis_cache.py

@@ -18,6 +18,7 @@ import logging
 import time
 from collections.abc import Awaitable, Callable, Sequence
 from contextvars import ContextVar
+from dataclasses import dataclass
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any, Final, Protocol, TypeVar, cast
 
@@ -198,6 +199,9 @@ class RedisCircuitBreaker:
         self._state = self.CLOSED
         _breaker_metrics().record_state_change(None, self._state)
 
+    def is_half_open(self) -> bool:
+        return self._state == self.HALF_OPEN
+
     def is_open(self) -> bool:
         """Returns True if Redis calls should be skipped."""
         if not self.enabled:
@@ -405,21 +409,32 @@ def log_redis_failure(
     logger.log(level, "%s: %s", message, exc, exc_info=exc if with_traceback else None)
 
 
-def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> int:
-    """Reject the call if the breaker is open, else return the swallowed-failure count to compare against."""
+@dataclass(frozen=True, slots=True)
+class _BreakerAdmission:
+    swallowed_before: int
+    is_probe: bool
+
+
+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 _swallowed_redis_failures.get()
+    return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), is_probe=breaker.is_half_open())
 
 
-def _exit_circuit_breaker(breaker: RedisCircuitBreaker, swallowed_before: int) -> None:
-    """Record success only when nothing failed while the call ran.
+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.
 
     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.
+    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.
     """
-    if _swallowed_redis_failures.get() == swallowed_before:
-        breaker.record_success()
+    if _swallowed_redis_failures.get() != admission.swallowed_before:
+        return
+    if breaker.is_half_open() and not admission.is_probe:
+        return
+    breaker.record_success()
 
 
 async def _run_under_circuit_breaker(
@@ -432,14 +447,14 @@ async def _run_under_circuit_breaker(
     Shared by the method decorator and the Lua script executor so both feed the same
     health signal.
     """
-    swallowed_before: Final = _enter_circuit_breaker(breaker, name)
+    admission: Final = _enter_circuit_breaker(breaker, name)
     try:
         result: Final = await call()
     except Exception as e:
         if _is_redis_health_failure(e):
             breaker.record_failure(is_timeout=_is_redis_timeout_failure(e))
         raise
-    _exit_circuit_breaker(breaker, swallowed_before)
+    _exit_circuit_breaker(breaker, admission)
     return result
 
 
@@ -449,14 +464,14 @@ def _run_under_circuit_breaker_sync(
     call: Callable[[], _RedisCallResult],
 ) -> _RedisCallResult:
     """Run one blocking Redis call under a circuit breaker, feeding the same health signal as the async path."""
-    swallowed_before: Final = _enter_circuit_breaker(breaker, name)
+    admission: Final = _enter_circuit_breaker(breaker, name)
     try:
         result: Final = call()
     except Exception as e:
         if _is_redis_health_failure(e):
             breaker.record_failure(is_timeout=_is_redis_timeout_failure(e))
         raise
-    _exit_circuit_breaker(breaker, swallowed_before)
+    _exit_circuit_breaker(breaker, admission)
     return result
 
 
@@ -1395,12 +1410,12 @@ class RedisCache(BaseCache):
         key_value_dict = {}
         _key_list: Final = [key for key in key_list if key is not None]
         start_time: Final = time.time()
-        swallowed_before: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache")
+        admission: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache")
 
         try:
             _keys: Final = [self.check_and_fix_namespace(key=cache_key or "") for cache_key in _key_list]
             results: Final = self._run_redis_mget_operation(keys=_keys)
-            _exit_circuit_breaker(self._circuit_breaker, swallowed_before)
+            _exit_circuit_breaker(self._circuit_breaker, admission)
             end_time: Final = time.time()
             _duration: Final = end_time - start_time
             self.service_logger_obj.service_success_hook(

litellm/proxy/db/db_transaction_queue/pod_lock_manager.py

@@ -1,10 +1,11 @@
 import asyncio
 import json
+import logging
 from typing import TYPE_CHECKING, Any, Final
 
 from litellm._logging import verbose_proxy_logger
 from litellm._uuid import uuid
-from litellm.caching.redis_cache import RedisCache
+from litellm.caching.redis_cache import RedisCache, log_redis_failure
 from litellm.constants import DEFAULT_CRON_JOB_LOCK_TTL_SECONDS
 from litellm.proxy.db.db_transaction_queue.base_update_queue import service_logger_obj
 from litellm.types.services import ServiceTypes
@@ -109,7 +110,7 @@ end
                     )
             return False
         except Exception as e:
-            verbose_proxy_logger.error("Error acquiring Redis lock for %s: %s", cronjob_id, e)
+            log_redis_failure(verbose_proxy_logger, logging.ERROR, f"Error acquiring Redis lock for {cronjob_id}", e)
             return False
 
     async def release_lock(
@@ -151,7 +152,7 @@ end
                     cronjob_id,
                 )
         except Exception as e:
-            verbose_proxy_logger.error("Error releasing Redis lock for %s: %s", cronjob_id, e)
+            log_redis_failure(verbose_proxy_logger, logging.ERROR, f"Error releasing Redis lock for {cronjob_id}", e)
 
     async def _compare_and_delete_lock(self, lock_key: str) -> int:
         """

tests/test_litellm/caching/test_redis_cache.py

@@ -1103,3 +1103,52 @@ def test_recovery_probe_still_closes_the_breaker():
 
     assert breaker._state == breaker.CLOSED
     assert breaker.is_open() is False
+
+
+@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.
+
+    Only the one call designated as the recovery probe has actually reached Redis after the
+    outage, so closing on the straggler's success resumed full Redis traffic before the probe
+    had proven anything.
+    """
+    from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker
+
+    breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
+    stale_admitted = asyncio.Event()
+    stale_release = asyncio.Event()
+    probe_admitted = asyncio.Event()
+    probe_release = asyncio.Event()
+
+    async def stale_call() -> str:
+        stale_admitted.set()
+        await stale_release.wait()
+        return "stale"
+
+    async def probe_call() -> str:
+        probe_admitted.set()
+        await probe_release.wait()
+        return "probe"
+
+    stale = asyncio.ensure_future(_run_under_circuit_breaker(breaker, "op", stale_call))
+    await stale_admitted.wait()
+    for _ in range(3):
+        breaker.record_failure()
+    assert breaker._state == breaker.OPEN
+    breaker._opened_at = time.time() - 9999
+    probe = asyncio.ensure_future(_run_under_circuit_breaker(breaker, "op", probe_call))
+    await probe_admitted.wait()
+    assert breaker._state == breaker.HALF_OPEN
+
+    stale_release.set()
+    assert await stale == "stale"
+
+    assert breaker._state == breaker.HALF_OPEN, "the straggler must not close the breaker for the probe"
+    assert breaker.is_open() is True
+
+    probe_release.set()
+    assert await probe == "probe"
+
+    assert breaker._state == breaker.CLOSED
+    assert breaker.is_open() is False

tests/test_litellm/proxy/db/db_transaction_queue/test_pod_lock_manager.py

@@ -1,4 +1,5 @@
 import json
+import logging
 from datetime import datetime, timedelta, timezone
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -6,6 +7,7 @@ import pytest
 from fastapi.testclient import TestClient
 
 
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
 from litellm.constants import DEFAULT_CRON_JOB_LOCK_TTL_SECONDS
 from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
 
@@ -215,6 +217,22 @@ async def test_redis_error_handling(pod_lock_manager, mock_redis):
     )
 
 
+@pytest.mark.asyncio
+async def test_lock_refused_by_the_open_circuit_breaker_is_not_logged_as_an_error(pod_lock_manager, mock_redis, caplog):
+    """Every cron job retries its lock on a timer, so an open breaker must not add an error line per cycle."""
+    refused = RedisCircuitBreakerOpenError("Redis circuit breaker is open - skipping async_set_cache")
+    mock_redis.async_set_cache.side_effect = refused
+    mock_redis.async_get_cache.return_value = pod_lock_manager.pod_id
+    mock_redis.async_delete_cache.side_effect = refused
+
+    with caplog.at_level(logging.ERROR):
+        acquired = await pod_lock_manager.acquire_lock(cronjob_id="test_job")
+        await pod_lock_manager.release_lock(cronjob_id="test_job")
+
+    assert acquired is False
+    assert caplog.records == []
+
+
 @pytest.mark.asyncio
 async def test_bytes_handling(pod_lock_manager, mock_redis):
     """