fix(caching): backport Redis breaker recovery guards to rc/1.101.0 (#40624)

(cherry picked from commit ff4b558243)

litellm/caching/dual_cache.py

@@ -23,7 +23,7 @@ from litellm.constants import DEFAULT_MAX_REDIS_BATCH_CACHE_SIZE
 
 from .base_cache import BaseCache
 from .in_memory_cache import InMemoryCache
-from .redis_cache import RedisCache, log_redis_failure
+from .redis_cache import RedisCache, RedisCircuitBreakerOpenError, log_redis_failure
 
 if TYPE_CHECKING:
     from opentelemetry.trace import Span as _Span
@@ -206,9 +206,12 @@ class DualCache(BaseCache):
                 redis_result: Final = self.redis_cache.batch_get_cache(
                     key_list=sublist_keys, parent_otel_span=parent_otel_span
                 )
-            except Exception:
+            except Exception as e:
                 # Do not throttle subsequent callers if the Redis read fails.
                 self._rollback_redis_batch_key_reservations(previous_access_times)
+                if isinstance(e, RedisCircuitBreakerOpenError):
+                    verbose_logger.debug("LiteLLM Cache: batch_get_cache served from memory only: %s", e)
+                    return result
                 raise
 
             if self.in_memory_cache is not None:
@@ -325,9 +328,12 @@ class DualCache(BaseCache):
                         redis_result: Final = await self.redis_cache.async_batch_get_cache(
                             sublist_keys, parent_otel_span=parent_otel_span
                         )
-                    except Exception:
+                    except Exception as e:
                         # Do not throttle subsequent callers if the Redis read fails.
                         self._rollback_redis_batch_key_reservations(previous_access_times)
+                        if isinstance(e, RedisCircuitBreakerOpenError):
+                            verbose_logger.debug("LiteLLM Cache: async_batch_get_cache served from memory only: %s", e)
+                            return result
                         raise
 
                     # Short-circuit if redis_result is None or contains only None values

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
 
@@ -176,8 +177,14 @@ 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)
 
+    @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."""
         if not self.enabled:
@@ -230,7 +237,7 @@ class RedisCircuitBreaker:
             self._set_state(self.OPEN)
 
     def record_success(self) -> None:
-        if not self.enabled:
+        if not self.enabled or self._state == self.OPEN:
             return
         if self._state == self.HALF_OPEN:
             verbose_logger.info("Redis circuit breaker CLOSED — Redis recovered")
@@ -246,6 +253,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")
@@ -385,21 +393,33 @@ 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
+    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 _swallowed_redis_failures.get()
+    return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), generation=breaker.generation)
 
 
-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 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.
+    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() == swallowed_before:
-        breaker.record_success()
+    if _swallowed_redis_failures.get() != admission.swallowed_before:
+        return
+    if breaker.generation != admission.generation:
+        return
+    breaker.record_success()
 
 
 async def _run_under_circuit_breaker(
@@ -412,14 +432,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
 
 
@@ -429,14 +449,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
 
 
@@ -1272,6 +1292,7 @@ class RedisCache(BaseCache):
         except Exception:
             return ast.literal_eval(decoded)
 
+    @_redis_circuit_breaker_guard_sync
     def get_cache(self, key, parent_otel_span: Span | None = None, **kwargs):
         try:
             key = self.check_and_fix_namespace(key=key)
@@ -1291,8 +1312,8 @@ class RedisCache(BaseCache):
             print_verbose(f"Got Redis Cache: key: {key}, cached_response {cached_response}")
             return self._get_cache_logic(cached_response=cached_response)
         except Exception as e:
-            # NON blocking - notify users Redis is throwing an exception
-            verbose_logger.error("litellm.caching.caching: get() - Got exception from REDIS: ", e)
+            verbose_logger.error("litellm.caching.caching: get() - Got exception from REDIS: %s", e)
+            _record_swallowed_redis_failure(self._circuit_breaker, e)
 
     def _run_redis_mget_operation(self, keys: list[str]) -> Sequence[bytes | str | None]:
         """
@@ -1329,12 +1350,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()
+        admission: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache")
 
         try:
