From 0f66803058008c06385c53e4649ef5088dadbe26 Mon Sep 17 00:00:00 2001 From: mubashir1osmani Date: Thu, 13 Aug 2026 13:28:15 -0700 Subject: [PATCH] fix(redis): rebuild the cluster client on breaker open and HALF_OPEN probe Fail-fast skips Redis while the cluster is down, but the next probe must not reuse the wedged NodesManager that hung on CLUSTER SLOTS. Evict the cached async client (and rebuild the sync one) when the breaker opens and again when the recovery probe starts, so ElastiCache coming back can accept traffic again. --- litellm/caching/redis_cache.py | 37 +++++++++- .../test_litellm/caching/test_redis_cache.py | 68 +++++++++++++++++++ 2 files changed, 104 insertions(+), 1 deletion(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index d6b416af923..21bf9029910 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -122,10 +122,14 @@ class RedisCircuitBreaker: failure_threshold: int, recovery_timeout: int, enabled: bool = True, + on_open: Callable[[], None] | None = None, + on_probe: Callable[[], None] | None = None, ) -> None: self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.enabled = enabled + self._on_open = on_open + self._on_probe = on_probe self._failure_count = 0 self._opened_at: float | None = None self._state = self.CLOSED @@ -142,6 +146,10 @@ class RedisCircuitBreaker: if self._state == self.OPEN: if time.time() - (self._opened_at or 0) > self.recovery_timeout: self._state = self.HALF_OPEN + # Drop the cached cluster client so the probe rediscovers + # CLUSTER SLOTS instead of reusing a wedged NodesManager. + if self._on_probe is not None: + self._on_probe() return False # this caller is the designated probe return True return False @@ -158,6 +166,8 @@ class RedisCircuitBreaker: self._failure_count, self.recovery_timeout, ) + if self._on_open is not None: + self._on_open() self._state = self.OPEN def record_success(self) -> None: @@ -338,6 +348,7 @@ class RedisCache(BaseCache): redis_kwargs.update(kwargs) self.redis_client = get_redis_client(**redis_kwargs) self.redis_async_client: async_redis_client | async_redis_cluster_client | None = None + self._async_client_generation = 0 self.redis_kwargs = redis_kwargs self.async_redis_conn_pool = get_redis_connection_pool(**redis_kwargs) @@ -360,6 +371,8 @@ class RedisCache(BaseCache): failure_threshold=REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD, recovery_timeout=REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT, enabled=REDIS_CIRCUIT_BREAKER_ENABLED, + on_open=self._evict_redis_clients, + on_probe=self._evict_redis_clients, ) self._setup_health_pings() @@ -437,7 +450,29 @@ class RedisCache(BaseCache): sorted_kwargs: Final = sorted(self.redis_kwargs.items()) kwargs_str: Final = json.dumps(sorted_kwargs, sort_keys=True) kwargs_hash: Final = hashlib.sha256(kwargs_str.encode()).hexdigest()[:16] - return f"async-redis-client-{kwargs_hash}" + return f"async-redis-client-{kwargs_hash}-g{self._async_client_generation}" + + def _evict_redis_clients(self) -> None: + """Drop cached clients so the next Redis call rebuilds them. + + redis-py's cluster NodesManager keeps a dead CLUSTER SLOTS map after + ElastiCache becomes unreachable. Reusing that client on HALF_OPEN + would probe the same wedged connection and never recover. Bumping + generation also invalidates Lua script executors keyed off this + cache key. Do not disconnect here — that can block the request that + just tripped the breaker. + """ + self._async_client_generation += 1 + self.redis_async_client = None + try: + from .._redis import get_redis_client + + self.redis_client = get_redis_client(**self.redis_kwargs) + except Exception: + verbose_logger.debug( + "Redis circuit breaker: sync client rebuild failed; next call will retry", + exc_info=True, + ) def init_async_client( self, diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index d48c889f75a..5824798869d 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -704,3 +704,71 @@ async def test_cluster_connect_timeout_swallowed_set_opens_breaker(): with pytest.raises(Exception, match="circuit breaker is open"): await _run_under_circuit_breaker(breaker, "async_set_cache", swallowed_cluster_timeout) + + +def test_breaker_on_open_fires_once_on_transition(): + """on_open drops the wedged cluster client; it must not fire on every later miss.""" + from litellm.caching.redis_cache import RedisCircuitBreaker + + opened = [] + cb = RedisCircuitBreaker( + failure_threshold=3, + recovery_timeout=60, + on_open=lambda: opened.append("open"), + ) + for _ in range(3): + cb.record_failure() + assert opened == ["open"] + cb.record_failure() + cb.record_failure() + assert opened == ["open"] + + +def test_breaker_on_probe_fires_when_half_open_starts(): + """HALF_OPEN rebuilds the client so CLUSTER SLOTS is rediscovered.""" + import time as time_mod + + from litellm.caching.redis_cache import RedisCircuitBreaker + + probes = [] + cb = RedisCircuitBreaker( + failure_threshold=3, + recovery_timeout=60, + on_probe=lambda: probes.append("probe"), + ) + cb._state = cb.OPEN + cb._opened_at = time_mod.time() - 9999 + assert cb.is_open() is False + assert cb._state == cb.HALF_OPEN + assert probes == ["probe"] + assert cb.is_open() is True + assert probes == ["probe"] + + +def test_redis_cache_evicts_clients_when_breaker_opens(monkeypatch, redis_no_ping): + """A HALF_OPEN probe must not reuse the NodesManager that just hung.""" + import time as time_mod + + monkeypatch.setenv("REDIS_HOST", "127.0.0.1") + monkeypatch.delenv("REDIS_URL", raising=False) + cache = RedisCache(host="127.0.0.1", port=1, socket_timeout=0.1) + key_before = cache._get_async_client_cache_key() + gen_before = cache._async_client_generation + sentinel = object() + cache.redis_async_client = sentinel + + for _ in range(cache._circuit_breaker.failure_threshold): + cache._circuit_breaker.record_failure() + + assert cache._async_client_generation == gen_before + 1 + assert cache.redis_async_client is None + assert cache._get_async_client_cache_key() != key_before + + cache._circuit_breaker.record_failure() + assert cache._async_client_generation == gen_before + 1 + + cache.redis_async_client = sentinel + cache._circuit_breaker._opened_at = time_mod.time() - 9999 + assert cache._circuit_breaker.is_open() is False + assert cache.redis_async_client is None + assert cache._async_client_generation == gen_before + 2