diff --git a/components/src/dynamo/trtllm/health_check.py b/components/src/dynamo/trtllm/health_check.py index 437fd82675cf..1f8cb46f2d1e 100644 --- a/components/src/dynamo/trtllm/health_check.py +++ b/components/src/dynamo/trtllm/health_check.py @@ -8,9 +8,10 @@ """ import logging -from typing import Any +from typing import Any, Optional from dynamo.health_check import HealthCheckPayload +from dynamo.trtllm.constants import DisaggregationMode logger = logging.getLogger(__name__) @@ -89,3 +90,51 @@ def __init__(self, tokenizer: Any = None) -> None: }, } super().__init__() + + +#: Reserved request key that marks a canary probe request for disagg decode. +#: The handler detects this marker in `_setup_disaggregated_params_for_mode` +#: and constructs a synthetic `LlmDisaggregatedParams(request_type= +#: "context_and_generation")` so the engine runs the probe locally +#: (full prefill + 1-token decode) without engaging the cache transceiver. +#: Internal-only — not part of the public request protocol. +CANARY_PROBE_KEY = "_canary_probe" + + +class TrtllmDisaggDecodeHealthCheckPayload(TrtllmHealthCheckPayload): + """Canary payload for TRT-LLM disagg decode workers. + + Sets a ``CANARY_PROBE_KEY`` marker so the handler bypasses the strict + "Disaggregated params are required for decode mode" guard and routes + the probe through `request_type="context_and_generation"` (local + prefill + decode, transceiver-free). + + Mirrors SGLang's `SglangDisaggHealthCheckPayload` which uses + `FAKE_BOOTSTRAP_HOST` for the same purpose, and vLLM's natural + agg-style fallback when `prefill_result` is absent. + """ + + def __init__(self, tokenizer: Any = None) -> None: + super().__init__(tokenizer=tokenizer) + self.default_payload[CANARY_PROBE_KEY] = True + + +def build_worker_health_check_payload( + disaggregation_mode: DisaggregationMode, + tokenizer: Any = None, +) -> Optional[dict]: + """ + Decide the health_check_payload for a TRT-LLM worker based on its disagg role. + + Decode workers use a probe payload with ``CANARY_PROBE_KEY`` set; the + handler short-circuits in `_setup_disaggregated_params_for_mode` and + routes the probe as a local agg request (no cache transceiver, no real + prefill peer required). This gives disagg decode the same engine-level + canary coverage as aggregated / prefill workers. + + Aggregated and prefill workers accept generic probes and register the + standard canary payload. + """ + if disaggregation_mode == DisaggregationMode.DECODE: + return TrtllmDisaggDecodeHealthCheckPayload(tokenizer=tokenizer).to_dict() + return TrtllmHealthCheckPayload(tokenizer=tokenizer).to_dict() diff --git a/components/src/dynamo/trtllm/request_handlers/handler_base.py b/components/src/dynamo/trtllm/request_handlers/handler_base.py index f5424050f5f9..4037d5266924 100644 --- a/components/src/dynamo/trtllm/request_handlers/handler_base.py +++ b/components/src/dynamo/trtllm/request_handlers/handler_base.py @@ -711,6 +711,22 @@ def _setup_disaggregated_params_for_mode( disaggregated_params = None epd_metadata: dict[str, Any] = {} + # Canary probe short-circuit: decode workers receive probes with a + # reserved `_canary_probe` marker (see TrtllmDisaggDecodeHealthCheckPayload + # in dynamo/trtllm/health_check.py). Build synthetic params with + # request_type="context_and_generation" so the engine runs the probe as + # full local prefill+decode — the cache transceiver activates only for + # request_type="generation_only", so this path is transceiver-free and + # doesn't need a real prefill peer. Mirrors sglang's FAKE_BOOTSTRAP_HOST. + if self.disaggregation_mode == DisaggregationMode.DECODE and request.get( + "_canary_probe" + ): + return ( + LlmDisaggregatedParams(request_type="context_and_generation"), + None, + {}, + ) + # PREFILL mode: setup context_only params if self.disaggregation_mode == DisaggregationMode.PREFILL: if ep_disaggregated_params: diff --git a/components/src/dynamo/trtllm/tests/test_health_check_disagg.py b/components/src/dynamo/trtllm/tests/test_health_check_disagg.py new file mode 100644 index 000000000000..878e6faa930e --- /dev/null +++ b/components/src/dynamo/trtllm/tests/test_health_check_disagg.py @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import pytest + +from dynamo.trtllm.constants import DisaggregationMode +from dynamo.trtllm.health_check import ( + CANARY_PROBE_KEY, + TrtllmDisaggDecodeHealthCheckPayload, + TrtllmHealthCheckPayload, + build_worker_health_check_payload, +) + +pytestmark = [ + pytest.mark.unit, + pytest.mark.trtllm, + pytest.mark.gpu_1, # needs trtllm packages installed but does not use GPU + pytest.mark.profiled_vram_gib(0), + pytest.mark.pre_merge, +] + + +def test_trtllm_health_check_payload_has_no_disagg_params(): + """Standard TrtllmHealthCheckPayload should NOT include disaggregated_params.""" + payload = TrtllmHealthCheckPayload().to_dict() + assert "disaggregated_params" not in payload + assert "prefill_result" not in payload + assert "token_ids" in payload + + +@pytest.mark.parametrize( + "mode", + [DisaggregationMode.AGGREGATED, DisaggregationMode.PREFILL], +) +def test_non_decode_modes_register_canary_payload(mode): + """Aggregated and prefill workers register the standard canary payload.""" + payload = build_worker_health_check_payload(disaggregation_mode=mode) + assert payload is not None + assert "token_ids" in payload + assert "sampling_options" in payload + + +def test_decode_mode_registers_probe_payload(): + """Decode workers register a probe payload carrying CANARY_PROBE_KEY. + + The handler detects the marker in `_setup_disaggregated_params_for_mode` + and routes the probe through `request_type="context_and_generation"` + so the engine runs it as a local agg request (no cache transceiver). + Pattern mirrors SGLang's FAKE_BOOTSTRAP_HOST. + """ + payload = build_worker_health_check_payload( + disaggregation_mode=DisaggregationMode.DECODE + ) + assert payload is not None + assert payload.get(CANARY_PROBE_KEY) is True + # Standard fields must still be present so the handler's downstream logic + # (sampling, stop conditions, token_ids) runs normally. + assert "token_ids" in payload + assert "sampling_options" in payload + assert "stop_conditions" in payload + + +def test_disagg_decode_payload_class_sets_probe_marker(): + """Direct construction of TrtllmDisaggDecodeHealthCheckPayload carries the marker.""" + payload = TrtllmDisaggDecodeHealthCheckPayload().to_dict() + assert payload.get(CANARY_PROBE_KEY) is True + assert "disaggregated_params" not in payload # real params built by handler diff --git a/components/src/dynamo/trtllm/workers/llm_worker.py b/components/src/dynamo/trtllm/workers/llm_worker.py index c8cd50ff6199..c85ba09ae2d8 100644 --- a/components/src/dynamo/trtllm/workers/llm_worker.py +++ b/components/src/dynamo/trtllm/workers/llm_worker.py @@ -54,7 +54,7 @@ from dynamo.trtllm.args import Config from dynamo.trtllm.constants import DisaggregationMode, Modality from dynamo.trtllm.engine import Backend, get_llm_engine -from dynamo.trtllm.health_check import TrtllmHealthCheckPayload +from dynamo.trtllm.health_check import build_worker_health_check_payload from dynamo.trtllm.multimodal_processor import MultimodalRequestProcessor from dynamo.trtllm.publisher import DYNAMO_COMPONENT_REGISTRY, get_publisher from dynamo.trtllm.request_handlers.handlers import ( @@ -602,8 +602,18 @@ async def init_llm_worker( custom_template_path=config.custom_jinja_template, ) - # Get health check payload (checks env var and falls back to TensorRT-LLM default) - health_check_payload = TrtllmHealthCheckPayload(tokenizer=tokenizer).to_dict() + # Decode workers opt out of canary (the TRT-LLM decode handler strictly + # rejects canary probes without disaggregated_params); agg/prefill workers + # register the standard TRT-LLM payload. See build_worker_health_check_payload + # for the full reasoning. + health_check_payload = build_worker_health_check_payload( + disaggregation_mode=config.disaggregation_mode, + tokenizer=tokenizer, + ) + if health_check_payload is None: + logging.info( + "Decode worker: canary health check disabled (no payload registered)" + ) if config.publish_events_and_metrics: # Initialize and pass in the publisher to the request handler to diff --git a/components/src/dynamo/vllm/tests/test_vllm_health_check.py b/components/src/dynamo/vllm/tests/test_vllm_health_check.py new file mode 100644 index 000000000000..a8a692e63ea3 --- /dev/null +++ b/components/src/dynamo/vllm/tests/test_vllm_health_check.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Regression guard: vLLM workers must always register a canary payload. + +DIS-1737 originally routed disagg decode through a `payload = None` branch in +`worker_factory.py`. That silently opted decode workers out of canary, which +after DIS-1185 (canary = sole readiness authority) left them stuck NotReady. +These tests ensure the payload constructor works for both agg and decode paths +so no one re-introduces a DECODE-specific None branch. +""" + +import pytest + +from dynamo.vllm.health_check import VllmHealthCheckPayload + +pytestmark = [ + pytest.mark.unit, + pytest.mark.vllm, + pytest.mark.pre_merge, + pytest.mark.gpu_0, +] + + +@pytest.mark.parametrize("use_text_input", [False, True]) +def test_vllm_health_check_payload_is_non_none(use_text_input): + """VllmHealthCheckPayload.to_dict() returns a non-None dict regardless of + tokenizer mode. Worker code must always register a canary target; decode + workers rely on the vLLM handler's natural agg-style fallback when + `prefill_result` is absent (handlers.py::_generate_token_mode).""" + payload = VllmHealthCheckPayload( + engine_client=None, use_text_input=use_text_input + ).to_dict() + + assert payload is not None + assert isinstance(payload, dict) + # Payload must contain some form of input the engine can decode. + assert "token_ids" in payload or "prompt" in payload diff --git a/components/src/dynamo/vllm/worker_factory.py b/components/src/dynamo/vllm/worker_factory.py index f20b19c78440..0237255b2ebd 100644 --- a/components/src/dynamo/vllm/worker_factory.py +++ b/components/src/dynamo/vllm/worker_factory.py @@ -392,6 +392,10 @@ async def _create_decode_worker( vllm_config, ) + # vLLM's DecodeWorkerHandler._generate_token_mode handles requests without + # prefill_result by running them agg-style, so the standard canary payload + # works for decode workers too — the engine produces one token regardless + # of disaggregation mode. Always register a canary target. health_check_payload = VllmHealthCheckPayload( engine_client, use_text_input=config.use_vllm_tokenizer ).to_dict() diff --git a/lib/runtime/src/system_health.rs b/lib/runtime/src/system_health.rs index 7cf3924b2592..efc786b16ab3 100644 --- a/lib/runtime/src/system_health.rs +++ b/lib/runtime/src/system_health.rs @@ -100,10 +100,21 @@ impl SystemHealth { self.health_check_enabled } - /// Signal endpoint transport registration. Sets Ready when canary is disabled; - /// no-op when canary is enabled (canary will set Ready after verification). + /// Signal endpoint transport registration. + /// + /// Marks the endpoint Ready when canary is disabled globally, OR when the + /// endpoint has no registered canary target (i.e. the caller opted out of + /// canary by not passing a `health_check_payload` to `serve_endpoint`). + /// Endpoints that DID register a canary target stay NotReady until the + /// canary verifies them — preserving DIS-1185's "canary is the authoritative + /// readiness signal for endpoints that opt in" contract. pub fn set_endpoint_registered(&self, endpoint: &str) { - if !self.health_check_enabled { + let has_target = self + .health_check_targets + .read() + .unwrap() + .contains_key(endpoint); + if !self.health_check_enabled || !has_target { self.set_endpoint_health_status(endpoint, HealthStatus::Ready); } } @@ -141,20 +152,26 @@ impl SystemHealth { .get(endpoint) .is_some_and(|status| *status == HealthStatus::Ready) }) + } else if !health_check_targets.is_empty() { + // Canary-opt-in endpoints exist: every one must be Ready. + health_check_targets + .iter() + .all(|(endpoint_subject, _target)| { + endpoint_health + .get(endpoint_subject) + .is_some_and(|status| *status == HealthStatus::Ready) + }) + } else if !endpoint_health.is_empty() { + // No canary targets, but endpoints have registered. Healthy when + // every registered endpoint is Ready. This covers workers that + // opt every endpoint out of canary (e.g. disagg decode workers, + // secondary operational endpoints). + endpoint_health + .values() + .all(|status| *status == HealthStatus::Ready) } else { - // If we have registered health check targets, use them to determine health - if !health_check_targets.is_empty() { - health_check_targets - .iter() - .all(|(endpoint_subject, _target)| { - endpoint_health - .get(endpoint_subject) - .is_some_and(|status| *status == HealthStatus::Ready) - }) - } else { - // No health check targets registered, use simple system health - self.system_health == HealthStatus::Ready - } + // No endpoints registered at all — use simple system health. + self.system_health == HealthStatus::Ready }; (healthy, endpoints) @@ -298,3 +315,94 @@ impl SystemHealth { &self.live_path } } + +#[cfg(test)] +mod tests { + use super::*; + + fn sh(canary_enabled: bool) -> SystemHealth { + SystemHealth::new( + HealthStatus::NotReady, + vec![], + canary_enabled, + "/health".into(), + "/live".into(), + ) + } + + #[test] + fn endpoint_registered_auto_readies_without_target_when_canary_enabled() { + // Canary enabled but the endpoint did not register a canary target — + // it opted out of canary, so registration alone marks it Ready. + let h = sh(true); + h.set_endpoint_registered("generate"); + assert_eq!( + h.get_endpoint_health_status("generate"), + Some(HealthStatus::Ready), + "endpoint with no canary target must auto-Ready" + ); + } + + #[test] + fn endpoint_registered_stays_notready_with_target_when_canary_enabled() { + // Canary enabled and the endpoint registered a target — registration + // does NOT flip Ready; canary verifies first (DIS-1185 contract). + let h = sh(true); + let instance = component::Instance { + component: "test".into(), + endpoint: "generate".into(), + namespace: "test".into(), + instance_id: 0, + transport: component::TransportType::Tcp("localhost:0".into()), + device_type: None, + }; + h.health_check_targets.write().unwrap().insert( + "generate".to_string(), + HealthCheckTarget { + instance, + payload: serde_json::json!({}), + }, + ); + h.set_endpoint_registered("generate"); + // With a registered target, set_endpoint_registered must NOT mark + // the endpoint Ready — canary is responsible. Either "no entry" or + // an explicit NotReady is acceptable (both make /health path #2 report + // unhealthy until canary succeeds). + assert_ne!( + h.get_endpoint_health_status("generate"), + Some(HealthStatus::Ready), + "endpoint with a canary target must wait for canary, not auto-Ready" + ); + } + + #[test] + fn endpoint_registered_readies_when_canary_disabled() { + let h = sh(false); + h.set_endpoint_registered("generate"); + assert_eq!( + h.get_endpoint_health_status("generate"), + Some(HealthStatus::Ready), + ); + } + + #[test] + fn get_health_status_uses_endpoint_health_when_no_target() { + // No canary targets, no use_endpoint_health_status override. + // A single registered endpoint in Ready state should make /health healthy. + let h = sh(true); + h.set_endpoint_registered("generate"); + let (healthy, _eps) = h.get_health_status(); + assert!( + healthy, + "when no canary target exists and all registered endpoints are Ready, /health is healthy" + ); + } + + #[test] + fn get_health_status_notready_when_no_endpoints_and_system_notready() { + // No endpoints registered at all — falls back to system_health (NotReady by default). + let h = sh(true); + let (healthy, _eps) = h.get_health_status(); + assert!(!healthy); + } +} diff --git a/tests/fault_tolerance/test_canary_rank_pause.py b/tests/fault_tolerance/test_canary_rank_pause.py new file mode 100644 index 000000000000..d82424b0da67 --- /dev/null +++ b/tests/fault_tolerance/test_canary_rank_pause.py @@ -0,0 +1,325 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Rank-pause test: prove the canary detects real engine hangs. + +We launch a real engine worker, wait for it to be Ready, then SIGSTOP the +engine-rank child process to simulate a hang (the engine can't produce +tokens, but the Python dispatch layer is still alive). We then poll +``/health`` on the worker's system port and assert: + + * When canary is ENABLED AND the endpoint has a registered payload, + ``/health`` flips to 503 — the canary probe times out against the + paused engine and marks the endpoint NotReady. This is the positive + case: *"canary can detect the issues we have seen."* + + * When canary is DISABLED (or the endpoint opts out by registering no + payload), ``/health`` STAYS at 200 — there is no active liveness + probe. This is the negative control: it proves the harness + distinguishes the "canary working" path from the "no canary" path, + rather than always reporting success. + +After each case we ``SIGCONT`` the rank and (for detect cases) verify +``/health`` returns to Ready as a round-trip sanity check. + +Coverage matrix — ``{trtllm, vllm, sglang} × {agg, disagg-prefill, +disagg-decode}`` — confirms all three backends ship a canary that +catches engine-level hangs in disagg setups: + * vllm decode uses the handler's natural agg-style fallback when + ``prefill_result`` is absent (handlers.py::_generate_token_mode). + * sglang decode uses ``FAKE_BOOTSTRAP_HOST`` via + ``SglangDisaggHealthCheckPayload`` to bypass real KV transfer. + * trtllm decode uses a ``_canary_probe`` marker + handler short-circuit + to ``request_type="context_and_generation"`` + (handler_base.py::_setup_disaggregated_params_for_mode). + +Style mirrors ``tests/serve/test_trtllm.py::test_deployment`` — same +``EngineProcess.from_script`` launcher, same ``/health`` polling. We do +not use the ``tests/fault_tolerance/hardware/fault_injection_service/`` +helpers; just POSIX signals. +""" + +from __future__ import annotations + +import dataclasses +import logging +import os +import signal +import time +from dataclasses import dataclass +from typing import Any + +import psutil +import pytest +import requests + +from tests.serve.test_sglang import sglang_configs +from tests.serve.test_trtllm import trtllm_configs +from tests.serve.test_vllm import vllm_configs +from tests.utils.engine_process import EngineProcess + +logger = logging.getLogger(__name__) + +# Seconds: how long to wait after SIGSTOP for the canary to notice. +# Canary interval in the runtime is typically 10 s; give it generous slack. +PAUSE_DETECT_BUDGET_S = 45 + +# Seconds: after SIGCONT, how long until /health is back. +RESUME_RECOVER_BUDGET_S = 30 + +# Seconds: how long to wait for /health to initially come up Ready. +STARTUP_READY_BUDGET_S = 120 + + +# Per-backend engine-rank discovery patterns. These match the substrings that +# appear in the engine subprocess's command line. Keep in sync with the +# `stragglers` declared on each backend's test config. +_RANK_PATTERNS: dict[str, tuple[str, ...]] = { + # trtllm's engine ranks are mpi4py.futures.server children of the + # `python3 -m dynamo.trtllm` worker (verified locally on Qwen3-0.6B + # disagg_same_gpu). The other substrings are defensive in case a + # future TRT-LLM release renames. + "trtllm": ("mpi4py.futures.server", "TRTLLM:EngineCore", "tensorrt_llm"), + "vllm": ("VLLM::EngineCore", "EngineCoreProc"), + "sglang": ("SGLANG:EngineCore", "sgl_scheduler"), +} + +# Per-backend config dicts keyed by backend name. +_CONFIGS_BY_BACKEND = { + "trtllm": trtllm_configs, + "vllm": vllm_configs, + "sglang": sglang_configs, +} + + +@dataclass +class RankPauseScenario: + """One row in the test matrix.""" + + # Label used in pytest param IDs. + label: str + # Backend: trtllm / vllm / sglang. Selects config dict + rank-discovery pattern. + backend: str + # Which {backend}_configs entry to launch. + base_config_key: str + # Which system-port index to monitor (0 = prefill/single worker, 1 = decode). + system_port_index: int + # Whether to enable canary in the worker env. + canary_enabled: bool + # "detect" → /health must flip to 503 after SIGSTOP. + # "miss" → /health must stay 200 after SIGSTOP. + expected: str + + +# Coverage: {backend} × {agg, disagg-prefill, disagg-decode} with canary ENABLED +# (expected=detect), plus one canary-OFF negative control on trtllm to prove the +# harness doesn't false-positive. Agg uses the "aggregated" config (port 0 is +# the single worker's system port); prefill/decode use "disaggregated_same_gpu" +# which spawns both workers on one GPU (ports 0 and 1). +SCENARIOS: list[RankPauseScenario] = [] +for _backend in ("trtllm", "vllm", "sglang"): + SCENARIOS.extend( + [ + RankPauseScenario( + label=f"{_backend}-agg-canary-on", + backend=_backend, + base_config_key="aggregated", + system_port_index=0, + canary_enabled=True, + expected="detect", + ), + RankPauseScenario( + label=f"{_backend}-disagg-prefill-canary-on", + backend=_backend, + base_config_key="disaggregated_same_gpu", + system_port_index=0, # DYN_SYSTEM_PORT1 = prefill + canary_enabled=True, + expected="detect", + ), + RankPauseScenario( + label=f"{_backend}-disagg-decode-canary-on", + backend=_backend, + base_config_key="disaggregated_same_gpu", + system_port_index=1, # DYN_SYSTEM_PORT2 = decode + canary_enabled=True, + expected="detect", + ), + ] + ) + +# Negative control — canary off; /health must NOT flip. One representative +# scenario is enough to prove the harness distinguishes detect/miss. +SCENARIOS.append( + RankPauseScenario( + label="trtllm-agg-canary-off", + backend="trtllm", + base_config_key="aggregated", + system_port_index=0, + canary_enabled=False, + expected="miss", + ) +) + + +def _find_engine_rank_pid( + parent_pid: int, patterns: tuple[str, ...], timeout_s: float = 30.0 +) -> int: + """Locate the engine-rank subprocess under the test's worker parent. + + Each backend spawns a child process whose command line contains a + recognizable marker (see ``_RANK_PATTERNS``). We wait up to + ``timeout_s`` for it to appear so callers don't race startup. + """ + deadline = time.monotonic() + timeout_s + last_err: Exception | None = None + while time.monotonic() < deadline: + try: + parent = psutil.Process(parent_pid) + for child in parent.children(recursive=True): + try: + cmd = " ".join(child.cmdline()) + except psutil.Error: + continue + if any(p in cmd for p in patterns): + return child.pid + except psutil.NoSuchProcess as e: + last_err = e + time.sleep(0.5) + raise RuntimeError( + f"Could not find engine-rank child of pid={parent_pid} matching " + f"{patterns} within {timeout_s}s. Last error: {last_err}" + ) + + +def _health_status(url: str, timeout: float = 2.0) -> int: + """Return HTTP status from /health, or 0 on connection error.""" + try: + r = requests.get(url, timeout=timeout) + return r.status_code + except requests.exceptions.RequestException: + return 0 + + +def _wait_for_status(url: str, target: int, deadline_s: float) -> int: + """Poll /health until status == target or deadline reached. Return last status.""" + deadline = time.monotonic() + deadline_s + last = -1 + while time.monotonic() < deadline: + last = _health_status(url) + if last == target: + return last + time.sleep(1.0) + return last + + +@pytest.mark.e2e +@pytest.mark.nightly +@pytest.mark.gpu_1 +@pytest.mark.parametrize( + "scenario", + SCENARIOS, + ids=lambda s: s.label if isinstance(s, RankPauseScenario) else str(s), +) +@pytest.mark.parametrize("num_system_ports", [2], indirect=True) +def test_canary_detects_rank_pause( + scenario: RankPauseScenario, + request: Any, + runtime_services_dynamic_ports, # noqa: ANN001 + dynamo_dynamic_ports, # noqa: ANN001 + num_system_ports, # noqa: ANN001 + predownload_models, # noqa: ANN001 +) -> None: + configs = _CONFIGS_BY_BACKEND[scenario.backend] + base = configs[scenario.base_config_key] + config = dataclasses.replace(base, frontend_port=dynamo_dynamic_ports.frontend_port) + config.env.update( + { + "MODEL_PATH": config.model, + "SERVED_MODEL_NAME": config.model, + "DYN_HEALTH_CHECK_ENABLED": "true" if scenario.canary_enabled else "false", + } + ) + + system_ports = [int(p) for p in dynamo_dynamic_ports.system_ports] + assert len(system_ports) > scenario.system_port_index, ( + f"scenario wants system_port_index={scenario.system_port_index} " + f"but only {len(system_ports)} ports are allocated" + ) + target_port = system_ports[scenario.system_port_index] + health_url = f"http://localhost:{target_port}/health" + logger.info( + "[%s] backend=%s health_url=%s canary=%s expected=%s", + scenario.label, + scenario.backend, + health_url, + scenario.canary_enabled, + scenario.expected, + ) + + extra_env: dict[str, str] = {} + for i, p in enumerate(system_ports, start=1): + extra_env[f"DYN_SYSTEM_PORT{i}"] = str(p) + extra_env["DYN_SYSTEM_PORT"] = str(system_ports[0]) + extra_env["DYN_HTTP_PORT"] = str(dynamo_dynamic_ports.frontend_port) + extra_env["DYN_HEALTH_CHECK_ENABLED"] = ( + "true" if scenario.canary_enabled else "false" + ) + + patterns = _RANK_PATTERNS[scenario.backend] + + with EngineProcess.from_script(config, request, extra_env=extra_env) as proc: + # 1. Wait for the worker to become healthy (Ready) before we pause. + status = _wait_for_status(health_url, 200, STARTUP_READY_BUDGET_S) + assert status == 200, ( + f"[{scenario.label}] worker never became healthy " + f"(last status={status}) within {STARTUP_READY_BUDGET_S}s" + ) + + # 2. Find the engine rank subprocess and SIGSTOP it. + rank_pid = _find_engine_rank_pid(proc.proc.pid, patterns) + logger.info( + "[%s] pausing engine rank pid=%d (patterns=%s)", + scenario.label, + rank_pid, + patterns, + ) + os.kill(rank_pid, signal.SIGSTOP) + try: + # 3. Give the canary (or lack thereof) time to notice. + if scenario.expected == "detect": + status = _wait_for_status(health_url, 503, PAUSE_DETECT_BUDGET_S) + assert status == 503, ( + f"[{scenario.label}] canary FAILED to detect rank pause: " + f"/health returned {status} (expected 503) within " + f"{PAUSE_DETECT_BUDGET_S}s" + ) + else: # miss + # /health must stay 200 throughout the window. Sample + # periodically so a transient glitch doesn't masquerade as + # "miss". + deadline = time.monotonic() + PAUSE_DETECT_BUDGET_S + flips = 0 + while time.monotonic() < deadline: + s = _health_status(health_url) + if s != 200: + flips += 1 + time.sleep(2.0) + assert flips == 0, ( + f"[{scenario.label}] expected canary to MISS rank pause " + f"(no active probe), but /health flipped away from 200 " + f"{flips} time(s). Harness may be false-detecting." + ) + finally: + # 4. Always resume the rank so teardown can complete cleanly. + try: + os.kill(rank_pid, signal.SIGCONT) + except ProcessLookupError: + pass + + # 5. After resume, for the detect case, confirm /health returns to 200. + if scenario.expected == "detect": + status = _wait_for_status(health_url, 200, RESUME_RECOVER_BUDGET_S) + assert status == 200, ( + f"[{scenario.label}] /health did not recover after SIGCONT " + f"(last status={status}) within {RESUME_RECOVER_BUDGET_S}s" + )