-            swallowed_before: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache")
             _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:
         """

litellm/proxy/management_endpoints/ui_sso.py

@@ -48,6 +48,7 @@ import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.caching.dual_cache import DualCache
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
 from litellm.constants import (
     CLI_SSO_CLAIM_MAP,
     CLI_SSO_CLAIM_MAX_SCALAR_LENGTH,
@@ -353,6 +354,16 @@ def _check_cli_sso_start_rate_limit(
         )
 
 
+def _read_cli_sso_flow(cache: DualCache, cache_key: str) -> object:
+    redis_cache: Final = cache.redis_cache
+    if redis_cache is None:
+        return cache.get_cache(key=cache_key)
+    try:
+        return redis_cache.get_cache(key=cache_key)
+    except RedisCircuitBreakerOpenError:
+        return None
+
+
 def _get_cli_sso_flow_or_raise(login_id: str | None, cache: DualCache) -> dict:
     if isinstance(login_id, str) and login_id.startswith("sk-"):
         raise HTTPException(
@@ -365,12 +376,7 @@ def _get_cli_sso_flow_or_raise(login_id: str | None, cache: DualCache) -> dict:
     if not _is_valid_cli_sso_login_id(login_id):
         raise HTTPException(status_code=400, detail="Invalid CLI login session id")
 
-    cache_key: Final = _get_cli_sso_flow_cache_key(cast(str, login_id))
-    redis_cache: Final = cache.redis_cache
-    if redis_cache is not None:
-        flow = redis_cache.get_cache(key=cache_key)
-    else:
-        flow = cache.get_cache(key=cache_key)
+    flow = _read_cli_sso_flow(cache, _get_cli_sso_flow_cache_key(cast(str, login_id)))
     if isinstance(flow, str):
         try:
             flow = _as_object(json.loads(flow))

litellm/proxy/proxy_server.py

@@ -236,6 +236,7 @@ import litellm._redis
 from litellm import Router
 from litellm._logging import _redact_string, verbose_proxy_logger, verbose_router_logger
 from litellm.caching.caching import DualCache, RedisCache
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
 from litellm.caching.redis_cluster_cache import RedisClusterCache
 from litellm.constants import (
     _REALTIME_BODY_CACHE_SIZE,
@@ -3332,8 +3333,10 @@ async def _apply_spend_counter_increments(pending: Sequence[_PendingSpendIncreme
     ]
     try:
         results: Final = await redis_cache.async_increment_pipeline(increment_list=increment_list)
-    except Exception:
+    except Exception as e:
         await asyncio.gather(*(_invalidate_spend_counter(counter_key=item.counter_key) for item in pending))
+        if isinstance(e, RedisCircuitBreakerOpenError):
+            return
         raise
     for item, current_value in zip(pending, results or ()):
         spend_counter_cache.in_memory_cache.set_cache(key=item.counter_key, value=current_value)

litellm/router_strategy/base_routing_strategy.py

@@ -3,12 +3,13 @@ Base class across routing strategies to abstract commmon functions like batch in
 """
 
 import asyncio
+import logging
 from abc import ABC
 from typing import Final
 
 from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
-from litellm.caching.redis_cache import RedisPipelineIncrementOperation
+from litellm.caching.redis_cache import RedisPipelineIncrementOperation, log_redis_failure
 from litellm.constants import DEFAULT_REDIS_SYNC_INTERVAL
 
 
@@ -147,7 +148,7 @@ class BaseRoutingStrategy(ABC):
                 return return_result
 
         except Exception as e:
-            verbose_router_logger.error("Error syncing in-memory cache with Redis: %s", e)
+            log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e)
             self.redis_increment_operation_queue = []
 
     def add_to_in_memory_keys_to_update(self, key: str):

litellm/router_strategy/budget_limiter.py

@@ -20,6 +20,7 @@ anthropic:
 
 import asyncio
 import builtins
+import logging
 from collections.abc import Mapping
 from datetime import datetime, timedelta, timezone
 from typing import Any, Final
@@ -27,7 +28,7 @@ from typing import Any, Final
 import litellm
 from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
