fix(caching): let the outer breaker guard judge failures a nested guarded call swallowed

batch_cache_write flushes through async_set_cache_pipeline, both guarded. The inner
guard consumed the swallowed pipeline failure and the outer then recorded a success,
so a dead Redis never tripped the breaker on the batch write path. Only the outermost
admission now reports, and the sync batch read runs under the same guard.

Also covers the sync add_cache and embedding pipeline wrappers, the increment pipeline,
sadd, and the stale raise path in tests.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/caching/redis_cache.py

@@ -288,6 +288,7 @@ _RedisCallResult = TypeVar("_RedisCallResult")
 _swallowed_redis_failures: Final[ContextVar[tuple[bool, ...]]] = ContextVar(
     "litellm_swallowed_redis_failures", default=()
 )
+_breaker_depth: Final[ContextVar[int]] = ContextVar("litellm_redis_breaker_depth", default=0)
 
 
 def _opaque_kwarg_key(value: object) -> str:
@@ -425,13 +426,20 @@ def _record_swallowed_redis_failure(exc: BaseException) -> None:
 class _BreakerAdmission:
     swallowed_before: int
     generation: int
+    nested: bool
 
 
 def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> _BreakerAdmission:
     """Reject the call if the breaker is open, else snapshot what its outcome will be judged against."""
     if breaker.is_open():
         raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open, skipping {name}")
-    return _BreakerAdmission(swallowed_before=len(_swallowed_redis_failures.get()), generation=breaker.generation)
+    depth: Final = _breaker_depth.get()
+    _breaker_depth.set(depth + 1)
+    return _BreakerAdmission(
+        swallowed_before=len(_swallowed_redis_failures.get()),
+        generation=breaker.generation,
+        nested=depth > 0,
+    )
 
 
 def _take_swallowed_failures(admission: _BreakerAdmission) -> tuple[bool, ...]:
@@ -448,7 +456,14 @@ def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmis
     method that returned is not on its own proof of a healthy Redis. A call admitted
     before the breaker opened reports nothing: its failures would refresh the open
     timer or knock out the recovery probe, and its success would close it early.
+
+    A guarded method calling another guarded method is one Redis interaction, so only
+    the outermost admission reports. The inner one leaves its swallowed failures in the
+    context for the outer to judge, otherwise the outer would read a clean context and
+    reset the streak the inner just fed.
     """
+    if admission.nested:
+        return
     swallowed: Final = _take_swallowed_failures(admission)
     if admission.generation != breaker.generation:
         return
@@ -459,7 +474,13 @@ def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmis
         breaker.record_failure(is_timeout=is_timeout)
 
 
+def _leave_circuit_breaker() -> None:
+    _breaker_depth.set(_breaker_depth.get() - 1)
+
+
 def _fail_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission, exc: BaseException) -> None:
+    if admission.nested:
+        return
     swallowed: Final = _take_swallowed_failures(admission)
     if admission.generation != breaker.generation:
         return
@@ -485,6 +506,8 @@ async def _run_under_circuit_breaker(
     except Exception as e:
         _fail_circuit_breaker(breaker, admission, e)
         raise
+    finally:
+        _leave_circuit_breaker()
     _exit_circuit_breaker(breaker, admission)
     return result
 
@@ -501,6 +524,8 @@ def _run_under_circuit_breaker_sync(
     except Exception as e:
         _fail_circuit_breaker(breaker, admission, e)
         raise
+    finally:
+        _leave_circuit_breaker()
     _exit_circuit_breaker(breaker, admission)
     return result
 
@@ -1441,12 +1466,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:
             _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, admission)
+            results: Final = _run_under_circuit_breaker_sync(
+                self._circuit_breaker, "batch_get_cache", lambda: self._run_redis_mget_operation(keys=_keys)
+            )
             end_time: Final = time.time()
             _duration: Final = end_time - start_time
             self.service_logger_obj.service_success_hook(
@@ -1470,6 +1495,8 @@ class RedisCache(BaseCache):
                 decoded_results[k] = v
 
             return decoded_results
+        except RedisCircuitBreakerOpenError:
+            raise
         except Exception as e:
             failed_at: Final = time.time()
             self.service_logger_obj.service_failure_hook(
@@ -1482,7 +1509,6 @@ class RedisCache(BaseCache):
                 parent_otel_span=parent_otel_span,
             )
             verbose_logger.error("Error occurred in batch get cache - %s", e)
-            _fail_circuit_breaker(self._circuit_breaker, admission, e)
             return key_value_dict
 
     @_redis_circuit_breaker_guard

tests/test_litellm/caching/test_caching.py

@@ -255,26 +255,41 @@ def test_exact_cache_key_includes_anthropic_messages_params(anthropic_param):
 
 
 @pytest.mark.asyncio
-async def test_async_add_cache_treats_an_open_breaker_as_a_quiet_skip(caplog):
-    """The top-level write wrapper logs a full ERROR traceback for any failure. An open Redis
-    breaker fails every write instantly, so under load that wrapper alone was hundreds of
+@pytest.mark.parametrize("write", ["add_cache", "async_add_cache", "async_add_cache_pipeline"])
+async def test_add_cache_wrappers_treat_an_open_breaker_as_a_quiet_skip(caplog, write: str):
+    """The top-level write wrappers log a full ERROR traceback for any failure. An open Redis
+    breaker fails every write instantly, so under load those wrappers alone were hundreds of
     stack formats per second per replica. Unexpected failures must still get the traceback.
     """
