mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-14 04:09:09 +00:00
bf9f59d76b
* fix(scheduler): remove a request's queue entry once it stops waiting Requests admitted while a healthy deployment existed never left the priority queue, and neither did requests cancelled while waiting. The stale entries blocked later requests during cooldown and, with Redis, made add_request raise a TypeError on queues read back as JSON lists Both scheduling paths now share one polling helper that removes the entry in a finally block, whether the request was admitted, timed out or cancelled Related to #43059 * test(router): allowlist _wait_for_scheduler_turn in the router coverage check The coverage script only counts direct calls in test files. The helper is exercised through prioritized acompletion and atext_completion in test_router.py, like the other allowlisted entries * fix(scheduler): admit healthy requests before reading the queue, and clean up cancelled enqueues poll() raised on an empty queue before checking for healthy deployments. With the cleanup now rewriting the queue after every admission, a concurrent write from another replica can erase a waiting request's entry, and that request then failed while a deployment was healthy. poll() now admits as soon as a deployment is healthy and only reads the queue during cooldown add_request also moved inside the try block, so a request cancelled while its queue write is in flight still has its entry removed * refactor(scheduler): move the wait loop into Scheduler.wait_for_turn The router passes a healthy-deployments callable into the scheduler, so tests inject a Scheduler directly instead of replacing the router's scheduler attribute. The cancelled-mid-enqueue test moves to test_scheduler.py with the other scheduler tests, and poll() takes the deployments as a Sequence since it only checks whether any are healthy * test(scheduler): move scheduler tests into tests/unit * fix(scheduler): finish the queue removal when cancellation is delivered again during cleanup A second cancel, or an anyio cancel scope that re-cancels on every await, interrupted remove_request mid-write and left the entry in Redis. * fix(scheduler): read queue entries back from redis as tuples * fix(scheduler): admit a request whose queue entry vanished during cooldown * fix(scheduler): re-enqueue a request whose queue entry vanished during cooldown * test(integration): cover priority scheduler queue cleanup across instances Adds tests/integration/routing/test_priority_scheduler_queue_cleanup.py: 27 cells against the real proxy (two workers, Postgres, Redis, a scripted upstream) for every prioritized surface (/v1/chat/completions, /v1/completions, /queue/chat/completions, streaming and not, OpenAI SDK sync and async, raw httpx), the non-integer priority pass-through, the in-memory queue on one proxy, two proxies sharing a Redis queue (served requests leave no entry, a dead replica's entry is skipped or expires, a waiter that times out or disconnects removes only itself), a Redis outage mid burst and a SIGKILLed worker. Each cell asserts the caller's response, the upstream's requests by marker and the Redis queue contents. On the merge base the cross-instance cells fail with the list-of-lists TypeError and the in-memory cell with 408s behind the leaked entry; on this branch every cell passes --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com> |
||
|---|---|---|
| .. | ||
| either_role.json | ||
| test_advisor_failure_cooldown.py | ||
| test_audit_any_sweep_chaos.py | ||
| test_complexity_router_llm_classifier.py | ||
| test_deployment_num_retries_generic_routes_wire.py | ||
| test_encrypted_content_affinity_no_peer.py | ||
| test_end_user_region_routing.py | ||
| test_key_max_parallel_requests.py | ||
| test_key_tpm_reservation.py | ||
| test_messages_stream_retry_budget_sources_owned_proxy.py | ||
| test_observed_routing.py | ||
| test_payment_required_cooldown.py | ||
| test_payment_required_mapping.py | ||
| test_priority_model_tpm_enforcement.py | ||
| test_priority_rate_limit_headers.py | ||
| test_priority_scheduler_queue_cleanup.py | ||
| test_prompt_caching_affinity_tool_call_marks.py | ||
| test_redis_increment_with_floor.py | ||
| test_redis_recovery.py | ||
| test_stale_cost_map_boot.py | ||
| test_team_model_tpm_limit.py | ||
| test_team_tag_routing.py | ||
| test_usage_based_routing_redis_reads.py | ||
| test_user_config_routing.py | ||