diff --git a/litellm/proxy/hooks/batch_rate_limiter.py b/litellm/proxy/hooks/batch_rate_limiter.py index fafe8555405..ffbcfc0f445 100644 --- a/litellm/proxy/hooks/batch_rate_limiter.py +++ b/litellm/proxy/hooks/batch_rate_limiter.py @@ -17,7 +17,7 @@ Quick summary: - async_log_success_event() fires on GET /v1/batches/{id} (batch completion) """ -from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union +from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union, cast from fastapi import HTTPException from pydantic import BaseModel @@ -164,18 +164,13 @@ class _PROXY_BatchRateLimiter(CustomLogger): batch_usage: BatchFileUsage, ) -> None: """ - Check rate limits and increment counters by the batch amounts. + Atomically check + increment rate-limit counters by the batch amounts. - Raises HTTPException if any limit would be exceeded. - - Holds the limiter's check-and-increment lock across the read-only - check and the increment to prevent concurrent batches from each - observing the same pre-increment state and collectively exceeding - the limit (TOCTOU). + Raises HTTPException if any descriptor would exceed its limit; in that + case no counter is modified. Backed by `atomic_check_and_increment_by_n` + which uses a Redis Lua script when available (multi-process atomic) and + falls back to a per-process asyncio.Lock + in-memory operation. """ - from litellm.types.caching import RedisPipelineIncrementOperation - - # Create descriptors and check if batch would exceed limits descriptors = self.parallel_request_limiter._create_rate_limit_descriptors( user_api_key_dict=user_api_key_dict, data=data, @@ -184,75 +179,35 @@ class _PROXY_BatchRateLimiter(CustomLogger): model_has_failures=False, ) - async with self.parallel_request_limiter._check_and_increment_lock: - # Check current usage without incrementing - rate_limit_response = await self.parallel_request_limiter.should_rate_limit( + increments = cast( + List[Dict[Literal["requests", "tokens"], int]], + [ + { + "requests": batch_usage.request_count, + "tokens": batch_usage.total_tokens, + } + for _ in descriptors + ], + ) + + rate_limit_response = ( + await self.parallel_request_limiter.atomic_check_and_increment_by_n( descriptors=descriptors, + increments=increments, parent_otel_span=user_api_key_dict.parent_otel_span, - read_only=True, ) + ) - # Verify batch won't exceed any limits + if rate_limit_response["overall_code"] == "OVER_LIMIT": for status in rate_limit_response["statuses"]: - rate_limit_type = status["rate_limit_type"] - limit_remaining = status["limit_remaining"] - - required_capacity = ( - batch_usage.request_count - if rate_limit_type == "requests" - else batch_usage.total_tokens if rate_limit_type == "tokens" else 0 - ) - - if required_capacity > limit_remaining: + if status["code"] == "OVER_LIMIT": self._raise_rate_limit_error( - status, descriptors, batch_usage, rate_limit_type + status, + descriptors, + batch_usage, + status["rate_limit_type"], ) - # Build pipeline operations for batch increments - # Reuse the same keys that descriptors check - pipeline_operations: List[RedisPipelineIncrementOperation] = [] - - for descriptor in descriptors: - key = descriptor["key"] - value = descriptor["value"] - rate_limit = descriptor.get("rate_limit") - - if rate_limit is None: - continue - - # Add RPM increment if limit is set - if rate_limit.get("requests_per_unit") is not None: - rpm_key = self.parallel_request_limiter.create_rate_limit_keys( - key=key, value=value, rate_limit_type="requests" - ) - pipeline_operations.append( - RedisPipelineIncrementOperation( - key=rpm_key, - increment_value=batch_usage.request_count, - ttl=self.parallel_request_limiter.window_size, - ) - ) - - # Add TPM increment if limit is set - if rate_limit.get("tokens_per_unit") is not None: - tpm_key = self.parallel_request_limiter.create_rate_limit_keys( - key=key, value=value, rate_limit_type="tokens" - ) - pipeline_operations.append( - RedisPipelineIncrementOperation( - key=tpm_key, - increment_value=batch_usage.total_tokens, - ttl=self.parallel_request_limiter.window_size, - ) - ) - - # Execute increments - if pipeline_operations: - await self.parallel_request_limiter.async_increment_tokens_with_ttl_preservation( - pipeline_operations=pipeline_operations, - parent_otel_span=user_api_key_dict.parent_otel_span, - ) - async def count_input_file_usage( self, file_id: str, diff --git a/litellm/proxy/hooks/dynamic_rate_limiter_v3.py b/litellm/proxy/hooks/dynamic_rate_limiter_v3.py index 4c0ea89554b..d827cf3067c 100644 --- a/litellm/proxy/hooks/dynamic_rate_limiter_v3.py +++ b/litellm/proxy/hooks/dynamic_rate_limiter_v3.py @@ -4,7 +4,7 @@ Dynamic rate limiter v3 - Saturation-aware priority-based rate limiting import os from datetime import datetime -from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Union +from typing import TYPE_CHECKING, Callable, Dict, List, Literal, Optional, Union from fastapi import HTTPException @@ -460,101 +460,90 @@ class _PROXY_DynamicRateLimitHandlerV3(CustomLogger): if priority_descriptors: descriptors_to_check.extend(priority_descriptors) - # Phases 1-3 must run as a single atomic critical section. Without - # this lock, concurrent requests all observe the same Phase 1 state, - # all pass enforcement, then all increment in Phase 3 — bypassing - # the limit (TOCTOU). Multi-replica deployments additionally rely on - # Redis Lua atomicity for cross-process safety. - async with self.v3_limiter._check_and_increment_lock: - # PHASE 1: Read-only check of ALL limits (no increments) - check_response = await self.v3_limiter.should_rate_limit( - descriptors=descriptors_to_check, - parent_otel_span=user_api_key_dict.parent_otel_span, - read_only=True, # CRITICAL: Don't increment counters yet - ) + # Atomic check-and-increment for the ENFORCED descriptor set: + # - model_saturation_check is always enforced + # - priority_model is enforced only when saturation crosses threshold + # + # Backed by a Redis Lua script (multi-process atomic) with an + # asyncio.Lock + in-memory fallback for single-process deployments. + # All-or-nothing: if any enforced descriptor would exceed its limit, + # no counter is modified and the response carries "OVER_LIMIT". + enforced_descriptors: List[RateLimitDescriptor] = [model_wide_descriptor] + if priority_descriptors and should_enforce_priority: + enforced_descriptors.extend(priority_descriptors) - verbose_proxy_logger.debug( - f"Read-only check: {json.dumps(check_response, indent=2)}" - ) + per_request_increment: Dict[Literal["requests", "tokens"], int] = { + "requests": 1, + "tokens": 0, + } + atomic_response = await self.v3_limiter.atomic_check_and_increment_by_n( + descriptors=enforced_descriptors, + increments=[per_request_increment for _ in enforced_descriptors], + parent_otel_span=user_api_key_dict.parent_otel_span, + ) - # PHASE 2: Decide which limits to enforce - if check_response["overall_code"] == "OVER_LIMIT": - for status in check_response["statuses"]: - if status["code"] == "OVER_LIMIT": - descriptor_key = status["descriptor_key"] + verbose_proxy_logger.debug( + f"Atomic check+increment response: {json.dumps(atomic_response, indent=2)}" + ) - # Model-wide limit exceeded (ALWAYS enforce) - if descriptor_key == "model_saturation_check": - raise HTTPException( - status_code=429, - detail={ - "error": f"Model capacity reached for {model}. " - f"Priority: {priority}, " - f"Rate limit type: {status['rate_limit_type']}, " - f"Remaining: {status['limit_remaining']}" - }, - headers={ - "retry-after": str(self.v3_limiter.window_size), - "rate_limit_type": str(status["rate_limit_type"]), - "x-litellm-priority": priority or "default", - }, - ) + if atomic_response["overall_code"] == "OVER_LIMIT": + for status in atomic_response["statuses"]: + if status["code"] != "OVER_LIMIT": + continue + descriptor_key = status["descriptor_key"] + if descriptor_key == "model_saturation_check": + raise HTTPException( + status_code=429, + detail={ + "error": f"Model capacity reached for {model}. " + f"Priority: {priority}, " + f"Rate limit type: {status['rate_limit_type']}, " + f"Remaining: {status['limit_remaining']}" + }, + headers={ + "retry-after": str(self.v3_limiter.window_size), + "rate_limit_type": str(status["rate_limit_type"]), + "x-litellm-priority": priority or "default", + }, + ) + if descriptor_key == "priority_model": + verbose_proxy_logger.debug( + f"Enforcing priority limits for {model}, saturation: {saturation:.1%}, " + f"priority: {priority}" + ) + raise HTTPException( + status_code=429, + detail={ + "error": f"Priority-based rate limit exceeded. " + f"Priority: {priority}, " + f"Rate limit type: {status['rate_limit_type']}, " + f"Remaining: {status['limit_remaining']}, " + f"Model saturation: {saturation:.1%}" + }, + headers={ + "retry-after": str(self.v3_limiter.window_size), + "rate_limit_type": str(status["rate_limit_type"]), + "x-litellm-priority": priority or "default", + "x-litellm-saturation": f"{saturation:.2%}", + }, + ) - # Priority limit exceeded (ONLY enforce when saturated) - elif ( - descriptor_key == "priority_model" - and should_enforce_priority - ): - verbose_proxy_logger.debug( - f"Enforcing priority limits for {model}, saturation: {saturation:.1%}, " - f"priority: {priority}" - ) - raise HTTPException( - status_code=429, - detail={ - "error": f"Priority-based rate limit exceeded. " - f"Priority: {priority}, " - f"Rate limit type: {status['rate_limit_type']}, " - f"Remaining: {status['limit_remaining']}, " - f"Model saturation: {saturation:.1%}" - }, - headers={ - "retry-after": str(self.v3_limiter.window_size), - "rate_limit_type": str(status["rate_limit_type"]), - "x-litellm-priority": priority or "default", - "x-litellm-saturation": f"{saturation:.2%}", - }, - ) - - # PHASE 3: Increment counters separately to avoid early-exit issues - # Model counter must ALWAYS increment, but priority counter might be over limit - # If we increment them together, v3_limiter's in-memory check will exit early - # and skip incrementing the model counter - - # Step 3a: Increment model-wide counter (always) - model_increment_response = await self.v3_limiter.should_rate_limit( - descriptors=[model_wide_descriptor], + # If priority is NOT enforced (saturation below threshold) but + # priority_descriptors exist, increment them for tracking only — no + # check, no rollback. This matches the prior tracking semantics. + if priority_descriptors and not should_enforce_priority: + priority_tracking_response = await self.v3_limiter.should_rate_limit( + descriptors=priority_descriptors, parent_otel_span=user_api_key_dict.parent_otel_span, read_only=False, ) - - # Step 3b: Increment priority counter (may be over limit, but we still track it) - if priority_descriptors: - priority_increment_response = await self.v3_limiter.should_rate_limit( - descriptors=priority_descriptors, - parent_otel_span=user_api_key_dict.parent_otel_span, - read_only=False, - ) - - # Combine responses for post-call hook - combined_response = { - "overall_code": model_increment_response["overall_code"], - "statuses": model_increment_response["statuses"] - + priority_increment_response["statuses"], - } - data["litellm_proxy_rate_limit_response"] = combined_response - else: - data["litellm_proxy_rate_limit_response"] = model_increment_response + data["litellm_proxy_rate_limit_response"] = { + "overall_code": atomic_response["overall_code"], + "statuses": atomic_response["statuses"] + + priority_tracking_response["statuses"], + } + else: + data["litellm_proxy_rate_limit_response"] = atomic_response async def async_pre_call_hook( self, diff --git a/litellm/proxy/hooks/parallel_request_limiter_v3.py b/litellm/proxy/hooks/parallel_request_limiter_v3.py index 5fea31608fa..9de14f55c94 100644 --- a/litellm/proxy/hooks/parallel_request_limiter_v3.py +++ b/litellm/proxy/hooks/parallel_request_limiter_v3.py @@ -81,6 +81,85 @@ end return results """ +CHECK_AND_INCREMENT_BY_N_SCRIPT = """ +-- Atomic check-and-increment-by-N across one or more descriptors. +-- All-or-nothing: if any descriptor would exceed its limit, no counter is +-- modified. +-- +-- KEYS layout: pairs of (window_key, counter_key), one pair per descriptor. +-- ARGV layout: +-- ARGV[1] = now (unix seconds) +-- ARGV[2] = window_size (seconds) +-- For each descriptor i (1..N), starting at ARGV[3]: +-- ARGV[3 + (i-1)*3 + 0] = limit +-- ARGV[3 + (i-1)*3 + 1] = increment +-- ARGV[3 + (i-1)*3 + 2] = ttl (counter TTL when window resets) +-- +-- Return on success: { 0, new_counter_1, new_counter_2, ... } +-- Return on over-limit: { 1, descriptor_index, current_counter, limit } +local now = tonumber(ARGV[1]) +local window_size = tonumber(ARGV[2]) +local descriptor_count = #KEYS / 2 + +-- Pass 1: read state, validate. Abort without writing if any over limit. +local descriptor_state = {} +for i = 1, descriptor_count do + local window_key = KEYS[(i - 1) * 2 + 1] + local counter_key = KEYS[(i - 1) * 2 + 2] + local arg_base = 3 + (i - 1) * 3 + local limit = tonumber(ARGV[arg_base]) + local increment = tonumber(ARGV[arg_base + 1]) + + local window_start = redis.call('GET', window_key) + local window_expired = (not window_start) or + ((now - tonumber(window_start)) >= window_size) + + local current_counter + if window_expired then + current_counter = 0 + else + current_counter = tonumber(redis.call('GET', counter_key) or 0) + end + + if current_counter + increment > limit then + return { 1, i, current_counter, limit } + end + + descriptor_state[i] = { window_expired, current_counter } +end + +-- Pass 2: all checks passed. Apply increments. +local results = { 0 } +for i = 1, descriptor_count do + local window_key = KEYS[(i - 1) * 2 + 1] + local counter_key = KEYS[(i - 1) * 2 + 2] + local arg_base = 3 + (i - 1) * 3 + local increment = tonumber(ARGV[arg_base + 1]) + local ttl = tonumber(ARGV[arg_base + 2]) + + local window_expired = descriptor_state[i][1] + + if window_expired then + redis.call('SET', window_key, tostring(now)) + redis.call('SET', counter_key, increment) + redis.call('EXPIRE', window_key, window_size) + if ttl > 0 then + redis.call('EXPIRE', counter_key, ttl) + end + table.insert(results, increment) + else + local new_counter = redis.call('INCRBY', counter_key, increment) + local current_ttl = redis.call('TTL', counter_key) + if current_ttl == -1 and ttl > 0 then + redis.call('EXPIRE', counter_key, ttl) + end + table.insert(results, new_counter) + end +end + +return results +""" + TOKEN_INCREMENT_SCRIPT = """ local results = {} @@ -163,9 +242,15 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): TOKEN_INCREMENT_SCRIPT ) ) + self.check_and_increment_by_n_script = ( + self.internal_usage_cache.dual_cache.redis_cache.async_register_script( + CHECK_AND_INCREMENT_BY_N_SCRIPT + ) + ) else: self.batch_rate_limiter_script = None self.token_increment_script = None + self.check_and_increment_by_n_script = None self.window_size = int(os.getenv("LITELLM_RATE_LIMIT_WINDOW_SIZE", 60)) @@ -595,6 +680,233 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): ) return rate_limit_response + async def atomic_check_and_increment_by_n( + self, + descriptors: List[RateLimitDescriptor], + increments: List[Dict[Literal["requests", "tokens"], int]], + parent_otel_span: Optional[Span] = None, + ) -> RateLimitResponse: + """ + Atomic check-and-increment-by-N across one or more descriptors. + + All-or-nothing: if any descriptor would exceed its limit, no counter is + modified and the response carries `overall_code = "OVER_LIMIT"` with + the offending descriptor's status. Closes the TOCTOU window between + read and increment in both single-process and multi-process (Redis) + deployments. + + Args: + descriptors: rate-limit descriptors to check + increments: per-descriptor increment amounts, indexed parallel to + `descriptors`. Each entry is `{"requests": int, "tokens": int}` + — values default to 0 when a descriptor has no matching limit. + + Returns: + RateLimitResponse with one status per (descriptor, rate_limit_type) + counter, mirroring `should_rate_limit`'s shape. + """ + if len(descriptors) != len(increments): + raise ValueError( + "atomic_check_and_increment_by_n: descriptors and increments " + "must have the same length" + ) + + keys: List[str] = [] + per_counter_meta: List[Dict[str, Any]] = [] + script_args: List[Any] = [] + + for descriptor, increment_amounts in zip(descriptors, increments): + descriptor_key = descriptor["key"] + descriptor_value = descriptor["value"] + rate_limit: RateLimitDescriptorRateLimitObject = ( + descriptor.get("rate_limit") or RateLimitDescriptorRateLimitObject() + ) + window_size = rate_limit.get("window_size") or self.window_size + window_key = f"{{{descriptor_key}:{descriptor_value}}}:window" + + for rate_limit_type in ("requests", "tokens"): + rlt: Literal["requests", "tokens"] = cast( + Literal["requests", "tokens"], rate_limit_type + ) + if rlt == "requests": + limit_value = rate_limit.get("requests_per_unit") + inc_amount = int(increment_amounts.get("requests", 0) or 0) + else: + limit_value = rate_limit.get("tokens_per_unit") + inc_amount = int(increment_amounts.get("tokens", 0) or 0) + if limit_value is None or inc_amount <= 0: + continue + counter_key = self.create_rate_limit_keys( + descriptor_key, descriptor_value, rlt + ) + keys.extend([window_key, counter_key]) + script_args.extend([int(limit_value), inc_amount, int(window_size)]) + per_counter_meta.append( + { + "descriptor_key": descriptor_key, + "current_limit": int(limit_value), + "rate_limit_type": rlt, + "window_key": window_key, + "counter_key": counter_key, + "increment": inc_amount, + "ttl": int(window_size), + } + ) + + if not keys: + return RateLimitResponse(overall_code="OK", statuses=[]) + + current_time = self._get_current_time() + now_int = int(current_time.timestamp()) + + # Multi-process atomicity via Redis Lua. Single-process atomicity + # falls back to the asyncio.Lock + in-memory sliding window below. + if self.check_and_increment_by_n_script is not None: + try: + raw = await self.check_and_increment_by_n_script( + keys=keys, + args=[now_int, self.window_size] + script_args, + ) + return self._build_atomic_response(raw, per_counter_meta) + except Exception as e: + verbose_proxy_logger.warning( + f"atomic_check_and_increment_by_n Lua failed, falling back " + f"to in-memory: {str(e)}" + ) + + async with self._check_and_increment_lock: + return await self._atomic_check_and_increment_in_memory( + per_counter_meta=per_counter_meta, + now_int=now_int, + parent_otel_span=parent_otel_span, + ) + + def _build_atomic_response( + self, + raw: List[Any], + per_counter_meta: List[Dict[str, Any]], + ) -> RateLimitResponse: + """Convert Lua script return value to RateLimitResponse.""" + if not raw: + return RateLimitResponse(overall_code="OK", statuses=[]) + + status_code = int(raw[0]) + if status_code == 1: + # Over limit: { 1, descriptor_index (1-based), current_counter, limit } + descriptor_index = int(raw[1]) - 1 + current_counter = int(raw[2]) + limit = int(raw[3]) + meta = per_counter_meta[descriptor_index] + return RateLimitResponse( + overall_code="OVER_LIMIT", + statuses=[ + RateLimitStatus( + code="OVER_LIMIT", + current_limit=limit, + limit_remaining=max(0, limit - current_counter), + rate_limit_type=meta["rate_limit_type"], + descriptor_key=meta["descriptor_key"], + ) + ], + ) + + statuses: List[RateLimitStatus] = [] + for meta, new_counter in zip(per_counter_meta, raw[1:]): + statuses.append( + RateLimitStatus( + code="OK", + current_limit=meta["current_limit"], + limit_remaining=max(0, meta["current_limit"] - int(new_counter)), + rate_limit_type=meta["rate_limit_type"], + descriptor_key=meta["descriptor_key"], + ) + ) + return RateLimitResponse(overall_code="OK", statuses=statuses) + + async def _atomic_check_and_increment_in_memory( + self, + per_counter_meta: List[Dict[str, Any]], + now_int: int, + parent_otel_span: Optional[Span] = None, + ) -> RateLimitResponse: + """In-memory all-or-nothing check-and-increment. Caller holds lock.""" + # Pass 1: read state, validate. + descriptor_state: List[Dict[str, Any]] = [] + for meta in per_counter_meta: + window_start = await self.internal_usage_cache.async_get_cache( + key=meta["window_key"], + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + window_expired = ( + window_start is None + or (now_int - int(window_start)) >= self.window_size + ) + current_counter = ( + 0 + if window_expired + else int( + await self.internal_usage_cache.async_get_cache( + key=meta["counter_key"], + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + or 0 + ) + ) + if current_counter + meta["increment"] > meta["current_limit"]: + return RateLimitResponse( + overall_code="OVER_LIMIT", + statuses=[ + RateLimitStatus( + code="OVER_LIMIT", + current_limit=meta["current_limit"], + limit_remaining=max( + 0, meta["current_limit"] - current_counter + ), + rate_limit_type=meta["rate_limit_type"], + descriptor_key=meta["descriptor_key"], + ) + ], + ) + descriptor_state.append( + {"window_expired": window_expired, "current": current_counter} + ) + + # Pass 2: apply increments. + statuses: List[RateLimitStatus] = [] + for meta, state in zip(per_counter_meta, descriptor_state): + new_counter = ( + meta["increment"] + if state["window_expired"] + else state["current"] + meta["increment"] + ) + if state["window_expired"]: + await self.internal_usage_cache.async_set_cache( + key=meta["window_key"], + value=str(now_int), + ttl=self.window_size, + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + await self.internal_usage_cache.async_set_cache( + key=meta["counter_key"], + value=new_counter, + ttl=meta["ttl"], + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + statuses.append( + RateLimitStatus( + code="OK", + current_limit=meta["current_limit"], + limit_remaining=max(0, meta["current_limit"] - new_counter), + rate_limit_type=meta["rate_limit_type"], + descriptor_key=meta["descriptor_key"], + ) + ) + return RateLimitResponse(overall_code="OK", statuses=statuses) + def create_organization_rate_limit_descriptor( self, user_api_key_dict: UserAPIKeyAuth, requested_model: Optional[str] = None ) -> List[RateLimitDescriptor]: diff --git a/tests/test_litellm/proxy/hooks/test_rate_limiter_toctou.py b/tests/test_litellm/proxy/hooks/test_rate_limiter_toctou.py index 006dac04498..076cb9d3f13 100644 --- a/tests/test_litellm/proxy/hooks/test_rate_limiter_toctou.py +++ b/tests/test_litellm/proxy/hooks/test_rate_limiter_toctou.py @@ -19,7 +19,7 @@ check-and-increment becomes atomic. import asyncio import os import sys -from typing import Any, Dict, List, Optional +from typing import List import pytest @@ -102,9 +102,7 @@ async def test_batch_limiter_concurrent_bypasses_tpm_via_toctou(): tpm_limit=TPM_LIMIT, rpm_limit=1000, ) - batch_usage = BatchFileUsage( - total_tokens=BATCH_TOKENS, request_count=1 - ) + batch_usage = BatchFileUsage(total_tokens=BATCH_TOKENS, request_count=1) barrier = _make_phase1_barrier(NUM_CONCURRENT) rate_limiter.should_rate_limit = barrier(rate_limiter.should_rate_limit) @@ -136,15 +134,13 @@ async def test_batch_limiter_concurrent_bypasses_tpm_via_toctou(): @pytest.mark.asyncio -async def test_batch_limiter_check_and_increment_is_two_separate_calls(): +async def test_batch_limiter_uses_atomic_check_and_increment(): """ - Structural test: _check_and_increment_batch_counters issues a read_only=True - check followed by a separate increment call — non-atomic by construction. + Regression test: batch limiter routes through + `atomic_check_and_increment_by_n` rather than the legacy two-phase + pattern (read_only=True check + separate async_increment_tokens_with_ttl_preservation). - Records call ordering on parallel_request_limiter to prove Phase 1 (check) - and Phase 3 (increment) are not wrapped in a single Redis transaction / - Lua script. After the fix, this pattern should be replaced with one - atomic_check_and_increment call. + Ensures future refactors don't reintroduce the TOCTOU window. """ dual_cache = DualCache() internal_usage_cache = InternalUsageCache(dual_cache=dual_cache) @@ -154,25 +150,23 @@ async def test_batch_limiter_check_and_increment_is_two_separate_calls(): batch_limiter = rate_limiter._get_batch_rate_limiter() assert batch_limiter is not None - call_log: List[Dict[str, Any]] = [] + call_log: List[str] = [] + original_atomic = rate_limiter.atomic_check_and_increment_by_n original_should = rate_limiter.should_rate_limit - original_inc = rate_limiter.async_increment_tokens_with_ttl_preservation + + async def logging_atomic(*args, **kwargs): + call_log.append("atomic_check_and_increment_by_n") + return await original_atomic(*args, **kwargs) async def logging_should(*args, **kwargs): - call_log.append( - {"method": "should_rate_limit", "read_only": kwargs.get("read_only")} - ) + call_log.append(f"should_rate_limit(read_only={kwargs.get('read_only')})") return await original_should(*args, **kwargs) - async def logging_inc(*args, **kwargs): - call_log.append({"method": "async_increment_tokens_with_ttl_preservation"}) - return await original_inc(*args, **kwargs) - + rate_limiter.atomic_check_and_increment_by_n = logging_atomic rate_limiter.should_rate_limit = logging_should - rate_limiter.async_increment_tokens_with_ttl_preservation = logging_inc user_api_key_dict = UserAPIKeyAuth( - api_key=hash_token("structural-test-key"), + api_key=hash_token("atomic-test-key"), tpm_limit=10000, rpm_limit=1000, ) @@ -183,29 +177,14 @@ async def test_batch_limiter_check_and_increment_is_two_separate_calls(): batch_usage=BatchFileUsage(total_tokens=50, request_count=1), ) - method_sequence = [c["method"] for c in call_log] - assert "should_rate_limit" in method_sequence, "Expected Phase 1 check" - phase1 = [ - c - for c in call_log - if c["method"] == "should_rate_limit" and c.get("read_only") is True - ] - phase3 = [ - c - for c in call_log - if c["method"] == "async_increment_tokens_with_ttl_preservation" - ] - assert len(phase1) >= 1 and len(phase3) >= 1, ( - f"Expected non-atomic Phase1+Phase3 pattern. call_log={call_log}" + assert "atomic_check_and_increment_by_n" in call_log, ( + f"Batch limiter must route through atomic_check_and_increment_by_n. " + f"Calls observed: {call_log}" ) - phase1_idx = method_sequence.index("should_rate_limit") - phase3_idx = method_sequence.index( - "async_increment_tokens_with_ttl_preservation" - ) - assert phase1_idx < phase3_idx, ( - "TOCTOU evidence: read-only check precedes increment as separate awaits — " - "no atomic Lua script wraps both. Sequence: " - f"{method_sequence}" + legacy_calls = [c for c in call_log if c.startswith("should_rate_limit(")] + assert not legacy_calls, ( + f"Batch limiter must not call should_rate_limit directly (legacy " + f"two-phase pattern). Observed: {legacy_calls}" ) @@ -295,15 +274,15 @@ async def test_dynamic_rate_limiter_v3_concurrent_bypasses_model_capacity(): @pytest.mark.asyncio -async def test_dynamic_rate_limiter_v3_phase1_phase3_are_separate_awaits(): +async def test_dynamic_rate_limiter_v3_uses_atomic_check_and_increment(): """ - Structural proof of TOCTOU: dynamic_rate_limiter_v3._check_rate_limits - issues a read_only=True call (Phase 1, line 464-468) followed by separate - read_only=False calls (Phase 3, lines 526-530 + 534-538). + Regression test: dynamic limiter's enforced descriptors flow through + `atomic_check_and_increment_by_n`, not the legacy + read_only=True check followed by a separate read_only=False increment. - Records each invocation of v3_limiter.should_rate_limit and asserts the - Phase1→Phase3 sequence. After fix, both phases must collapse into a single - atomic operation. + When priority is enforced (saturation >= threshold), priority_model is + bundled into the atomic call alongside model_saturation_check. When not + enforced, priority counter is incremented for tracking only. """ os.environ["LITELLM_LICENSE"] = "test-license-key" litellm.priority_reservation = {"high": 0.9, "low": 0.1} @@ -311,7 +290,7 @@ async def test_dynamic_rate_limiter_v3_phase1_phase3_are_separate_awaits(): dual_cache = DualCache() handler = DynamicRateLimitHandler(internal_usage_cache=dual_cache) - model = "structural-dyn-model" + model = "atomic-dyn-model" llm_router = Router( model_list=[ { @@ -327,18 +306,19 @@ async def test_dynamic_rate_limiter_v3_phase1_phase3_are_separate_awaits(): ) handler.update_variables(llm_router=llm_router) - read_only_flags: List[Optional[bool]] = [] - original = handler.v3_limiter.should_rate_limit + atomic_descriptors_observed: List[List[str]] = [] + original_atomic = handler.v3_limiter.atomic_check_and_increment_by_n - async def logging_should(*args, **kwargs): - read_only_flags.append(kwargs.get("read_only")) - return await original(*args, **kwargs) + async def logging_atomic(*args, **kwargs): + ds = kwargs.get("descriptors") or (args[0] if args else []) + atomic_descriptors_observed.append([d["key"] for d in ds]) + return await original_atomic(*args, **kwargs) - handler.v3_limiter.should_rate_limit = logging_should + handler.v3_limiter.atomic_check_and_increment_by_n = logging_atomic from litellm.types.router import ModelGroupInfo - user = UserAPIKeyAuth(api_key=hash_token("dyn-structural-key")) + user = UserAPIKeyAuth(api_key=hash_token("dyn-atomic-key")) user.metadata = {"priority": "high"} await handler._check_rate_limits( @@ -355,21 +335,12 @@ async def test_dynamic_rate_limiter_v3_phase1_phase3_are_separate_awaits(): data={}, ) - assert True in read_only_flags or any( - f is True for f in read_only_flags - ), f"Expected read_only=True (Phase 1) call. Got: {read_only_flags}" - assert any(f is False for f in read_only_flags), ( - f"Expected read_only=False (Phase 3) call. Got: {read_only_flags}" + assert atomic_descriptors_observed, ( + "Dynamic limiter must route enforced descriptors through " + "atomic_check_and_increment_by_n (no legacy read_only=True / " + "separate-increment pattern)." ) - - first_read_only = next( - (i for i, f in enumerate(read_only_flags) if f is True), None - ) - first_write = next( - (i for i, f in enumerate(read_only_flags) if f is False), None - ) - assert first_read_only is not None and first_write is not None - assert first_read_only < first_write, ( - f"TOCTOU evidence: Phase 1 (read_only) precedes Phase 3 (increment) " - f"as separate non-atomic calls. read_only sequence: {read_only_flags}" + assert "model_saturation_check" in atomic_descriptors_observed[0], ( + f"Expected model_saturation_check in atomic descriptor set. " + f"Got: {atomic_descriptors_observed}" )