feat(rate-limit): atomic check-and-increment-by-N for multi-process safety

The previous fix for the TOCTOU bypass relied on a per-instance asyncio.Lock,
which closed the window only within a single proxy worker. Multi-replica
deployments still raced across processes — A and B both read counter=99,
both passed validation, both incremented to 100/100 → effective limit doubled.

Add `CHECK_AND_INCREMENT_BY_N_SCRIPT` Lua script that processes any number of
(window_key, counter_key, limit, increment, ttl) descriptors atomically with
all-or-nothing semantics: if any descriptor would exceed its limit, no counter
is modified and the script returns OVER_LIMIT with the offending descriptor's
state. When Redis isn't configured, the in-memory fallback uses the existing
asyncio.Lock for single-process atomicity.

Expose this as `_PROXY_MaxParallelRequestsHandler_v3.atomic_check_and_increment_by_n`
and rewire both call sites:

- batch_rate_limiter._check_and_increment_batch_counters: replace the
  read_only=True check + separate async_increment_tokens_with_ttl_preservation
  with a single atomic call passing the batch's (request_count, total_tokens)
  as the increment.
- dynamic_rate_limiter_v3._check_rate_limits: bundle model_saturation_check
  (always enforced) and priority_model (enforced only when saturated) into
  one atomic call. When priority is unenforced, increment its counter via
  the existing should_rate_limit(read_only=False) path for tracking only.

Update structural regression tests to assert the new atomic path is used
rather than the legacy two-phase pattern.

Tests: 4/4 TOCTOU tests pass, 59 existing rate-limiter tests pass, no
regressions.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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,

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,

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

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}"
     )