-from litellm.caching.redis_cache import RedisPipelineIncrementOperation
+from litellm.caching.redis_cache import RedisCache, RedisPipelineIncrementOperation, log_redis_failure
 from litellm.integrations.custom_logger import CustomLogger, Span
 from litellm.litellm_core_utils.core_helpers import (
     get_metadata_variable_name_from_kwargs,
@@ -92,6 +93,13 @@ class _LiteLLMParamsDictView:
         return dict(self._params)
 
 
+async def _push_increments_to_redis(redis_cache: RedisCache, queued: list[RedisPipelineIncrementOperation]) -> None:
+    try:
+        await redis_cache.async_increment_pipeline(increment_list=queued)
+    except Exception as e:
+        log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e)
+
+
 class RouterBudgetLimiting(CustomLogger):
     def __init__(
         self,
@@ -536,17 +544,13 @@ class RouterBudgetLimiting(CustomLogger):
                 "Pushing Redis Increment Pipeline for queue: %s",
                 self.redis_increment_operation_queue,
             )
-            if len(self.redis_increment_operation_queue) > 0:
-                asyncio.create_task(
-                    self.dual_cache.redis_cache.async_increment_pipeline(
-                        increment_list=self.redis_increment_operation_queue,
-                    )
-                )
-
+            queued: Final = self.redis_increment_operation_queue
             self.redis_increment_operation_queue = []
+            if queued:
+                asyncio.create_task(_push_increments_to_redis(self.dual_cache.redis_cache, queued))
 
         except Exception as e:
-            verbose_router_logger.error("Error syncing in-memory cache with Redis: %s", e)
+            log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e)
 
     async def _sync_in_memory_spend_with_redis(self):
         """
@@ -601,7 +605,7 @@ class RouterBudgetLimiting(CustomLogger):
                         verbose_router_logger.debug("Updated in-memory cache for %s: %s", key, value)
 
         except Exception as e:
-            verbose_router_logger.error("Error syncing in-memory cache with Redis: %s", e)
+            log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e)
 
     def _get_budget_config_for_deployment(
         self,

litellm/router_utils/health_state_cache.py

@@ -12,6 +12,7 @@ from typing_extensions import TypedDict
 
 from litellm import verbose_logger
 from litellm.caching.caching import DualCache
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
 
 if TYPE_CHECKING:
     from opentelemetry.trace import Span as _Span
@@ -27,6 +28,16 @@ class DeploymentHealthStateValue(TypedDict):
     reason: str
 
 
+def _read_shared_health_snapshot(cache: DualCache, key: str) -> object:
+    redis_cache: Final = cache.redis_cache
+    if redis_cache is None:
+        return None
+    try:
+        return redis_cache.get_cache(key)
+    except RedisCircuitBreakerOpenError:
+        return None
+
+
 class DeploymentHealthCache:
     """
     Cache for deployment health states produced by background health checks.
@@ -50,13 +61,12 @@ class DeploymentHealthCache:
         coexist on the one shared entry without erasing each other's results.
         The snapshot is read from Redis when available, since a pod-local read
         would only ever see this writer's own previous merge. When the Redis
