perf(scheduler): skip the queue when a healthy deployment can take the request

wait_for_turn wrote a queue entry to Redis and removed it again for every
prioritized request, even when poll would admit it straight away because a
deployment was healthy. With router_settings.default_priority set, that is two
Redis writes on every request for a queue entry nobody waits behind.

Check health before enqueueing so only requests that actually have to wait
touch the shared queue. Admission is unchanged.

litellm/scheduler.py

@@ -101,6 +101,8 @@ class Scheduler:
         timeout: float,
         get_healthy_deployments: Callable[[], Awaitable[Sequence[object]]],
     ) -> None:
+        if len(await get_healthy_deployments()) > 0:
+            return
         try:
             await self.add_request(request=request)
             end_time: Final = time.monotonic() + timeout

tests/unit/test_scheduler.py

@@ -227,6 +227,23 @@ async def test_wait_for_turn_during_cooldown_survives_the_queue_key_expiring():
     assert await scheduler.get_queue("sched-model") == []
 
 
+@pytest.mark.asyncio
+async def test_wait_for_turn_never_queues_a_request_a_healthy_deployment_can_take():
+    scheduler: Final = Scheduler()
+
+    async def one_healthy_deployment_seen_with_an_empty_queue() -> Sequence[object]:
+        queue: Final = await scheduler.get_queue("sched-model")
+        assert queue == [], f"the admitted request was queued while a deployment was healthy: {queue}"
+        return ({"model_info": {"id": "a"}},)
+
+    await scheduler.wait_for_turn(
+        request=FlowItem(priority=1, request_id="admitted", model_name="sched-model"),
+        timeout=5,
+        get_healthy_deployments=one_healthy_deployment_seen_with_an_empty_queue,
+    )
+
+    assert await scheduler.get_queue("sched-model") == []
+
 
 class _JsonRoundTripRedisCache:
     def __init__(self) -> None: