From 21c08ac4589493291afa9c1471d08fb8a07a901a Mon Sep 17 00:00:00 2001 From: jackLei <42642542+jackLei0901@users.noreply.github.com> Date: Wed, 30 Sep 2026 17:10:03 -0700 Subject: [PATCH] [CI] Deflake shutdown wait-timeout test by synchronizing on request admission (#58125) Signed-off-by: jackLei0901 <42642542+jackLei0901@users.noreply.github.com> --- tests/entrypoints/launchers/test_shutdown.py | 48 ++++++++++++++++++-- 1 file changed, 44 insertions(+), 4 deletions(-) diff --git a/tests/entrypoints/launchers/test_shutdown.py b/tests/entrypoints/launchers/test_shutdown.py index 6d18d825eb..f03afbff09 100644 --- a/tests/entrypoints/launchers/test_shutdown.py +++ b/tests/entrypoints/launchers/test_shutdown.py @@ -14,6 +14,7 @@ import httpx import openai import psutil import pytest +from prometheus_client.parser import text_string_to_metric_families from tests.utils import RemoteOpenAIServer from vllm.utils.network_utils import get_open_port @@ -127,6 +128,40 @@ async def _concurrent_request_loop( t.cancel() +def _num_running_requests(metrics_text: str) -> float: + """Sum ``vllm:num_requests_running`` across the server's label sets.""" + total = 0.0 + for family in text_string_to_metric_families(metrics_text): + if family.name == "vllm:num_requests_running": + for sample in family.samples: + if sample.name == "vllm:num_requests_running": + total += sample.value + return total + + +async def _wait_for_running_request( + metrics_url: str, + timeout: float = _INFLIGHT_REQUEST_START_TIMEOUT, +) -> float: + """Wait until the server reports a running request, returning the delay. + + ``ShutdownState.inflight_requests`` is incremented before the HTTP call is + awaited, so it only shows that a client coroutine started. Draining can + only complete work the engine has already admitted, so synchronize on the + server's own scheduler gauge instead. + """ + start = time.monotonic() + deadline = start + timeout + async with httpx.AsyncClient() as metrics_client: + while time.monotonic() < deadline: + response = await metrics_client.get(metrics_url) + response.raise_for_status() + if _num_running_requests(response.text) > 0: + return time.monotonic() - start + await asyncio.sleep(_INFLIGHT_REQUEST_POLL_INTERVAL) + raise AssertionError(f"No request reached the running state within {timeout}s") + + @pytest.mark.asyncio async def test_shutdown_on_engine_failure(tmp_path: Path): """Verify that API returns connection error when server process is killed. @@ -240,11 +275,15 @@ async def test_wait_timeout_completes_requests(): _concurrent_request_loop(client, state, sigterm_sent, concurrency=10) ) - await asyncio.sleep(0.5) - proc.send_signal(signal.SIGTERM) - sigterm_sent.set() - try: + # Drain can only complete requests the engine already admitted, so + # wait for the server to report one instead of guessing with a sleep. + admission_delay = await _wait_for_running_request( + remote_server.url_for("metrics") + ) + proc.send_signal(signal.SIGTERM) + sigterm_sent.set() + await asyncio.wait_for(request_task, timeout=_SHUTDOWN_DETECTION_TIMEOUT) except asyncio.TimeoutError: pass @@ -257,6 +296,7 @@ async def test_wait_timeout_completes_requests(): # wait timeout should complete in-flight requests assert state.requests_after_sigterm > 0, ( f"Wait timeout should complete in-flight requests. " + f"admission_delay: {admission_delay:.3f}s, " f"503: {state.got_503}, 500: {state.got_500}, " f"conn_errors: {state.connection_errors}, errors: {state.errors}" )