-        read comes back empty (a miss, or a swallowed connection error), the
-        pod-local copy of the last merge is used so peers are not erased.
+        read comes back empty (a miss, a swallowed connection error, or a read
+        refused by the open circuit breaker), the pod-local copy of the last
+        merge is used so peers are not erased.
         """
         try:
-            redis_raw: Final = (
-                self.cache.redis_cache.get_cache(self.CACHE_KEY) if self.cache.redis_cache is not None else None
-            )
+            redis_raw: Final = _read_shared_health_snapshot(self.cache, self.CACHE_KEY)
             raw: Final = redis_raw if isinstance(redis_raw, dict) else self.cache.get_cache(key=self.CACHE_KEY)
             existing: Final = raw if isinstance(raw, dict) else {}
             expiry_seconds: Final = self.staleness_threshold * 1.5

tests/test_litellm/caching/test_dual_cache.py

@@ -677,3 +677,30 @@ async def test_a_real_redis_failure_still_logs_an_error(caplog):
     errors = [record for record in caplog.records if record.levelno == logging.ERROR]
     assert [record.getMessage() for record in errors] == ["LiteLLM Cache: exception in async_get_cache: redis is down"]
     assert errors[0].exc_info is not None
+
+
+def _dual_cache_with_open_breaker_and_a_memory_hit() -> DualCache:
+    in_memory = InMemoryCache()
+    in_memory.set_cache("k1", "v1")
+    return DualCache(in_memory_cache=in_memory, redis_cache=_OpenBreakerRedis(), default_redis_batch_cache_expiry=10)  # pyright: ignore[reportArgumentType]  # duck-typed Redis double
+
+
+def test_open_breaker_keeps_sync_batch_read_memory_hits_and_releases_reservations():
+    """A refused Redis batch read must still answer with the in-memory hits and hold no reservation.
+
+    The refusal was logged and turned into a bare None, so a caller lost its in-memory hits
+    for as long as the breaker stayed open, and the reserved keys stayed throttled until
+    the batch expiry passed even though nothing was ever read for them.
+    """
+    cache = _dual_cache_with_open_breaker_and_a_memory_hit()
+
+    assert list(cache.batch_get_cache(["k1", "k2"])) == ["v1", None]
+    assert "k2" not in cache.last_redis_batch_access_time
+
+
+@pytest.mark.asyncio
+async def test_open_breaker_keeps_async_batch_read_memory_hits_and_releases_reservations():
+    cache = _dual_cache_with_open_breaker_and_a_memory_hit()
+
+    assert list(await cache.async_batch_get_cache(["k1", "k2"])) == ["v1", None]
+    assert "k2" not in cache.last_redis_batch_access_time

tests/test_litellm/caching/test_redis_cache.py

@@ -1,11 +1,12 @@
 import asyncio
+import time
 from collections.abc import Iterator
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 from litellm._service_logger import ServiceLogging
-from litellm.caching.redis_cache import RedisCache
+from litellm.caching.redis_cache import RedisCache, RedisCircuitBreakerOpenError
 
 
 @pytest.fixture
@@ -515,14 +516,46 @@ async def test_circuit_breaker_opens_when_method_swallows_redis_failure(call_met
         await call_method(cache)
 
 
-def test_circuit_breaker_open_keeps_sync_batch_get_cache_as_a_miss(sync_batch_redis_cache):
-    """An open breaker must preserve the sync batch read's dictionary fallback."""
+def test_circuit_breaker_open_makes_sync_batch_get_cache_fast_fail(sync_batch_redis_cache, caplog):
+    """Once the breaker is open the sync batch read refuses with the typed error instead of a miss.
+
+    Swallowing the refusal into `{}` made every sync batch read on an open breaker emit an ERROR
+    log and a service failure event per call, and the DualCache caller could not tell the
+    refusal from a dead Redis, so it dropped its in-memory hits too.
+    """
     from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
 
     for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
         assert sync_batch_redis_cache.batch_get_cache(key_list=["lit6729"]) == {}
 
-    assert sync_batch_redis_cache.batch_get_cache(key_list=["lit6729"]) == {}
+    caplog.clear()
+    with caplog.at_level("INFO"):
+        with pytest.raises(RedisCircuitBreakerOpenError):
+            sync_batch_redis_cache.batch_get_cache(key_list=["lit6729"])
+    sync_batch_redis_cache.redis_client.mget.assert_called()
+    assert caplog.records == []
+
+
+def test_sync_get_cache_failure_feeds_the_breaker_and_logs_a_well_formed_record(sync_batch_redis_cache, caplog):
+    """The sync get path swallowed its Redis error without recording it, and its log call was malformed.
+
+    `verbose_logger.error("...: ", e)` passes the exception as a format argument to a message
+    with no placeholder, so the record carried no error text. Nothing fed the breaker either,
+    so a dead Redis read through this path never opened it.
+    """
+    from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
+
+    sync_batch_redis_cache.redis_client.get.side_effect = OSError("redis unavailable")
+
+    with caplog.at_level("ERROR"):
+        for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
+            assert sync_batch_redis_cache.get_cache("lit7468") is None
+
+    assert all("redis unavailable" in record.getMessage() for record in caplog.records)
+    assert len(caplog.records) == REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
+    assert sync_batch_redis_cache._circuit_breaker.is_open() is True
+    with pytest.raises(RedisCircuitBreakerOpenError):
+        sync_batch_redis_cache.get_cache("lit7468")
 
 
 @pytest.fixture