-    from unittest.mock import AsyncMock
+    from unittest.mock import AsyncMock, MagicMock
 
     from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
+    from litellm.types.utils import Embedding, EmbeddingResponse
 
     caplog.set_level(logging.DEBUG, logger="LiteLLM")
     cache = Cache(type=LiteLLMCacheType.LOCAL)
-    cache.cache.async_set_cache = AsyncMock(side_effect=RedisCircuitBreakerOpenError("open"))
+    embedding = EmbeddingResponse(model="text-embedding-3-small", data=[Embedding(embedding=[0.1], index=0, object="embedding")])
 
-    await cache.async_add_cache("result", model="gpt-5.4-nano", messages=[{"role": "user", "content": "hi"}])
+    async def run_write() -> None:
+        if write == "add_cache":
+            cache.add_cache("result", model="gpt-5.4-nano", messages=[{"role": "user", "content": "hi"}])
+        elif write == "async_add_cache":
+            await cache.async_add_cache("result", model="gpt-5.4-nano", messages=[{"role": "user", "content": "hi"}])
+        else:
+            await cache.async_add_cache_pipeline(embedding, model="text-embedding-3-small", input="hi")
+
+    def install_backend(exc: Exception) -> None:
+        cache.cache.set_cache = MagicMock(side_effect=exc)
+        cache.cache.async_set_cache = AsyncMock(side_effect=exc)
+        cache.cache.async_set_cache_pipeline = AsyncMock(side_effect=exc)
+
+    install_backend(RedisCircuitBreakerOpenError("open"))
+    await run_write()
 
     assert [r.levelno for r in caplog.records if "add_cache" in r.getMessage()] == [logging.DEBUG]
-    cache.cache.async_set_cache.assert_awaited_once()
 
-    cache.cache.async_set_cache = AsyncMock(side_effect=OSError("disk full"))
-    await cache.async_add_cache("result", model="gpt-5.4-nano", messages=[{"role": "user", "content": "hi"}])
+    install_backend(OSError("disk full"))
+    await run_write()
 
     errors = [r for r in caplog.records if r.levelno == logging.ERROR]
     assert len(errors) == 1 and errors[0].exc_info is not None

tests/test_litellm/caching/test_dual_cache.py

