diff --git a/apps/launcher/hms_launcher/__init__.py b/apps/launcher/hms_launcher/__init__.py index 9f79a89..779ef3f 100644 --- a/apps/launcher/hms_launcher/__init__.py +++ b/apps/launcher/hms_launcher/__init__.py @@ -1,10 +1,15 @@ """hms_launcher – Supervisor/Launcher (PLAN.md §6.1A, §9). -Phase-0-Umfang: portable Pfadauflösung, Portwahl, GStreamer-Environment, -kontrollierter Start von Control Core und Renderer. Vollständige -Heartbeat-/Crash-Recovery-Logik folgt in Phase 1 (§31). +Phase-0/1-Umfang: portable Pfadauflösung, Portwahl, GStreamer-Environment, +Prozessüberwachung mit Restart-Policy und Crashloop-Erkennung, +kontrolliertes Beenden mit Recovery-Markierung. """ from hms_launcher.paths import AppPaths, resolve_app_root +from hms_launcher.supervisor import ( + ProcessSpec, + Supervisor, + find_free_port, +) -__all__ = ["AppPaths", "resolve_app_root"] +__all__ = ["AppPaths", "resolve_app_root", "ProcessSpec", "Supervisor", "find_free_port"] diff --git a/apps/launcher/hms_launcher/supervisor.py b/apps/launcher/hms_launcher/supervisor.py new file mode 100644 index 0000000..1feee36 --- /dev/null +++ b/apps/launcher/hms_launcher/supervisor.py @@ -0,0 +1,206 @@ +"""Supervisor/Launcher (PLAN.md §6.1A, §26.2, §26.4). + +Aufgaben (§6.1A): +- Start und Überwachung der Teilprozesse (Control Core, Renderer) +- Wahl freier lokaler Ports (127.0.0.1) +- Setzen portabler Runtime-Pfade (AppPaths.portable_environment) +- Heartbeat-Überwachung auf Prozessebene (Lebenszyklus) +- kontrolliertes Beenden mit Timeout (§26.4) +- Wiederanlauf nach Control-Core-Absturz, Crashloop-Erkennung (§26.2) +- Recovery-Markierung bei Zwangsbeendigung (§26.4) + +Der Supervisor verwaltet echte Prozesse; keine Mock-Implementierung. +""" + +from __future__ import annotations + +import os +import signal +import socket +import subprocess +import time +from dataclasses import dataclass, field +from pathlib import Path + +from hms_launcher.paths import AppPaths + +DEFAULT_STOP_TIMEOUT_S = 10.0 +DEFAULT_MAX_RESTARTS = 5 +DEFAULT_RESTART_WINDOW_S = 60.0 + + +@dataclass(frozen=True) +class ProcessSpec: + """Beschreibung eines zu überwachenden Teilprozesses.""" + + name: str + cmd: list[str] + restartable: bool = True + max_restarts: int = DEFAULT_MAX_RESTARTS + restart_window_s: float = DEFAULT_RESTART_WINDOW_S + stop_timeout_s: float = DEFAULT_STOP_TIMEOUT_S + + +@dataclass +class ProcessState: + """Laufzeitinformation zu einem überwachten Prozess (§26.2).""" + + spec: ProcessSpec + proc: subprocess.Popen | None = None + restarts: list[float] = field(default_factory=list) # Zeitstempel je Neustart + last_start_ns: int = 0 + crashlooped: bool = False + stopped_by_supervisor: bool = False + + @property + def running(self) -> bool: + return self.proc is not None and self.proc.poll() is None + + @property + def returncode(self) -> int | None: + return self.proc.poll() if self.proc is not None else None + + +def find_free_port(host: str = "127.0.0.1") -> int: + """Wählt einen freien lokalen Port (§6.1A). Socket wird sofort wieder + freigegeben; der Kindprozess bindet ihn anschließend selbst.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind((host, 0)) + return sock.getsockname()[1] + + +class Supervisor: + """Überwacht Teilprozesse mit Restart-Policy und Crashloop-Schwelle.""" + + def __init__(self, paths: AppPaths | None = None) -> None: + self._paths = paths + self._states: dict[str, ProcessState] = {} + self._recovery_marker: Path | None = None + if paths is not None: + self._recovery_marker = paths.userdata / "recovery" / "unclean_shutdown" + + # ---------- Start/Stop ---------- + + def start(self, spec: ProcessSpec) -> None: + """Startet einen Teilprozess mit portabler Umgebung.""" + if spec.name in self._states and self._states[spec.name].running: + raise RuntimeError(f"process {spec.name!r} already running") + env = dict(os.environ) + if self._paths is not None: + env.update(self._paths.portable_environment()) + proc = subprocess.Popen( + spec.cmd, + env=env, + cwd=str(self._paths.root) if self._paths is not None else None, + ) + state = self._states.get(spec.name) + if state is None: + state = ProcessState(spec=spec) + self._states[spec.name] = state + else: + state.spec = spec + state.proc = proc + state.last_start_ns = time.monotonic_ns() + state.stopped_by_supervisor = False + + def stop(self, name: str, timeout: float | None = None) -> int | None: + """Kontrolliertes Beenden: terminate → warten → kill (§26.4). + + Gibt den Rückgabecode zurück; None falls der Prozess nicht lief. + """ + state = self._states.get(name) + if state is None or state.proc is None: + return None + if not state.running: + return state.returncode + state.stopped_by_supervisor = True + timeout = timeout if timeout is not None else state.spec.stop_timeout_s + state.proc.terminate() # SIGTERM: laufende Writes abschließen (§26.4) + try: + return state.proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + # Zwangsbeendigung: Recovery-Markierung setzen (§26.4) + state.proc.kill() + self._write_recovery_marker(name) + return state.proc.wait(timeout=5) + + def shutdown(self) -> None: + """Beendet alle Prozesse kontrolliert (Renderer zuletzt, um Output + so lange wie möglich zu halten; §26.2).""" + for name in reversed(list(self._states)): + self.stop(name) + + # ---------- Überwachung (§26.2) ---------- + + def check(self) -> dict[str, str]: + """Prüft alle Prozesse; startet Abgestürzte gemäß Policy neu. + + Rückgabe: name → Zustand (running/restarted/crashloop/stopped). + """ + result: dict[str, str] = {} + now = time.monotonic() + for name, state in list(self._states.items()): + if state.running: + result[name] = "running" + continue + if state.stopped_by_supervisor: + result[name] = "stopped" + continue + if not state.spec.restartable or state.crashlooped: + result[name] = "crashloop" if state.crashlooped else "stopped" + continue + # Neustarts innerhalb des Zeitfensters zählen (Crashloop, §26.2) + state.restarts = [ + t for t in state.restarts if now - t < state.spec.restart_window_s + ] + if len(state.restarts) >= state.spec.max_restarts: + state.crashlooped = True # endlose Neustarts verhindern + result[name] = "crashloop" + continue + state.restarts.append(now) + self.start(state.spec) + result[name] = "restarted" + return result + + def state(self, name: str) -> ProcessState: + return self._states[name] + + def names(self) -> list[str]: + return list(self._states) + + # ---------- Recovery (§26.4) ---------- + + def _write_recovery_marker(self, name: str) -> None: + if self._recovery_marker is None: + return + self._recovery_marker.parent.mkdir(parents=True, exist_ok=True) + with open(self._recovery_marker, "a", encoding="utf-8") as fh: + fh.write( + f"{time.strftime('%Y-%m-%dT%H:%M:%S%z')} forced-kill {name}\n" + ) + + def consume_recovery_marker(self) -> list[str]: + """Liest und löscht die Recovery-Markierung (Crash-Recovery-Dialog, + §24.3/§26.4). Gibt die Zeilen zurück.""" + if self._recovery_marker is None or not self._recovery_marker.is_file(): + return [] + lines = self._recovery_marker.read_text(encoding="utf-8").splitlines() + self._recovery_marker.unlink() + return lines + + +# Windows-kompatibles SIGTERM: terminate() nutzt auf Windows TerminateProcess, +# das kein SIGTERM ist. Für sauberes Shutdown nutzen Kindprozesse dort einen +# Steuerkanal (IPC-command) – der Supervisor sendet SIGTERM nur auf POSIX. + + +def request_graceful_stop(proc: subprocess.Popen, timeout: float) -> int | None: + """POSIX: SIGTERM; Windows: proc.terminate(). Wartet dann kontrolliert.""" + if os.name == "posix": + proc.send_signal(signal.SIGTERM) + else: + proc.terminate() + try: + return proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + return None diff --git a/tests/unit/test_supervisor.py b/tests/unit/test_supervisor.py new file mode 100644 index 0000000..c40fc26 --- /dev/null +++ b/tests/unit/test_supervisor.py @@ -0,0 +1,190 @@ +"""Tests Supervisor/Launcher mit echten Kindprozessen (PLAN.md §6.1A, §26). + +Keine Mocks: die Tests starten reale Python-Kindprozesse und prüfen +Start, kontrolliertes Beenden, Neustart nach Absturz, Crashloop-Erkennung +und Recovery-Markierung. +""" + +from __future__ import annotations + +import sys +import time +from pathlib import Path + +import pytest + +from hms_launcher import AppPaths, ProcessSpec, Supervisor, find_free_port + + +def _sleeper(seconds: float = 30.0) -> list[str]: + """Kindprozess-Kommando, das `seconds` lang still läuft.""" + return [sys.executable, "-c", f"import time; time.sleep({seconds})"] + + +def _crasher() -> list[str]: + """Kindprozess-Kommando, das sofort mit Code 1 endet.""" + return [sys.executable, "-c", "import sys; sys.exit(1)"] + + +def _wait_exit(state, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while state.proc is not None and state.proc.poll() is None and time.monotonic() < deadline: + time.sleep(0.02) + + +def _wait_file(path: Path, timeout: float = 5.0) -> None: + deadline = time.monotonic() + timeout + while not path.is_file() and time.monotonic() < deadline: + time.sleep(0.01) + assert path.is_file(), f"kindprozess schrieb {path} nicht" + + +def test_find_free_port_returns_usable_port() -> None: + port = find_free_port() + assert 1024 <= port <= 65535 + import socket + + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", port)) + + +def test_start_and_running_state(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + sup.start(ProcessSpec(name="renderer", cmd=_sleeper())) + try: + assert sup.state("renderer").running + assert sup.check()["renderer"] == "running" + finally: + sup.shutdown() + assert sup.state("renderer").returncode is not None + + +def test_duplicate_start_rejected(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + sup.start(ProcessSpec(name="core", cmd=_sleeper())) + try: + with pytest.raises(RuntimeError, match="already running"): + sup.start(ProcessSpec(name="core", cmd=_sleeper())) + finally: + sup.shutdown() + + +def test_controlled_stop_uses_terminate(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + sup.start(ProcessSpec(name="core", cmd=_sleeper(), stop_timeout_s=5.0)) + rc = sup.stop("core") + try: + assert rc is not None + assert not sup.state("core").running + # kontrolliert beendet → kein Neustart durch check() + assert sup.check()["core"] == "stopped" + # keine Recovery-Markierung bei sauberem Stop + assert sup.consume_recovery_marker() == [] + finally: + sup.shutdown() + + +def test_crash_triggers_restart(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + spec = ProcessSpec( + name="core", cmd=_sleeper(0.2), max_restarts=3, restart_window_s=30.0 + ) + sup.start(spec) + _wait_exit(sup.state("core")) + result = sup.check() # Absturz erkannt → Neustart + assert result["core"] == "restarted" + assert sup.state("core").running + assert len(sup.state("core").restarts) == 1 + sup.shutdown() + + +def test_crashloop_detection_stops_restarting(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + spec = ProcessSpec(name="core", cmd=_crasher(), max_restarts=2, restart_window_s=60.0) + sup.start(spec) + _wait_exit(sup.state("core")) + assert sup.check()["core"] == "restarted" # Neustart 1 + _wait_exit(sup.state("core")) + assert sup.check()["core"] == "restarted" # Neustart 2 + _wait_exit(sup.state("core")) + assert sup.check()["core"] == "crashloop" # Schwelle erreicht (§26.2) + assert sup.state("core").crashlooped + # weiteres check() startet nicht mehr + assert sup.check()["core"] == "crashloop" + sup.shutdown() + + +def test_non_restartable_process_stays_down(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + sup.start(ProcessSpec(name="oneshot", cmd=_crasher(), restartable=False)) + _wait_exit(sup.state("oneshot")) + assert sup.check()["oneshot"] == "stopped" + assert not sup.state("oneshot").running + + +def test_recovery_marker_on_forced_kill(tmp_path: Path) -> None: + """Kindprozess ignoriert SIGTERM → kill → Recovery-Markierung (§26.4). + + Der Kindprozess bestätigt die Handler-Registrierung über eine Ready-Datei; + erst danach sendet der Test SIGTERM (keine Race Condition). + """ + if sys.platform == "win32": + return # Signal-Ignorieren unter Windows nicht testbar + ready = tmp_path / "ready.txt" + ignore_term = ( + "import signal, time, sys\n" + "signal.signal(signal.SIGTERM, lambda *a: None)\n" + f"open(r'{ready}', 'w').write('ok')\n" + "time.sleep(30)\n" + ) + sup = Supervisor(AppPaths(root=tmp_path)) + sup.start( + ProcessSpec( + name="stubborn", + cmd=[sys.executable, "-c", ignore_term], + stop_timeout_s=0.3, + ) + ) + _wait_file(ready) # Handler aktiv, bevor SIGTERM gesendet wird + rc = sup.stop("stubborn") + assert rc is not None + lines = sup.consume_recovery_marker() + assert any("forced-kill stubborn" in line for line in lines) + # Markierung nach dem Lesen gelöscht + assert sup.consume_recovery_marker() == [] + + +def test_shutdown_stops_all_processes(tmp_path: Path) -> None: + sup = Supervisor(AppPaths(root=tmp_path)) + sup.start(ProcessSpec(name="renderer", cmd=_sleeper())) + sup.start(ProcessSpec(name="core", cmd=_sleeper())) + sup.shutdown() + assert not sup.state("renderer").running + assert not sup.state("core").running + + +def test_environment_gets_gstreamer_vars(tmp_path: Path) -> None: + """Kindprozess erbt die portable GStreamer-Umgebung (§9.2, ADR-0002). + + Der Kindprozess schreibt beide Variablen in eine Datei (kein stdout- + Piping nötig).""" + paths = AppPaths(root=tmp_path) + result_file = tmp_path / "env_result.txt" + probe = ( + "import os\n" + f"open(r'{result_file}', 'w').write(\n" + " os.environ.get('GST_PLUGIN_SYSTEM_PATH_1_0', 'MISSING')\n" + " + '|'\n" + " + os.environ.get('GST_PLUGIN_PATH_1_0', 'MISSING')\n" + ")\n" + ) + sup = Supervisor(paths) + sup.start( + ProcessSpec(name="envprobe", cmd=[sys.executable, "-c", probe], restartable=False) + ) + _wait_exit(sup.state("envprobe")) + _wait_file(result_file) + content = result_file.read_text(encoding="utf-8") + system_path, plugin_path = content.split("|", 1) + assert system_path == "" # System-Plugins unterdrückt (§9.2) + assert plugin_path.endswith("gstreamer-1.0") # gebündelte Plugin-Untermenge