@@ -617,7 +650,8 @@ def test_sync_batch_get_cache_survives_a_service_callback_that_raises(
         with ThreadPoolExecutor(max_workers=1) as pool:
             assert pool.submit(cache.batch_get_cache, key_list=["lit6729"]).result() == {}
 
-    assert cache.batch_get_cache(key_list=["lit6729"]) == {}
+    with pytest.raises(RedisCircuitBreakerOpenError):
+        cache.batch_get_cache(key_list=["lit6729"])
 
 
 def test_call_stack_info_skips_breaker_guard_frames():
@@ -966,6 +1000,8 @@ async def test_breaker_metrics_track_state_and_failure_class():
     assert sample("litellm_redis_circuit_breaker_state", {"state": "open"}) == open_gauge_before + 1
     assert sample("litellm_redis_circuit_breaker_state", {"state": "closed"}) == closed_gauge_before
 
+    breaker._opened_at = time.time() - 9999
+    assert breaker.is_open() is False
     breaker.record_success()
     assert sample("litellm_redis_circuit_breaker_state", {"state": "open"}) == open_gauge_before
     assert sample("litellm_redis_circuit_breaker_state", {"state": "closed"}) == closed_gauge_before + 1
@@ -986,3 +1022,139 @@ 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
+
+
+def test_success_admitted_before_the_breaker_opened_cannot_close_it():
+    """A stale in-flight success must not close a breaker that opened while it ran.
+
+    Calls admitted while the breaker was still closed finish after later failures opened it.
+    Recording their success unconditionally closed the breaker again, skipping the recovery
+    timeout and the single half-open probe, so the breaker flapped between open and closed
+    on every straggler while Redis was still down.
+    """
+    from litellm.caching.redis_cache import RedisCircuitBreaker
+
+    breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
+    for _ in range(3):
+        breaker.record_failure()
+    assert breaker._state == breaker.OPEN
+
+    breaker.record_success()
+
+    assert breaker._state == breaker.OPEN
+    assert breaker.is_open() is True
+
+
+def test_recovery_probe_still_closes_the_breaker():
+    from litellm.caching.redis_cache import RedisCircuitBreaker
+
+    breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
+    for _ in range(3):
+        breaker.record_failure()
+    breaker._opened_at = time.time() - 9999
+    assert breaker.is_open() is False
+    assert breaker._state == breaker.HALF_OPEN
+
+    breaker.record_success()
+
+    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
+
+
+@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

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):
     """

tests/test_litellm/proxy/management_endpoints/test_ui_sso.py

@@ -2616,6 +2616,28 @@ class TestCLIKeyRegenerationFlow:
         )
         cache.set_cache.assert_not_called()
 
+    def test_cli_sso_flow_lookup_treats_an_open_redis_breaker_as_a_miss(self):
+        """A Redis read refused by the open circuit breaker is a missing session, not a server error.
+
+        The direct Redis read is what keeps the flow authoritative across workers, so the
+        refusal must not fall back to a possibly stale in-memory copy either.
+        """
+        from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
+        from litellm.proxy.management_endpoints.ui_sso import _get_cli_sso_flow_or_raise
+
+        redis_cache = MagicMock()
+        redis_cache.get_cache.side_effect = RedisCircuitBreakerOpenError("Redis circuit breaker is open")
+        cache = MagicMock()
+        cache.redis_cache = redis_cache
+        cache.get_cache.return_value = {"poll_secret_hash": "stale", "sso_complete": False}
+
+        with pytest.raises(HTTPException) as exc_info:
+            _get_cli_sso_flow_or_raise(login_id="cli-breaker_open_1234567890", cache=cache)
+
+        assert exc_info.value.status_code == 400
+        assert "not found or expired" in exc_info.value.detail
+        cache.get_cache.assert_not_called()
+
     def test_cli_sso_flow_with_enum_survives_redis_round_trip(self):
         """
         RedisCache stores values via str(value) and reads them back through

