[Core] Add a generation readiness check /health_generate

.buildkite/test_areas/distributed.yaml

@@ -90,6 +90,8 @@ steps:
   - vllm/v1/engine/
   - vllm/v1/worker/
   - tests/v1/distributed
+  - tests/v1/engine/readiness/
+  - vllm/entrypoints/serve/instrumentator/health.py
   - tests/entrypoints/launchers/api_server/test_multi_api_servers.py
   - tests/v1/e2e/general/test_sharded_sampling.py
   - tests/v1/e2e/spec_decode/test_sharded_sampling.py
@@ -98,6 +100,7 @@ steps:
   # https://github.com/NVIDIA/nccl/issues/1838
   - export NCCL_CUMEM_HOST_ENABLE=0
   - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py
+  - pytest -v -s v1/engine/readiness/test_ready_multi_gpu.py
   - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
   - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py
   - DP_SIZE=2 pytest -v -s entrypoints/launchers/api_server/test_multi_api_servers.py

.buildkite/test_areas/engine.yaml

@@ -44,6 +44,7 @@ steps:
   timeout_in_minutes: 45
   source_file_dependencies:
     - vllm/v1/engine/
+    - vllm/entrypoints/serve/instrumentator/health.py
     - tests/v1/engine/
     - tests/v1/test_tensor_ipc_queue.py
   commands:

docs/usage/security.md

@@ -214,6 +214,7 @@ The following endpoints **do not require authentication** even when `--api-key`
 - `/tokenize` - Tokenize text (not gated by `--enable-scale-out`)
 - `/detokenize` - Detokenize tokens
 - `/health` - Health check
+- `/health_generate` - GPU execution and EngineCore forward-progress check
 - `/ping` - SageMaker health check
 - `/version` - Version information
 - `/load` - Server load metrics

tests/entrypoints/serve/instrumentator/test_basic.py

@@ -2,6 +2,7 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
 import asyncio
+import json
 from http import HTTPStatus
 from unittest.mock import AsyncMock, Mock
 
@@ -13,7 +14,11 @@
 
 from tests.utils import RemoteOpenAIServer
 from vllm import envs
-from vllm.v1.engine.exceptions import EngineDeadError
+from vllm.v1.engine.exceptions import (
+    EngineDeadError,
+    EngineSleepingError,
+    EngineUnhealthyError,
+)
 from vllm.version import __version__ as VLLM_VERSION
 
 MODEL_NAME = "Qwen/Qwen3-0.6B"
@@ -224,3 +229,51 @@ async def test_health_check_engine_dead_error():
 
     # Assert that it returns 503 Service Unavailable
     assert response.status_code == 503
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("error", "expected_body"),
+    [
+        (None, None),
+        (EngineDeadError(), {"status": "not_ready", "reason": "engine_dead"}),
+        (
+            EngineSleepingError("paused", engine_ranks=[0], in_progress=[]),
+            {
+                "status": "not_ready",
+                "reason": "sleeping",
+                "message": "paused",
+                "engine_ranks": [0],
+                "in_progress": [],
+            },
+        ),
+        (
+            EngineUnhealthyError(
+                "timed out",
+                reason="probe_timeout",
+                in_progress=[{"operation": "collective_rpc:update_weights"}],
+            ),
+            {
+                "status": "not_ready",
+                "reason": "probe_timeout",
+                "message": "timed out",
+                "in_progress": [{"operation": "collective_rpc:update_weights"}],
+            },
+        ),
+    ],
+)
+async def test_health_generate_response(error, expected_body):
+    from vllm.entrypoints.serve.instrumentator.health import health_generate
+
+    mock_request = Mock(spec=Request)
+    mock_engine_client = AsyncMock()
+    mock_engine_client.check_health_gpu.side_effect = error
+    mock_request.app.state.engine_client = mock_engine_client
+
+    response = await health_generate(mock_request)
+
+    if expected_body is None:
+        assert response.status_code == 200
+        return
+    assert response.status_code == 503
+    assert json.loads(response.body) == expected_body

tests/v1/engine/readiness/fault_injection.py

