diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 4b83c2e..b35249d 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -9,6 +9,7 @@ on: jobs: test: runs-on: ubuntu-latest + timeout-minutes: 15 steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 @@ -26,6 +27,7 @@ jobs: matrix: os: [ubuntu-latest, macos-latest, windows-latest] runs-on: ${{ matrix.os }} + timeout-minutes: 15 steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 @@ -35,5 +37,9 @@ jobs: run: | python -m pip install --upgrade pip pip install .[test] - - name: Run platform-neutral Launcher lifecycle tests - run: pytest tests/test_launcher_background.py + - name: Run platform-neutral Launcher lifecycle and fail-stop tests + run: >- + pytest + tests/test_launcher_background.py + tests/test_launcher_cleanup.py + tests/test_launcher_fail_stop.py diff --git a/README.md b/README.md index a1e3c0e..f4bb251 100644 --- a/README.md +++ b/README.md @@ -315,7 +315,8 @@ asset definition. { "version": "0.1", "runtime": { - "cleanup_mmap_on_start": false + "health_check_interval_sec": 1.0, + "health_check_timeout_sec": 0.8 }, "defaults": { "cwd": ".", @@ -350,7 +351,7 @@ asset definition. | Key | Description | | ---------- | ----------- | | `version` | Free-form version string for the specification.【F:src/hakoniwa_pdu/apps/launcher/model.py†L82-L90】 | -| `runtime` | Optional Launcher-owned runtime preparation. `cleanup_mmap_on_start` is disabled by default and removes only Hakoniwa mmap/lock files before any asset starts. Enable it only when this Launcher exclusively owns the configured runtime directory. | +| `runtime` | Runtime integrity supervision intervals. `cleanup_mmap_on_start` is deprecated and refused; recovery is always an explicit operator action. | | `defaults` | Shared defaults applied to every asset when the field is omitted at the asset level. Paths are resolved relative to the launch file.【F:src/hakoniwa_pdu/apps/launcher/loader.py†L47-L91】 | | `assets` | Array of process definitions. Launch order is automatically sorted by `depends_on` while preserving the original order when there is no dependency.【F:src/hakoniwa_pdu/apps/launcher/model.py†L112-L167】 | | `notify` | Optional notification that fires when an asset exits or the launcher aborts. Supports `webhook` and `exec` variants.【F:src/hakoniwa_pdu/apps/launcher/model.py†L58-L80】【F:src/hakoniwa_pdu/apps/launcher/hako_monitor.py†L83-L121】 | @@ -399,12 +400,30 @@ keeps its slightly longer Python subprocess timeout as an outer watchdog. Older or unrecognized hako-cmd versions remain on the legacy command line, so existing installations continue to work without the new option. -When `runtime.cleanup_mmap_on_start` is `true`, the Launcher resolves -`HAKO_CONFIG_PATH`, reads `core_mmap_path`, and removes only `mmap-0x*.bin`, -`flock.bin`, `pdu_init.lock`, and `pro_init.lock` before starting any asset. It -does not delete runtime files on shutdown. This option is an explicit ownership -declaration: do not enable it when another Launcher or Hakoniwa process may use -the same mmap directory. +With hako-cmd 1.1.0 or newer, the Launcher runs `hako-cmd check` before startup +and periodically while active. The consumer requires the versioned v1 report +with a `checks[]` array; an unsupported or malformed report is `UNKNOWN`. +`ABORT` changes the Launcher state to `FAILED` +and terminates every managed process. `UNKNOWN` is reported as a warning and +does not trigger destructive action. Every check has an outer subprocess +timeout, configured by `runtime.health_check_timeout_sec`; the periodic interval +is `runtime.health_check_interval_sec`. + +The Launcher never repairs or deletes a failed runtime automatically. The old +`runtime.cleanup_mmap_on_start` field is retained only for input compatibility; +setting it to `true` now refuses startup. Inspect the evidence with +`hako-cmd check`, stop all related processes, and run `hako-cmd recover` +explicitly. Launcher session files and asset logs remain available after a +fail-stop event. + +An unexpected asset exit is also fail-stop: the asset name, PID, exit code or +signal, cleanup exception, and post-cleanup process diagnostics are stored in +`failure_report`. Successful cleanup does not replace `FAILED` with +`TERMINATED`. `failure_report.cleanup.recover_safe` is true only when every +managed process and descendant scope was verified stopped. In particular, +Windows uses `taskkill /T /F` for native process trees but cannot independently +enumerate descendants after a parent has already disappeared; that limitation +is recorded and recovery remains unsafe until the operator checks it. #### Notify section diff --git a/docs/launcher-background-lifecycle.md b/docs/launcher-background-lifecycle.md index 81b660a..b17b319 100644 --- a/docs/launcher-background-lifecycle.md +++ b/docs/launcher-background-lifecycle.md @@ -40,6 +40,20 @@ python -m hakoniwa_pdu.apps.launcher.hako_launcher_ctl \ `terminate` calls the running Launcher's `LauncherService.terminate()` through a localhost control endpoint. Child-process cleanup therefore remains owned by the Launcher, including the existing OS-specific process-group behavior. +With hako-cmd 1.1.0 or newer, the background Launcher also supervises runtime +integrity. An `ABORT` result moves the session to `FAILED` and stops all managed +children. It does not remove the session file, logs, mmap files, or lock files; +recovery requires a separate, explicit `hako-cmd recover` after every related +process has stopped. An `UNKNOWN` result is diagnostic only and is logged as a +warning. + +Startup-time ABORT details are retained in the background session even when no +control server was created. Runtime ABORT and unexpected asset exit reports +include cleanup status, exceptions, survivor diagnostics, and a conservative +`recover_safe` flag. Cleanup never changes an existing `FAILED` state to +`TERMINATED`; only an explicit normal termination of a healthy session produces +`TERMINATED`. + The session file is retained after termination with `state: "TERMINATED"` for post-run inspection. ## Session ownership and stale files diff --git a/src/hakoniwa_pdu/apps/launcher/hako_asset_runner.py b/src/hakoniwa_pdu/apps/launcher/hako_asset_runner.py index 7e52db3..d57e7a4 100644 --- a/src/hakoniwa_pdu/apps/launcher/hako_asset_runner.py +++ b/src/hakoniwa_pdu/apps/launcher/hako_asset_runner.py @@ -95,6 +95,8 @@ def __init__(self, *, env: Optional[Mapping[str, str]] = None) -> None: self.handle: Optional[ProcHandle] = None self._win_pid: Optional[int] = None self._posix_pgid: Optional[int] = None + self._managed_pid: Optional[int] = None + self._windows_tree_cleanup = False # ---------- helpers ---------- @staticmethod @@ -180,6 +182,8 @@ def spawn( timeout=15.0 ) self._win_pid = int(out.strip()) + self._managed_pid = self._win_pid + self._windows_tree_cleanup = True dummy = subprocess.Popen(["/bin/true"]) if IS_POSIX else subprocess.Popen(["cmd.exe", "/c", "exit", "0"]) self.handle = ProcHandle( popen=dummy, @@ -211,6 +215,8 @@ def spawn( # even after the direct child exits: launch scripts may leave a # long-running descendant in the same group. self._posix_pgid = popen.pid + self._managed_pid = popen.pid + self._windows_tree_cleanup = IS_WIN self.handle = ProcHandle( popen=popen, @@ -327,6 +333,19 @@ def kill(self) -> None: try: if IS_POSIX: os.killpg(self._posix_pgid or h.popen.pid, signal.SIGKILL) + elif IS_WIN: + completed = subprocess.run( + ["taskkill", "/PID", str(h.popen.pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=5.0, + check=False, + ) + if completed.returncode not in (0, 128): + raise RuntimeError( + f"taskkill /T failed for pid={h.popen.pid}: " + f"rc={completed.returncode}" + ) else: h.popen.kill() finally: @@ -338,6 +357,33 @@ def kill(self) -> None: f"pgid={self._posix_pgid}" ) + def diagnostics(self) -> dict: + """Return non-mutating lifecycle evidence for failure reports.""" + info = self.exit_info() + if self._windows_tree_cleanup: + cleanup_scope = "windows_taskkill_tree" + descendants_verified = False + elif IS_POSIX: + cleanup_scope = "posix_process_group" + descendants_verified = not self.is_alive() + elif IS_WIN: + cleanup_scope = "windows_taskkill_tree" + # Windows taskkill /T is requested while the parent exists, but + # this runner does not enumerate an already-orphaned descendant. + descendants_verified = False + else: + cleanup_scope = "direct_process" + descendants_verified = False + return { + "pid": self._managed_pid if self._managed_pid is not None else info.pid, + "pgid": self._posix_pgid, + "alive": self.is_alive(), + "exit_code": info.exit_code, + "signal": info.signal, + "cleanup_scope": cleanup_scope, + "descendants_verified_stopped": descendants_verified, + } + def exit_info(self) -> ExitInfo: h = self.handle # Windows プロセス優先で情報を埋める diff --git a/src/hakoniwa_pdu/apps/launcher/hako_cli.py b/src/hakoniwa_pdu/apps/launcher/hako_cli.py index 1c2e928..ff2da78 100644 --- a/src/hakoniwa_pdu/apps/launcher/hako_cli.py +++ b/src/hakoniwa_pdu/apps/launcher/hako_cli.py @@ -5,12 +5,21 @@ import shutil import subprocess import re +import json +from dataclasses import dataclass from typing import Optional, Literal from .effective_model import EffectiveSpec from .envmerge import merge_env +@dataclass(frozen=True) +class RuntimeHealth: + status: Literal["OK", "ABORT", "UNKNOWN"] + reason: str + details: dict + + class HakoCli: """ Thin wrapper to run hako-cmd in the foreground. @@ -81,6 +90,67 @@ def list_assets(self, *, timeout: Optional[float] = None) -> tuple[int, set[str] } return int(proc.returncode), names + def check(self, *, timeout: float = 1.0) -> RuntimeHealth: + """Run the non-mutating native integrity check with an outer timeout.""" + env = merge_env( + defaults_env=self.defaults_env_ops, + asset_env=None, + asset_name="hako_cli", + ) + try: + resolved = self._resolve_cmd(env) + except OSError as exc: + return RuntimeHealth("UNKNOWN", str(exc), {}) + if not self._supports_health_check(resolved, env): + return RuntimeHealth( + "UNKNOWN", + "hako-cmd does not support runtime integrity checks", + {"version": self._detected_version}, + ) + try: + proc = subprocess.run( + [resolved, "check"], + cwd=str(self.spec.base_dir), + env=env, + check=False, + timeout=timeout, + capture_output=True, + text=True, + ) + except subprocess.TimeoutExpired: + return RuntimeHealth("UNKNOWN", "hako-cmd check timed out", {}) + except OSError as exc: + return RuntimeHealth("UNKNOWN", str(exc), {}) + try: + payload = json.loads(proc.stdout) + except (json.JSONDecodeError, TypeError): + return RuntimeHealth( + "UNKNOWN", + "hako-cmd check returned invalid JSON", + {"returncode": proc.returncode, "stderr": proc.stderr}, + ) + if payload.get("schema_version") != 1: + return RuntimeHealth( + "UNKNOWN", "hako-cmd check returned an unsupported schema", payload + ) + checks = payload.get("checks") + if not isinstance(checks, list) or any( + not isinstance(check, dict) + or not isinstance(check.get("name"), str) + or check.get("status") not in ("OK", "ABORT", "UNKNOWN") + or not isinstance(check.get("reason"), str) + for check in checks + ): + return RuntimeHealth( + "UNKNOWN", "hako-cmd check returned invalid checks", payload + ) + status = payload.get("status") + if status not in ("OK", "ABORT", "UNKNOWN"): + return RuntimeHealth( + "UNKNOWN", "hako-cmd check returned an invalid status", payload + ) + return RuntimeHealth(status, str(payload.get("reason", "")), payload) + # ---- internals ---- def _resolve_cmd(self, env: dict[str, str]) -> str: path = env.get("PATH") @@ -125,6 +195,13 @@ def _supports_bounded_lock_wait(self, resolved: str, env: dict[str, str]) -> boo ) return self._bounded_lock_wait_supported + def _supports_health_check(self, resolved: str, env: dict[str, str]) -> bool: + self._supports_bounded_lock_wait(resolved, env) + return bool( + self._detected_version is not None + and self._detected_version >= (1, 1, 0) + ) + def _command( self, resolved: str, diff --git a/src/hakoniwa_pdu/apps/launcher/hako_launcher.py b/src/hakoniwa_pdu/apps/launcher/hako_launcher.py index 371abab..f2177f0 100644 --- a/src/hakoniwa_pdu/apps/launcher/hako_launcher.py +++ b/src/hakoniwa_pdu/apps/launcher/hako_launcher.py @@ -16,7 +16,6 @@ from .hako_monitor import HakoMonitor from .hako_cli import HakoCli from .envmerge import merge_env -from .hako_runtime_cleanup import cleanup_runtime_mmap from .hako_launcher_control import ( LauncherControlError, LauncherControlServer, @@ -52,6 +51,16 @@ def __init__(self, *, launch_path: str) -> None: self._watch_thread: Optional[threading.Thread] = None self._stop_watch = threading.Event() self._cli_prepared = False + runtime = self.launcher_spec.runtime + self._health_check_interval_sec = float( + runtime.health_check_interval_sec if runtime is not None else 1.0 + ) + self._health_check_timeout_sec = float( + runtime.health_check_timeout_sec if runtime is not None else 0.8 + ) + self._last_unknown_health_reason: Optional[str] = None + self._last_health_report: Optional[dict] = None + self.failure_report: Optional[dict] = None def _prepare_cli(self) -> None: if self._cli_prepared: @@ -74,16 +83,104 @@ def _prepare_runtime(self) -> None: runtime = self.launcher_spec.runtime if runtime is None or not runtime.cleanup_mmap_on_start: return - env = merge_env( - defaults_env=self.defaults_env_ops, - asset_env=None, - asset_name="hako_launcher", + raise RuntimeError( + "runtime.cleanup_mmap_on_start is disabled by fail-stop policy; " + "inspect with 'hako-cmd check' and explicitly run " + "'hako-cmd recover' only after confirming all processes stopped" ) - removed = cleanup_runtime_mmap(env=env, base_dir=self.spec.base_dir) - print( - f"[INFO] removed {len(removed)} stale Hakoniwa runtime file(s) " - "before asset startup" + + def _check_runtime_health(self, *, startup: bool = False) -> str: + health = self.cli.check(timeout=self._health_check_timeout_sec) + self._last_health_report = { + "status": health.status, + "reason": health.reason, + "details": health.details, + "detected_at": time.time(), + } + if health.status == "ABORT": + print( + f"[ERROR] Hakoniwa runtime integrity ABORT: {health.reason}", + file=sys.stderr, + ) + return "ABORT" + if health.status == "UNKNOWN": + if startup or health.reason != self._last_unknown_health_reason: + print( + f"[WARN] Hakoniwa runtime integrity UNKNOWN: {health.reason}", + file=sys.stderr, + ) + self._last_unknown_health_reason = health.reason + return "UNKNOWN" + self._last_unknown_health_reason = None + return "OK" + + def _managed_process_diagnostics(self) -> list[dict]: + diagnostics: list[dict] = [] + for running in list(self.monitor.procs): + runner = running.runner + try: + detail = runner.diagnostics() if hasattr(runner, "diagnostics") else { + "alive": bool(runner.is_alive()), + } + except Exception as exc: + detail = { + "alive": None, + "diagnostic_error": str(exc), + "descendants_verified_stopped": False, + } + diagnostics.append({"asset": running.asset.name, **detail}) + return diagnostics + + def _cleanup_after_failure(self, reason: str) -> dict: + cleanup_error: str | None = None + try: + self.monitor.abort(reason) + except Exception as exc: + cleanup_error = str(exc) + print(f"[ERROR] fail-stop cleanup failed: {exc}", file=sys.stderr) + + processes = self._managed_process_diagnostics() + survivors = [ + process for process in processes if process.get("alive") is not False + ] + unverified_descendants = [ + process for process in processes + if process.get("descendants_verified_stopped") is False + ] + recover_safe = ( + cleanup_error is None + and not survivors + and not unverified_descendants ) + return { + "status": "OK" if cleanup_error is None and not survivors else "FAILED", + "reason": cleanup_error or ( + "managed processes stopped" + if not survivors + else "one or more managed processes may still be alive" + ), + "exception": cleanup_error, + "processes": processes, + "survivors": survivors, + "recover_safe": recover_safe, + "recovery_guidance": ( + "all managed processes and descendants were verified stopped" + if recover_safe + else "do not run hako-cmd recover until remaining or unverified " + "processes are checked and stopped" + ), + } + + def _fail_stop(self, reason: str, *, evidence: Optional[dict] = None) -> None: + self.failure_report = { + **(getattr(self, "_last_health_report", None) or {}), + "trigger": reason, + **(evidence or {}), + } + self.state = "FAILED" + self._stop_watch.set() + self.failure_report["cleanup"] = self._cleanup_after_failure(reason) + self.state = "FAILED" # -------- 状態遷移API -------- def activate(self, *, require_hako_cmd: bool = True) -> None: @@ -103,6 +200,9 @@ def activate(self, *, require_hako_cmd: bool = True) -> None: # process-only/Core-free launch files do not need Hakoniwa Core. if require_hako_cmd: self._prepare_cli() + if self._check_runtime_health(startup=True) == "ABORT": + self._fail_stop("startup_runtime_integrity_abort") + raise RuntimeError("Hakoniwa runtime integrity check returned ABORT") self._prepare_runtime() print("[INFO] activating 'before_start' assets...") @@ -114,9 +214,9 @@ def activate(self, *, require_hako_cmd: bool = True) -> None: raise self._stop_watch.clear() self._watch_thread = threading.Thread(target=self._watch_loop, daemon=True) - self._watch_thread.start() self.state = "ACTIVATED" print("[INFO] state -> ACTIVATED") + self._watch_thread.start() def cmd(self, command: str) -> int: if self.state not in ("ACTIVATED", "RUNNING", "STOPPED"): @@ -152,6 +252,7 @@ def terminate(self) -> None: if self.state == "IDLE" and not self.monitor.procs: print("[launcher] terminate: already IDLE") return + preserve_failed = self.state == "FAILED" if self.state == "TERMINATED": print("[launcher] terminate: verifying terminated assets...") else: @@ -165,8 +266,8 @@ def terminate(self) -> None: print("[ERROR] asset cleanup failed; state -> FAILED", file=sys.stderr) raise else: - self.state = "TERMINATED" - print("[INFO] state -> TERMINATED") + self.state = "FAILED" if preserve_failed else "TERMINATED" + print(f"[INFO] state -> {self.state}") def status(self) -> str: return self.state @@ -174,19 +275,41 @@ def status(self) -> str: # -------- 内部：監視ループ（非ブロッキング） -------- def _watch_loop(self): try: - while not self._stop_watch.is_set() and self.monitor.procs: + next_health_check = time.monotonic() + while not self._stop_watch.is_set(): for rp in list(self.monitor.procs): if not rp.runner.is_alive(): print(f"[WARN] asset exited: {rp.asset.name} -> abort all") - self.terminate() + exit_info = ( + rp.runner.exit_info() + if hasattr(rp.runner, "exit_info") + else None + ) + self._fail_stop( + "unexpected_asset_exit", + evidence={ + "asset_exit": { + "asset": rp.asset.name, + "pid": getattr(exit_info, "pid", None), + "exit_code": getattr(exit_info, "exit_code", None), + "signal": getattr(exit_info, "signal", None), + } + }, + ) return + now = time.monotonic() + if getattr(self, "_cli_prepared", False) and now >= next_health_check: + if self._check_runtime_health() == "ABORT": + self._fail_stop("runtime_integrity_abort") + return + next_health_check = now + self._health_check_interval_sec time.sleep(0.2) except Exception as e: print(f"[watch] exception: {e}", file=sys.stderr) - try: - self.terminate() - except Exception as cleanup_error: - print(f"[watch] cleanup exception: {cleanup_error}", file=sys.stderr) + self._fail_stop( + "watch_exception", + evidence={"watch_exception": str(e)}, + ) # -------- background lifecycle -------- @@ -418,18 +541,22 @@ def _write_failed_session( *, session_id: str, pid: int | None = None, + failure_report: dict | None = None, ) -> None: + payload = { + "version": 1, + "session_id": session_id, + "pid": os.getpid() if pid is None else pid, + "launch_file": str(launch_file), + "log_file": str(log_file), + "state": "FAILED", + "error": error, + } + if failure_report is not None: + payload["failure_report"] = failure_report write_owned_session( session_file, - { - "version": 1, - "session_id": session_id, - "pid": os.getpid() if pid is None else pid, - "launch_file": str(launch_file), - "log_file": str(log_file), - "state": "FAILED", - "error": error, - }, + payload, session_id=session_id, ) @@ -461,16 +588,18 @@ def _run_background_worker( server.serve_until_terminated() return 0 except Exception as exc: - try: - service.terminate() - except Exception: - pass + if service.status() != "FAILED": + try: + service.terminate() + except Exception: + pass _write_failed_session( session_path, launch_path, log_path, str(exc), session_id=session_id, + failure_report=service.failure_report, ) print(f"[launcher] background worker failed: {exc}", file=sys.stderr) return 1 @@ -562,23 +691,27 @@ async def main(argv: list[str] | None = None) -> int: service.activate() rc = service.cmd("start") if not args.no_watch: - while service.status() not in ("TERMINATED",): + while service.status() not in ("TERMINATED", "FAILED"): time.sleep(0.5) + if service.status() == "FAILED": + return 1 return 0 if rc == 0 else rc except Exception as e: print(f"[launcher] Exception: {e}", file=sys.stderr) - service.terminate() + if service.status() != "FAILED": + service.terminate() return 1 elif args.mode == "activate-only": try: service.activate(require_hako_cmd=False) - while service.status() not in ("TERMINATED",): + while service.status() not in ("TERMINATED", "FAILED"): time.sleep(0.5) - return 0 + return 1 if service.status() == "FAILED" else 0 except Exception as e: print(f"[launcher] Exception: {e}", file=sys.stderr) - service.terminate() + if service.status() != "FAILED": + service.terminate() return 1 elif args.mode == "serve": diff --git a/src/hakoniwa_pdu/apps/launcher/hako_launcher_control.py b/src/hakoniwa_pdu/apps/launcher/hako_launcher_control.py index d31fecd..f235fc4 100644 --- a/src/hakoniwa_pdu/apps/launcher/hako_launcher_control.py +++ b/src/hakoniwa_pdu/apps/launcher/hako_launcher_control.py @@ -388,7 +388,7 @@ def control_port(self) -> int: return int(self.server_address[1]) def session_payload(self) -> dict[str, Any]: - return { + payload = { "version": SESSION_VERSION, "session_id": self.session_id, "pid": os.getpid(), @@ -399,6 +399,10 @@ def session_payload(self) -> dict[str, Any]: "log_file": str(self.log_file), "state": self.service.status(), } + failure_report = getattr(self.service, "failure_report", None) + if failure_report is not None: + payload["failure_report"] = failure_report + return payload def persist(self, *, force: bool = False) -> None: state = self.service.status() @@ -455,7 +459,10 @@ def serve_until_terminated(self, poll_interval: float = 0.2) -> None: self.timeout = poll_interval self.persist(force=True) try: - while not self.stop_requested and self.service.status() != "TERMINATED": + while ( + not self.stop_requested + and self.service.status() not in _TERMINAL_STATES + ): self.handle_request() self.persist() finally: diff --git a/src/hakoniwa_pdu/apps/launcher/model.py b/src/hakoniwa_pdu/apps/launcher/model.py index 79be60f..72d200d 100644 --- a/src/hakoniwa_pdu/apps/launcher/model.py +++ b/src/hakoniwa_pdu/apps/launcher/model.py @@ -63,6 +63,8 @@ class RuntimeOptions(BaseModel): # Opt-in because removing mmap files assumes this Launcher has exclusive # ownership of the configured Hakoniwa runtime directory. cleanup_mmap_on_start: bool = False + health_check_interval_sec: PositiveFloat = 1.0 + health_check_timeout_sec: PositiveFloat = 0.8 class HakoAssetReadiness(BaseModel): diff --git a/src/hakoniwa_pdu/apps/launcher/schemas/launcher.schema.json b/src/hakoniwa_pdu/apps/launcher/schemas/launcher.schema.json index 9fa4eab..b133c78 100644 --- a/src/hakoniwa_pdu/apps/launcher/schemas/launcher.schema.json +++ b/src/hakoniwa_pdu/apps/launcher/schemas/launcher.schema.json @@ -12,7 +12,20 @@ "cleanup_mmap_on_start": { "type": "boolean", "default": false, - "description": "Before assets start, remove Hakoniwa mmap and lock files from core_mmap_path. Enable only when this Launcher exclusively owns that runtime directory." + "deprecated": true, + "description": "Deprecated and refused by fail-stop policy. Use hako-cmd check and explicit hako-cmd recover." + }, + "health_check_interval_sec": { + "type": "number", + "exclusiveMinimum": 0, + "default": 1.0, + "description": "Interval in seconds between runtime integrity checks." + }, + "health_check_timeout_sec": { + "type": "number", + "exclusiveMinimum": 0, + "default": 0.8, + "description": "Outer subprocess timeout for each hako-cmd check." } }, "additionalProperties": false diff --git a/tests/test_launcher_background.py b/tests/test_launcher_background.py index 11c3595..83d33f1 100644 --- a/tests/test_launcher_background.py +++ b/tests/test_launcher_background.py @@ -399,6 +399,7 @@ def abort(self, reason): service._watch_thread = None service._prepare_runtime = lambda: calls.append("runtime") service._prepare_cli = lambda: calls.append("prepare") + service._check_runtime_health = lambda **kwargs: "OK" service.activate() @@ -497,6 +498,7 @@ def abort(self, reason): service._watch_thread = None service._prepare_cli = lambda: None service._prepare_runtime = lambda: None + service._check_runtime_health = lambda **kwargs: "OK" with pytest.raises(RuntimeError, match="second asset failed"): service.activate() diff --git a/tests/test_launcher_cleanup.py b/tests/test_launcher_cleanup.py index 66274a9..ad7cc79 100644 --- a/tests/test_launcher_cleanup.py +++ b/tests/test_launcher_cleanup.py @@ -1,4 +1,6 @@ import os +import signal +import subprocess import sys import threading import time @@ -11,6 +13,38 @@ from hakoniwa_pdu.apps.launcher.hako_monitor import HakoMonitor, Running +def _pid_exists(pid: int) -> bool: + if os.name == "nt": + result = subprocess.run( + ["tasklist", "/FO", "CSV", "/NH", "/FI", f"PID eq {pid}"], + capture_output=True, + text=True, + timeout=3, + check=False, + ) + return f'"{pid}"' in result.stdout + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + + +def _force_kill_pid(pid: int) -> None: + if not _pid_exists(pid): + return + if os.name == "nt": + subprocess.run( + ["taskkill", "/PID", str(pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=5, + check=False, + ) + else: + os.kill(pid, signal.SIGKILL) + + @pytest.mark.skipif(os.name != "posix", reason="POSIX process-group behavior") def test_asset_runner_terminates_descendant_after_group_leader_exits(tmp_path): child_code = "import time; time.sleep(60)" @@ -171,3 +205,63 @@ def abort(self, reason): service.terminate() assert service.state == "FAILED" + + +def test_real_process_abort_stops_process_tree_and_preserves_evidence(tmp_path): + child_pid_file = tmp_path / "child.pid" + evidence = tmp_path / "flock.bin" + evidence.write_bytes(b"diagnostic evidence") + child_code = "import time; time.sleep(60)" + parent_code = ( + "import pathlib, subprocess, sys, time; " + f"child=subprocess.Popen([sys.executable, '-c', {child_code!r}]); " + f"pathlib.Path({str(child_pid_file)!r}).write_text(str(child.pid)); " + "time.sleep(60)" + ) + runner = AssetRunner() + runner.spawn(sys.executable, ["-c", parent_code], cwd=tmp_path) + child_pid = None + try: + deadline = time.time() + 5 + while not child_pid_file.exists() and time.time() < deadline: + time.sleep(0.02) + assert child_pid_file.exists() + child_pid = int(child_pid_file.read_text()) + assert runner.is_alive() + assert _pid_exists(child_pid) + + monitor = _monitor_with(runner) + service = object.__new__(hako_launcher.LauncherService) + service.state = "RUNNING" + service.monitor = monitor + service._stop_watch = threading.Event() + service._last_health_report = { + "status": "ABORT", + "reason": "owner died", + "details": {"owner_pid": 42}, + "detected_at": time.time(), + } + service.failure_report = None + + service._fail_stop("runtime_integrity_abort") + + deadline = time.time() + 5 + while _pid_exists(child_pid) and time.time() < deadline: + time.sleep(0.05) + assert service.state == "FAILED" + assert runner.is_alive() is False + assert _pid_exists(child_pid) is False + assert evidence.read_bytes() == b"diagnostic evidence" + cleanup = service.failure_report["cleanup"] + assert cleanup["survivors"] == [] + assert cleanup["processes"][0]["asset"] == "asset" + if os.name == "nt": + assert cleanup["recover_safe"] is False + assert cleanup["processes"][0]["cleanup_scope"] == "windows_taskkill_tree" + else: + assert cleanup["recover_safe"] is True + finally: + if runner.is_alive(): + runner.kill() + if child_pid is not None: + _force_kill_pid(child_pid) diff --git a/tests/test_launcher_fail_stop.py b/tests/test_launcher_fail_stop.py new file mode 100644 index 0000000..3e7ab4b --- /dev/null +++ b/tests/test_launcher_fail_stop.py @@ -0,0 +1,335 @@ +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace +import subprocess +import threading +import time + +import pytest + +from hakoniwa_pdu.apps.launcher import hako_launcher +from hakoniwa_pdu.apps.launcher.hako_cli import HakoCli, RuntimeHealth +from hakoniwa_pdu.apps.launcher.hako_launcher import LauncherService +from hakoniwa_pdu.apps.launcher.hako_launcher_control import ( + LauncherControlServer, + read_session, + reserve_session, +) + + +class Result: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def test_hako_cli_check_parses_abort(monkeypatch, tmp_path): + cli = HakoCli(SimpleNamespace(base_dir=tmp_path)) + monkeypatch.setattr(cli, "_resolve_cmd", lambda env: "hako-cmd") + + def fake_run(command, **kwargs): + if "--version" in command: + return Result(stdout="hako-cmd version 1.1.0\n") + assert command == ["hako-cmd", "check"] + assert kwargs["timeout"] == 0.25 + return Result( + returncode=2, + stdout=json.dumps({ + "schema_version": 1, + "status": "ABORT", + "reason": "owner died", + "checks": [ + {"name": "master_lock", "status": "ABORT", "reason": "owner died"} + ], + }), + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + health = cli.check(timeout=0.25) + assert health.status == "ABORT" + assert health.reason == "owner died" + + +def test_hako_cli_check_rejects_unversioned_report(monkeypatch, tmp_path): + cli = HakoCli(SimpleNamespace(base_dir=tmp_path)) + cli._detected_version = (1, 1, 0) + cli._bounded_lock_wait_supported = True + monkeypatch.setattr(cli, "_resolve_cmd", lambda env: "hako-cmd") + monkeypatch.setattr( + subprocess, + "run", + lambda *args, **kwargs: Result( + returncode=0, + stdout=json.dumps({"status": "OK", "reason": "old", "checks": {}}), + ), + ) + health = cli.check(timeout=0.25) + assert health.status == "UNKNOWN" + assert "unsupported schema" in health.reason + + +def test_hako_cli_check_timeout_is_unknown(monkeypatch, tmp_path): + cli = HakoCli(SimpleNamespace(base_dir=tmp_path)) + cli._detected_version = (1, 1, 0) + cli._bounded_lock_wait_supported = True + monkeypatch.setattr(cli, "_resolve_cmd", lambda env: "hako-cmd") + + def fake_run(command, **kwargs): + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + + monkeypatch.setattr(subprocess, "run", fake_run) + assert cli.check(timeout=0.1).status == "UNKNOWN" + + +class Monitor: + def __init__(self): + self.procs = [] + self.start_calls = [] + self.abort_calls = [] + + def start_assets(self, timing): + self.start_calls.append(timing) + + def abort(self, reason): + self.abort_calls.append(reason) + + +def bare_service(health: RuntimeHealth) -> LauncherService: + service = object.__new__(LauncherService) + service.state = "IDLE" + service.monitor = Monitor() + service.cli = SimpleNamespace(check=lambda **kwargs: health) + service._stop_watch = threading.Event() + service._watch_thread = None + service._cli_prepared = True + service._health_check_interval_sec = 1.0 + service._health_check_timeout_sec = 0.1 + service._last_unknown_health_reason = None + service._last_health_report = None + service.failure_report = None + service._prepare_cli = lambda: None + service._prepare_runtime = lambda: None + return service + + +def test_startup_abort_fails_before_spawning_assets(): + service = bare_service(RuntimeHealth("ABORT", "owner died", {})) + with pytest.raises(RuntimeError, match="returned ABORT"): + service.activate() + assert service.state == "FAILED" + assert service.monitor.start_calls == [] + assert service.failure_report["reason"] == "owner died" + + +def test_startup_unknown_warns_and_continues(capsys): + service = bare_service(RuntimeHealth("UNKNOWN", "legacy metadata", {})) + service.activate() + assert service.state == "ACTIVATED" + assert service.monitor.start_calls == ["before_start"] + assert "UNKNOWN" in capsys.readouterr().err + service.terminate() + + +def test_fail_stop_preserves_failed_state_and_stops_assets(): + service = bare_service(RuntimeHealth("ABORT", "owner died", {})) + service.state = "RUNNING" + service._fail_stop("runtime_integrity_abort") + assert service.state == "FAILED" + assert service.monitor.abort_calls == ["runtime_integrity_abort"] + assert service._stop_watch.is_set() + assert service.failure_report["trigger"] == "runtime_integrity_abort" + + +def test_periodic_abort_transitions_to_failed_and_stops_monitor(): + service = bare_service(RuntimeHealth("ABORT", "owner died", {"generation": 7})) + service.state = "RUNNING" + service._health_check_interval_sec = 0.01 + watcher = threading.Thread(target=service._watch_loop) + watcher.start() + watcher.join(timeout=2) + assert not watcher.is_alive() + assert service.state == "FAILED" + assert service.monitor.abort_calls == ["runtime_integrity_abort"] + assert service.failure_report["details"]["generation"] == 7 + + +def test_fail_stop_does_not_delete_diagnostic_files(tmp_path): + evidence = tmp_path / "flock.bin" + evidence.write_bytes(b"diagnostic evidence") + service = bare_service(RuntimeHealth("ABORT", "owner died", {})) + service._last_health_report = { + "status": "ABORT", "reason": "owner died", "details": {} + } + service._fail_stop("runtime_integrity_abort") + assert evidence.read_bytes() == b"diagnostic evidence" + + +def test_background_session_persists_failure_report(tmp_path): + service = bare_service(RuntimeHealth("ABORT", "owner died", {"owner_pid": 42})) + service._check_runtime_health() + service._fail_stop("runtime_integrity_abort") + session = tmp_path / "session.json" + reserve_session(session, {"session_id": "test", "state": "STARTING"}) + server = LauncherControlServer( + service=service, + session_path=session, + launch_file=tmp_path / "launch.json", + log_file=tmp_path / "launcher.log", + session_id="test", + ) + try: + server.persist(force=True) + payload = read_session(session) + assert payload["state"] == "FAILED" + assert payload["failure_report"]["reason"] == "owner died" + assert payload["failure_report"]["details"]["owner_pid"] == 42 + finally: + server.server_close() + + +def test_cleanup_on_start_is_refused_by_fail_stop_policy(): + service = object.__new__(LauncherService) + service.launcher_spec = SimpleNamespace( + runtime=SimpleNamespace(cleanup_mmap_on_start=True) + ) + with pytest.raises(RuntimeError, match="hako-cmd recover"): + service._prepare_runtime() + + +def test_unexpected_asset_exit_stays_failed_and_records_exit(): + class DeadRunner: + def is_alive(self): + return False + + def exit_info(self): + return SimpleNamespace(pid=321, exit_code=9, signal=None) + + def diagnostics(self): + return { + "pid": 321, + "alive": False, + "exit_code": 9, + "signal": None, + "cleanup_scope": "posix_process_group", + "descendants_verified_stopped": True, + } + + service = bare_service(RuntimeHealth("OK", "healthy", {})) + service.state = "RUNNING" + service.monitor.procs = [ + SimpleNamespace(asset=SimpleNamespace(name="failed-asset"), runner=DeadRunner()) + ] + + service._watch_loop() + + assert service.state == "FAILED" + assert service.failure_report["trigger"] == "unexpected_asset_exit" + assert service.failure_report["asset_exit"] == { + "asset": "failed-asset", + "pid": 321, + "exit_code": 9, + "signal": None, + } + assert service.failure_report["cleanup"]["status"] == "OK" + + +def test_fail_stop_records_cleanup_exception_and_survivor(): + class LiveRunner: + def is_alive(self): + return True + + def diagnostics(self): + return { + "pid": 654, + "alive": True, + "exit_code": None, + "signal": None, + "cleanup_scope": "direct_process", + "descendants_verified_stopped": False, + } + + class FailingMonitor(Monitor): + def __init__(self): + super().__init__() + self.procs = [ + SimpleNamespace(asset=SimpleNamespace(name="survivor"), runner=LiveRunner()) + ] + + def abort(self, reason): + raise RuntimeError("kill denied") + + service = bare_service(RuntimeHealth("ABORT", "owner died", {})) + service.monitor = FailingMonitor() + service._last_health_report = { + "status": "ABORT", + "reason": "owner died", + "details": {"owner_pid": 42}, + "detected_at": time.time(), + } + + service._fail_stop("runtime_integrity_abort") + + cleanup = service.failure_report["cleanup"] + assert service.state == "FAILED" + assert cleanup["status"] == "FAILED" + assert cleanup["exception"] == "kill denied" + assert cleanup["survivors"][0]["asset"] == "survivor" + assert cleanup["recover_safe"] is False + assert "do not run hako-cmd recover" in cleanup["recovery_guidance"] + + +def test_terminate_after_failure_does_not_overwrite_failed_state(): + service = bare_service(RuntimeHealth("ABORT", "owner died", {})) + service.state = "FAILED" + service.failure_report = {"trigger": "runtime_integrity_abort"} + + service.terminate() + + assert service.state == "FAILED" + assert service.failure_report["trigger"] == "runtime_integrity_abort" + + +def test_background_startup_abort_preserves_failure_report(tmp_path): + class StartupAbortService: + state = "IDLE" + failure_report = None + + def activate(self, *, require_hako_cmd=True): + self.state = "FAILED" + self.failure_report = { + "status": "ABORT", + "reason": "owner died", + "details": {"owner_pid": 42}, + "detected_at": 123.0, + "trigger": "startup_runtime_integrity_abort", + "cleanup": {"status": "OK", "recover_safe": True}, + } + raise RuntimeError("Hakoniwa runtime integrity check returned ABORT") + + def status(self): + return self.state + + def terminate(self): + raise AssertionError("FAILED startup must not be overwritten by terminate") + + session = tmp_path / "session.json" + reserve_session(session, {"session_id": "startup-abort", "state": "STARTING"}) + service = StartupAbortService() + + rc = hako_launcher._run_background_worker( + service, + str(tmp_path / "launch.json"), + str(session), + "startup-abort", + "immediate", + ) + + payload = read_session(session) + assert rc == 1 + assert payload["state"] == "FAILED" + assert payload["failure_report"]["status"] == "ABORT" + assert payload["failure_report"]["details"]["owner_pid"] == 42 + assert payload["failure_report"]["cleanup"]["status"] == "OK"