Merge 96563e947c into d8c0e2c715

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -1676,6 +1676,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"]],
@@ -2380,6 +2392,50 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
                     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],

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