fix(router): keep the routing and budget sync loops quiet while the Redis breaker is open

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):

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,

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 == []

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