From 32b7daf69116907678c03f4d78f713080db52317 Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 20:58:41 +0000 Subject: [PATCH 01/14] fix(caching): make an open Redis circuit breaker a quiet cache miss An open breaker raised a generic Exception on every skipped call, and DualCache caught it and logged a full ERROR traceback each time. Under load that became hundreds of traceback formats per second on every replica and pinned the proxies at 100% CPU. Raise a typed RedisCircuitBreakerOpenError instead and have DualCache return its in-memory result without logging for it. Classify redis-py pool exhaustion (ConnectionError chained from TimeoutError) as a timeout so a latency blip goes through the duration gate. Track a breaker generation so a call admitted before the breaker opened cannot close it, leaving that to the HALF_OPEN probe. Rate limit the LoggingWorker callback-error traceback to one per interval so a stalled logging backend cannot start a second traceback storm. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/dual_cache.py | 15 +- litellm/caching/redis_cache.py | 76 ++++++++--- litellm/constants.py | 1 + litellm/litellm_core_utils/logging_worker.py | 23 +++- tests/test_litellm/caching/test_dual_cache.py | 62 +++++++++ .../test_litellm/caching/test_redis_cache.py | 128 ++++++++++++++++++ .../litellm_core_utils/test_logging_worker.py | 45 ++++++ 7 files changed, 326 insertions(+), 24 deletions(-) diff --git a/litellm/caching/dual_cache.py b/litellm/caching/dual_cache.py index ec17cc1d809..98c6da0f02f 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 +from .redis_cache import RedisCache, RedisCircuitBreakerOpenError if TYPE_CHECKING: from opentelemetry.trace import Span as _Span @@ -250,6 +250,8 @@ class DualCache(BaseCache): print_verbose(f"get cache: cache result: {result}") return result + except RedisCircuitBreakerOpenError: + return None except Exception: verbose_logger.error(traceback.format_exc()) @@ -319,6 +321,9 @@ class DualCache(BaseCache): redis_result: Final = await self.redis_cache.async_batch_get_cache( sublist_keys, parent_otel_span=parent_otel_span ) + except RedisCircuitBreakerOpenError: + self._rollback_redis_batch_key_reservations(previous_access_times) + return result except Exception: # Do not throttle subsequent callers if the Redis read fails. self._rollback_redis_batch_key_reservations(previous_access_times) @@ -352,6 +357,8 @@ class DualCache(BaseCache): if self.redis_cache is not None and local_only is False: await self.redis_cache.async_set_cache(key, value, **kwargs) + except RedisCircuitBreakerOpenError: + return except Exception as e: verbose_logger.exception("LiteLLM Cache: Excepton async add_cache: %s", e) @@ -371,6 +378,8 @@ class DualCache(BaseCache): await self.redis_cache.async_set_cache_pipeline( cache_list=cache_list, ttl=kwargs.pop("ttl", None), **kwargs ) + except RedisCircuitBreakerOpenError: + return except Exception as e: verbose_logger.exception("LiteLLM Cache: Excepton async add_cache: %s", e) @@ -408,6 +417,8 @@ class DualCache(BaseCache): refresh_ttl=refresh_ttl, ) + return result + except RedisCircuitBreakerOpenError: return result except Exception as e: verbose_logger.warning( @@ -437,6 +448,8 @@ class DualCache(BaseCache): parent_otel_span=parent_otel_span, ) + return result + except RedisCircuitBreakerOpenError: return result except Exception as e: verbose_logger.warning( diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 106c1580110..de4dcb5aa27 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -17,6 +17,7 @@ import json 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 @@ -148,6 +149,10 @@ def _get_call_stack_info(num_frames: int = 2) -> str: return "unknown" +class RedisCircuitBreakerOpenError(Exception): + """Expected fast-fail while the breaker is open; optional-cache callers treat it as a miss.""" + + class RedisCircuitBreaker: """ Tracks Redis health for a RedisCache instance. @@ -163,8 +168,12 @@ class RedisCircuitBreaker: (no success or hard failure in between) that reaches failure_threshold and spans timeout_min_duration seconds OPEN -> HALF_OPEN after recovery_timeout seconds - HALF_OPEN -> CLOSED on success - HALF_OPEN -> OPEN on failure (resets timer) + HALF_OPEN -> CLOSED on the recovery probe's success + HALF_OPEN -> OPEN on the recovery probe's failure (resets timer) + + Every OPEN transition starts a new generation. A call reports its outcome only for the + generation it was admitted under, so a success from a call that was already in flight + when the breaker opened cannot close it and the HALF_OPEN probe is the only call that can. Timeouts are accounted separately from hard connectivity failures because the async Redis timeout includes time waiting for the worker event loop to resume: one loop @@ -194,9 +203,14 @@ class RedisCircuitBreaker: self._timeout_count = 0 self._timeout_streak_started_at: float | None = None self._opened_at: float | None = None + self._generation = 0 self._state = self.CLOSED _breaker_metrics().record_state_change(None, self._state) + @property + def generation(self) -> int: + return self._generation + def is_open(self) -> bool: """Returns True if Redis calls should be skipped.""" if not self.enabled: @@ -241,18 +255,19 @@ class RedisCircuitBreaker: if self._state != self.OPEN: verbose_logger.warning( "Redis circuit breaker OPENED after %d consecutive failures" - " (%d hard connectivity) — fast-failing Redis calls for %ds", + " (%d hard connectivity), fast-failing Redis calls for %ds", self._failure_count, self._hard_failure_count, self.recovery_timeout, ) + self._generation += 1 self._set_state(self.OPEN) def record_success(self) -> None: if not self.enabled: return if self._state == self.HALF_OPEN: - verbose_logger.info("Redis circuit breaker CLOSED — Redis recovered") + verbose_logger.info("Redis circuit breaker CLOSED, Redis recovered") self._failure_count = 0 self._hard_failure_count = 0 self._timeout_count = 0 @@ -320,7 +335,10 @@ def _redis_timeout_error_types() -> tuple[type, ...]: def _is_redis_timeout_failure(exc: BaseException) -> bool: - return isinstance(exc, _redis_timeout_error_types()) + """Follows __cause__: a blocking pool wait raises ConnectionError from asyncio.TimeoutError.""" + if isinstance(exc, _redis_timeout_error_types()): + return True + return exc.__cause__ is not None and _is_redis_timeout_failure(exc.__cause__) class _BreakerMetrics: @@ -391,23 +409,37 @@ def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseExcep _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1) -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 snapshot what its outcome will be judged against.""" if breaker.is_open(): - raise Exception(f"Redis circuit breaker is open — skipping {name}") - return _swallowed_redis_failures.get() + raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open, skipping {name}") + 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 opened 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. """ - if _swallowed_redis_failures.get() == swallowed_before: + if admission.generation != breaker.generation: + return + if _swallowed_redis_failures.get() == admission.swallowed_before: breaker.record_success() +def _fail_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission, exc: BaseException) -> None: + if admission.generation != breaker.generation or not _is_redis_health_failure(exc): + return + breaker.record_failure(is_timeout=_is_redis_timeout_failure(exc)) + + async def _run_under_circuit_breaker( breaker: RedisCircuitBreaker, name: str, @@ -418,14 +450,13 @@ 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)) + _fail_circuit_breaker(breaker, admission, e) raise - _exit_circuit_breaker(breaker, swallowed_before) + _exit_circuit_breaker(breaker, admission) return result @@ -435,14 +466,13 @@ 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() + _fail_circuit_breaker(breaker, admission, e) raise - _exit_circuit_breaker(breaker, swallowed_before) + _exit_circuit_breaker(breaker, admission) return result @@ -1382,10 +1412,10 @@ class RedisCache(BaseCache): start_time: Final = time.time() try: - swallowed_before: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache") + admission: 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( @@ -1409,6 +1439,8 @@ class RedisCache(BaseCache): decoded_results[k] = v return decoded_results + except RedisCircuitBreakerOpenError: + return key_value_dict except Exception as e: failed_at: Final = time.time() self.service_logger_obj.service_failure_hook( diff --git a/litellm/constants.py b/litellm/constants.py index 0a7ef363e92..421419fecc0 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -571,6 +571,7 @@ ANTHROPIC_MESSAGES_MAX_DETACHED_STREAM_DRAINS: Final = int( LOGGING_WORKER_CONCURRENCY: Final = int(os.getenv("LOGGING_WORKER_CONCURRENCY", 100)) # Must be above 0 LOGGING_WORKER_MAX_QUEUE_SIZE: Final = int(os.getenv("LOGGING_WORKER_MAX_QUEUE_SIZE", 50_000)) LOGGING_WORKER_MAX_TIME_PER_COROUTINE: Final = float(os.getenv("LOGGING_WORKER_MAX_TIME_PER_COROUTINE", 20.0)) +LOGGING_WORKER_ERROR_TRACEBACK_INTERVAL_SECONDS: Final = 60.0 LOGGING_WORKER_CLEAR_PERCENTAGE: Final = int( os.getenv("LOGGING_WORKER_CLEAR_PERCENTAGE", 50) ) # Percentage of queue to clear (default: 50%) diff --git a/litellm/litellm_core_utils/logging_worker.py b/litellm/litellm_core_utils/logging_worker.py index 1d74595781a..dfaace1bb10 100644 --- a/litellm/litellm_core_utils/logging_worker.py +++ b/litellm/litellm_core_utils/logging_worker.py @@ -6,6 +6,7 @@ import atexit import contextvars import inspect import logging +import time from collections.abc import Coroutine, Iterator from typing import Final @@ -16,6 +17,7 @@ from litellm.constants import ( LOGGING_WORKER_AGGRESSIVE_CLEAR_COOLDOWN_SECONDS, LOGGING_WORKER_CLEAR_PERCENTAGE, LOGGING_WORKER_CONCURRENCY, + LOGGING_WORKER_ERROR_TRACEBACK_INTERVAL_SECONDS, LOGGING_WORKER_MAX_QUEUE_SIZE, LOGGING_WORKER_MAX_TIME_PER_COROUTINE, MAX_ITERATIONS_TO_CLEAR_QUEUE, @@ -47,10 +49,14 @@ class LoggingWorker: timeout: float = LOGGING_WORKER_MAX_TIME_PER_COROUTINE, max_queue_size: int = LOGGING_WORKER_MAX_QUEUE_SIZE, concurrency: int = LOGGING_WORKER_CONCURRENCY, + error_traceback_interval: float = LOGGING_WORKER_ERROR_TRACEBACK_INTERVAL_SECONDS, ): self.timeout = timeout self.max_queue_size = max_queue_size self.concurrency = concurrency + self.error_traceback_interval = error_traceback_interval + self._last_error_traceback_at: float | None = None + self._errors_since_traceback: int = 0 self._queue: asyncio.Queue[LoggingTask] | None = None self._worker_task: asyncio.Task | None = None self._running_tasks: set[asyncio.Task] = set() @@ -163,7 +169,7 @@ class LoggingWorker: timeout=self.timeout, ) except Exception as e: - verbose_logger.exception("LoggingWorker error: %s", e) + self._log_task_error(e) finally: self._untrack_dequeued(task) self._queue.task_done() @@ -171,6 +177,21 @@ class LoggingWorker: # Always release semaphore, even if queue is None sem.release() + def _log_task_error(self, error: Exception) -> None: + """One traceback per interval: a stalled backend fails every in-flight task at once.""" + now: Final = time.monotonic() + last_traceback_at: Final = self._last_error_traceback_at + if last_traceback_at is not None and now - last_traceback_at < self.error_traceback_interval: + self._errors_since_traceback += 1 + return + verbose_logger.exception( + "LoggingWorker error (%d more suppressed since the last traceback): %r", + self._errors_since_traceback, + error, + ) + self._last_error_traceback_at = now + self._errors_since_traceback = 0 + async def _worker_loop(self) -> None: """Main worker loop that gets tasks and schedules them to run concurrently.""" try: diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index ded3be26630..46dd2687e48 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/tests/test_litellm/caching/test_dual_cache.py @@ -1,4 +1,5 @@ import asyncio +import logging import time import uuid from unittest.mock import AsyncMock, MagicMock, patch @@ -576,3 +577,64 @@ async def test_dual_cache_late_attach_redis_wires_writes_and_ttl_async(): assert mock_redis.async_set_cache.call_args[0][:2] == (key_after, val_after) assert in_memory.get_cache(key_after) == val_after + + +@pytest.fixture +def dual_cache_with_open_breaker(): + """A DualCache whose Redis tier is behind an already-open circuit breaker. + + The Redis client is a mock that fails the test if anything reaches it, so every + guarded call has to be short-circuited by the breaker. + """ + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.caching.redis_cache import _is_redis_timeout_failure + from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD + + with ( + patch("asyncio.get_running_loop", side_effect=RuntimeError("No running event loop")), + patch( # test-quality-ok: RedisCache.__init__ builds its client eagerly, with no injection point + "litellm._redis.get_redis_client", return_value=MagicMock() + ), + ): + redis_cache = RedisCache(host="127.0.0.1", port=6379) + unreachable = AsyncMock() + unreachable.get.side_effect = AssertionError("an open breaker must not touch Redis") + unreachable.mget.side_effect = AssertionError("an open breaker must not touch Redis") + unreachable.set.side_effect = AssertionError("an open breaker must not touch Redis") + unreachable.pipeline.side_effect = AssertionError("an open breaker must not touch Redis") + for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD): + redis_cache._circuit_breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused"))) + assert redis_cache._circuit_breaker.is_open() is True + with patch.object(redis_cache, "init_async_client", return_value=unreachable): + yield DualCache(in_memory_cache=InMemoryCache(), redis_cache=redis_cache) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "call, expected", + [ + pytest.param(lambda c: c.async_get_cache("lit7468"), lambda n: None, id="async_get_cache"), + pytest.param( + lambda c: c.async_batch_get_cache(["lit7468", "lit7460"]), lambda n: [None, None], id="async_batch_get_cache" + ), + 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"), + ], +) +async def test_open_breaker_is_a_quiet_cache_miss(dual_cache_with_open_breaker, call, expected, caplog): + """While the breaker is open, every cache operation must degrade to the in-memory result + without logging anything above DEBUG. + + Before this, each skipped call raised a generic exception that DualCache caught and logged + as a full ERROR traceback. Under production request rates that was hundreds of stack + formats per second per replica, enough to pin every proxy at 100% CPU on a Redis blip. + """ + caplog.set_level(logging.DEBUG, logger="LiteLLM") + + for n in range(1, 51): + assert await call(dual_cache_with_open_breaker) == expected(n) + + noisy = [r for r in caplog.records if r.levelno > logging.DEBUG] + assert noisy == [], f"an open breaker must be silent per call, got {[r.getMessage() for r in noisy]}" diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 6b2df118611..856f9b49c56 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1013,3 +1013,131 @@ async def test_breaker_metrics_track_state_and_failure_class(): 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 + + +@pytest.mark.asyncio +async def test_pool_exhaustion_counts_as_a_timeout_not_a_hard_failure(): + """redis-py reports a blocking pool that waited out its timeout as a ConnectionError + chained from the underlying TimeoutError. That is Redis being slow, the same signal as + a read timeout, so a burst of them must go through the duration gate instead of opening + the breaker on the fifth one as though Redis had refused the connection. + """ + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.caching.redis_cache import ( + RedisCircuitBreaker, + _is_redis_timeout_failure, + _run_under_circuit_breaker, + ) + + def pool_exhausted() -> RedisConnectionError: + """Built the way redis-py's BlockingConnectionPool.get_connection raises it.""" + try: + try: + raise asyncio.TimeoutError() + except asyncio.TimeoutError as err: + raise RedisConnectionError("No connection available.") from err + except RedisConnectionError as chained: + return chained + + assert _is_redis_timeout_failure(pool_exhausted()) is True + assert _is_redis_timeout_failure(RedisConnectionError("Connection refused")) is False + + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=5.0) + + async def pool_exhausted_call(): + raise pool_exhausted() + + for _ in range(breaker.failure_threshold + 1): + with pytest.raises(RedisConnectionError, match="No connection available"): + await _run_under_circuit_breaker(breaker, "op", pool_exhausted_call) + + assert breaker.is_open() is False, "an instantaneous burst of pool waits must not open the breaker" + + +@pytest.mark.asyncio +async def test_success_admitted_before_the_breaker_opened_cannot_close_it(): + """Only the HALF_OPEN recovery probe may close the breaker. + + A call that was already in flight when the breaker opened knows nothing about whether + Redis has recovered. Letting its late success close the breaker made the state flap + OPEN -> CLOSED -> OPEN under load, and every OPEN transition re-logged the warning while + the next five failures each paid the full socket timeout again. + """ + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.caching.redis_cache import ( + RedisCircuitBreaker, + RedisCircuitBreakerOpenError, + _run_under_circuit_breaker, + ) + + breaker = RedisCircuitBreaker(failure_threshold=2, recovery_timeout=60) + release_slow_success = asyncio.Event() + + async def slow_success(): + await release_slow_success.wait() + return "ok" + + async def refused(): + raise RedisConnectionError("refused") + + in_flight = asyncio.create_task(_run_under_circuit_breaker(breaker, "slow", slow_success)) + await asyncio.sleep(0) + for _ in range(breaker.failure_threshold): + with pytest.raises(RedisConnectionError): + await _run_under_circuit_breaker(breaker, "op", refused) + assert breaker.is_open() is True + + release_slow_success.set() + assert await in_flight == "ok" + + assert breaker.is_open() is True, "a pre-open success is not a recovery probe" + with pytest.raises(RedisCircuitBreakerOpenError): + await _run_under_circuit_breaker(breaker, "op", slow_success) + + +@pytest.mark.asyncio +async def test_open_breaker_raises_its_own_exception_type(): + """Callers with an optional cache need to tell the expected fast-fail apart from a real error.""" + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.caching.redis_cache import ( + RedisCircuitBreaker, + RedisCircuitBreakerOpenError, + _run_under_circuit_breaker, + ) + + breaker = RedisCircuitBreaker(failure_threshold=1, recovery_timeout=60) + + async def refused(): + raise RedisConnectionError("refused") + + with pytest.raises(RedisConnectionError): + await _run_under_circuit_breaker(breaker, "op", refused) + + with pytest.raises(RedisCircuitBreakerOpenError, match="circuit breaker is open"): + await _run_under_circuit_breaker(breaker, "op", refused) + + +@pytest.mark.asyncio +async def test_recovery_probe_still_closes_the_breaker(): + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker + + breaker = RedisCircuitBreaker(failure_threshold=1, recovery_timeout=0.05) + + async def refused(): + raise RedisConnectionError("refused") + + async def recovered(): + return "ok" + + with pytest.raises(RedisConnectionError): + await _run_under_circuit_breaker(breaker, "op", refused) + assert breaker.is_open() is True + + await asyncio.sleep(0.06) + assert await _run_under_circuit_breaker(breaker, "probe", recovered) == "ok" + assert breaker.is_open() is False diff --git a/tests/test_litellm/litellm_core_utils/test_logging_worker.py b/tests/test_litellm/litellm_core_utils/test_logging_worker.py index eb4e893adb8..f93a2e0b08a 100644 --- a/tests/test_litellm/litellm_core_utils/test_logging_worker.py +++ b/tests/test_litellm/litellm_core_utils/test_logging_worker.py @@ -525,3 +525,48 @@ class TestLoggingWorker: asyncio.run(rebind_on_second_loop()) assert sorted(executed) == [0, 1, 2, 3, 4] + + @pytest.mark.asyncio + async def test_callback_timeout_burst_logs_one_traceback_per_interval(self, caplog): + """A slow logging backend times out every in-flight callback at once. Logging a full + traceback for each of them turns that stall into a CPU-bound log storm on every replica, + so the worker must emit one traceback per interval and count the rest. + """ + caplog.set_level(logging.DEBUG, logger="LiteLLM") + worker = LoggingWorker(timeout=0.05, max_queue_size=200, concurrency=100, error_traceback_interval=60.0) + worker.start() + + async def stalled_callback(): + await asyncio.sleep(10) + + for _ in range(40): + worker.enqueue(stalled_callback()) + + await asyncio.sleep(0.5) + await worker.stop() + + errors = [r for r in caplog.records if r.levelno >= logging.ERROR and "LoggingWorker error" in r.getMessage()] + assert len(errors) == 1, f"expected one traceback for the burst, got {len(errors)}" + assert errors[0].exc_info is not None + + @pytest.mark.asyncio + async def test_error_traceback_resumes_after_interval_with_suppressed_count(self, caplog): + caplog.set_level(logging.DEBUG, logger="LiteLLM") + worker = LoggingWorker(timeout=0.05, max_queue_size=200, concurrency=100, error_traceback_interval=0.2) + worker.start() + + async def stalled_callback(): + await asyncio.sleep(10) + + for _ in range(5): + worker.enqueue(stalled_callback()) + await asyncio.sleep(0.3) + for _ in range(3): + worker.enqueue(stalled_callback()) + await asyncio.sleep(0.3) + await worker.stop() + + messages = [r.getMessage() for r in caplog.records if "LoggingWorker error" in r.getMessage()] + assert len(messages) == 2, messages + assert "(0 more suppressed" in messages[0] + assert "(4 more suppressed" in messages[1] From 1957bd388ed1c8771f767c01ccf25aabaf19ac37 Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 21:15:02 +0000 Subject: [PATCH 02/14] fix(proxy): treat an open Redis breaker as a dropped spend counter update, not a cost tracking failure Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/proxy/proxy_server.py | 7 +++- .../proxy/proxy_server/test_spend_counters.py | 42 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 9269fd48e6c..ff8f8ac8647 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -240,6 +240,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, @@ -3369,9 +3370,11 @@ 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)) - raise + if not isinstance(e, RedisCircuitBreakerOpenError): + raise + return 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/proxy/proxy_server/test_spend_counters.py b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py index c343652efd9..f5965838f05 100644 --- a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py +++ b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py @@ -1348,6 +1348,48 @@ async def test_is_spend_counter_cache_warm_redis_error_falls_back_to_in_memory( assert result is False +# --------------------------------------------------------------------------- +# _apply_spend_counter_increments +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_apply_spend_counter_increments_open_breaker_invalidates_and_returns(monkeypatch): + """An open Redis breaker fast-fails the pipeline on every request, so the callback + must not turn each one into a tracking-cost failure: drop the stale local copies + and return like a miss, without raising.""" + 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, skipping async_increment_pipeline") + ) + monkeypatch.setattr(ps, "spend_counter_cache", fake_cache) + pending: Final = ( + ps._PendingSpendIncrement(counter_key="spend:key:k", increment=1.0), + ps._PendingSpendIncrement(counter_key="spend:team:t", increment=1.0), + ) + + await ps._apply_spend_counter_increments(pending=pending) + + deleted: Final = sorted(call.kwargs["key"] for call in fake_cache.in_memory_cache.delete_cache.call_args_list) + assert deleted == ["spend:key:k", "spend:team:t"] + assert fake_cache.in_memory_cache.set_cache.called is False + + +@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=RuntimeError("incr fail")) + monkeypatch.setattr(ps, "spend_counter_cache", fake_cache) + pending: Final = (ps._PendingSpendIncrement(counter_key="spend:key:k", increment=1.0),) + + with pytest.raises(RuntimeError): + await ps._apply_spend_counter_increments(pending=pending) + + assert fake_cache.in_memory_cache.delete_cache.called is True + + # --------------------------------------------------------------------------- # _increment_spend_counter_cache # --------------------------------------------------------------------------- From 14b5915d4c74dcfc1490fb4e58caf3af42110253 Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 21:24:29 +0000 Subject: [PATCH 03/14] fix(proxy): log the open-breaker TTL preservation fallback at debug instead of per request warning Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../hooks/parallel_request_limiter_v3.py | 9 +++++-- .../hooks/test_parallel_request_limiter_v3.py | 24 +++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/hooks/parallel_request_limiter_v3.py b/litellm/proxy/hooks/parallel_request_limiter_v3.py index c6c3dde4b6e..7fdc919de2c 100644 --- a/litellm/proxy/hooks/parallel_request_limiter_v3.py +++ b/litellm/proxy/hooks/parallel_request_limiter_v3.py @@ -26,6 +26,7 @@ from typing_extensions import NotRequired, ReadOnly from litellm import DualCache from litellm._logging import verbose_proxy_logger +from litellm.caching.redis_cache import RedisCircuitBreakerOpenError from litellm.constants import DYNAMIC_RATE_LIMIT_ERROR_THRESHOLD_PER_MINUTE, INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.prompt_templates.common_utils import ( @@ -3856,8 +3857,12 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): ) except Exception as e: - verbose_proxy_logger.warning("TTL preservation failed, falling back to regular pipeline: %s", e) - # Fallback to regular pipeline on error + log: Final = ( + verbose_proxy_logger.debug + if isinstance(e, RedisCircuitBreakerOpenError) + else verbose_proxy_logger.warning + ) + log("TTL preservation failed, falling back to regular pipeline: %s", e) await self.internal_usage_cache.dual_cache.async_increment_cache_pipeline( increment_list=pipeline_operations, litellm_parent_otel_span=parent_otel_span, diff --git a/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py b/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py index 6d382370f5f..2d49338753d 100644 --- a/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py +++ b/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py @@ -1795,6 +1795,30 @@ async def test_async_increment_tokens_fallback_behavior(): ), "Fallback method should be called when Lua script is not available" +@pytest.mark.asyncio +async def test_async_increment_tokens_open_breaker_falls_back_without_warning(caplog): + """An open Redis breaker fast-fails the Lua script on every request, so it must fall + back to the regular pipeline quietly instead of emitting a WARNING per request.""" + from unittest.mock import AsyncMock + + from litellm.caching.redis_cache import RedisCircuitBreakerOpenError + + handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(DualCache())) + handler.token_increment_script = object() + handler._execute_token_increment_script = AsyncMock( + side_effect=RedisCircuitBreakerOpenError("Redis circuit breaker is open, skipping run_script") + ) + fallback = AsyncMock() + handler.internal_usage_cache.dual_cache.async_increment_cache_pipeline = fallback + pipeline_operations = [RedisPipelineIncrementOperation(key="test_breaker_key", increment_value=10.0, ttl=60)] + + with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"): + await handler.async_increment_tokens_with_ttl_preservation(pipeline_operations=pipeline_operations) + + assert fallback.await_count == 1 + assert [r.levelno for r in caplog.records if "TTL preservation failed" in r.getMessage()] == [logging.DEBUG] + + # Redis Cluster Compatibility Tests def test_group_keys_by_hash_tag_regular_redis(): """ From ad78a8f8d0a4e6200a98d5ef0ddae252b1e5c154 Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 21:35:49 +0000 Subject: [PATCH 04/14] refactor(caching): walk exception causes iteratively for the breaker timeout check Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/redis_cache.py | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index de4dcb5aa27..742c7144784 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -15,7 +15,7 @@ import hashlib import inspect import json import time -from collections.abc import Awaitable, Callable, Sequence +from collections.abc import Awaitable, Callable, Iterator, Sequence from contextvars import ContextVar from dataclasses import dataclass from datetime import timedelta @@ -334,11 +334,22 @@ def _redis_timeout_error_types() -> tuple[type, ...]: return (RedisTimeoutError, TimeoutError) +_MAX_EXCEPTION_CAUSE_DEPTH: Final = 20 + + +def _exception_cause_chain(exc: BaseException) -> Iterator[BaseException]: + current = exc # rebind-ok: advances one link per iteration of the bounded walk + for _ in range(_MAX_EXCEPTION_CAUSE_DEPTH): + yield current + if current.__cause__ is None: + return + current = current.__cause__ + + def _is_redis_timeout_failure(exc: BaseException) -> bool: - """Follows __cause__: a blocking pool wait raises ConnectionError from asyncio.TimeoutError.""" - if isinstance(exc, _redis_timeout_error_types()): - return True - return exc.__cause__ is not None and _is_redis_timeout_failure(exc.__cause__) + """Walks __cause__: a blocking pool wait raises ConnectionError from asyncio.TimeoutError.""" + timeout_types: Final = _redis_timeout_error_types() + return any(isinstance(cause, timeout_types) for cause in _exception_cause_chain(exc)) class _BreakerMetrics: From df8a9c72ab44c9fa5552b1757f36d2fe99e39a7f Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 21:45:46 +0000 Subject: [PATCH 05/14] fix(caching): keep the sync Redis read and the top-level cache wrapper quiet when the breaker is open The sync RedisCache.get_cache ran outside the breaker and logged with a stray positional argument, so every failure printed a logging-module stack dump. Cache.add_cache and its async twins logged a full traceback for the expected open-breaker fast fail Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/caching.py | 8 +++++- litellm/caching/dual_cache.py | 2 ++ litellm/caching/redis_cache.py | 4 ++- tests/test_litellm/caching/test_caching.py | 26 +++++++++++++++++++ tests/test_litellm/caching/test_dual_cache.py | 15 +++++++++++ .../test_litellm/caching/test_redis_cache.py | 23 +++++++++++++++- 6 files changed, 75 insertions(+), 3 deletions(-) diff --git a/litellm/caching/caching.py b/litellm/caching/caching.py index 884d095793c..b5423998401 100644 --- a/litellm/caching/caching.py +++ b/litellm/caching/caching.py @@ -32,7 +32,7 @@ from .dual_cache import DualCache # noqa: F401 from .gcs_cache import GCSCache from .in_memory_cache import InMemoryCache from .qdrant_semantic_cache import QdrantSemanticCache -from .redis_cache import RedisCache +from .redis_cache import RedisCache, RedisCircuitBreakerOpenError from .redis_cluster_cache import RedisClusterCache from .redis_semantic_cache import RedisSemanticCache from .s3_cache import S3Cache @@ -677,6 +677,8 @@ class Cache: return cache_key, cached_data, kwargs = self._add_cache_logic(result=result, **kwargs) self.cache.set_cache(cache_key, cached_data, **kwargs) + except RedisCircuitBreakerOpenError as e: + verbose_logger.debug("LiteLLM Cache: skipped add_cache: %s", e) except Exception as e: verbose_logger.exception("LiteLLM Cache: Excepton add_cache: %s", e) @@ -696,6 +698,8 @@ class Cache: await dynamic_cache_object.async_set_cache(cache_key, cached_data, **kwargs) else: await self.cache.async_set_cache(cache_key, cached_data, **kwargs) + except RedisCircuitBreakerOpenError as e: + verbose_logger.debug("LiteLLM Cache: skipped add_cache: %s", e) except Exception as e: verbose_logger.exception("LiteLLM Cache: Excepton add_cache: %s", e) @@ -875,6 +879,8 @@ class Cache: await dynamic_cache_object.async_set_cache_pipeline(cache_list=cache_list, **kwargs) else: await self.cache.async_set_cache_pipeline(cache_list=cache_list, **kwargs) + except RedisCircuitBreakerOpenError as e: + verbose_logger.debug("LiteLLM Cache: skipped add_cache: %s", e) except Exception as e: verbose_logger.exception("LiteLLM Cache: Excepton add_cache: %s", e) diff --git a/litellm/caching/dual_cache.py b/litellm/caching/dual_cache.py index 98c6da0f02f..6fb918ce748 100644 --- a/litellm/caching/dual_cache.py +++ b/litellm/caching/dual_cache.py @@ -177,6 +177,8 @@ class DualCache(BaseCache): print_verbose(f"get cache: cache result: {result}") return result + except RedisCircuitBreakerOpenError: + return None except Exception: verbose_logger.error(traceback.format_exc()) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 742c7144784..6c19b018c1d 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -1364,6 +1364,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) @@ -1384,7 +1385,8 @@ class RedisCache(BaseCache): 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]: """ diff --git a/tests/test_litellm/caching/test_caching.py b/tests/test_litellm/caching/test_caching.py index 4d0ec0fb677..5d326a9ece0 100644 --- a/tests/test_litellm/caching/test_caching.py +++ b/tests/test_litellm/caching/test_caching.py @@ -252,3 +252,29 @@ def test_exact_cache_key_includes_anthropic_messages_params(anthropic_param): assert baseline != cache.get_cache_key( model="claude-sonnet-4-5", messages=messages, **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 + stack formats per second per replica. Unexpected failures must still get the traceback. + """ + from unittest.mock import AsyncMock + + from litellm.caching.redis_cache import RedisCircuitBreakerOpenError + + caplog.set_level(logging.DEBUG, logger="LiteLLM") + cache = Cache(type=LiteLLMCacheType.LOCAL) + cache.cache.async_set_cache = AsyncMock(side_effect=RedisCircuitBreakerOpenError("open")) + + await cache.async_add_cache("result", model="gpt-5.4-nano", messages=[{"role": "user", "content": "hi"}]) + + 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"}]) + + errors = [r for r in caplog.records if r.levelno == logging.ERROR] + assert len(errors) == 1 and errors[0].exc_info is not None diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index 46dd2687e48..3e21362b3f0 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/tests/test_litellm/caching/test_dual_cache.py @@ -638,3 +638,18 @@ async def test_open_breaker_is_a_quiet_cache_miss(dual_cache_with_open_breaker, noisy = [r for r in caplog.records if r.levelno > logging.DEBUG] assert noisy == [], f"an open breaker must be silent per call, got {[r.getMessage() for r in noisy]}" + + +def test_open_breaker_is_a_quiet_cache_miss_on_the_sync_read_path(dual_cache_with_open_breaker, caplog): + """The sync read runs in the request thread pool for /v1/messages and /v1/responses, so it + must short-circuit on an open breaker like the async path instead of dialing Redis per call. + """ + caplog.set_level(logging.DEBUG, logger="LiteLLM") + redis_client = dual_cache_with_open_breaker.redis_cache.redis_client + redis_client.get.side_effect = AssertionError("an open breaker must not touch Redis") + + for _ in range(50): + assert dual_cache_with_open_breaker.get_cache("lit7468") is None + + noisy = [r for r in caplog.records if r.levelno > logging.DEBUG] + assert noisy == [], f"an open breaker must be silent per call, got {[r.getMessage() for r in noisy]}" diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 856f9b49c56..1a884d44db1 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1,4 +1,5 @@ import asyncio +import logging from collections.abc import Iterator from unittest.mock import AsyncMock, MagicMock, patch @@ -646,7 +647,6 @@ def test_sync_batch_get_cache_survives_a_service_callback_that_raises( from concurrent.futures import ThreadPoolExecutor import litellm - from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD cache, service_logger = sync_batch_cache_with_service_logger @@ -1141,3 +1141,24 @@ async def test_recovery_probe_still_closes_the_breaker(): await asyncio.sleep(0.06) assert await _run_under_circuit_breaker(breaker, "probe", recovered) == "ok" assert breaker.is_open() is False + + +def test_sync_get_cache_failure_feeds_the_breaker_and_logs_a_well_formed_record(sync_batch_redis_cache, caplog): + """The sync read used to log with a stray positional arg, so every Redis failure produced a + `--- Logging error ---` stack dump on stderr, and it sat outside the breaker so it kept dialing + Redis on every request even after the async paths had opened it. + """ + from redis.exceptions import ConnectionError as RedisConnectionError + + from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD + + caplog.set_level(logging.ERROR, logger="LiteLLM") + sync_batch_redis_cache.redis_client.get.side_effect = RedisConnectionError("refused") + + for _ in range(REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD): + assert sync_batch_redis_cache.get_cache("lit7468") is None + + assert sync_batch_redis_cache._circuit_breaker.is_open() is True + assert sync_batch_redis_cache.redis_client.get.call_count == REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD + assert all("refused" in record.getMessage() for record in caplog.records) + assert len(caplog.records) == REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD From 7658cd53aaf5f8cc217b9a9a25891851eaa64cbc Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 22:25:03 +0000 Subject: [PATCH 06/14] fix(caching): judge swallowed Redis failures per admission and throttle worker tracebacks per error type Swallowed failures are now reported to the breaker when the admitted call exits, so a call admitted before the breaker opened cannot refresh the open timer or knock out the recovery probe. The sync batch read raises the typed open-breaker error like its async twin so DualCache releases its batch reservations, and LoggingWorker throttles tracebacks per exception class instead of worker-wide Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/dual_cache.py | 3 + litellm/caching/redis_cache.py | 60 ++++++++++------- litellm/litellm_core_utils/logging_worker.py | 20 +++--- tests/test_litellm/caching/test_dual_cache.py | 20 ++++++ .../test_litellm/caching/test_redis_cache.py | 65 +++++++++++++++++-- .../litellm_core_utils/test_logging_worker.py | 32 ++++++++- 6 files changed, 162 insertions(+), 38 deletions(-) diff --git a/litellm/caching/dual_cache.py b/litellm/caching/dual_cache.py index 6fb918ce748..5dd0493cd65 100644 --- a/litellm/caching/dual_cache.py +++ b/litellm/caching/dual_cache.py @@ -206,6 +206,9 @@ class DualCache(BaseCache): redis_result: Final = self.redis_cache.batch_get_cache( key_list=sublist_keys, parent_otel_span=parent_otel_span ) + except RedisCircuitBreakerOpenError: + self._rollback_redis_batch_key_reservations(previous_access_times) + return result except Exception: # Do not throttle subsequent callers if the Redis read fails. self._rollback_redis_batch_key_reservations(previous_access_times) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 6c19b018c1d..3ecf805ed0e 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -285,7 +285,9 @@ class RedisCircuitBreaker: _RedisCallResult = TypeVar("_RedisCallResult") -_swallowed_redis_failures: Final[ContextVar[int]] = ContextVar("litellm_swallowed_redis_failures", default=0) +_swallowed_redis_failures: Final[ContextVar[tuple[bool, ...]]] = ContextVar( + "litellm_swallowed_redis_failures", default=() +) def _opaque_kwarg_key(value: object) -> str: @@ -405,8 +407,8 @@ def _breaker_metrics() -> _BreakerMetrics: return _BreakerMetrics() -def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseException) -> None: - """Record a Redis failure that the calling method is about to swallow. +def _record_swallowed_redis_failure(exc: BaseException) -> None: + """Note a Redis failure that the calling method is about to swallow, for the breaker exit to judge. The marker is a ContextVar rather than a counter on the breaker because breakers are shared by every concurrent caller. A plain shared counter cannot tell "my call failed" @@ -416,8 +418,7 @@ def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseExcep """ if not _is_redis_health_failure(exc): return - breaker.record_failure(is_timeout=_is_redis_timeout_failure(exc)) - _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1) + _swallowed_redis_failures.set((*_swallowed_redis_failures.get(), _is_redis_timeout_failure(exc))) @dataclass(frozen=True, slots=True) @@ -430,25 +431,42 @@ def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> _BreakerA """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=_swallowed_redis_failures.get(), generation=breaker.generation) + return _BreakerAdmission(swallowed_before=len(_swallowed_redis_failures.get()), generation=breaker.generation) + + +def _take_swallowed_failures(admission: _BreakerAdmission) -> tuple[bool, ...]: + """Return the is_timeout flag of every failure this call swallowed, and drop them from the context.""" + all_swallowed: Final = _swallowed_redis_failures.get() + _swallowed_redis_failures.set(all_swallowed[: admission.swallowed_before]) + return all_swallowed[admission.swallowed_before :] def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission) -> None: - """Record success only when nothing failed while the call ran and the breaker has not opened since. + """Report the call's outcome to the breaker generation it was admitted under. 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 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. """ + swallowed: Final = _take_swallowed_failures(admission) if admission.generation != breaker.generation: return - if _swallowed_redis_failures.get() == admission.swallowed_before: + if not swallowed: breaker.record_success() + return + for is_timeout in swallowed: + breaker.record_failure(is_timeout=is_timeout) def _fail_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission, exc: BaseException) -> None: - if admission.generation != breaker.generation or not _is_redis_health_failure(exc): + swallowed: Final = _take_swallowed_failures(admission) + if admission.generation != breaker.generation: return - breaker.record_failure(is_timeout=_is_redis_timeout_failure(exc)) + for is_timeout in swallowed: + breaker.record_failure(is_timeout=is_timeout) + if _is_redis_health_failure(exc): + breaker.record_failure(is_timeout=_is_redis_timeout_failure(exc)) async def _run_under_circuit_breaker( @@ -1056,7 +1074,7 @@ class RedisCache(BaseCache): str(e), value, ) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) async def _pipeline_helper( self, @@ -1143,7 +1161,7 @@ class RedisCache(BaseCache): str(e), cache_value, ) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) async def _set_cache_sadd_helper( self, @@ -1228,7 +1246,7 @@ class RedisCache(BaseCache): str(e), value, ) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) @_redis_circuit_breaker_guard async def batch_cache_write(self, key, value, **kwargs): @@ -1386,7 +1404,7 @@ class RedisCache(BaseCache): except Exception as e: # NON blocking - notify users Redis is throwing an exception verbose_logger.error("litellm.caching.caching: get() - Got exception from REDIS: %s", e) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) def _run_redis_mget_operation(self, keys: list[str]) -> Sequence[bytes | str | None]: """ @@ -1423,9 +1441,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() + admission: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache") try: - admission: 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, admission) @@ -1452,8 +1470,6 @@ class RedisCache(BaseCache): decoded_results[k] = v return decoded_results - except RedisCircuitBreakerOpenError: - return key_value_dict except Exception as e: failed_at: Final = time.time() self.service_logger_obj.service_failure_hook( @@ -1466,7 +1482,7 @@ class RedisCache(BaseCache): parent_otel_span=parent_otel_span, ) verbose_logger.error("Error occurred in batch get cache - %s", e) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _fail_circuit_breaker(self._circuit_breaker, admission, e) return key_value_dict @_redis_circuit_breaker_guard @@ -1513,7 +1529,7 @@ class RedisCache(BaseCache): ) ) print_verbose(f"litellm.caching.caching: async get() - Got exception from REDIS: {e}") - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) @_redis_circuit_breaker_guard async def async_batch_get_cache( @@ -1585,7 +1601,7 @@ class RedisCache(BaseCache): ) ) verbose_logger.error("Error occurred in async batch get cache - %s", e) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) return key_value_dict def sync_ping(self) -> bool: @@ -1837,7 +1853,7 @@ class RedisCache(BaseCache): return ttl except Exception as e: verbose_logger.debug("Redis TTL Error: %s", e) - _record_swallowed_redis_failure(self._circuit_breaker, e) + _record_swallowed_redis_failure(e) return None @_redis_circuit_breaker_guard diff --git a/litellm/litellm_core_utils/logging_worker.py b/litellm/litellm_core_utils/logging_worker.py index dfaace1bb10..de3999d63ea 100644 --- a/litellm/litellm_core_utils/logging_worker.py +++ b/litellm/litellm_core_utils/logging_worker.py @@ -55,8 +55,8 @@ class LoggingWorker: self.max_queue_size = max_queue_size self.concurrency = concurrency self.error_traceback_interval = error_traceback_interval - self._last_error_traceback_at: float | None = None - self._errors_since_traceback: int = 0 + self._last_error_traceback_at: dict[type[BaseException], float] = {} + self._errors_since_traceback: dict[type[BaseException], int] = {} self._queue: asyncio.Queue[LoggingTask] | None = None self._worker_task: asyncio.Task | None = None self._running_tasks: set[asyncio.Task] = set() @@ -178,19 +178,21 @@ class LoggingWorker: sem.release() def _log_task_error(self, error: Exception) -> None: - """One traceback per interval: a stalled backend fails every in-flight task at once.""" + """One traceback per error type per interval: a stalled backend fails every in-flight task at once.""" now: Final = time.monotonic() - last_traceback_at: Final = self._last_error_traceback_at + error_type: Final = type(error) + last_traceback_at: Final = self._last_error_traceback_at.get(error_type) if last_traceback_at is not None and now - last_traceback_at < self.error_traceback_interval: - self._errors_since_traceback += 1 + self._errors_since_traceback[error_type] = self._errors_since_traceback.get(error_type, 0) + 1 return verbose_logger.exception( - "LoggingWorker error (%d more suppressed since the last traceback): %r", - self._errors_since_traceback, + "LoggingWorker error (%d more %s suppressed since the last traceback): %r", + self._errors_since_traceback.get(error_type, 0), + error_type.__name__, error, ) - self._last_error_traceback_at = now - self._errors_since_traceback = 0 + self._last_error_traceback_at[error_type] = now + self._errors_since_traceback[error_type] = 0 async def _worker_loop(self) -> None: """Main worker loop that gets tasks and schedules them to run concurrently.""" diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index 3e21362b3f0..b2462b297fc 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/tests/test_litellm/caching/test_dual_cache.py @@ -653,3 +653,23 @@ def test_open_breaker_is_a_quiet_cache_miss_on_the_sync_read_path(dual_cache_wit noisy = [r for r in caplog.records if r.levelno > logging.DEBUG] assert noisy == [], f"an open breaker must be silent per call, got {[r.getMessage() for r in noisy]}" + + +def test_open_breaker_does_not_leave_sync_batch_reservations_behind(dual_cache_with_open_breaker, caplog): + """A batch read skipped by the breaker must not hold its keys for the batch expiry window. + + The sync read reserves keys before dialing Redis so concurrent callers do not all hit it. + When the breaker rejects the read, those reservations have to be released, otherwise the + first read after Redis recovers is still throttled for up to default_redis_batch_cache_expiry. + """ + caplog.set_level(logging.DEBUG, logger="LiteLLM") + dual_cache_with_open_breaker.redis_cache.redis_client.mget.side_effect = AssertionError( + "an open breaker must not touch Redis" + ) + + assert dual_cache_with_open_breaker.batch_get_cache(keys=["lit7468", "lit7460"]) == [None, None] + + assert "lit7468" not in dual_cache_with_open_breaker.last_redis_batch_access_time + assert "lit7460" not in dual_cache_with_open_breaker.last_redis_batch_access_time + noisy = [r for r in caplog.records if r.levelno > logging.DEBUG] + assert noisy == [], f"an open breaker must be silent per call, got {[r.getMessage() for r in noisy]}" diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 1a884d44db1..3bc3aa1e047 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -516,14 +516,20 @@ 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_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 + can release the batch reservations it took before the call. + """ + from litellm.caching.redis_cache import RedisCircuitBreakerOpenError 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"]) == {} + with pytest.raises(RedisCircuitBreakerOpenError): + sync_batch_redis_cache.batch_get_cache(key_list=["lit6729"]) + assert sync_batch_redis_cache.redis_client.mget.call_count == REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD def test_batch_get_counts_raises_where_batch_get_cache_reports_a_miss(sync_batch_redis_cache): @@ -647,6 +653,7 @@ def test_sync_batch_get_cache_survives_a_service_callback_that_raises( from concurrent.futures import ThreadPoolExecutor import litellm + from litellm.caching.redis_cache import RedisCircuitBreakerOpenError from litellm.constants import REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD cache, service_logger = sync_batch_cache_with_service_logger @@ -661,7 +668,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(): @@ -801,7 +809,7 @@ async def test_concurrent_success_is_not_cancelled_by_another_calls_failure(): # starts would leave its snapshot correct and prove nothing. async def swallows_a_failure(): await asyncio.sleep(0.02) - _record_swallowed_redis_failure(breaker, RedisConnectionError("redis unreachable")) + _record_swallowed_redis_failure(RedisConnectionError("redis unreachable")) async def succeeds_while_the_other_fails(): await asyncio.sleep(0.05) @@ -1097,6 +1105,53 @@ async def test_success_admitted_before_the_breaker_opened_cannot_close_it(): await _run_under_circuit_breaker(breaker, "op", slow_success) +@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. + + 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 + back by its own socket timeout, and knock out the HALF_OPEN probe if it landed then. + """ + 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=2, recovery_timeout=0.05) + release_stale_call = asyncio.Event() + + async def stale_call_that_swallows_its_failure(): + await release_stale_call.wait() + _record_swallowed_redis_failure(RedisConnectionError("refused")) + return {} + + async def refused(): + raise RedisConnectionError("refused") + + async def recovered(): + return "ok" + + in_flight = asyncio.create_task(_run_under_circuit_breaker(breaker, "stale", stale_call_that_swallows_its_failure)) + await asyncio.sleep(0) + for _ in range(breaker.failure_threshold): + with pytest.raises(RedisConnectionError): + await _run_under_circuit_breaker(breaker, "op", refused) + assert breaker.is_open() is True + + await asyncio.sleep(0.04) + release_stale_call.set() + assert await in_flight == {} + await asyncio.sleep(0.03) + + assert await _run_under_circuit_breaker(breaker, "probe", recovered) == "ok" + assert breaker.is_open() is False + + @pytest.mark.asyncio async def test_open_breaker_raises_its_own_exception_type(): """Callers with an optional cache need to tell the expected fast-fail apart from a real error.""" diff --git a/tests/test_litellm/litellm_core_utils/test_logging_worker.py b/tests/test_litellm/litellm_core_utils/test_logging_worker.py index f93a2e0b08a..66d49b22c6b 100644 --- a/tests/test_litellm/litellm_core_utils/test_logging_worker.py +++ b/tests/test_litellm/litellm_core_utils/test_logging_worker.py @@ -568,5 +568,33 @@ class TestLoggingWorker: messages = [r.getMessage() for r in caplog.records if "LoggingWorker error" in r.getMessage()] assert len(messages) == 2, messages - assert "(0 more suppressed" in messages[0] - assert "(4 more suppressed" in messages[1] + assert "(0 more TimeoutError suppressed" in messages[0] + assert "(4 more TimeoutError suppressed" in messages[1] + + @pytest.mark.asyncio + async def test_traceback_throttle_is_per_error_type(self, caplog): + """A timeout burst from one stalled backend must not hide the first traceback of a different + failure, otherwise a misconfigured callback stays invisible for the whole interval. + """ + caplog.set_level(logging.DEBUG, logger="LiteLLM") + worker = LoggingWorker(timeout=0.05, max_queue_size=200, concurrency=100, error_traceback_interval=60.0) + worker.start() + + async def stalled_callback(): + await asyncio.sleep(10) + + async def misconfigured_callback(): + raise KeyError("missing api key") + + for _ in range(20): + worker.enqueue(stalled_callback()) + await asyncio.sleep(0.2) + for _ in range(3): + worker.enqueue(misconfigured_callback()) + await asyncio.sleep(0.2) + await worker.stop() + + messages = [r.getMessage() for r in caplog.records if "LoggingWorker error" in r.getMessage()] + assert len(messages) == 2, messages + assert "TimeoutError" in messages[0] + assert "KeyError" in messages[1] and "missing api key" in messages[1] From fbd923190e04583e56184c6ff38e1b2e41ad6c36 Mon Sep 17 00:00:00 2001 From: mateo Date: Thu, 10 Sep 2026 22:53:46 +0000 Subject: [PATCH 07/14] 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 | 36 ++++++-- tests/test_litellm/caching/test_caching.py | 33 +++++-- tests/test_litellm/caching/test_dual_cache.py | 5 ++ .../test_litellm/caching/test_redis_cache.py | 90 +++++++++++++++++-- 4 files changed, 141 insertions(+), 23 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 3ecf805ed0e..2c1ca75212c 100644 --- a/litellm/caching/redis_cache.py +++ b/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 diff --git a/tests/test_litellm/caching/test_caching.py b/tests/test_litellm/caching/test_caching.py index 5d326a9ece0..d135d2892c5 100644 --- a/tests/test_litellm/caching/test_caching.py +++ b/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 diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index b2462b297fc..4177c215d0b 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/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): diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 3bc3aa1e047..8c994a4742e 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/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" 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 08/14] 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 # --------------------------------------------------------------------------- From 7169ddaef6ad4847f630f2c2bcefa2799db19c92 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:16:59 -0700 Subject: [PATCH 09/14] fix(router): treat a breaker-refused Redis read as a miss in the health state cache The sync Redis read now raises while the circuit breaker is open, and the health state merge caught that as a generic error, skipping the local write and logging an error on every background health check cycle. Read the shared snapshot through a helper that treats the refused read as a miss so the merge falls back to the pod-local copy the way a swallowed connection error already did --- litellm/router_utils/health_state_cache.py | 20 ++++++++++---- .../router_utils/test_health_state_cache.py | 27 +++++++++++++++++++ 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/litellm/router_utils/health_state_cache.py b/litellm/router_utils/health_state_cache.py index 22d816e13e9..c8ca7105392 100644 --- a/litellm/router_utils/health_state_cache.py +++ b/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 diff --git a/tests/test_litellm/router_utils/test_health_state_cache.py b/tests/test_litellm/router_utils/test_health_state_cache.py index ffd031f9b7d..aa976bb1002 100644 --- a/tests/test_litellm/router_utils/test_health_state_cache.py +++ b/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"} From dcdd884352b1f39f51dd0719bba4eca0046859f8 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:36:19 -0700 Subject: [PATCH 10/14] fix(caching): let only the recovery probe close a half-open Redis breaker A call admitted before the breaker opened could finish while the breaker was HALF_OPEN and close it before the designated probe reported, so Redis traffic resumed on a stale answer. The admission now records whether the call is the probe and only the probe's success closes a half-open breaker. The cron job lock manager also logged an error every cycle the open breaker refused its Redis call, one line per job per pod. That refusal is now a debug line like every other guarded call, while real Redis errors still log at error --- litellm/caching/redis_cache.py | 43 ++++++++++------ .../db_transaction_queue/pod_lock_manager.py | 7 +-- .../test_litellm/caching/test_redis_cache.py | 49 +++++++++++++++++++ .../test_pod_lock_manager.py | 18 +++++++ 4 files changed, 100 insertions(+), 17 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 5b1a9739eb8..812b435ae5f 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -18,6 +18,7 @@ import logging import time from collections.abc import Awaitable, Callable, Sequence from contextvars import ContextVar +from dataclasses import dataclass from datetime import timedelta from typing import TYPE_CHECKING, Any, Final, Protocol, TypeVar, cast @@ -198,6 +199,9 @@ class RedisCircuitBreaker: self._state = self.CLOSED _breaker_metrics().record_state_change(None, self._state) + def is_half_open(self) -> bool: + return self._state == self.HALF_OPEN + def is_open(self) -> bool: """Returns True if Redis calls should be skipped.""" if not self.enabled: @@ -405,21 +409,32 @@ def log_redis_failure( logger.log(level, "%s: %s", message, exc, exc_info=exc if with_traceback else None) -def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> int: - """Reject the call if the breaker is open, else return the swallowed-failure count to compare against.""" +@dataclass(frozen=True, slots=True) +class _BreakerAdmission: + swallowed_before: int + is_probe: bool + + +def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> _BreakerAdmission: + """Reject the call if the breaker is open, else record what its success may later prove.""" if breaker.is_open(): raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open — skipping {name}") - return _swallowed_redis_failures.get() + return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), is_probe=breaker.is_half_open()) -def _exit_circuit_breaker(breaker: RedisCircuitBreaker, swallowed_before: int) -> None: - """Record success only when nothing failed while the call ran. +def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission) -> None: + """Record success only when nothing failed while the call ran and the call may vouch for Redis. Several Redis methods catch their own connection errors and return a default, so a - method that returned is not on its own proof of a healthy Redis. + method that returned is not on its own proof of a healthy Redis. While the breaker is + half open only the designated recovery probe may close it: a call admitted before the + breaker opened that finishes late says nothing about whether Redis recovered. """ - if _swallowed_redis_failures.get() == swallowed_before: - breaker.record_success() + if _swallowed_redis_failures.get() != admission.swallowed_before: + return + if breaker.is_half_open() and not admission.is_probe: + return + breaker.record_success() async def _run_under_circuit_breaker( @@ -432,14 +447,14 @@ async def _run_under_circuit_breaker( Shared by the method decorator and the Lua script executor so both feed the same health signal. """ - swallowed_before: Final = _enter_circuit_breaker(breaker, name) + admission: Final = _enter_circuit_breaker(breaker, name) try: result: Final = await call() except Exception as e: if _is_redis_health_failure(e): breaker.record_failure(is_timeout=_is_redis_timeout_failure(e)) raise - _exit_circuit_breaker(breaker, swallowed_before) + _exit_circuit_breaker(breaker, admission) return result @@ -449,14 +464,14 @@ def _run_under_circuit_breaker_sync( call: Callable[[], _RedisCallResult], ) -> _RedisCallResult: """Run one blocking Redis call under a circuit breaker, feeding the same health signal as the async path.""" - swallowed_before: Final = _enter_circuit_breaker(breaker, name) + admission: Final = _enter_circuit_breaker(breaker, name) try: result: Final = call() except Exception as e: if _is_redis_health_failure(e): breaker.record_failure(is_timeout=_is_redis_timeout_failure(e)) raise - _exit_circuit_breaker(breaker, swallowed_before) + _exit_circuit_breaker(breaker, admission) return result @@ -1395,12 +1410,12 @@ class RedisCache(BaseCache): key_value_dict = {} _key_list: Final = [key for key in key_list if key is not None] start_time: Final = time.time() - swallowed_before: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache") + admission: Final = _enter_circuit_breaker(self._circuit_breaker, "batch_get_cache") try: _keys: Final = [self.check_and_fix_namespace(key=cache_key or "") for cache_key in _key_list] results: Final = self._run_redis_mget_operation(keys=_keys) - _exit_circuit_breaker(self._circuit_breaker, swallowed_before) + _exit_circuit_breaker(self._circuit_breaker, admission) end_time: Final = time.time() _duration: Final = end_time - start_time self.service_logger_obj.service_success_hook( diff --git a/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py b/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py index 4be1331e955..bc67617e444 100644 --- a/litellm/proxy/db/db_transaction_queue/pod_lock_manager.py +++ b/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: """ diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index c6f0caa1dca..9df9c44a30f 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1103,3 +1103,52 @@ def test_recovery_probe_still_closes_the_breaker(): assert breaker._state == breaker.CLOSED assert breaker.is_open() is False + + +@pytest.mark.asyncio +async def test_stale_success_during_the_recovery_probe_leaves_the_breaker_to_the_probe(): + """A call admitted before the trip that finishes while HALF_OPEN must not close the breaker. + + Only the one call designated as the recovery probe has actually reached Redis after the + outage, so closing on the straggler's success resumed full Redis traffic before the probe + had proven anything. + """ + from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker + + breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60) + stale_admitted = asyncio.Event() + stale_release = asyncio.Event() + probe_admitted = asyncio.Event() + probe_release = asyncio.Event() + + async def stale_call() -> str: + stale_admitted.set() + await stale_release.wait() + return "stale" + + async def probe_call() -> str: + probe_admitted.set() + await probe_release.wait() + return "probe" + + stale = asyncio.ensure_future(_run_under_circuit_breaker(breaker, "op", stale_call)) + await stale_admitted.wait() + for _ in range(3): + breaker.record_failure() + assert breaker._state == breaker.OPEN + breaker._opened_at = time.time() - 9999 + probe = asyncio.ensure_future(_run_under_circuit_breaker(breaker, "op", probe_call)) + await probe_admitted.wait() + assert breaker._state == breaker.HALF_OPEN + + stale_release.set() + assert await stale == "stale" + + assert breaker._state == breaker.HALF_OPEN, "the straggler must not close the breaker for the probe" + assert breaker.is_open() is True + + probe_release.set() + assert await probe == "probe" + + assert breaker._state == breaker.CLOSED + assert breaker.is_open() is False diff --git a/tests/test_litellm/proxy/db/db_transaction_queue/test_pod_lock_manager.py b/tests/test_litellm/proxy/db/db_transaction_queue/test_pod_lock_manager.py index ecd5c5f50c0..4684c3213d6 100644 --- a/tests/test_litellm/proxy/db/db_transaction_queue/test_pod_lock_manager.py +++ b/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): """ From 01c6b50564c16375023dd240fd376ccca11fa42b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:57:24 -0700 Subject: [PATCH 11/14] fix(caching): let a Redis breaker success count only for the state that admitted the call --- litellm/caching/redis_cache.py | 23 +++++---- .../test_litellm/caching/test_redis_cache.py | 50 +++++++++++++++++++ 2 files changed, 64 insertions(+), 9 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 812b435ae5f..2c36995c4f8 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -197,10 +197,13 @@ 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) - def is_half_open(self) -> bool: - return self._state == self.HALF_OPEN + @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.""" @@ -270,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") @@ -412,27 +416,28 @@ def log_redis_failure( @dataclass(frozen=True, slots=True) class _BreakerAdmission: swallowed_before: int - is_probe: bool + 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 _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), is_probe=breaker.is_half_open()) + return _BreakerAdmission(swallowed_before=_swallowed_redis_failures.get(), generation=breaker.generation) def _exit_circuit_breaker(breaker: RedisCircuitBreaker, admission: _BreakerAdmission) -> None: - """Record success only when nothing failed while the call ran and the call may vouch for Redis. + """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. While the breaker is - half open only the designated recovery probe may close it: a call admitted before the - breaker opened that finishes late says nothing about whether Redis recovered. + 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() != admission.swallowed_before: return - if breaker.is_half_open() and not admission.is_probe: + if breaker.generation != admission.generation: return breaker.record_success() diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 9df9c44a30f..bcae33b976e 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1152,3 +1152,53 @@ async def test_stale_success_during_the_recovery_probe_leaves_the_breaker_to_the 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 From 0ffe6512de83f4907ff8a387d673b04fb14eb7db Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 19:15:40 -0700 Subject: [PATCH 12/14] fix(router): keep the routing and budget sync loops quiet while the Redis breaker is open --- .../router_strategy/base_routing_strategy.py | 5 ++- litellm/router_strategy/budget_limiter.py | 17 ++++----- .../test_base_routing_strategy.py | 18 ++++++++- .../test_budget_limiter_hotpath.py | 37 +++++++++++++++++++ 4 files changed, 64 insertions(+), 13 deletions(-) diff --git a/litellm/router_strategy/base_routing_strategy.py b/litellm/router_strategy/base_routing_strategy.py index 8235761ca98..686d57e2b77 100644 --- a/litellm/router_strategy/base_routing_strategy.py +++ b/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): diff --git a/litellm/router_strategy/budget_limiter.py b/litellm/router_strategy/budget_limiter.py index a8d51f95e45..acdc57d706d 100644 --- a/litellm/router_strategy/budget_limiter.py +++ b/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 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, @@ -536,17 +537,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: + await self.dual_cache.redis_cache.async_increment_pipeline(increment_list=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 +598,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, diff --git a/tests/test_litellm/router_strategy/test_base_routing_strategy.py b/tests/test_litellm/router_strategy/test_base_routing_strategy.py index 154042692d0..dc75c3d2916 100644 --- a/tests/test_litellm/router_strategy/test_base_routing_strategy.py +++ b/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 == [] diff --git a/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py b/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py index 36fa38bacb5..128b2dbd842 100644 --- a/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py +++ b/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,34 @@ 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 From 2fc520329ffcbf6ec98b86d8f1c21e0da314e337 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 19:26:28 -0700 Subject: [PATCH 13/14] fix(router): keep the budget push off the request callback path The provider budget push runs inside the request success callback, so awaiting the Redis pipeline there made every request wait for the round trip. Hand it back to a task whose failure is logged through the breaker aware logger, so an open breaker stays a debug line and a real Redis error is one error line instead of an unretrieved task traceback --- litellm/router_strategy/budget_limiter.py | 11 +++- .../test_budget_limiter_hotpath.py | 56 +++++++++++++++++++ 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/litellm/router_strategy/budget_limiter.py b/litellm/router_strategy/budget_limiter.py index acdc57d706d..3e094df7ac8 100644 --- a/litellm/router_strategy/budget_limiter.py +++ b/litellm/router_strategy/budget_limiter.py @@ -28,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, log_redis_failure +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, @@ -93,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, @@ -540,7 +547,7 @@ class RouterBudgetLimiting(CustomLogger): queued: Final = self.redis_increment_operation_queue self.redis_increment_operation_queue = [] if queued: - await self.dual_cache.redis_cache.async_increment_pipeline(increment_list=queued) + asyncio.create_task(_push_increments_to_redis(self.dual_cache.redis_cache, queued)) except Exception as e: log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e) diff --git a/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py b/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py index 128b2dbd842..4cc8fe78811 100644 --- a/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py +++ b/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py @@ -340,3 +340,59 @@ async def test_sync_refused_by_the_open_circuit_breaker_is_quiet_and_leaks_no_ta 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() From 25ed0abfc955fe886fe214af580153b6976dde8b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 10 Sep 2026 19:42:27 -0700 Subject: [PATCH 14/14] chore(ui): regenerate schema.d.ts after merging the base --- ui/litellm-dashboard/src/lib/http/schema.d.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 6826cded6f5..29435c31aee 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/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;