From 7ca8e2313c76ea63510f0b64653ebbcf4e829b96 Mon Sep 17 00:00:00 2001 From: Alphaxiaoteng <230277249+Alphaxiaoteng@users.noreply.github.com> Date: Thu, 3 Sep 2026 18:50:28 +0800 Subject: [PATCH 1/2] fix: refund RPM when max_parallel_requests rejects Windowed request counters were incremented before the concurrency gauge. Refund those increments when the parallel slot cannot be acquired so a rejected request does not burn RPM quota (#39309). Co-authored-by: Cursor --- .../hooks/parallel_request_limiter_v3.py | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/litellm/proxy/hooks/parallel_request_limiter_v3.py b/litellm/proxy/hooks/parallel_request_limiter_v3.py index 63129602082..d07d6b37ac0 100644 --- a/litellm/proxy/hooks/parallel_request_limiter_v3.py +++ b/litellm/proxy/hooks/parallel_request_limiter_v3.py @@ -1355,6 +1355,18 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): parent_otel_span=parent_otel_span, read_only=read_only, ) + # Windowed RPM/TPM counters are incremented before the concurrency + # gauge. A request that never acquires a parallel slot must not keep + # those increments, otherwise rejected traffic burns RPM (#39309). + if ( + not read_only + and gauge_response["overall_code"] == "OVER_LIMIT" + and keys_to_fetch + ): + await self._refund_windowed_increments( + keys_to_fetch=keys_to_fetch, + parent_otel_span=parent_otel_span, + ) return RateLimitResponse( overall_code=gauge_response["overall_code"], statuses=[*windowed_response["statuses"], *gauge_response["statuses"]], @@ -1865,6 +1877,50 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): "Failed to refund %s on cross-descriptor rollback: %s", entry["counter_key"], e ) + async def _refund_windowed_increments( + self, + keys_to_fetch: list[str], + parent_otel_span: Span | None = None, + ) -> None: + """ + Undo the +1 windowed request/token increments from this admission + attempt after max_parallel_requests rejects the call. + """ + for i in range(0, len(keys_to_fetch), 2): + counter_key = keys_to_fetch[i + 1] + try: + current: CacheCounterValue | None = await self.internal_usage_cache.async_get_cache( + key=counter_key, + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + if current is None: + continue + new_value = max(0, int(current) - 1) + await self.internal_usage_cache.async_set_cache( + key=counter_key, + value=new_value, + ttl=self.window_size, + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + redis_cache = self.internal_usage_cache.dual_cache.redis_cache + if redis_cache is not None: + try: + await redis_cache.async_increment(key=counter_key, value=-1) + except Exception as redis_err: # noqa: BLE001 + verbose_proxy_logger.warning( + "Failed to refund Redis windowed counter %s: %s", + counter_key, + redis_err, + ) + except Exception as e: # noqa: BLE001 + verbose_proxy_logger.warning( + "Failed to refund windowed counter %s after parallel reject: %s", + counter_key, + e, + ) + def _build_atomic_response( self, raw: list[CacheCounterValue], From 96563e947c25b978a2f4c022050488571928e936 Mon Sep 17 00:00:00 2001 From: Alphaxiaoteng <230277249+Alphaxiaoteng@users.noreply.github.com> Date: Thu, 3 Sep 2026 18:51:11 +0800 Subject: [PATCH 2/2] test: cover RPM refund after parallel reject Co-authored-by: Cursor --- .../test_parallel_reject_refunds_rpm.py | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 tests/local_testing/test_parallel_reject_refunds_rpm.py diff --git a/tests/local_testing/test_parallel_reject_refunds_rpm.py b/tests/local_testing/test_parallel_reject_refunds_rpm.py new file mode 100644 index 00000000000..8d3b897c25f --- /dev/null +++ b/tests/local_testing/test_parallel_reject_refunds_rpm.py @@ -0,0 +1,47 @@ +"""Regression for #39309: parallel rejects must not burn RPM quota.""" + +from pathlib import Path + +import pytest + + +def test_should_rate_limit_refunds_on_parallel_over_limit(): + source = Path("litellm/proxy/hooks/parallel_request_limiter_v3.py").read_text() + assert "_refund_windowed_increments" in source + assert 'gauge_response["overall_code"] == "OVER_LIMIT"' in source + gauge_idx = source.index("await self._check_parallel_request_gauges") + refund_idx = source.index("await self._refund_windowed_increments", gauge_idx) + return_idx = source.index( + "return RateLimitResponse(\n overall_code=gauge_response", + gauge_idx, + ) + assert gauge_idx < refund_idx < return_idx + + +@pytest.mark.asyncio +async def test_refund_windowed_increments_decrements_counter(): + from unittest.mock import AsyncMock, MagicMock + + from litellm.proxy.hooks import parallel_request_limiter_v3 as mod + + stub = MagicMock() + stub.window_size = 60 + stub.internal_usage_cache = MagicMock() + stub.internal_usage_cache.dual_cache.redis_cache = None + store = {"{api_key:k}:requests": 2} + + async def get_cache(key, litellm_parent_otel_span=None, local_only=False): + return store.get(key) + + async def set_cache(key, value, ttl=None, litellm_parent_otel_span=None, local_only=False): + store[key] = value + + stub.internal_usage_cache.async_get_cache = AsyncMock(side_effect=get_cache) + stub.internal_usage_cache.async_set_cache = AsyncMock(side_effect=set_cache) + + await mod._PROXY_MaxParallelRequestsHandler_v3._refund_windowed_increments( + stub, + keys_to_fetch=["{api_key:k}:window", "{api_key:k}:requests"], + parent_otel_span=None, + ) + assert store["{api_key:k}:requests"] == 1