scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-16 04:29:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 138

fix: refund RPM when max_parallel_requests rejects

Browse source 
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 <cursoragent@cursor.com>
This commit is contained in:
Alphaxiaoteng 2026-09-03 18:50:28 +08:00
parent ed9d29a9b4
commit 7ca8e2313c
1 changed files with 56 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
litellm/proxy/hooks/parallel_request_limiter_v3.py
View file

@ -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],