@@ -0,0 +1,112 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Runtime fault injection for `/health_generate` end-to-end tests.
+
+Loaded into the server via ``--scheduler-cls`` / ``--worker-extension-cls``.
+Scheduler faults are toggled with flag files under ``$VLLM_READY_FAULT_DIR``
+(the scheduler lives in the EngineCore process). Worker faults are triggered
+through the dev ``/collective_rpc`` endpoint, which only forwards strings.
+"""
+
+import os
+import time
+from typing import Any
+
+import torch
+
+from vllm.config import VllmConfig
+from vllm.v1.core.sched.async_scheduler import AsyncScheduler
+
+FAULT_DIR_ENV = "VLLM_READY_FAULT_DIR"
+SCHEDULER_STALL = "scheduler_stall"
+UTILITY_HANG = "utility_hang"
+
+
+def _launch_illegal_memory_access() -> None:
+    import triton
+    import triton.language as tl
+
+    @triton.jit
+    def _oob_store(ptr, offset):
+        tl.store(ptr + offset, 1.0)
+
+    buf = torch.empty(1, device="cuda", dtype=torch.float32)
+    _oob_store[(1,)](buf, 1 << 42)
+
+
+class FaultInjectionScheduler(AsyncScheduler):
+    """While the ``scheduler_stall`` flag exists, schedule zero tokens.
+
+    Mimics #45388: requests stay waiting but no tokens are ever scheduled,
+    while the EngineCore busy loop keeps spinning.
+    """
+
+    max_num_scheduled_tokens: int
+
+    def schedule(self, *args, **kwargs):
+        fault_dir = os.environ.get(FAULT_DIR_ENV)
+        if fault_dir is None or not os.path.exists(
+            os.path.join(fault_dir, SCHEDULER_STALL)
+        ):
+            return super().schedule(*args, **kwargs)
+        saved = self.max_num_scheduled_tokens
+        self.max_num_scheduled_tokens = 0
+        try:
+            return super().schedule(*args, **kwargs)
+        finally:
+            self.max_num_scheduled_tokens = saved
+
+
+class FaultInjectionWorkerExtension:
+    vllm_config: VllmConfig
+    model_runner: Any
+    rank: int
+
+    def _fault_targets_me(self, dp_ranks: str) -> bool:
+        # Dense DP engines keep data_parallel_rank=0; the index is the DP rank.
+        rank = self.vllm_config.parallel_config.data_parallel_index
+        return rank in {int(r) for r in dp_ranks.split(",")}
+
+    def fault_hang_while_flag_set(self, flag: str) -> None:
+        """Block this control call while the flag exists, like a weight
+        sync stuck waiting on its peers."""
+        path = os.path.join(os.environ[FAULT_DIR_ENV], flag)
+        while os.path.exists(path):
+            time.sleep(0.1)
+
+    def fault_ima_now(self, dp_ranks: str) -> None:
+        """Launch an illegal memory access without synchronizing.
+
+        Models a latent async CUDA fault that has not surfaced yet because
+        nothing synchronized afterwards.
+        """
+        if not self._fault_targets_me(dp_ranks):
+            return
+        _launch_illegal_memory_access()
+
+    def fault_ima_now_on_worker(self, worker_rank: str) -> None:
+        """Like ``fault_ima_now``, on one TP/PP worker of the engine."""
+        if self.rank == int(worker_rank):
+            _launch_illegal_memory_access()
+
+    def fault_ima_on_next_forward(self, dp_ranks: str) -> None:
+        """Launch an illegal memory access at the end of the next forward
+        pass, so the fault belongs to the readiness probe's own GPU work."""
+        if not self._fault_targets_me(dp_ranks):
+            return
+
+        def hook(*_):
+            handle.remove()
+            _launch_illegal_memory_access()
+
+        handle = self.model_runner.get_model().register_forward_hook(hook)
+
+    def fault_dp_mismatched_collective(self, dp_ranks: str) -> None:
+        """Enter a DP all-reduce that the other ranks never join (#36594)."""
+        if not self._fault_targets_me(dp_ranks):
+            return
+        from vllm.distributed.parallel_state import get_dp_group
+
+        t = torch.ones(1, device="cuda")
+        torch.distributed.all_reduce(t, group=get_dp_group().device_group)
+        torch.accelerator.synchronize()

tests/v1/engine/readiness/test_ready_multi_gpu.py