tests/test_litellm/proxy/proxy_server/test_spend_counters.py

@@ -1146,6 +1146,54 @@ async def test_prepare_window_spend_counter_increment_missing_window_start_inval
     assert fake_cache.redis_cache.async_increment.called is False
 
 
+# ---------------------------------------------------------------------------
+# _apply_spend_counter_increments
+# ---------------------------------------------------------------------------
+
+
+def _two_pending_increments() -> tuple[ps._PendingSpendIncrement, ...]:
+    return (
+        ps._PendingSpendIncrement(counter_key="spend:key:k", increment=1.5),
+        ps._PendingSpendIncrement(counter_key="spend:team:t", increment=1.5),
+    )
+
+
+@pytest.mark.asyncio
+async def test_apply_spend_counter_increments_open_breaker_invalidates_and_returns(monkeypatch):
+    """An open Redis circuit breaker is a known, already-logged state, not a per-request tracking failure.
+
+    Re-raising the refusal sent every request through the cost callback's error path, which
+    logged an ERROR and fired the failed-tracking alert once per request for the whole outage.
+    """
+    from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
+
+    fake_cache = _make_spend_counter_cache()
+    fake_cache.redis_cache.async_increment_pipeline = AsyncMock(
+        side_effect=RedisCircuitBreakerOpenError("Redis circuit breaker is open")
+    )
+    monkeypatch.setattr(ps, "spend_counter_cache", fake_cache)
+
+    await ps._apply_spend_counter_increments(_two_pending_increments())
+
+    deleted_keys = sorted(call.kwargs["key"] for call in fake_cache.in_memory_cache.delete_cache.call_args_list)
+    assert deleted_keys == ["spend:key:k", "spend:team:t"]
+    fake_cache.in_memory_cache.set_cache.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_apply_spend_counter_increments_other_redis_error_invalidates_and_raises(monkeypatch):
+    fake_cache = _make_spend_counter_cache()
+    fake_cache.redis_cache.async_increment_pipeline = AsyncMock(side_effect=ConnectionError("redis down"))
+    monkeypatch.setattr(ps, "spend_counter_cache", fake_cache)
+
+    with pytest.raises(ConnectionError, match="redis down"):
+        await ps._apply_spend_counter_increments(_two_pending_increments())
+
+    deleted_keys = sorted(call.kwargs["key"] for call in fake_cache.in_memory_cache.delete_cache.call_args_list)
+    assert deleted_keys == ["spend:key:k", "spend:team:t"]
+    fake_cache.in_memory_cache.set_cache.assert_not_called()
+
+
 # ---------------------------------------------------------------------------
 # _ensure_spend_counter_initialized
 # ---------------------------------------------------------------------------

tests/test_litellm/router_strategy/test_base_routing_strategy.py

@@ -1,4 +1,5 @@
 import json
+import logging
 from typing import Any, Dict, List, Optional, Set, Union
 
 import pytest
@@ -9,7 +10,7 @@ from unittest.mock import MagicMock, patch
 
 
 from litellm.caching.caching import DualCache
-from litellm.caching.redis_cache import RedisPipelineIncrementOperation
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError, RedisPipelineIncrementOperation
 from litellm.router_strategy.base_routing_strategy import BaseRoutingStrategy
 
 
@@ -146,3 +147,18 @@ async def test_cache_keys_management(base_strategy):
     # Test resetting cache keys
     base_strategy.reset_in_memory_keys_to_update()
     assert len(base_strategy.get_in_memory_keys_to_update()) == 0
