Phase 1: Supervisor/Launcher mit Prozessüberwachung (§6.1A, §26)
- Supervisor: echte Kindprozesse, portable Umgebung, freie Portwahl - Kontrolliertes Beenden: terminate -> wait -> kill mit Recovery-Markierung - Restart-Policy mit Crashloop-Erkennung (Zeitfenster-Schwelle) - Recovery-Marker: forced-kill wird persistent dokumentiert (§26.4) - 10 Tests mit echten Prozessen (keine Mocks): Start/Stop, Crash-Restart, Crashloop, SIGTERM-resistenter Prozess, Environment-Vererbung
This commit is contained in:
HMS MediaEngine Agent
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user