From 706134d1f74401a6e70f80fdf317c5d2049ed7f4 Mon Sep 17 00:00:00 2001 From: amarrtech <272048731+amarrtech@users.noreply.github.com> Date: Thu, 8 Oct 2026 08:13:20 -0700 Subject: [PATCH 1/3] fix(proxy): resync config after Redis reconnect Signed-off-by: amarrtech <272048731+amarrtech@users.noreply.github.com> --- litellm/proxy/common_utils/config_sync_pubsub.py | 15 ++++++++++++--- .../proxy/common_utils/test_config_sync_pubsub.py | 12 +++++++++--- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/litellm/proxy/common_utils/config_sync_pubsub.py b/litellm/proxy/common_utils/config_sync_pubsub.py index 579db290190..aaacf4da42a 100644 --- a/litellm/proxy/common_utils/config_sync_pubsub.py +++ b/litellm/proxy/common_utils/config_sync_pubsub.py @@ -177,6 +177,7 @@ class ConfigSyncSubscriber: "_monotonic", "_redis_cache", "_resync_callbacks", + "_resync_on_subscribe", "_rng", "_sleep", "_task", @@ -197,6 +198,7 @@ class ConfigSyncSubscriber: ) -> None: self._redis_cache = redis_cache self._resync_callbacks = resync_callbacks + self._resync_on_subscribe = False self._debounce_seconds = debounce_seconds self._jitter_max_seconds = jitter_max_seconds self._min_resync_interval_seconds = min_resync_interval_seconds @@ -233,12 +235,16 @@ class ConfigSyncSubscriber: try: await pubsub.subscribe(config_sync_channel(self._redis_cache)) backoff_seconds = self._backoff_initial_seconds + if self._resync_on_subscribe: + await self._resync() + self._resync_on_subscribe = False await self._consume(pubsub) finally: await self._close_pubsub(pubsub) except asyncio.CancelledError: raise except Exception as e: # noqa: BLE001 # any redis failure falls through to backoff and reconnect + self._resync_on_subscribe = True verbose_proxy_logger.warning( "config sync subscriber redis error: %s; reconnecting in %.0fs", e, @@ -253,10 +259,13 @@ class ConfigSyncSubscriber: if message is None: continue await self._sleep(self._debounce_seconds + self._rng.uniform(0.0, self._jitter_max_seconds)) - await self._wait_for_min_resync_interval() await self._drain_pending(pubsub) - await self._run_resync_callbacks() - self._last_resync_at = self._monotonic() + await self._resync() + + async def _resync(self) -> None: + await self._wait_for_min_resync_interval() + await self._run_resync_callbacks() + self._last_resync_at = self._monotonic() async def _wait_for_min_resync_interval(self) -> None: if self._last_resync_at is None: diff --git a/tests/unit/proxy/common_utils/test_config_sync_pubsub.py b/tests/unit/proxy/common_utils/test_config_sync_pubsub.py index 83ed3afa293..c4b414f099e 100644 --- a/tests/unit/proxy/common_utils/test_config_sync_pubsub.py +++ b/tests/unit/proxy/common_utils/test_config_sync_pubsub.py @@ -493,15 +493,21 @@ async def test_second_start_does_not_open_a_second_subscription() -> None: assert pubsub.subscribed_channels == [CONFIG_SYNC_CHANNEL] -async def test_redis_error_leads_to_backoff_and_resubscribe() -> None: +async def test_redis_error_resubscribes_then_resyncs_without_new_message() -> None: broken = _BrokenPubSub() - healthy = _QueuePubSub(initial_messages=[json.dumps({"object_type": "litellm_credentialstable"})]) + healthy = _QueuePubSub() cache = _FakeRedisCache(_ScriptedPubSubRedisClient([broken, healthy])) resyncs: List[str] = [] fired = asyncio.Event() + + async def resync() -> None: + resyncs.append("resync") + if healthy.subscribed_channels == [CONFIG_SYNC_CHANNEL]: + fired.set() + subscriber = ConfigSyncSubscriber( redis_cache=cache, - resync_callbacks=(_recording_callback(resyncs, "resync", fired),), + resync_callbacks=(resync,), debounce_seconds=0.01, jitter_max_seconds=0.0, backoff_initial_seconds=0.02, From 9c7ab15f153ac203ed81542c3401ca47837ecee2 Mon Sep 17 00:00:00 2001 From: amarrtech <272048731+amarrtech@users.noreply.github.com> Date: Thu, 8 Oct 2026 09:03:36 -0700 Subject: [PATCH 2/3] fix(proxy): coalesce reconnect resync writes Signed-off-by: amarrtech <272048731+amarrtech@users.noreply.github.com> --- .../proxy/common_utils/config_sync_pubsub.py | 7 +-- .../common_utils/test_config_sync_pubsub.py | 46 +++++++++++++++++++ 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/common_utils/config_sync_pubsub.py b/litellm/proxy/common_utils/config_sync_pubsub.py index aaacf4da42a..e3de9809f02 100644 --- a/litellm/proxy/common_utils/config_sync_pubsub.py +++ b/litellm/proxy/common_utils/config_sync_pubsub.py @@ -259,11 +259,12 @@ class ConfigSyncSubscriber: if message is None: continue await self._sleep(self._debounce_seconds + self._rng.uniform(0.0, self._jitter_max_seconds)) - await self._drain_pending(pubsub) - await self._resync() + await self._resync(pubsub) - async def _resync(self) -> None: + async def _resync(self, pubsub: ConfigSyncPubSub | None = None) -> None: await self._wait_for_min_resync_interval() + if pubsub is not None: + await self._drain_pending(pubsub) await self._run_resync_callbacks() self._last_resync_at = self._monotonic() diff --git a/tests/unit/proxy/common_utils/test_config_sync_pubsub.py b/tests/unit/proxy/common_utils/test_config_sync_pubsub.py index c4b414f099e..28c5ac9aed1 100644 --- a/tests/unit/proxy/common_utils/test_config_sync_pubsub.py +++ b/tests/unit/proxy/common_utils/test_config_sync_pubsub.py @@ -433,6 +433,52 @@ async def test_writes_during_the_throttle_wait_collapse_into_the_next_resync() - assert pubsub.queue.empty() +async def test_write_arriving_during_throttle_wait_does_not_trigger_an_extra_resync() -> None: + pubsub = _QueuePubSub() + cache = _FakeRedisCache(_ScriptedPubSubRedisClient([pubsub])) + clock = _FakeClock() + first_resync = asyncio.Event() + second_resync = asyncio.Event() + inserted_during_wait = asyncio.Event() + resyncs: List[str] = [] + + async def recording_sleep(seconds: float) -> None: + if seconds > 0 and not inserted_during_wait.is_set(): + inserted_during_wait.set() + pubsub.queue.put_nowait("change-during-wait") + await asyncio.sleep(0) + + async def resync() -> None: + resyncs.append("resync") + if len(resyncs) == 1: + first_resync.set() + if len(resyncs) == 2: + second_resync.set() + + subscriber = ConfigSyncSubscriber( + redis_cache=cache, + resync_callbacks=(resync,), + debounce_seconds=0.0, + jitter_max_seconds=0.0, + min_resync_interval_seconds=10.0, + sleep=recording_sleep, + monotonic=clock, + ) + + subscriber.start() + pubsub.queue.put_nowait("first-change") + await asyncio.wait_for(first_resync.wait(), timeout=5) + clock.now += 4.0 + pubsub.queue.put_nowait("second-change") + await asyncio.wait_for(second_resync.wait(), timeout=5) + await asyncio.sleep(0.05) + await subscriber.stop() + + assert inserted_during_wait.is_set() + assert resyncs == ["resync", "resync"] + assert pubsub.queue.empty() + + async def test_polls_without_messages_do_not_trigger_resyncs() -> None: pubsub = _EmptyPollsThenMessagePubSub(empty_polls=3, initial_messages=["change"]) cache = _FakeRedisCache(_ScriptedPubSubRedisClient([pubsub])) From fa573674997e57bdcc70ce6d7b2e4ff097ec5e73 Mon Sep 17 00:00:00 2001 From: amarrtech <272048731+amarrtech@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:15:57 -0700 Subject: [PATCH 3/3] test(proxy): finalize reconnect coalescing Assisted-by: OpenAI Codex Signed-off-by: amarrtech <272048731+amarrtech@users.noreply.github.com> --- .../proxy/common_utils/config_sync_pubsub.py | 2 +- .../common_utils/test_config_sync_pubsub.py | 17 +++++++++-------- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/litellm/proxy/common_utils/config_sync_pubsub.py b/litellm/proxy/common_utils/config_sync_pubsub.py index e3de9809f02..3e6c3f8028a 100644 --- a/litellm/proxy/common_utils/config_sync_pubsub.py +++ b/litellm/proxy/common_utils/config_sync_pubsub.py @@ -236,7 +236,7 @@ class ConfigSyncSubscriber: await pubsub.subscribe(config_sync_channel(self._redis_cache)) backoff_seconds = self._backoff_initial_seconds if self._resync_on_subscribe: - await self._resync() + await self._resync(pubsub) self._resync_on_subscribe = False await self._consume(pubsub) finally: diff --git a/tests/unit/proxy/common_utils/test_config_sync_pubsub.py b/tests/unit/proxy/common_utils/test_config_sync_pubsub.py index 28c5ac9aed1..7c31c1aeb9f 100644 --- a/tests/unit/proxy/common_utils/test_config_sync_pubsub.py +++ b/tests/unit/proxy/common_utils/test_config_sync_pubsub.py @@ -1,6 +1,7 @@ import asyncio import json import random +import typing from typing import Callable, Coroutine, Iterable, List, Optional, Tuple from unittest.mock import AsyncMock, MagicMock, patch @@ -434,13 +435,13 @@ async def test_writes_during_the_throttle_wait_collapse_into_the_next_resync() - async def test_write_arriving_during_throttle_wait_does_not_trigger_an_extra_resync() -> None: - pubsub = _QueuePubSub() - cache = _FakeRedisCache(_ScriptedPubSubRedisClient([pubsub])) - clock = _FakeClock() - first_resync = asyncio.Event() - second_resync = asyncio.Event() - inserted_during_wait = asyncio.Event() - resyncs: List[str] = [] + pubsub: typing.Final = _QueuePubSub() + cache: typing.Final = _FakeRedisCache(_ScriptedPubSubRedisClient([pubsub])) + clock: typing.Final = _FakeClock() + first_resync: typing.Final = asyncio.Event() + second_resync: typing.Final = asyncio.Event() + inserted_during_wait: typing.Final = asyncio.Event() + resyncs: typing.Final[List[str]] = [] async def recording_sleep(seconds: float) -> None: if seconds > 0 and not inserted_during_wait.is_set(): @@ -455,7 +456,7 @@ async def test_write_arriving_during_throttle_wait_does_not_trigger_an_extra_res if len(resyncs) == 2: second_resync.set() - subscriber = ConfigSyncSubscriber( + subscriber: typing.Final = ConfigSyncSubscriber( redis_cache=cache, resync_callbacks=(resync,), debounce_seconds=0.0,