+
+
+@pytest.mark.asyncio
+async def test_push_refused_by_the_open_circuit_breaker_is_not_logged_as_an_error(base_strategy, mock_dual_cache, caplog):
+    """The sync loop pushes every 100 ms under usage-based routing, so an open breaker must not add an error line per cycle."""
+    mock_dual_cache.redis_cache.async_increment_pipeline.side_effect = RedisCircuitBreakerOpenError(
+        "Redis circuit breaker is open - skipping async_increment_pipeline"
+    )
+    base_strategy.redis_increment_operation_queue = [{"key": "k", "increment_value": 1.0, "ttl": 60}]
+
+    with caplog.at_level(logging.ERROR):
+        await base_strategy._push_in_memory_increments_to_redis()
+
+    assert caplog.records == []
+    assert base_strategy.redis_increment_operation_queue == []

tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py

@@ -1,7 +1,13 @@
+import asyncio
+import gc
+import logging
+from unittest.mock import AsyncMock, MagicMock
+
 import pytest
 
 import litellm
 from litellm.caching.caching import DualCache
+from litellm.caching.redis_cache import RedisCache, RedisCircuitBreakerOpenError
 from litellm.router_strategy.budget_limiter import RouterBudgetLimiting
 from litellm.types.router import LiteLLM_Params
 from litellm.types.utils import BudgetConfig
@@ -303,3 +309,90 @@ def test_router_add_deployment_registers_deployment_budget(
     )
     assert config is not None
     assert config.max_budget == 0.000000000001
+
+
+@pytest.mark.asyncio
+async def test_sync_refused_by_the_open_circuit_breaker_is_quiet_and_leaks_no_task(disable_budget_sync, caplog):
+    """The budget sync runs every second, so an open breaker must not add an error line or an unretrieved task exception per cycle."""
+    refused = RedisCircuitBreakerOpenError("Redis circuit breaker is open - skipping async_increment_pipeline")
+    redis_cache = MagicMock(spec=RedisCache)
+    redis_cache.async_increment_pipeline = AsyncMock(side_effect=refused)
+    redis_cache.async_batch_get_cache = AsyncMock(side_effect=refused)
+    limiter = RouterBudgetLimiting(
+        dual_cache=DualCache(redis_cache=redis_cache),
+        provider_budget_config={"openai": BudgetConfig(max_budget=1.0, budget_duration="1d")},
+    )
+    await asyncio.gather(*(task for task in asyncio.all_tasks() if task is not asyncio.current_task()))
+    limiter.redis_increment_operation_queue = [{"key": "provider_spend:openai:1d", "increment_value": 0.5, "ttl": 60}]
+    loop = asyncio.get_running_loop()
+    unretrieved = MagicMock()
+    loop.set_exception_handler(unretrieved)
+
+    try:
+        with caplog.at_level(logging.ERROR):
+            await limiter._sync_in_memory_spend_with_redis()
+            await asyncio.sleep(0)
+            gc.collect()
+    finally:
+        loop.set_exception_handler(None)
+
+    assert caplog.records == []
+    unretrieved.assert_not_called()
+    assert limiter.redis_increment_operation_queue == []
+    assert redis_cache.async_increment_pipeline.await_count == 1
+
+
+async def _limiter_with_redis(redis_cache: MagicMock) -> RouterBudgetLimiting:
+    limiter = RouterBudgetLimiting(
+        dual_cache=DualCache(redis_cache=redis_cache),
+        provider_budget_config={"openai": BudgetConfig(max_budget=1.0, budget_duration="1d")},
+    )
+    await asyncio.gather(*(task for task in asyncio.all_tasks() if task is not asyncio.current_task()))
+    limiter.redis_increment_operation_queue = [{"key": "provider_spend:openai:1d", "increment_value": 0.5, "ttl": 60}]
+    return limiter
+
+
+@pytest.mark.asyncio
+async def test_push_returns_before_redis_answers(disable_budget_sync):
+    """The push runs inside the request success callback, so it must hand the Redis round trip to a task instead of waiting on it."""
+    redis_answered = asyncio.Event()
+
+    async def wait_for_redis(**_: object) -> None:
+        await redis_answered.wait()
+
+    redis_cache = MagicMock(spec=RedisCache)
+    redis_cache.async_increment_pipeline = AsyncMock(side_effect=wait_for_redis)
+    limiter = await _limiter_with_redis(redis_cache)
+
+    await asyncio.wait_for(limiter._push_in_memory_increments_to_redis(), timeout=1)
+    await asyncio.sleep(0)
+
+    assert not redis_answered.is_set()
+    assert redis_cache.async_increment_pipeline.await_count == 1
+    assert limiter.redis_increment_operation_queue == []
+    redis_answered.set()
+    await asyncio.gather(*(task for task in asyncio.all_tasks() if task is not asyncio.current_task()))
+
+
+@pytest.mark.asyncio
+async def test_push_task_failure_is_logged_once_and_not_leaked(disable_budget_sync, caplog):
+    """A real Redis failure on the background push must surface as one error line, never as an unretrieved task exception."""
+    redis_cache = MagicMock(spec=RedisCache)
+    redis_cache.async_increment_pipeline = AsyncMock(side_effect=ConnectionError("Error 61 connecting to 127.0.0.1:6379"))
+    limiter = await _limiter_with_redis(redis_cache)
+    loop = asyncio.get_running_loop()
+    unretrieved = MagicMock()
+    loop.set_exception_handler(unretrieved)
+
+    try:
+        with caplog.at_level(logging.ERROR):
+            await limiter._push_in_memory_increments_to_redis()
+            await asyncio.sleep(0)
+            gc.collect()
+    finally:
+        loop.set_exception_handler(None)
+
+    assert [record.getMessage() for record in caplog.records] == [
+        "Error syncing in-memory cache with Redis: Error 61 connecting to 127.0.0.1:6379"
+    ]
+    unretrieved.assert_not_called()

tests/test_litellm/router_utils/test_health_state_cache.py

@@ -7,6 +7,7 @@ import time
 import pytest
 
 from litellm.caching.caching import DualCache
+from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
 from litellm.router_utils.health_state_cache import DeploymentHealthCache
 
 
@@ -145,8 +146,11 @@ class _SharedRedisFake:
     def __init__(self):
         self.store = {}
         self.fail_get = False
+        self.breaker_open = False
 
     def get_cache(self, key, parent_otel_span=None, **kwargs):
+        if self.breaker_open:
+            raise RedisCircuitBreakerOpenError("Redis circuit breaker is open - skipping get_cache")
         if self.fail_get:
             return None  # RedisCache.get_cache swallows connection errors and returns None
         return self.store.get(key)
@@ -192,3 +196,26 @@ def test_failed_redis_read_falls_back_to_local_copy():
         {"prod-bad": {"is_healthy": False, "timestamp": time.time(), "reason": "check_failed"}}
     )
     assert set(redis_fake.store[DeploymentHealthCache.CACHE_KEY]) == {"prod-bad", "internal-bad"}
+
+
+def test_open_circuit_breaker_read_still_merges_into_local_copy(caplog):
+    """A read refused by the open breaker is a miss, so the merge and local write still happen quietly."""
+    redis_fake = _SharedRedisFake()
+    pod_a = DeploymentHealthCache(cache=DualCache(redis_cache=redis_fake), staleness_threshold=60.0)
+    pod_b = DeploymentHealthCache(cache=DualCache(redis_cache=redis_fake), staleness_threshold=60.0)
+    pod_a.set_deployment_health_states(
+        {"prod-bad": {"is_healthy": False, "timestamp": time.time(), "reason": "check_failed"}}
+    )
+    pod_b.set_deployment_health_states(
+        {"internal-bad": {"is_healthy": False, "timestamp": time.time(), "reason": "timeout"}}
+    )
+    pod_a.set_deployment_health_states(
+        {"prod-bad": {"is_healthy": False, "timestamp": time.time(), "reason": "check_failed"}}
+    )
+    redis_fake.breaker_open = True
+    with caplog.at_level("ERROR"):
+        pod_a.set_deployment_health_states(
+            {"prod-new-bad": {"is_healthy": False, "timestamp": time.time(), "reason": "check_failed"}}
+        )
+    assert caplog.records == []
+    assert pod_a.get_unhealthy_deployment_ids() == {"prod-bad", "internal-bad", "prod-new-bad"}