@@ -0,0 +1,98 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""`/health_generate` against injected failures across two GPUs (DP=2 and PP=2).
+
+MoE DP ranks step in lockstep waves where idle ranks run dummy batches, and
+must not be probed with an ad-hoc dummy batch; dense DP ranks are independent
+engines that can be probed like a single engine.
+"""
+
+import time
+
+import pytest
+
+from tests.utils import multi_gpu_marks
+
+from .fault_injection import SCHEDULER_STALL
+from .utils import STALL_TIMEOUT_S, ready_probe_server
+
+pytestmark = multi_gpu_marks(num_gpus=2)
+
+MOE_MODEL = "ibm-research/PowerMoE-3b"
+DENSE_MODEL = "hmellor/tiny-random-LlamaForCausalLM"
+MOE_DP_ARGS = [
+    "--data-parallel-size",
+    "2",
+    "--enable-expert-parallel",
+    # Weights are dummy; two layers keep startup and memory small.
+    "--hf-overrides",
+    '{"num_hidden_layers": 2}',
+]
+DENSE_DP_ARGS = ["--data-parallel-size", "2"]
+
+
+@pytest.fixture(scope="module")
+def moe_server(tmp_path_factory):
+    with ready_probe_server(
+        MOE_MODEL, MOE_DP_ARGS, tmp_path_factory.mktemp("faults")
+    ) as srv:
+        yield srv
+
+
+def test_moe_no_false_positive(moe_server):
+    """An idle rank inside a wave keeps counting as progressing, and probes
+    racing with wave starts do not flap."""
+    with moe_server.busy(dp_ranks=(0,)):
+        assert moe_server.stays_ready(STALL_TIMEOUT_S + 3)
+    for i in range(10):
+        moe_server.generate_in_background(8, dp_rank=i % 2)
+        assert moe_server.status() == 200
+        time.sleep(0.2 * (i % 3))
+
+
+def test_moe_scheduler_stall(moe_server):
+    """#45388 under DP: ranks hold requests but schedule nothing while the
+    wave keeps running dummy batches."""
+    moe_server.set_fault(SCHEDULER_STALL)
+    try:
+        moe_server.generate_in_background(8, dp_rank=0)
+        moe_server.generate_in_background(8, dp_rank=1)
+        assert moe_server.wait_until_unready() is not None
+        assert moe_server.status("/health") == 200
+    finally:
+        moe_server.clear_fault(SCHEDULER_STALL)
+    assert moe_server.wait_until_ready()
+
+
+def test_moe_collective_mismatch_while_idle(tmp_path):
+    """#36594 class: rank 0 is wedged in a DP collective inside a control RPC
+    while there is no traffic."""
+    with ready_probe_server(MOE_MODEL, MOE_DP_ARGS, tmp_path) as srv:
+        assert srv.status() == 200
+        srv.rpc("fault_dp_mismatched_collective", "0", background=True)
+        assert srv.wait_until_unready() is not None
+        assert srv.status("/health") == 200
+        # Any API server can name the wedged operation and rank.
+        for _ in range(4):
+            body = srv.ready_body()
+            assert body["reason"] == "probe_timeout"
+            stuck = {(op["operation"], op["engine_rank"]) for op in body["in_progress"]}
+            assert ("collective_rpc:fault_dp_mismatched_collective", 0) in stuck
+
+
+def test_dense_latent_cuda_error_while_idle(tmp_path):
+    """A latent CUDA fault on an idle dense DP rank is caught by its probe."""
+    with ready_probe_server(DENSE_MODEL, DENSE_DP_ARGS, tmp_path) as srv:
+        assert srv.status() == 200
+        srv.rpc("fault_ima_now", "1")
+        assert srv.wait_until_unready() is not None
+
+
+def test_cuda_error_on_non_output_pp_stage(tmp_path):
+    """The probe checks every worker, not only the last PP stage that returns
+    model outputs; PP dummy runs do not communicate across stages."""
+    args = ["--pipeline-parallel-size", "2"]
+    with ready_probe_server(DENSE_MODEL, args, tmp_path) as srv:
+        assert srv.status() == 200
+        srv.rpc("fault_ima_now_on_worker", "0")
+        assert srv.status() == 503

tests/v1/engine/readiness/test_ready_single_gpu.py

@@ -0,0 +1,66 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""`/health_generate` against injected failures on a single GPU.
+
+`/health` stays 200 for all of these; `/health_generate` must not.
+"""
+
+import pytest
+
+from .fault_injection import SCHEDULER_STALL, UTILITY_HANG
+from .utils import STALL_TIMEOUT_S, ready_probe_server
+
+MODEL = "hmellor/tiny-random-LlamaForCausalLM"
+
+
+@pytest.fixture(scope="module")
+def server(tmp_path_factory):
+    with ready_probe_server(MODEL, [], tmp_path_factory.mktemp("faults")) as srv:
+        yield srv
+
+
+def test_no_false_positive(server):
+    """A healthy engine stays ready when idle and under sustained load."""
+    assert server.stays_ready(STALL_TIMEOUT_S + 3)
+    with server.busy():
+        assert server.stays_ready(STALL_TIMEOUT_S + 3)
+
+
+def test_scheduler_stall(server):
+    """#45388: requests outstanding but zero tokens scheduled."""
+    server.set_fault(SCHEDULER_STALL)
+    try:
+        server.generate_in_background(8)
+        assert server.wait_until_unready() is not None
+        assert server.status("/health") == 200
+    finally:
+        server.clear_fault(SCHEDULER_STALL)
+    assert server.wait_until_ready()
+
+
+def test_blocked_utility_call_reports_operation(server):
+    """A busy loop blocked inside a control call (e.g. a stuck weight sync)
+    times out the probe, and the response names the call it is blocked in."""
+    server.set_fault(UTILITY_HANG)
+    try:
+        server.rpc("fault_hang_while_flag_set", UTILITY_HANG, background=True)
+        assert server.wait_until_unready() is not None
+        body = server.ready_body()
+        assert body["reason"] == "probe_timeout"
+        (operation,) = body["in_progress"]
+        assert operation["operation"] == "collective_rpc:fault_hang_while_flag_set"
+        assert server.status("/health") == 200
+    finally:
+        server.clear_fault(UTILITY_HANG)
+    assert server.wait_until_ready()
+
+
+@pytest.mark.parametrize("fault", ["fault_ima_now", "fault_ima_on_next_forward"])
+def test_cuda_error_while_idle(tmp_path, fault: str):
+    """An async CUDA fault that nothing has synchronized on yet, raised before
+    or during the probe's own forward pass, fails that same `/health_generate` call."""
+    with ready_probe_server(MODEL, [], tmp_path) as srv:
+        assert srv.status() == 200
+        srv.rpc(fault, "0")
+        assert srv.status("/health") == 200
+        assert srv.status() == 503