@@ -621,6 +621,11 @@ def dual_cache_with_open_breaker():
         pytest.param(lambda c: c.async_set_cache("lit7468", "v"), lambda n: None, id="async_set_cache"),
         pytest.param(lambda c: c.async_set_cache_pipeline([("lit7468", "v")]), lambda n: None, id="async_set_cache_pipeline"),
         pytest.param(lambda c: c.async_increment_cache("lit7468", 1.0, ttl=60), float, id="async_increment_cache"),
+        pytest.param(
+            lambda c: c.async_increment_cache_pipeline([{"key": "lit7468", "increment_value": 1.0, "ttl": 60}]),
+            lambda n: [float(n)],
+            id="async_increment_cache_pipeline",
+        ),
     ],
 )
 async def test_open_breaker_is_a_quiet_cache_miss(dual_cache_with_open_breaker, call, expected, caplog):

tests/test_litellm/caching/test_redis_cache.py

@@ -494,6 +494,7 @@ def _closed_port() -> int:
         pytest.param(lambda c: c.async_batch_get_cache(["lit4930"]), id="async_batch_get_cache"),
         pytest.param(lambda c: c.async_set_cache("lit4930", "v"), id="async_set_cache"),
         pytest.param(lambda c: c.async_get_ttl("lit4930"), id="async_get_ttl"),
+        pytest.param(lambda c: c.async_set_cache_sadd("lit4930", ["v"], ttl=None), id="async_set_cache_sadd"),
     ],
 )
 async def test_circuit_breaker_opens_when_method_swallows_redis_failure(call_method):
