From e9c388d79917cd4018a568181f31969ab223cf4e Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 17:49:20 -0700 Subject: [PATCH] fix(caching): keep an open Redis circuit breaker quiet on the sync read and spend counter paths The sync get path was unguarded, logged with a stray format argument, and never fed the breaker. The sync batch read swallowed the breaker's refusal as an ERROR plus a service failure event per call, so DualCache dropped its in-memory hits and left batch reservations behind. record_success closed an OPEN breaker on stale in-flight successes, skipping the recovery timeout and the half-open probe. The spend counter pipeline re-raised the refusal into the cost callback, which logged an ERROR and fired the failed-tracking alert per request. --- litellm/caching/dual_cache.py | 12 ++- litellm/caching/redis_cache.py | 9 +- litellm/proxy/management_endpoints/ui_sso.py | 18 ++-- litellm/proxy/proxy_server.py | 5 +- tests/test_litellm/caching/test_dual_cache.py | 27 ++++++ .../test_litellm/caching/test_redis_cache.py | 83 +++++++++++++++++-- .../proxy/management_endpoints/test_ui_sso.py | 22 +++++ .../proxy/proxy_server/test_spend_counters.py | 48 +++++++++++ 8 files changed, 205 insertions(+), 19 deletions(-) diff --git a/litellm/caching/dual_cache.py b/litellm/caching/dual_cache.py index baabfad6852..be761e1258b 100644 --- a/litellm/caching/dual_cache.py +++ b/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 diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 6b93529e456..5b1a9739eb8 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -250,7 +250,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") @@ -1337,6 +1337,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 +1357,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,9 +1395,9 @@ class RedisCache(BaseCache): key_value_dict = {} _key_list: Final = [key for key in key_list if key is not None] start_time: Final = time.time() + swallowed_before: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache") 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) diff --git a/litellm/proxy/management_endpoints/ui_sso.py b/litellm/proxy/management_endpoints/ui_sso.py index c60888e298f..1ba90725eff 100644 --- a/litellm/proxy/management_endpoints/ui_sso.py +++ b/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)) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 0f24acb8bb4..94e74b20297 100644 --- a/litellm/proxy/proxy_server.py +++ b/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) diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index eae8bbfdaff..4c9068722b8 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/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 diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index bcaa58c9c40..c6f0caa1dca 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/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,40 @@ 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 diff --git a/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py b/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py index 8d8bc15f9be..2050e65d2a1 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_ui_sso.py +++ b/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 diff --git a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py index c343652efd9..2f47736a398 100644 --- a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py +++ b/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 # ---------------------------------------------------------------------------