Merge pull request #40624 from BerriAI/litellm_redis_breaker_open_silent_miss

fix(caching): keep an open Redis circuit breaker open and quiet on the sync read and spend counter paths

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
 
@@ -196,8 +197,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:
@@ -250,7 +257,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")
@@ -266,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")
@@ -405,21 +413,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(
@@ -432,14 +452,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 +469,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
 
 
@@ -1337,6 +1357,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)
@@ -1356,8 +1377,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]:
         """
@@ -1394,12 +1415,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

@@ -47,6 +47,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,
@@ -336,6 +337,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(
@@ -348,12 +359,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

@@ -251,6 +251,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,
@@ -3412,8 +3413,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")
 
 
 def test_batch_get_counts_raises_where_batch_get_cache_reports_a_miss(sync_batch_redis_cache):
@@ -661,7 +694,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():
@@ -1010,6 +1044,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
@@ -1030,3 +1066,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"}

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -35064,7 +35064,7 @@ export interface components {
             classification_prompt?: string | null;
             /**
              * Classifier Context Budget Chars
-             * @description Maximum characters of prior-turn text quoted to the LLM classifier, across the whole context window, per classification call. Turns are taken newest first and quoted whole while they fit, so a conversation small enough to quote entirely is never cut; once the budget runs out the older turns are dropped whole and only the turn straddling the boundary is truncated, into whatever space is left. The current ask and the caller's system prompt sit outside this budget and are always sent in full, as does the numbering each quoted turn carries. A budget under 120 leaves no room to quote a turn and suppresses the block; set classifier_context_window_size to 0 to turn context off deliberately. Only applies when classifier_type is 'llm'.
+             * @description Maximum characters of prior-turn text quoted to the LLM classifier, across the whole context window, per classification call. Turns are taken newest first and quoted whole while they fit, so a conversation small enough to quote entirely is never cut; once the budget runs out the older turns are dropped whole and only the turn straddling the boundary is truncated, into whatever space is left. The current ask and, except for Claude Code requests, the extracted system-role text sit outside this budget and are sent in full, as does the numbering each quoted turn carries. A budget under 120 leaves no room to quote a turn and suppresses the block; set classifier_context_window_size to 0 to turn context off deliberately. Only applies when classifier_type is 'llm'.
              * @default 8000
              */
             classifier_context_budget_chars: number;
@@ -35081,7 +35081,7 @@ export interface components {
             classifier_context_per_turn_chars?: number | null;
             /**
              * Classifier Context Window Size
-             * @description Number of prior user turns (tool output and harness reminders excluded) to include as context in the LLM classifier prompt, so a follow-up like 'now do the same for the streaming path' is classified against what it refers to. Counts turns of both roles when classifier_context_include_assistant_turns is enabled. These turns are sent to the classifier model, which may be a different deployment or provider than the routed completion model; that call already carries the current user ask and the caller's system prompt in full. Set to 0 to send neither prior turns nor any conversation context beyond the current ask. Only applies when classifier_type is 'llm'.
+             * @description Number of prior user turns (tool output and harness reminders excluded) to include as context in the LLM classifier prompt, so a follow-up like 'now do the same for the streaming path' is classified against what it refers to. Counts turns of both roles when classifier_context_include_assistant_turns is enabled. These turns are sent to the classifier model, which may be a different deployment or provider than the routed completion model; that call carries the current user ask and, except for Claude Code requests, the extracted system-role text in full. Claude Code system text is omitted to avoid classifying harness instructions; the routed completion still receives it. Set to 0 to send neither prior turns nor any conversation context beyond the current ask. Only applies when classifier_type is 'llm'.
              * @default 3
              */
             classifier_context_window_size: number;