@@ -505,6 +506,7 @@ async def test_circuit_breaker_opens_when_method_swallows_redis_failure(call_met
     breaker could never open. An unreachable Redis then stayed in the pool and every
     request kept paying the full socket timeout on it.
     """
+    from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
     from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
 
     cache = await asyncio.to_thread(RedisCache, host="127.0.0.1", port=_closed_port(), socket_timeout=0.5)
@@ -512,10 +514,73 @@ async def test_circuit_breaker_opens_when_method_swallows_redis_failure(call_met
     for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
         await call_method(cache)
 
-    with pytest.raises(Exception, match="circuit breaker is open"):
+    with pytest.raises(RedisCircuitBreakerOpenError):
         await call_method(cache)
 
 
+@pytest.mark.asyncio
+async def test_nested_guarded_flush_failures_still_open_the_breaker():
+    """A guarded method that delegates to another guarded method is one Redis call, not two.
+
+    batch_cache_write is guarded and flushes through the guarded async_set_cache_pipeline,
+    which swallows the pipeline error. If the inner guard consumes that failure, the outer
+    guard sees a clean run and records a success, so the streak resets on every write and
+    a dead Redis keeps receiving flushes instead of tripping the breaker.
+    """
+    from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
+    from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
+
+    cache = await asyncio.to_thread(
+        RedisCache, host="127.0.0.1", port=_closed_port(), socket_timeout=0.5, redis_flush_size=1
+    )
+
+    for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD):
+        await cache.batch_cache_write("lit7468", "v")
+
+    with pytest.raises(RedisCircuitBreakerOpenError):
+        await cache.batch_cache_write("lit7468", "v")
+    assert cache._circuit_breaker._failure_count == REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD
+
+
+@pytest.mark.asyncio
+async def test_nested_guard_that_raises_counts_one_failure_for_the_outer_call():
+    """An inner guarded call that raises through the outer one is a single Redis failure, and a
+    swallowed inner failure followed by an outer raise is two, so the count follows what Redis
+    actually refused rather than how many guard frames the error crossed.
+    """
+    from redis.exceptions import ConnectionError as RedisConnectionError
+
+    from litellm.caching.redis_cache import (
+        RedisCircuitBreaker,
+        _record_swallowed_redis_failure,
+        _run_under_circuit_breaker,
+    )
+
+    breaker = RedisCircuitBreaker(failure_threshold=10, recovery_timeout=60)
+
+    async def refused():
+        raise RedisConnectionError("refused")
+
+    async def outer_delegating_to_inner():
+        return await _run_under_circuit_breaker(breaker, "inner", refused)
+
+    async def outer_swallowing_then_raising():
+        async def inner_swallowing():
+            _record_swallowed_redis_failure(RedisConnectionError("refused"))
+            return {}
+
+        await _run_under_circuit_breaker(breaker, "inner", inner_swallowing)
+        raise RedisConnectionError("refused")
+
+    with pytest.raises(RedisConnectionError):
+        await _run_under_circuit_breaker(breaker, "outer", outer_delegating_to_inner)
+    assert breaker._failure_count == 1
+
+    with pytest.raises(RedisConnectionError):
+        await _run_under_circuit_breaker(breaker, "outer", outer_swallowing_then_raising)
+    assert breaker._failure_count == 3
+
+
 def test_sync_batch_get_cache_swallowed_failures_open_the_breaker_and_then_fast_fail(sync_batch_redis_cache):
     """The sync batch read hides its Redis error behind an empty dict, but the breaker must still
     count it, and once open the read has to raise the typed error like its async twin so DualCache
@@ -1106,12 +1171,13 @@ async def test_success_admitted_before_the_breaker_opened_cannot_close_it():
 
 
 @pytest.mark.asyncio
-async def test_swallowed_failure_admitted_before_the_breaker_opened_cannot_delay_recovery():
-    """A stale in-flight call that swallows its Redis error must not restart the open timer.
+@pytest.mark.parametrize("stale_call_raises", [False, True], ids=["swallows", "raises"])
+async def test_failure_admitted_before_the_breaker_opened_cannot_delay_recovery(stale_call_raises: bool):
+    """A stale in-flight call that fails after the breaker opened must not restart the open timer.
 
-    Guarded methods that return a default instead of raising still report their failure to
-    the breaker on exit. If that report landed against a newer breaker generation, every
-    slow call that was already dialing Redis when the breaker opened would push recovery
+    Whether the method swallows its Redis error and returns a default or lets it propagate,
+    the failure is reported on exit. If that report landed against a newer breaker generation,
+    every slow call that was already dialing Redis when the breaker opened would push recovery
     back by its own socket timeout, and knock out the HALF_OPEN probe if it landed then.
     """
     from redis.exceptions import ConnectionError as RedisConnectionError
@@ -1125,8 +1191,10 @@ async def test_swallowed_failure_admitted_before_the_breaker_opened_cannot_delay
     breaker = RedisCircuitBreaker(failure_threshold=2, recovery_timeout=0.05)
     release_stale_call = asyncio.Event()
 
-    async def stale_call_that_swallows_its_failure():
+    async def stale_call():
         await release_stale_call.wait()
+        if stale_call_raises:
+            raise RedisConnectionError("refused")
         _record_swallowed_redis_failure(RedisConnectionError("refused"))
         return {}
 
@@ -1136,7 +1204,7 @@ async def test_swallowed_failure_admitted_before_the_breaker_opened_cannot_delay
     async def recovered():
         return "ok"
 
-    in_flight = asyncio.create_task(_run_under_circuit_breaker(breaker, "stale", stale_call_that_swallows_its_failure))
+    in_flight = asyncio.create_task(_run_under_circuit_breaker(breaker, "stale", stale_call))
     await asyncio.sleep(0)
     for _ in range(breaker.failure_threshold):
         with pytest.raises(RedisConnectionError):
@@ -1145,7 +1213,11 @@ async def test_swallowed_failure_admitted_before_the_breaker_opened_cannot_delay
 
     await asyncio.sleep(0.04)
     release_stale_call.set()
-    assert await in_flight == {}
+    if stale_call_raises:
+        with pytest.raises(RedisConnectionError):
+            await in_flight
+    else:
+        assert await in_flight == {}
     await asyncio.sleep(0.03)
 
     assert await _run_under_circuit_breaker(breaker, "probe", recovered) == "ok"