From 8d9a4afed93646b9e7cb262540663a90ad7f0d9f Mon Sep 17 00:00:00 2001 From: zhumingjue138 Date: Thu, 28 May 2026 21:41:33 +0800 Subject: [PATCH] [Test] Add scenarios for L5 reliability test (#3729) Signed-off-by: zhumingjue --- tests/dfx/reliability/helpers.py | 477 +++++++++++++++++- .../test_reliability_qwen3_omni.py | 384 +++++++++++--- .../dfx/reliability/test_reliability_wan22.py | 324 ++++++++++-- 3 files changed, 1045 insertions(+), 140 deletions(-) diff --git a/tests/dfx/reliability/helpers.py b/tests/dfx/reliability/helpers.py index 9ebd0ce97..fd01e8a8e 100644 --- a/tests/dfx/reliability/helpers.py +++ b/tests/dfx/reliability/helpers.py @@ -4,10 +4,16 @@ This module keeps fault injection callable from tests directly: - GPU OOM (CUDA sidecar memory hog) - process kill by pattern and signal - post-ready hooks via ``fault_injector`` / ``omni_server_after_fault`` fixtures + +Worker PID classification (``worker_pids`` in snapshots) is still used for logging / +markers; post-fault **process** cleanup assertions use the full captured +``tree_pids`` (serve root + descendants at fault time) — see +:func:`assert_no_server_tree_process_residual_and_gpu_release`. """ from __future__ import annotations +import concurrent.futures import http.client import json import logging @@ -27,6 +33,16 @@ import pytest logger = logging.getLogger(__name__) +# Substrings matched against ``psutil.Process.name`` + argv text for classifying GPU +# workers under the test server tree. Extend per-suite via +# ``server.reliability_worker_markers_extra`` or replace via +# ``server.reliability_worker_markers`` (see ``_resolve_runtime_worker_markers``). +DEFAULT_RUNTIME_WORKER_MARKERS: tuple[str, ...] = ( + "multiprocessing.spawn", + # Covers ``VLLM::Worker``, ``VLLM::StageEngineCoreProc_*`` (Omni stage engines), etc. + "VLLM::", +) + @dataclass class OomHandle: @@ -452,6 +468,61 @@ def _safe_proc_info(pid: int) -> tuple[str, str]: return "", "" +def _normalize_worker_marker_substrings(seq: Sequence[Any], *, context: str) -> tuple[str, ...]: + """Return deduped non-empty marker substrings; raises if any entry is blank.""" + out: list[str] = [] + for item in seq: + s = str(item).strip() + if not s: + raise ValueError(f"{context}: worker marker must be a non-empty string, got {item!r}") + out.append(s) + return tuple(dict.fromkeys(out)) + + +def _resolve_runtime_worker_markers( + server: Any, + injector_markers_extra: Sequence[str] | None, +) -> tuple[str, ...]: + """Resolve which substrings classify worker PIDs for fault snapshots. + + If ``server.reliability_worker_markers`` is set to a sequence, **only** those + markers are used (no defaults). Otherwise defaults from + :data:`DEFAULT_RUNTIME_WORKER_MARKERS` are merged with: + + - ``server.reliability_worker_markers_extra`` when present, and + - ``injector_markers_extra`` from :func:`make_server_root_kill_fault_injector` / + :func:`make_server_tree_kill_fault_injector`. + """ + replace = getattr(server, "reliability_worker_markers", None) + if replace is not None: + return _normalize_worker_marker_substrings(replace, context="server.reliability_worker_markers") + + merged: list[str] = [] + merged.extend(DEFAULT_RUNTIME_WORKER_MARKERS) + extra_attr = getattr(server, "reliability_worker_markers_extra", None) + if extra_attr is not None: + merged.extend( + _normalize_worker_marker_substrings(extra_attr, context="server.reliability_worker_markers_extra") + ) + if injector_markers_extra is not None: + merged.extend( + _normalize_worker_marker_substrings( + injector_markers_extra, + context="injector worker_markers_extra", + ) + ) + return tuple(dict.fromkeys(merged)) + + +def _pid_looks_like_runtime_worker(pid: int, markers: Sequence[str]) -> bool: + """True when process name or argv contains any resolved worker marker substring.""" + if not markers: + return False + name, cmdline = _safe_proc_info(pid) + text = f"{name} {cmdline}" + return any(marker in text for marker in markers) + + def _list_server_process_tree(server: Any) -> list[int]: """Return [root, descendants...] PIDs for the current test server instance.""" root_proc = getattr(server, "proc", None) @@ -468,6 +539,29 @@ def _list_server_process_tree(server: Any) -> list[int]: return [root_pid, *descendants] +def _pids_in_server_tree_substring_match(server: Any, pattern: str, *, limit: int) -> list[int]: + """Return up to ``limit`` PIDs in the server tree whose name+cmdline contains ``pattern`` literally. + + ``pgrep -f`` matches the process command line used by procps and does not consult the + short process name; vLLM-Omni often exposes titles like ``VLLM::Worker`` or + ``VLLM::StageEngineCoreProc_*`` via prctl/setproctitle in ``ps``/``name()`` only. + This helper keeps kill injection aligned with :func:`_safe_proc_info` / fault snapshots. + """ + if limit <= 0: + return [] + tree = _list_server_process_tree(server) + if not tree or not pattern: + return [] + hits: list[int] = [] + for pid in tree: + if len(hits) >= limit: + break + name, cmdline = _safe_proc_info(int(pid)) + if pattern in f"{name} {cmdline}": + hits.append(int(pid)) + return hits + + def _log_server_process_tree(server: Any) -> None: """Print server process tree for debugging fault injection targets.""" pids = _list_server_process_tree(server) @@ -486,6 +580,315 @@ FaultInjector = Callable[[Any], None] """Callable invoked with the live ``OmniServer`` after it is ready (see ``omni_server_after_fault``).""" +def run_fault_injection_with_rate_load( + *, + submit_request: Callable[[], Any], + inject_fault: Callable[[], None], + num_requests: int = 10, + request_rate: float = 0.3, + submit_interval_sec: float | None = None, + completion_timeout_sec: float = 120.0, +) -> dict[str, Any]: + """Submit requests with a fixed rate, then inject fault once an in-flight request is observed.""" + if num_requests <= 0: + raise ValueError("num_requests must be > 0") + if request_rate <= 0: + raise ValueError("request_rate must be > 0") + + interval_sec = submit_interval_sec if submit_interval_sec is not None else (1.0 / request_rate) + if interval_sec < 0: + raise ValueError("submit_interval_sec must be >= 0") + + max_workers = min(max(2, num_requests), 8) + injected = False + futures: list[concurrent.futures.Future[Any]] = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + for idx in range(num_requests): + futures.append(executor.submit(submit_request)) + in_flight = any(not future.done() for future in futures) + if in_flight and not injected: + inject_fault() + injected = True + if idx != num_requests - 1 and interval_sec > 0: + time.sleep(interval_sec) + + if not injected: + in_flight = any(not future.done() for future in futures) + if in_flight: + inject_fault() + injected = True + + if not injected: + pytest.skip( + "no in-flight request observed before fault injection; " + "increase request cost or request rate for this environment" + ) + + done, pending = concurrent.futures.wait( + futures, + timeout=completion_timeout_sec, + return_when=concurrent.futures.ALL_COMPLETED, + ) + assert not pending, ( + "some in-flight-load requests did not converge after fault injection: " + f"pending={len(pending)} done={len(done)}" + ) + failure_observed = False + completed = 0 + success = 0 + exceptions = 0 + for future in futures: + completed += 1 + try: + future.result() + success += 1 + except Exception: # noqa: BLE001 + exceptions += 1 + failure_observed = True + return { + "num_submitted": num_requests, + "completed": completed, + "success": success, + "exceptions": exceptions, + "failure_observed": failure_observed, + "inflight_observed": True, + } + + +def list_alive_pids(pids: Sequence[int]) -> list[int]: + """Return PIDs from ``pids`` that still exist in the kernel and are **not** zombies. + + After SIGTERM/SIGKILL the serve child may exit before the test harness calls + ``Popen.wait()``; until reaped, ``psutil.pid_exists`` stays true for the defunct + slot. Those PIDs must not count as "residual server processes" for leak checks. + """ + out: list[int] = [] + for pid in pids: + pid_i = int(pid) + if not psutil.pid_exists(pid_i): + continue + try: + if psutil.Process(pid_i).status() == psutil.STATUS_ZOMBIE: + continue + except psutil.Error: + continue + out.append(pid_i) + return out + + +def query_gpu_compute_pid_used_memory_mb() -> dict[int, int] | None: + """Query NVIDIA compute-process memory map (pid -> used MB). + + Returns ``None`` when ``nvidia-smi`` is unavailable/unreadable on current host. + """ + out = subprocess.run( + ["nvidia-smi", "--query-compute-apps=pid,used_memory", "--format=csv,noheader,nounits"], + check=False, + capture_output=True, + text=True, + ) + if out.returncode != 0: + logger.warning("[reliability][gpu] nvidia-smi query failed: %s", out.stderr.strip()) + return None + + pid_to_mem_mb: dict[int, int] = {} + for line in out.stdout.splitlines(): + row = line.strip() + if not row: + continue + pieces = [part.strip() for part in row.split(",", maxsplit=1)] + if len(pieces) != 2: + continue + pid_raw, mem_raw = pieces + if not pid_raw.isdigit(): + continue + mem_digits = "".join(ch for ch in mem_raw if ch.isdigit()) + if not mem_digits: + continue + pid_to_mem_mb[int(pid_raw)] = int(mem_digits) + return pid_to_mem_mb + + +def worker_residual_timeout_after_kill_signal(signal_name: str) -> float: + """Wall-clock budget for fault-snapshot PIDs / GPU to clear after ``signal_name``.""" + if signal_name == "SIGKILL": + return 30.0 + if signal_name == "SIGINT": + return 120.0 + if signal_name == "SIGTERM": + return 90.0 + return 30.0 + + +def assert_no_server_tree_process_residual_and_gpu_release( + server: Any, + *, + scenario: str, + timeout_sec: float = 30.0, + poll_interval_sec: float = 0.5, +) -> None: + """Assert no live PIDs remain from the fault-time server tree and no GPU use by those PIDs. + + Uses ``reliability_fault_snapshot["tree_pids"]`` (root + all descendants captured at + injection time), not the marker-filtered ``worker_pids`` subset — so helpers like + ``multiprocessing.resource_tracker`` are included in the residual check. + + **Zombie** PIDs (exited children not yet reaped by the test ``Popen``) are excluded + from the "alive" list; ``psutil.pid_exists`` alone would false-positive on them. + + Note: if a child is reparented to PID 1 while staying alive, its PID is unchanged + and remains detected; if it **exits and a new unrelated process reuses the same + numeric PID**, this check may false-positive (rare on short windows). + """ + snapshot = getattr(server, "reliability_fault_snapshot", None) + if not isinstance(snapshot, dict): + pytest.fail(f"[{scenario}] missing reliability fault snapshot on server") + tree_pids = [int(pid) for pid in snapshot.get("tree_pids", [])] + if not tree_pids: + pytest.skip(f"[{scenario}] no server process tree PIDs captured for this run") + + tree_pid_set = set(tree_pids) + deadline = time.monotonic() + timeout_sec + last_alive: list[int] = [] + last_gpu_leaks: dict[int, int] = {} + while time.monotonic() < deadline: + last_alive = list_alive_pids(tree_pids) + gpu_map = query_gpu_compute_pid_used_memory_mb() + if gpu_map is None: + pytest.skip(f"[{scenario}] nvidia-smi unavailable; skip GPU release assertion") + last_gpu_leaks = {pid: mem_mb for pid, mem_mb in gpu_map.items() if pid in tree_pid_set and mem_mb > 0} + if not last_alive and not last_gpu_leaks: + return + time.sleep(poll_interval_sec) + + assert not last_alive, f"[{scenario}] residual server-tree processes remain alive: {last_alive}" + assert not last_gpu_leaks, f"[{scenario}] server-tree PID GPU memory not released: {last_gpu_leaks}" + + +def assert_no_worker_residual_and_gpu_release( + server: Any, + *, + scenario: str, + timeout_sec: float = 30.0, + poll_interval_sec: float = 0.5, +) -> None: + """Deprecated: use :func:`assert_no_server_tree_process_residual_and_gpu_release`.""" + assert_no_server_tree_process_residual_and_gpu_release( + server, + scenario=scenario, + timeout_sec=timeout_sec, + poll_interval_sec=poll_interval_sec, + ) + + +def _capture_server_fault_snapshot( + server: Any, + *, + worker_markers_extra: Sequence[str] | None = None, +) -> dict[str, list[int]]: + """Capture current server process snapshot for post-fault assertions. + + ``tree_pids`` is ``[root, ...descendants]`` and is used by + :func:`assert_no_server_tree_process_residual_and_gpu_release`. + ``worker_pids`` is the marker-filtered subset for debugging only. + """ + markers = _resolve_runtime_worker_markers(server, worker_markers_extra) + tree_pids = _list_server_process_tree(server) + root_pid = tree_pids[0] if tree_pids else None + worker_pids: list[int] = [] + for pid in tree_pids: + if root_pid is not None and int(pid) == int(root_pid): + continue + if _pid_looks_like_runtime_worker(pid, markers): + worker_pids.append(pid) + snapshot = { + "root_pids": [root_pid] if root_pid is not None else [], + "tree_pids": tree_pids, + "worker_pids": worker_pids, + "worker_markers": list(markers), + } + setattr(server, "reliability_fault_snapshot", snapshot) + return snapshot + + +def make_server_root_kill_fault_injector( + *, + signal_name: str = "SIGKILL", + post_kill_wait_seconds: float = 0.0, + worker_markers_extra: Sequence[str] | None = None, +) -> FaultInjector: + """Kill only the current server root process (simulate first strike on PID1-like process). + + ``worker_markers_extra`` is merged into worker PID detection for the fault snapshot + (after :data:`DEFAULT_RUNTIME_WORKER_MARKERS` unless + ``server.reliability_worker_markers`` replaces the set entirely). + """ + + def _inject(server: Any) -> None: + _log_server_process_tree(server) + snapshot = _capture_server_fault_snapshot(server, worker_markers_extra=worker_markers_extra) + root_pids = snapshot["root_pids"] + if not root_pids: + pytest.skip("no server root process found for root-kill injection") + root_pid = int(root_pids[0]) + sig = getattr(signal, signal_name, None) + if sig is None: + raise ValueError(f"Unsupported signal_name: {signal_name}") + name, cmdline = _safe_proc_info(root_pid) + print( + f"[reliability][process-kill] root-kill pid={root_pid} name={name} signal={signal_name} cmdline={cmdline}", + flush=True, + ) + os.kill(root_pid, sig) + if post_kill_wait_seconds > 0: + time.sleep(post_kill_wait_seconds) + + return _inject + + +def make_server_tree_kill_fault_injector( + *, + signal_name: str = "SIGKILL", + post_kill_wait_seconds: float = 0.0, + inter_kill_wait_seconds: float = 0.2, + worker_markers_extra: Sequence[str] | None = None, +) -> FaultInjector: + """Kill current server root first, then kill all remaining descendants. + + See :func:`make_server_root_kill_fault_injector` for ``worker_markers_extra``. + """ + + def _inject(server: Any) -> None: + _log_server_process_tree(server) + snapshot = _capture_server_fault_snapshot(server, worker_markers_extra=worker_markers_extra) + tree_pids = [int(pid) for pid in snapshot["tree_pids"]] + if not tree_pids: + pytest.skip("no server process tree found for tree-kill injection") + + sig = getattr(signal, signal_name, None) + if sig is None: + raise ValueError(f"Unsupported signal_name: {signal_name}") + + root_pid = tree_pids[0] + kill_order = [root_pid, *[pid for pid in tree_pids if pid != root_pid]] + for pid in kill_order: + if not psutil.pid_exists(pid): + continue + name, cmdline = _safe_proc_info(pid) + print( + f"[reliability][process-kill] tree-kill pid={pid} name={name} signal={signal_name} cmdline={cmdline}", + flush=True, + ) + os.kill(pid, sig) + if inter_kill_wait_seconds > 0: + time.sleep(inter_kill_wait_seconds) + + if post_kill_wait_seconds > 0: + time.sleep(post_kill_wait_seconds) + + return _inject + + def make_process_kill_fault_injector( *, grep_patterns: str | Sequence[str], @@ -493,11 +896,18 @@ def make_process_kill_fault_injector( limit: int = 1, post_kill_wait_seconds: float = 0.0, ) -> FaultInjector: - """Build a post-ready injector that kills processes matched by ``pgrep -f``. + """Build a post-ready injector that kills processes matched by ordered ``grep_patterns``. - Tries each pattern in order until at least one PID is killed. If none match, - the returned callable issues ``pytest.skip`` (same behavior as the previous - inline reliability test). + Matching is tried in two phases per pattern (first hit wins): + + 1. **Server process tree** (``psutil``): literal substring match against + ``Process.name()`` plus argv text. This matches how :func:`_safe_proc_info` + logs processes and catches vLLM-style titles (``VLLM::...``) that often do not + appear in ``pgrep -f``\'s command-line view. + 2. **``pgrep -f``** (legacy): scoped to the server PID tree when it is known; + uses procps regular-expression rules for the pattern. + + If neither phase finds a target, the returned callable issues ``pytest.skip``. Args: grep_patterns: One pattern or an ordered list of patterns. @@ -514,9 +924,45 @@ def make_process_kill_fault_injector( logger.warning( "[reliability][process-kill] no server process tree found; fallback to global pgrep matching" ) + + def _kill_and_wait(filtered: list[int], pattern: str, *, source: str) -> None: + sig = getattr(signal, signal_name, None) + if sig is None: + raise ValueError(f"Unsupported signal_name: {signal_name}") + for pid in filtered: + name, cmdline = _safe_proc_info(pid) + print( + f"[reliability][process-kill] killing pid={pid} name={name} signal={signal_name} " + f"source={source} cmdline={cmdline}", + flush=True, + ) + os.kill(pid, sig) + print( + f"[reliability][process-kill] matched pattern={pattern!r} source={source} " + f"killed_pids={filtered} killed_count={len(filtered)}", + flush=True, + ) + if post_kill_wait_seconds > 0: + print( + f"[reliability][process-kill] waiting {post_kill_wait_seconds:.2f}s after kill", + flush=True, + ) + time.sleep(post_kill_wait_seconds) + for pattern in patterns: print( - f"[reliability][process-kill] trying pattern={pattern} signal={signal_name} limit={limit}", + f"[reliability][process-kill] trying server_tree substring pattern={pattern!r} " + f"signal={signal_name} limit={limit}", + flush=True, + ) + tree_hits = _pids_in_server_tree_substring_match(server, pattern, limit=limit) + if tree_hits: + _kill_and_wait(tree_hits, pattern, source="server_tree") + return + + for pattern in patterns: + print( + f"[reliability][process-kill] trying pgrep -f pattern={pattern!r} signal={signal_name} limit={limit}", flush=True, ) pids = inject_process_kill( @@ -535,26 +981,7 @@ def make_process_kill_fault_injector( ) continue if filtered: - sig = getattr(signal, signal_name, None) - if sig is None: - raise ValueError(f"Unsupported signal_name: {signal_name}") - for pid in filtered: - name, cmdline = _safe_proc_info(pid) - print( - f"[reliability][process-kill] killing pid={pid} name={name} signal={signal_name} cmdline={cmdline}", - flush=True, - ) - os.kill(pid, sig) - print( - f"[reliability][process-kill] matched pattern={pattern} killed_pids={filtered} killed_count={len(filtered)}", - flush=True, - ) - if post_kill_wait_seconds > 0: - print( - f"[reliability][process-kill] waiting {post_kill_wait_seconds:.2f}s after kill", - flush=True, - ) - time.sleep(post_kill_wait_seconds) + _kill_and_wait(filtered, pattern, source="pgrep") return logger.warning( "[reliability][process-kill] no process matched patterns=%s signal=%s limit=%s", diff --git a/tests/dfx/reliability/test_reliability_qwen3_omni.py b/tests/dfx/reliability/test_reliability_qwen3_omni.py index 37f63b942..49028a03f 100644 --- a/tests/dfx/reliability/test_reliability_qwen3_omni.py +++ b/tests/dfx/reliability/test_reliability_qwen3_omni.py @@ -21,13 +21,19 @@ from tests.dfx.conftest import ( resolve_oom_device_spec, ) from tests.dfx.reliability.helpers import ( + FaultInjector, + assert_no_server_tree_process_residual_and_gpu_release, extract_openai_error_contract_from_bytes, get_health_raw, inject_gpu_oom, make_process_kill_fault_injector, + make_server_root_kill_fault_injector, + make_server_tree_kill_fault_injector, post_chat_completions_raw, post_json_raw, + run_fault_injection_with_rate_load, stop_gpu_oom_hogs, + worker_residual_timeout_after_kill_signal, ) from tests.helpers.mark import hardware_test from tests.helpers.media import generate_synthetic_audio, generate_synthetic_image, generate_synthetic_video @@ -97,7 +103,73 @@ FAULT_ERROR_KEYWORDS = ( "500", "503", ) -RUNTIME_WORKER_PATTERN = "VLLM::Worker" +PROCESS_KILL_ERROR_KEYWORDS = ( + "timeout", + "did not complete within", + "connection", + "engine", + "orchestrator", + "dead", + "internal", + "500", + "503", +) +# Prefix match for vLLM-titled children (``VLLM::Worker``, ``VLLM::StageEngineCoreProc_*``, …). +# Must be a literal substring of ``name``/``cmdline`` — no trailing whitespace (would break match). +RUNTIME_WORKER_PATTERN = "VLLM::" +# RFC#2366 signal x target matrix: +# - worker: SIGTERM, SIGKILL +# - serve-root no-load: SIGINT; SIGTERM/SIGKILL skipped (vllm issue#43060) +# - serve-root with-load: SIGINT; SIGTERM skipped (issue#3683); SIGKILL skipped (issue#43060) +# - serve-tree no-load: SIGTERM/SIGKILL +# - serve-tree with-load: SIGKILL; SIGTERM skipped (issue#3683) +_SERVE_ROOT_NON_SIGINT_SKIP = pytest.mark.skip(reason="vllm issue#43060") +_WITH_LOAD_SIGTERM_SKIP = pytest.mark.skip(reason="issue#3683") +SERVE_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm"), + pytest.param("SIGINT", id="sigint"), + pytest.param("SIGKILL", id="sigkill", marks=_SERVE_ROOT_NON_SIGINT_SKIP), +] +SERVE_WITH_LOAD_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm", marks=_WITH_LOAD_SIGTERM_SKIP), + pytest.param("SIGINT", id="sigint", marks=_WITH_LOAD_SIGTERM_SKIP), + pytest.param("SIGKILL", id="sigkill", marks=_WITH_LOAD_SIGTERM_SKIP), +] +TREE_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm"), + pytest.param("SIGKILL", id="sigkill"), +] +TREE_WITH_LOAD_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm", marks=_WITH_LOAD_SIGTERM_SKIP), + pytest.param("SIGKILL", id="sigkill"), +] +WORKER_SIGNAL_FAULT_PARAMS = [ + pytest.param( + make_process_kill_fault_injector( + grep_patterns=RUNTIME_WORKER_PATTERN, + signal_name="SIGTERM", + limit=1, + post_kill_wait_seconds=2.0, + ), + id="runtime_process_chain_sigterm", + ), + pytest.param( + make_process_kill_fault_injector( + grep_patterns=RUNTIME_WORKER_PATTERN, + signal_name="SIGKILL", + limit=1, + post_kill_wait_seconds=2.0, + ), + id="runtime_process_chain_sigkill", + ), +] +INFLIGHT_INJECTION_REQUEST_RATE = 0.3 +INFLIGHT_INJECTION_REQUEST_COUNT = 10 + +# Post-fault chat probe: ``MAX_ELAPSED`` aligns with HTTP read timeout; assert allows +1s jitter +# when the client raises at timeout (e.g. ~30.03s). +_PROCESS_KILL_WORKER_CHAT_FF_HTTP_TIMEOUT_SEC = 30 +_PROCESS_KILL_WORKER_CHAT_FF_MAX_ELAPSED_SEC = 30 class _HasServeArgs(Protocol): @@ -194,6 +266,55 @@ def _looks_like_server_unreachable(exc: BaseException) -> bool: return "connection refused" in msg or "actively refused" in msg +def _chat_request_config(omni_server: _HasServeArgs) -> dict[str, Any]: + messages = dummy_messages_from_mix_data( + system_prompt=_get_system_prompt(), + content_text="What is the capital of China? Answer in one short sentence.", + ) + return { + "model": omni_server.model, + "messages": messages, + "stream": False, + "modalities": ["text"], + } + + +def _assert_post_fault_chat_fast_fail(host: str, port: int, *, model: str, scenario: str) -> None: + payload = { + "model": model, + "messages": [{"role": "user", "content": "post-fault fast-fail check"}], + "stream": False, + "modalities": ["text"], + } + start = time.monotonic() + try: + status, body = post_json_raw(host, port, "/v1/chat/completions", payload, timeout_sec=20) + elapsed = time.monotonic() - start + assert elapsed < 15, f"[{scenario} fast_fail] /v1/chat/completions did not fail fast: {elapsed:.2f}s" + assert status >= 500, ( + f"[{scenario} fast_fail] expected server-side error after fault, got status={status}, body={body[:200]!r}" + ) + except Exception: # noqa: BLE001 + elapsed = time.monotonic() - start + assert elapsed < 15, f"[{scenario} fast_fail] exception was too slow after fault: {elapsed:.2f}s" + + +def _assert_post_fault_health_terminal(host: str, port: int, *, scenario: str) -> None: + deadline = time.monotonic() + 20.0 + last_observation = "" + while time.monotonic() < deadline: + try: + status, body = get_health_raw(host, port, timeout_sec=5) + last_observation = f"http={status}, body={body[:200]!r}" + if status == 503: + return + except Exception as exc: # noqa: BLE001 + last_observation = f"exception={exc!r}" + return + time.sleep(0.5) + pytest.fail(f"[{scenario} health] no terminal post-fault health observed: {last_observation}") + + QWEN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, DEPLOY_CONFIGS_DIR) @@ -354,17 +475,7 @@ def test_reliability_fault_gpu_oom_concurrent_pressure_failure(omni_server_funct @pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") @pytest.mark.parametrize( "fault_injector", - [ - pytest.param( - make_process_kill_fault_injector( - grep_patterns="VLLM::Worker", - signal_name="SIGKILL", - limit=1, - post_kill_wait_seconds=2.0, - ), - id="runtime_process_chain", - ), - ], + WORKER_SIGNAL_FAULT_PARAMS, indirect=True, ) @pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) @@ -385,7 +496,7 @@ def test_reliability_fault_process_kill_request_failure( try: openai_client_function.send_omni_request(request_config, request_num=1) except Exception as exc: - assert_fault_exception(exc, FAULT_ERROR_KEYWORDS) + assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS) else: pytest.fail("expected request failure after process-kill injection") @@ -394,48 +505,22 @@ def test_reliability_fault_process_kill_request_failure( @pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") @pytest.mark.parametrize( "fault_injector", - [ - pytest.param( - make_process_kill_fault_injector( - grep_patterns="VLLM::Worker", - signal_name="SIGKILL", - limit=1, - post_kill_wait_seconds=2.0, - ), - id="runtime_process_chain", - ), - ], + WORKER_SIGNAL_FAULT_PARAMS, indirect=True, ) @pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) def test_reliability_fault_process_kill_health_fast_fail_and_concurrent( omni_server_after_fault_function, ) -> None: - """Black-box: after worker SIGKILL, /health→503, chat fails fast, concurrent chat does not hang.""" + """Black-box: after worker process kill, chat fails fast, concurrent chat does not hang; /health→503 last. + + ``SIGTERM`` on a stage engine often exceeds a 15s client-visible path; ``SIGKILL`` is + typically quicker. See module constants ``_PROCESS_KILL_WORKER_CHAT_FF_*``. + """ host = omni_server_after_fault_function.host port = omni_server_after_fault_function.port model = omni_server_after_fault_function.model - deadline = time.monotonic() + 20.0 - last_observation = "" - saw_503 = False - health_final_status: int | None = None - health_final_body = b"" - while time.monotonic() < deadline: - try: - status, body = get_health_raw(host, port, timeout_sec=5) - last_observation = f"http={status}, body={body[:200]!r}" - health_final_status, health_final_body = status, body - if status == 503: - saw_503 = True - break - except Exception as exc: # noqa: BLE001 - last_observation = f"exception={exc!r}" - time.sleep(0.5) - assert saw_503, ( - f"[process_kill health] expected /health to become 503 after fault injection, got {last_observation}" - ) - payload = { "model": model, "messages": [{"role": "user", "content": "Say hello in one short sentence."}], @@ -444,20 +529,28 @@ def test_reliability_fault_process_kill_health_fast_fail_and_concurrent( } ff_status: int | None = None ff_body = b"" - ff_exc: BaseException | None = None start = time.monotonic() try: - ff_status, ff_body = post_json_raw(host, port, "/v1/chat/completions", payload, timeout_sec=20) + ff_status, ff_body = post_json_raw( + host, + port, + "/v1/chat/completions", + payload, + timeout_sec=_PROCESS_KILL_WORKER_CHAT_FF_HTTP_TIMEOUT_SEC, + ) elapsed = time.monotonic() - start - assert elapsed < 15, f"[process_kill fast_fail] request did not fail fast after fault: {elapsed:.2f}s" + assert elapsed <= _PROCESS_KILL_WORKER_CHAT_FF_MAX_ELAPSED_SEC + 1.0, ( + f"[process_kill fast_fail] request did not fail fast after fault: {elapsed:.2f}s" + ) assert ff_status >= 500, ( f"[process_kill fast_fail] expected server-side failure after fault, " f"got status={ff_status}, body={ff_body[:200]!r}" ) - except Exception as exc: - ff_exc = exc + except Exception: elapsed = time.monotonic() - start - assert elapsed < 15, f"[process_kill fast_fail] request exception was too slow after fault: {elapsed:.2f}s" + assert elapsed <= _PROCESS_KILL_WORKER_CHAT_FF_MAX_ELAPSED_SEC + 1.0, ( + f"[process_kill fast_fail] request exception was too slow after fault: {elapsed:.2f}s" + ) payload_json = json.dumps( { @@ -490,32 +583,195 @@ def test_reliability_fault_process_kill_health_fast_fail_and_concurrent( assert elapsed < 30, f"[process_kill concurrent] fault-time request convergence is too slow: {elapsed:.2f}s" fault_observed = False - conc_debug: list[Any] = [] for future in done: try: status, body = future.result() - conc_debug.append((status, body[:200])) if status >= 500: fault_observed = True except Exception as exc: - conc_debug.append(repr(exc)) + assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS) fault_observed = True - # DEBUG: remove before merge - print( - health_final_status, - health_final_body[:200], - ff_status, - ff_body[:200], - ff_exc, - conc_debug, - ) assert fault_observed, ( "[process_kill concurrent] expected at least one request to fail after process-kill fault injection" ) + deadline = time.monotonic() + 20.0 + last_observation = "" + final_health_status: int | None = None + while time.monotonic() < deadline: + try: + status, body = get_health_raw(host, port, timeout_sec=5) + last_observation = f"http={status}, body={body[:200]!r}" + final_health_status = status + if status == 503: + break + except Exception as exc: # noqa: BLE001 + last_observation = f"exception={exc!r}" + time.sleep(0.5) + if final_health_status != 503: + pytest.skip("issue#3050") + assert final_health_status == 503, ( + "[process_kill health] expected /health 503 after fatal fault, " + f"got status={final_health_status}, last_observation={last_observation}" + ) + @pytest.mark.slow -@pytest.mark.skip(reason="issue#2327") +@pytest.mark.skip(reason="issue#3683") +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize( + "fault_injector", + WORKER_SIGNAL_FAULT_PARAMS, + indirect=True, +) +@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) +def test_reliability_fault_process_kill_worker_with_load_request_failure( + omni_server_function, + openai_client_function, + fault_injector: FaultInjector, +) -> None: + request_config = _chat_request_config(omni_server_function) + scenario = "kill_worker_with_load" + load_result = run_fault_injection_with_rate_load( + submit_request=lambda: openai_client_function.send_omni_request(request_config, request_num=1), + inject_fault=lambda: fault_injector(omni_server_function), + num_requests=INFLIGHT_INJECTION_REQUEST_COUNT, + request_rate=INFLIGHT_INJECTION_REQUEST_RATE, + completion_timeout_sec=120.0, + ) + assert load_result["failure_observed"], ( + f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}" + ) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + + +@pytest.mark.slow +@hardware_test(res={"cuda": "H100"}, num_cards=2) +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", SERVE_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) +def test_reliability_fault_process_kill_serve_root_no_load_fast_fail_and_cleanup( + omni_server_function, + signal_name: str, +) -> None: + """Black-box: kill serve root without load; verify fast-fail/health/cleanup.""" + scenario = f"kill_serve_root_no_load_{signal_name.lower()}" + injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0) + injector(omni_server_function) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@hardware_test(res={"cuda": "H100"}, num_cards=2) +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", SERVE_WITH_LOAD_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) +def test_reliability_fault_process_kill_serve_root_with_load_fast_fail_and_cleanup( + omni_server_function, + openai_client_function, + signal_name: str, +) -> None: + request_config = _chat_request_config(omni_server_function) + scenario = f"kill_serve_root_with_load_{signal_name.lower()}" + injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0) + load_result = run_fault_injection_with_rate_load( + submit_request=lambda: openai_client_function.send_omni_request(request_config, request_num=1), + inject_fault=lambda: injector(omni_server_function), + num_requests=INFLIGHT_INJECTION_REQUEST_COUNT, + request_rate=INFLIGHT_INJECTION_REQUEST_RATE, + completion_timeout_sec=300.0, + ) + assert load_result["failure_observed"], ( + f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}" + ) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", TREE_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) +def test_reliability_fault_process_kill_tree_no_load_fast_fail_and_cleanup( + omni_server_function, + signal_name: str, +) -> None: + """Black-box: kill server tree without load; verify fast-fail/health/cleanup.""" + scenario = f"kill_serve_tree_no_load_{signal_name.lower()}" + injector = make_server_tree_kill_fault_injector( + signal_name=signal_name, + post_kill_wait_seconds=2.0, + inter_kill_wait_seconds=0.1, + ) + injector(omni_server_function) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", TREE_WITH_LOAD_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) +def test_reliability_fault_process_kill_tree_with_load_fast_fail_and_cleanup( + omni_server_function, + openai_client_function, + signal_name: str, +) -> None: + request_config = _chat_request_config(omni_server_function) + scenario = f"kill_serve_tree_with_load_{signal_name.lower()}" + injector = make_server_tree_kill_fault_injector( + signal_name=signal_name, + post_kill_wait_seconds=2.0, + inter_kill_wait_seconds=0.1, + ) + load_result = run_fault_injection_with_rate_load( + submit_request=lambda: openai_client_function.send_omni_request(request_config, request_num=1), + inject_fault=lambda: injector(omni_server_function), + num_requests=INFLIGHT_INJECTION_REQUEST_COUNT, + request_rate=INFLIGHT_INJECTION_REQUEST_RATE, + completion_timeout_sec=120.0, + ) + assert load_result["failure_observed"], ( + f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}" + ) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@pytest.mark.skip(reason="OOM injection causes the serving instance to exit") @hardware_test(res={"cuda": "H100"}, num_cards=2) @pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True) def test_reliability_fault_gpu_oom_state_converges_after_fault_removed( diff --git a/tests/dfx/reliability/test_reliability_wan22.py b/tests/dfx/reliability/test_reliability_wan22.py index e27db616f..84175429d 100644 --- a/tests/dfx/reliability/test_reliability_wan22.py +++ b/tests/dfx/reliability/test_reliability_wan22.py @@ -20,28 +20,25 @@ from tests.dfx.conftest import ( supports_video_generation, ) from tests.dfx.reliability.helpers import ( + FaultInjector, + assert_no_server_tree_process_residual_and_gpu_release, get_health_raw, inject_gpu_oom, make_process_kill_fault_injector, + make_server_root_kill_fault_injector, + make_server_tree_kill_fault_injector, + run_fault_injection_with_rate_load, stop_gpu_oom_hogs, + worker_residual_timeout_after_kill_signal, ) from tests.helpers.mark import hardware_test from tests.helpers.media import generate_synthetic_image RELIABILITY_SCENARIOS: list[dict[str, Any]] = [ { - "test_name": "wan22_t2v_reliability_default", + "test_name": "wan22_i2v_reliability_default", "server_params": { - "model": "Wan-AI/Wan2.2-T2V-A14B-Diffusers", - "server_args": [ - "--num-gpus", - "1", - "--boundary-ratio", - "0.875", - "--flow-shift", - "5.0", - "--disable-log-stats", - ], + "model": "Wan-AI/Wan2.2-I2V-A14B-Diffusers", }, } ] @@ -73,9 +70,114 @@ PROCESS_KILL_ERROR_KEYWORDS = ( "500", "503", ) +# RFC#2366 signal x target matrix (aligned with test_reliability_qwen3_omni.py): +# - worker: SIGTERM, SIGKILL +# - serve-root: SIGTERM, SIGINT; SIGKILL skipped (issue#3725) +# - serve-tree no-load: SIGTERM, SIGKILL +# - serve-tree with-load: SIGTERM, SIGKILL +_SERVE_ROOT_SIGKILL_SKIP = pytest.mark.skip(reason="issue#3725") +SERVE_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm"), + pytest.param("SIGINT", id="sigint"), + pytest.param("SIGKILL", id="sigkill", marks=_SERVE_ROOT_SIGKILL_SKIP), +] +TREE_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm"), + pytest.param("SIGKILL", id="sigkill"), +] +TREE_WITH_LOAD_SIGNAL_PARAMS = [ + pytest.param("SIGTERM", id="sigterm"), + pytest.param("SIGKILL", id="sigkill"), +] +WORKER_SIGNAL_FAULT_PARAMS = [ + pytest.param( + make_process_kill_fault_injector( + grep_patterns="vLLM-Omni::", + signal_name="SIGTERM", + limit=1, + post_kill_wait_seconds=2.0, + ), + id="runtime_process_chain_sigterm", + ), + pytest.param( + make_process_kill_fault_injector( + grep_patterns="vLLM-Omni::", + signal_name="SIGKILL", + limit=1, + post_kill_wait_seconds=2.0, + ), + id="runtime_process_chain_sigkill", + ), +] WAN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, E2E_STAGE_CONFIGS_DIR) DIFFUSION_VIDEO_PARAMS = [param for param in WAN_PARAMS if supports_video_generation(param.model)] +INFLIGHT_INJECTION_REQUEST_RATE = 0.3 +INFLIGHT_INJECTION_REQUEST_COUNT = 10 + + +def _video_request_config() -> dict[str, Any]: + image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}" + return { + "form_data": { + "prompt": "Generate a realistic road-driving video with camera motion.", + "width": 512, + "height": 512, + "fps": 8, + "num_frames": 8, + "guidance_scale": 1.0, + "flow_shift": 5.0, + "num_inference_steps": 4, + "seed": 42, + }, + "image_reference": image_data_url, + "stream": False, + } + + +def _assert_post_fault_video_fast_fail(host: str, port: int, *, scenario: str) -> None: + url = f"http://{host}:{port}/v1/videos" + payload = { + "prompt": "post-fault fast-fail check", + "width": "512", + "height": "512", + "fps": "8", + "num_frames": "8", + "num_inference_steps": "4", + } + start = time.monotonic() + try: + response = requests.post( + url, + data=payload, + headers={"Accept": "application/json"}, + timeout=20, + ) + elapsed = time.monotonic() - start + assert elapsed < 15, f"[{scenario} fast_fail] /v1/videos did not fail fast: {elapsed:.2f}s" + assert response.status_code >= 500, ( + f"[{scenario} fast_fail] expected server-side error after fault, " + f"got status={response.status_code}, body={response.text[:200]!r}" + ) + except Exception: # noqa: BLE001 + elapsed = time.monotonic() - start + assert elapsed < 15, f"[{scenario} fast_fail] exception was too slow after fault: {elapsed:.2f}s" + + +def _assert_post_fault_health_terminal(host: str, port: int, *, scenario: str) -> None: + deadline = time.monotonic() + 20.0 + last_observation = "" + while time.monotonic() < deadline: + try: + status, body = get_health_raw(host, port, timeout_sec=5) + last_observation = f"http={status}, body={body[:200]!r}" + if status == 503: + return + except Exception as exc: # noqa: BLE001 + last_observation = f"exception={exc!r}" + return + time.sleep(0.5) + pytest.fail(f"[{scenario} health] no terminal post-fault health observed: {last_observation}") @pytest.mark.slow @@ -122,40 +224,15 @@ def test_reliability_fault_gpu_oom_video_large_request_failure(omni_server_funct @pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") @pytest.mark.parametrize( "fault_injector", - [ - pytest.param( - make_process_kill_fault_injector( - grep_patterns="multiprocessing.spawn", - signal_name="SIGKILL", - limit=1, - post_kill_wait_seconds=2.0, - ), - id="runtime_process_chain", - ), - ], + WORKER_SIGNAL_FAULT_PARAMS, indirect=True, ) @pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) -def test_reliability_fault_process_kill_video_request_failure( +def test_reliability_fault_process_kill_request_failure( omni_server_after_fault_function, openai_client_function, ) -> None: - image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}" - request_config = { - "form_data": { - "prompt": "Generate a realistic road-driving video with camera motion.", - "width": 512, - "height": 512, - "fps": 8, - "num_frames": 8, - "guidance_scale": 1.0, - "flow_shift": 5.0, - "num_inference_steps": 4, - "seed": 42, - }, - "image_reference": image_data_url, - "stream": False, - } + request_config = _video_request_config() try: openai_client_function.send_video_diffusion_request(request_config, request_num=1) except Exception as exc: @@ -168,21 +245,11 @@ def test_reliability_fault_process_kill_video_request_failure( @pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") @pytest.mark.parametrize( "fault_injector", - [ - pytest.param( - make_process_kill_fault_injector( - grep_patterns="multiprocessing.spawn", - signal_name="SIGKILL", - limit=1, - post_kill_wait_seconds=2.0, - ), - id="runtime_process_chain", - ), - ], + WORKER_SIGNAL_FAULT_PARAMS, indirect=True, ) @pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) -def test_reliability_fault_process_kill_video_health_fast_fail_and_concurrent( +def test_reliability_fault_process_kill_health_fast_fail_and_concurrent( omni_server_after_fault_function, openai_client_function, ) -> None: @@ -290,6 +357,37 @@ def test_reliability_fault_process_kill_video_health_fast_fail_and_concurrent( ) +@pytest.mark.slow +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize( + "fault_injector", + WORKER_SIGNAL_FAULT_PARAMS, + indirect=True, +) +@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) +def test_reliability_fault_process_kill_worker_with_load_request_failure( + omni_server_function, + openai_client_function, + fault_injector: FaultInjector, +) -> None: + request_config = _video_request_config() + scenario = "kill_worker_with_load" + load_result = run_fault_injection_with_rate_load( + submit_request=lambda: openai_client_function.send_video_diffusion_request(request_config, request_num=1), + inject_fault=lambda: fault_injector(omni_server_function), + num_requests=INFLIGHT_INJECTION_REQUEST_COUNT, + request_rate=INFLIGHT_INJECTION_REQUEST_RATE, + completion_timeout_sec=120.0, + ) + assert load_result["failure_observed"], ( + f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}" + ) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_video_fast_fail(host, port, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + + @pytest.mark.slow @hardware_test(res={"cuda": "H100"}, num_cards=1) @pytest.mark.skip(reason="issue#2327") @@ -409,3 +507,127 @@ def test_reliability_video_oom_recovers_after_fault_removed( "unhealthy terminal state should fail fast on /v1/videos, " f"got health={terminal_health}, request_status={recovery_status}, body={recovery_text_prefix!r}" ) + + +@pytest.mark.slow +@hardware_test(res={"cuda": "H100"}, num_cards=1) +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", SERVE_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) +def test_reliability_fault_process_kill_serve_root_with_load_fast_fail_and_cleanup( + omni_server_function, + openai_client_function, + signal_name: str, +) -> None: + """Black-box: during in-flight video requests, kill serve root and verify post-fault behavior/cleanup.""" + request_config = _video_request_config() + scenario = f"kill_serve_root_with_load_{signal_name.lower()}" + injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0) + load_result = run_fault_injection_with_rate_load( + submit_request=lambda: openai_client_function.send_video_diffusion_request(request_config, request_num=1), + inject_fault=lambda: injector(omni_server_function), + num_requests=INFLIGHT_INJECTION_REQUEST_COUNT, + request_rate=INFLIGHT_INJECTION_REQUEST_RATE, + completion_timeout_sec=120.0, + ) + assert load_result["failure_observed"], ( + f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}" + ) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_video_fast_fail(host, port, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@hardware_test(res={"cuda": "H100"}, num_cards=1) +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", SERVE_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) +def test_reliability_fault_process_kill_serve_root_no_load_fast_fail_and_cleanup( + omni_server_function, + signal_name: str, +) -> None: + """Black-box: kill serve root without load; verify fast-fail/health/cleanup.""" + scenario = f"kill_serve_root_no_load_{signal_name.lower()}" + injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0) + injector(omni_server_function) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_video_fast_fail(host, port, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", TREE_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) +def test_reliability_fault_process_kill_tree_no_load_fast_fail_and_cleanup( + omni_server_function, + signal_name: str, +) -> None: + """Black-box: kill server tree without load; verify fast-fail/health/cleanup.""" + scenario = f"kill_serve_tree_no_load_{signal_name.lower()}" + injector = make_server_tree_kill_fault_injector( + signal_name=signal_name, + post_kill_wait_seconds=2.0, + inter_kill_wait_seconds=0.1, + ) + injector(omni_server_function) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_video_fast_fail(host, port, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + ) + + +@pytest.mark.slow +@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only") +@pytest.mark.parametrize("signal_name", TREE_WITH_LOAD_SIGNAL_PARAMS) +@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True) +def test_reliability_fault_process_kill_tree_with_load_fast_fail_and_cleanup( + omni_server_function, + openai_client_function, + signal_name: str, +) -> None: + """Black-box: during in-flight video requests, kill server process tree and verify post-fault behavior/cleanup.""" + request_config = _video_request_config() + scenario = f"kill_serve_tree_with_load_{signal_name.lower()}" + injector = make_server_tree_kill_fault_injector( + signal_name=signal_name, + post_kill_wait_seconds=2.0, + inter_kill_wait_seconds=0.1, + ) + load_result = run_fault_injection_with_rate_load( + submit_request=lambda: openai_client_function.send_video_diffusion_request(request_config, request_num=1), + inject_fault=lambda: injector(omni_server_function), + num_requests=INFLIGHT_INJECTION_REQUEST_COUNT, + request_rate=INFLIGHT_INJECTION_REQUEST_RATE, + completion_timeout_sec=120.0, + ) + assert load_result["failure_observed"], ( + f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}" + ) + host = omni_server_function.host + port = omni_server_function.port + _assert_post_fault_video_fast_fail(host, port, scenario=scenario) + _assert_post_fault_health_terminal(host, port, scenario=scenario) + assert_no_server_tree_process_residual_and_gpu_release( + omni_server_function, + scenario=scenario, + timeout_sec=worker_residual_timeout_after_kill_signal(signal_name), + )