diff --git a/litellm/proxy/hooks/parallel_request_limiter_v3.py b/litellm/proxy/hooks/parallel_request_limiter_v3.py index 31437af7770..278e74a0f74 100644 --- a/litellm/proxy/hooks/parallel_request_limiter_v3.py +++ b/litellm/proxy/hooks/parallel_request_limiter_v3.py @@ -392,6 +392,29 @@ class _AsyncLuaScript(Protocol): def __call__(self, *, keys: Sequence[str], args: Sequence[object]) -> Awaitable[list[CacheCounterValue]]: ... +WindowKeyType: TypeAlias = Literal["requests", "tokens"] + + +def descriptor_window_key(descriptor_key: str, descriptor_value: str, rate_limit_type: WindowKeyType) -> str: + """ + Window-start key for a single descriptor counter. + + Each rate-limit type owns its window key. Sharing one window key between + the requests and tokens counters made a window roll on one pair leave the + sibling counter's stale previous-window value stranded under a freshly + reset window, which was then counted against the new window — guaranteed + false 429s on the first request after every window boundary (issue #24677). + With per-type windows, every tracked window rolls independently and no + branch can skip a sibling counter's reset. + """ + return f"{{{descriptor_key}:{descriptor_value}}}:window:{rate_limit_type}" + + +def legacy_descriptor_window_key(descriptor_key: str, descriptor_value: str) -> str: + """Return the shared window key used before RPM and TPM were separated.""" + return f"{{{descriptor_key}:{descriptor_value}}}:window" + + class RateLimitDescriptorRateLimitObject(TypedDict, total=False): requests_per_unit: int | None tokens_per_unit: int | None @@ -1053,6 +1076,75 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): return results + async def _backfill_legacy_window_keys( + self, + window_keys: Sequence[str], + parent_otel_span: Span | None = None, + ) -> None: + """Seed new per-type windows from the pre-split shared window key. + + The counter keys are intentionally unchanged, so copying the old + window start preserves the active window during a rolling upgrade. + Redis uses ``NX`` so concurrent replicas cannot overwrite a window + that another replica has already initialized. The legacy key remains + readable until its normal TTL expires. + """ + redis_cache = self.internal_usage_cache.dual_cache.redis_cache + for window_key in window_keys: + if not window_key.endswith((":window:requests", ":window:tokens")): + continue + + new_window_value = await self.internal_usage_cache.async_get_cache( + key=window_key, + litellm_parent_otel_span=parent_otel_span, + local_only=False, + ) + if new_window_value is not None: + continue + + legacy_window_key = f"{window_key.rsplit(':window:', 1)[0]}:window" + legacy_window_value = await self.internal_usage_cache.async_get_cache( + key=legacy_window_key, + litellm_parent_otel_span=parent_otel_span, + local_only=False, + ) + if legacy_window_value is None: + continue + + if redis_cache is None: + await self.internal_usage_cache.async_set_cache( + key=window_key, + value=legacy_window_value, + ttl=self.window_size, + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + continue + + inserted = await redis_cache.async_set_cache( + key=window_key, + value=legacy_window_value, + ttl=self.window_size, + nx=True, + parent_otel_span=parent_otel_span, + ) + current_window_value = ( + legacy_window_value + if inserted + else await redis_cache.async_get_cache( + key=window_key, + parent_otel_span=parent_otel_span, + ) + ) + if current_window_value is not None: + await self.internal_usage_cache.async_set_cache( + key=window_key, + value=current_window_value, + ttl=self.window_size, + litellm_parent_otel_span=parent_otel_span, + local_only=True, + ) + def create_rate_limit_keys( self, key: str, @@ -1071,9 +1163,17 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): keys_to_fetch: list[str], cache_values: CacheCounterValues, key_metadata: dict[str, WindowKeyMetadata], + now_int: int | None = None, ) -> RateLimitResponse: """ Check if the cache values are over the limit. + + ``now_int`` enables window-aware evaluation: a counter snapshot whose + window_start shows its window has already expired is treated as 0. + The local in-memory mirror written by a previous request can hold an + over-limit value from a window that has since rolled; rejecting on it + (before the authoritative pass rolls the window) produced 429s that + outlived their window by up to a full window_size (#24677). """ statuses: Final[list[RateLimitStatus]] = [] overall_code = "OK" @@ -1082,10 +1182,23 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): item_code = "OK" window_key = keys_to_fetch[i] counter_key = keys_to_fetch[i + 1] - counter_value = cache_values[i + 1] + counter_value: CacheCounterValue | None = cache_values[i + 1] requests_limit = key_metadata[window_key]["requests_limit"] tokens_limit = key_metadata[window_key]["tokens_limit"] + window_expired = False + if now_int is not None and counter_value is not None: + window_start = cache_values[i] + if window_start is not None: + try: + window_expired = (now_int - int(window_start)) >= key_metadata[window_key]["window_size"] + except (TypeError, ValueError): + window_expired = False + + # This counter belongs to a window that has already rolled over; + # it must not reject the request that starts the new window. + effective_counter_value: CacheCounterValue | None = 0 if window_expired else counter_value + # Determine which limit to use for current_limit and limit_remaining current_limit: int | None = None rate_limit_type: Literal["requests", "tokens", "max_parallel_requests"] | None = None @@ -1099,12 +1212,14 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if current_limit is None or rate_limit_type is None: continue - if counter_value is not None and int(counter_value) > current_limit: + if effective_counter_value is not None and int(effective_counter_value) > current_limit: overall_code = "OVER_LIMIT" item_code = "OVER_LIMIT" # Only compute limit_remaining if current_limit is not None - limit_remaining = current_limit - int(counter_value) if counter_value is not None else current_limit + limit_remaining = ( + current_limit - int(effective_counter_value) if effective_counter_value is not None else current_limit + ) statuses.append( { @@ -1280,6 +1395,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): windowed_response = RateLimitResponse(overall_code="OK", statuses=[]) if keys_to_fetch: + await self._backfill_legacy_window_keys( + window_keys=keys_to_fetch[::2], + parent_otel_span=parent_otel_span, + ) ## CHECK IN-MEMORY CACHE cache_values = await self._batch_get_counter_values( # rebind-ok: refreshed by the Redis read below when the in-memory pass is under limit keys=keys_to_fetch, @@ -1288,7 +1407,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): ) if cache_values is not None: - rate_limit_response: Final = self.is_cache_list_over_limit(keys_to_fetch, cache_values, key_metadata) + rate_limit_response: Final = self.is_cache_list_over_limit( + keys_to_fetch, cache_values, key_metadata, now_int=now_int + ) if rate_limit_response["overall_code"] == "OVER_LIMIT": return rate_limit_response @@ -1304,7 +1425,8 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): # For keys that don't exist yet, set them to 0 if cache_values is None: cache_values = [ # rebind-ok: missing keys default to a zeroed window snapshot - str(now_int) if key.endswith(":window") else 0 for key in keys_to_fetch + str(now_int) if key.endswith((":window:requests", ":window:tokens")) else 0 + for key in keys_to_fetch ] elif self.batch_rate_limiter_script is not None: # NORMAL MODE: Increment counters in Redis @@ -1342,7 +1464,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): window_size=self.window_size, ) - windowed_response = self.is_cache_list_over_limit(keys_to_fetch, cache_values, key_metadata) + windowed_response = self.is_cache_list_over_limit( + keys_to_fetch, cache_values, key_metadata, now_int=now_int + ) if windowed_response["overall_code"] == "OVER_LIMIT": return windowed_response @@ -1384,8 +1508,6 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): max_parallel_requests_limit = rate_limit.get("max_parallel_requests") window_size = rate_limit.get("window_size") or self.window_size - window_key = f"{{{descriptor_key}:{descriptor_value}}}:window" - if max_parallel_requests_limit is not None: gauges.append( ParallelRequestGauge( @@ -1398,24 +1520,46 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): ) rate_limit_set = False + requests_window_key: str | None = ( + descriptor_window_key(descriptor_key, descriptor_value, "requests") + if requests_limit is not None + else None + ) + tokens_window_key: str | None = ( + descriptor_window_key(descriptor_key, descriptor_value, "tokens") if tokens_limit is not None else None + ) if requests_limit is not None: rpm_key = self.create_rate_limit_keys(descriptor_key, descriptor_value, "requests") - keys_to_fetch.extend([window_key, rpm_key]) + if requests_window_key is not None: + keys_to_fetch.extend((requests_window_key, rpm_key)) rate_limit_set = True if tokens_limit is not None: tpm_key = self.create_rate_limit_keys(descriptor_key, descriptor_value, "tokens") - keys_to_fetch.extend([window_key, tpm_key]) + if tokens_window_key is not None: + keys_to_fetch.extend((tokens_window_key, tpm_key)) rate_limit_set = True if not rate_limit_set: continue - key_metadata[window_key] = { - "requests_limit": (int(requests_limit) if requests_limit is not None else None), - "tokens_limit": int(tokens_limit) if tokens_limit is not None else None, - "window_size": int(window_size), - "descriptor_key": descriptor_key, - } + # Per-type window metadata: each counter is evaluated against its + # own window, so the sibling limit stays None for the other type. + if requests_window_key is not None: + requests_metadata: WindowKeyMetadata = { + "requests_limit": int(requests_limit), + "tokens_limit": None, + "window_size": int(window_size), + "descriptor_key": descriptor_key, + } + key_metadata[requests_window_key] = requests_metadata + if tokens_window_key is not None: + tokens_metadata: WindowKeyMetadata = { + "requests_limit": None, + "tokens_limit": int(tokens_limit), + "window_size": int(window_size), + "descriptor_key": descriptor_key, + } + key_metadata[tokens_window_key] = tokens_metadata return keys_to_fetch, key_metadata, gauges def _gauge_status(self, gauge: ParallelRequestGauge, in_flight: int, code: str) -> RateLimitStatus: @@ -1701,6 +1845,11 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if not descriptor_groups: return RateLimitResponse(overall_code="OK", statuses=[]) + await self._backfill_legacy_window_keys( + window_keys=[meta["window_key"] for _keys, _args, group_meta in descriptor_groups for meta in group_meta], + parent_otel_span=parent_otel_span, + ) + # Multi-process atomicity via Redis Lua, per descriptor for slot # co-location. Single-process atomicity falls back to the # asyncio.Lock + in-memory sliding window below — there are no @@ -1736,7 +1885,6 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): descriptor.get("rate_limit") or RateLimitDescriptorRateLimitObject() ) window_size: Final = rate_limit.get("window_size") or self.window_size - window_key: Final = f"{{{descriptor_key}:{descriptor_value}}}:window" keys: Final[list[str]] = [] args: Final[list[int]] = [] @@ -1753,6 +1901,11 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): if limit_value is None or inc_amount < 0: continue counter_key = self.create_rate_limit_keys(descriptor_key, descriptor_value, rlt) + # Per-type window key: the requests and tokens counters roll their + # windows independently, so a roll on one can never leave the + # sibling counter's stale value stranded under a fresh window + # (issue #24677). + window_key = descriptor_window_key(descriptor_key, descriptor_value, rlt) # Counter-key TTL and window_size are conceptually distinct # ("how long the counter Redis key lives" vs "how long the # sliding window is"). Kept as separate values so a future @@ -4285,7 +4438,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger): key=counter_key, increment_value=increment, ttl=self.window_size, - window_key=f"{{{scope_key}:{scope_value}}}:window", + # The guarded increment is tokens-only, so it is pinned to the + # tokens window that reserve_io_tokens observed. + window_key=descriptor_window_key(scope_key, scope_value, "tokens"), expected_window_start=window_identity[0], reservation_backend=window_identity[1], ) diff --git a/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py b/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py index 4003286d887..9955c6440f4 100644 --- a/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py +++ b/tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py @@ -29,6 +29,8 @@ from litellm.proxy.hooks.parallel_request_limiter_v3 import ( ) from litellm.proxy.hooks.parallel_request_limiter_v3 import ( _PROXY_MaxParallelRequestsHandler_v3 as _PROXY_MaxParallelRequestsHandler, + descriptor_window_key, + legacy_descriptor_window_key, ) from litellm.proxy.utils import InternalUsageCache, ProxyLogging, hash_token from litellm.types.caching import RedisPipelineIncrementOperation @@ -255,7 +257,7 @@ async def test_rate_limiter_script_return_values_v3(monkeypatch, time_controller ) # Verify both counter and window values are stored in cache - window_key = f"{{api_key:{_api_key}}}:window" + window_key = f"{{api_key:{_api_key}}}:window:requests" counter_key = f"{{api_key:{_api_key}}}:requests" window_value = await local_cache.async_get_cache(key=window_key) @@ -6171,3 +6173,257 @@ async def test_success_hook_leaves_stash_untouched_for_non_batch_responses(): data={}, user_api_key_dict=user, response=ModelResponse(usage=Usage(total_tokens=5)) ) assert get_request_stash().batch_enqueued_reservation == reservation + + +############################################################ +# Issue #24677 regression tests: false 429s at window boundaries. +# +# Root cause: the requests and tokens counters of one descriptor shared a +# single window-start key. A window roll triggered by one counter (the RPM +# pair in the batch pass) left the sibling counter's stale previous-window +# value stranded under the freshly reset window, so the stale value was +# counted against the new window — guaranteed false 429s on the first +# request after every boundary, and window-blind local pre-checks kept +# rejecting for up to a full extra window after a genuine limit hit. +# +# The requests below are spaced like steady production traffic (last write +# well before the boundary) so the previous window's counters are still +# cached when the window rolls, matching the conditions under which the +# bug manifested. +############################################################ + + +@pytest.mark.asyncio +async def test_tpm_counter_resets_at_boundary_with_both_limits_reservation_disabled( + monkeypatch, time_controller +): + """ + Reservation disabled + BOTH rpm and tpm limits: crossing the window + boundary must reset BOTH counters. + + Pre-fix, the batch pass reset the shared window on the requests pair and + the tokens pair then saw the fresh window and incremented its stale + value (3 -> 4 > limit 3), so the boundary request got a false 429 that + persisted until the next boundary. + """ + monkeypatch.setenv("LITELLM_RATE_LIMIT_WINDOW_SIZE", "60") + monkeypatch.setenv("LITELLM_TPM_TOKEN_RESERVATION_ENABLED", "false") + _api_key = hash_token("sk-24677-reservation-disabled") + user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, rpm_limit=1000, tpm_limit=3) + local_cache = DualCache() + parallel_request_handler = _PROXY_MaxParallelRequestsHandler( + internal_usage_cache=InternalUsageCache(local_cache), + time_provider=time_controller.now, + ) + assert parallel_request_handler.tpm_reservation_enabled is False + + # Pin the fake clock so the boundary crossing lands in the same sliver + # production hits: the window-start value has rolled over (integer + # seconds), while the previous window's counters are still cached. + time_controller._current = datetime(2026, 1, 1, 0, 0, 0, 500000) + + tokens_key = parallel_request_handler.create_rate_limit_keys("api_key", _api_key, "tokens") + requests_key = parallel_request_handler.create_rate_limit_keys("api_key", _api_key, "requests") + + # Steady traffic: one request every 10s fills the TPM window to exactly + # its limit (3 requests, +1 token each); last write at t=20. + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + for _ in range(2): + time_controller.advance(10) + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + tokens_value = await local_cache.async_get_cache(key=tokens_key) + assert int(tokens_value) == 3, "TPM counter should sit exactly at its limit before the boundary" + + # Cross the window boundary (t=59.75, before the counters' TTL lapses): + # the new window must start empty. + time_controller.advance(39.75) + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + + tokens_value = await local_cache.async_get_cache(key=tokens_key) + requests_value = await local_cache.async_get_cache(key=requests_key) + assert int(tokens_value) == 1, "TPM counter must reset when its window rolls; stale value must not carry over" + assert int(requests_value) == 1, "RPM counter must reset when its window rolls" + + # Recovery is immediate, not deferred to the next boundary. + time_controller.advance(0.1) + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + tokens_value = await local_cache.async_get_cache(key=tokens_key) + assert int(tokens_value) == 2 + + +@pytest.mark.asyncio +async def test_tpm_reservation_not_poisoned_by_rpm_window_roll(monkeypatch, time_controller): + """ + Reservation enabled (default) + BOTH rpm and tpm limits: the atomic TPM + reservation must not be charged the stale tokens counter left behind by + the RPM pass's window roll. + + Pre-fix, the RPM pass rolled the shared window; reserve_tpm_tokens then + read the fresh window together with the previous window's token count + (186) and rejected the boundary request even though the new window was + empty. With per-type windows the tokens window rolls in the reservation + pass itself, so the boundary request is accounted to the new window. + """ + monkeypatch.setenv("LITELLM_RATE_LIMIT_WINDOW_SIZE", "60") + _api_key = hash_token("sk-24677-reservation-enabled") + user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, rpm_limit=1000, tpm_limit=186) + local_cache = DualCache() + parallel_request_handler = _PROXY_MaxParallelRequestsHandler( + internal_usage_cache=InternalUsageCache(local_cache), + time_provider=time_controller.now, + ) + assert parallel_request_handler.tpm_reservation_enabled is True + + # Pin the fake clock so the boundary crossing lands in the same sliver + # production hits: the tokens window-start value has rolled over, while + # the previous window's tokens counter is still cached. + time_controller._current = datetime(2026, 1, 1, 0, 0, 0, 500000) + + descriptors = parallel_request_handler._create_rate_limit_descriptors( + user_api_key_dict=user_api_key_dict, + data={}, + rpm_limit_type=None, + tpm_limit_type=None, + model_has_failures=False, + ) + tokens_key = parallel_request_handler.create_rate_limit_keys("api_key", _api_key, "tokens") + + # Steady traffic: six 31-token reservations exactly fill the 186-token + # window; last write at t=50. + for _ in range(6): + if _ > 0: + time_controller.advance(10) + response = await parallel_request_handler.should_rate_limit( + descriptors=descriptors, skip_tpm_check=True + ) + assert response["overall_code"] == "OK" + tpm_response = await parallel_request_handler.reserve_tpm_tokens( + descriptors=descriptors, estimated_tokens=31 + ) + assert tpm_response["overall_code"] == "OK" + + tokens_value = await local_cache.async_get_cache(key=tokens_key) + assert int(tokens_value) == 186, "TPM counter should sit exactly at its limit before the boundary" + + # First request after the window boundary (t=59.75, before the tokens + # counter's TTL lapses) must be admitted into the new window. + time_controller.advance(9.75) + response = await parallel_request_handler.should_rate_limit(descriptors=descriptors, skip_tpm_check=True) + assert response["overall_code"] == "OK" + tpm_response = await parallel_request_handler.reserve_tpm_tokens(descriptors=descriptors, estimated_tokens=31) + assert tpm_response["overall_code"] == "OK", ( + "First reservation after the window boundary must not be limited by the previous window's token count" + ) + + tokens_value = await local_cache.async_get_cache(key=tokens_key) + assert int(tokens_value) == 31, "Boundary request must be accounted to the new window" + + +@pytest.mark.asyncio +async def test_over_limit_mirror_does_not_reject_after_window_roll(monkeypatch, time_controller): + """ + The local in-memory over-limit pre-check must ignore a counter snapshot + whose window has already rolled. A genuine over-limit rejection froze an + over-limit value in the local mirror, and the window-blind pre-check then + kept returning 429s for up to a full extra window_size after the boundary. + """ + monkeypatch.setenv("LITELLM_RATE_LIMIT_WINDOW_SIZE", "60") + _api_key = hash_token("sk-24677-stale-mirror") + user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, rpm_limit=2) + local_cache = DualCache() + parallel_request_handler = _PROXY_MaxParallelRequestsHandler( + internal_usage_cache=InternalUsageCache(local_cache), + time_provider=time_controller.now, + ) + requests_key = parallel_request_handler.create_rate_limit_keys("api_key", _api_key, "requests") + + # Steady traffic: two requests admitted (t=0, t=10), third at t=20 + # genuinely exceeds RPM=2 -> 429, and the local mirror holds the + # over-limit value (3) written at t=20. + for _ in range(2): + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + time_controller.advance(10) + with pytest.raises(HTTPException) as exc_info: + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + assert exc_info.value.status_code == 429 + + # After the window boundary (strictly past 60s, while the stale mirror + # entry is still cached) the key must recover immediately instead of being + # rejected from the stale local mirror. + time_controller.advance(40.25) + await parallel_request_handler.async_pre_call_hook( + user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type="" + ) + requests_value = await local_cache.async_get_cache(key=requests_key) + assert int(requests_value) == 1, "Counter must reset to 1 for the new window's first request" + + +def test_is_cache_list_over_limit_ignores_counter_from_expired_window(time_controller): + """ + Unit test for the window-aware pre-check: a counter snapshot paired with + a window_start older than window_size is treated as 0, never as + over-limit. + """ + local_cache = DualCache() + parallel_request_handler = _PROXY_MaxParallelRequestsHandler( + internal_usage_cache=InternalUsageCache(local_cache), + time_provider=time_controller.now, + ) + window_key = "{api_key:sk-24677-unit}:window:requests" + counter_key = "{api_key:sk-24677-unit}:requests" + keys_to_fetch = [window_key, counter_key] + key_metadata = { + window_key: { + "requests_limit": 2, + "tokens_limit": None, + "window_size": 60, + "descriptor_key": "api_key", + } + } + + now_int = int(time_controller.now().timestamp()) + # Stale window (started 60s ago) with an over-limit counter value. + response = parallel_request_handler.is_cache_list_over_limit(keys_to_fetch, ["0", 3], key_metadata, now_int=now_int) + assert response["overall_code"] == "OK" + assert response["statuses"][0]["limit_remaining"] == 2 + + # Same over-limit counter inside a live window is still over-limit. + response = parallel_request_handler.is_cache_list_over_limit( + keys_to_fetch, [str(now_int), 3], key_metadata, now_int=now_int + ) + assert response["overall_code"] == "OVER_LIMIT" + + # Without now_int the legacy (window-blind) behavior is preserved. + response = parallel_request_handler.is_cache_list_over_limit(keys_to_fetch, ["0", 3], key_metadata) + assert response["overall_code"] == "OVER_LIMIT" + + +@pytest.mark.asyncio +async def test_new_window_keys_backfill_active_legacy_window(time_controller): + """A rolling upgrade preserves an active legacy window in the new keys.""" + local_cache = DualCache() + parallel_request_handler = _PROXY_MaxParallelRequestsHandler( + internal_usage_cache=InternalUsageCache(local_cache), + time_provider=time_controller.now, + ) + descriptor_key = "api_key" + descriptor_value = "sk-24677-migration" + legacy_key = legacy_descriptor_window_key(descriptor_key, descriptor_value) + new_key = descriptor_window_key(descriptor_key, descriptor_value, "requests") + + await local_cache.async_set_cache(key=legacy_key, value="100", ttl=60) + await parallel_request_handler._backfill_legacy_window_keys([new_key]) + + assert await local_cache.async_get_cache(key=new_key) == 100