scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-06 08:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

Merge 54b1560b15 into aea5358c48

Browse source
This commit is contained in:
ben7am1n 2026-09-05 05:26:15 +00:00 • committed by GitHub
commit 9f62a88b9b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 430 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

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

258
tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py
View file

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