From 6cb14954d4696270e12d4adb800bc337251dc631 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 24 Jul 2026 18:34:20 +0300 Subject: [PATCH] fix(simulation): stabilize PX4 readiness supervision --- simulation/s1/stock-rover-lifecycle.yaml | 1 - src/k1link/simulation/orchestrator.py | 9 ++- src/k1link/simulation/process_supervisor.py | 73 +++++++++++++-------- src/k1link/simulation/stock_rover.py | 1 + tests/test_simulation_s1_orchestrator.py | 15 ++++- 5 files changed, 67 insertions(+), 32 deletions(-) diff --git a/simulation/s1/stock-rover-lifecycle.yaml b/simulation/s1/stock-rover-lifecycle.yaml index c5e1ed4..110fa84 100644 --- a/simulation/s1/stock-rover-lifecycle.yaml +++ b/simulation/s1/stock-rover-lifecycle.yaml @@ -16,4 +16,3 @@ px4_ready_log_patterns: - 'world: rover, model: rover_ackermann_0' - Startup script returned successfully - successfully created rt/fmu/out/vehicle_status_v1 data writer - - Running, connected diff --git a/src/k1link/simulation/orchestrator.py b/src/k1link/simulation/orchestrator.py index 4cfcf59..ce409d3 100644 --- a/src/k1link/simulation/orchestrator.py +++ b/src/k1link/simulation/orchestrator.py @@ -419,7 +419,7 @@ class SimulationApplicationService: return self._terminal_failure( starting, operation="start", - detail=type(exc).__name__, + detail=_exception_detail(exc), observed_at_utc=observed_at_utc, host_monotonic_ns=host_monotonic_ns, sim_time_ns=0, @@ -471,7 +471,7 @@ class SimulationApplicationService: return self._terminal_failure( run, operation=operation, - detail=type(exc).__name__, + detail=_exception_detail(exc), observed_at_utc=observed_at_utc, host_monotonic_ns=host_monotonic_ns, sim_time_ns=sim_time_ns, @@ -537,6 +537,11 @@ class SimulationApplicationService: ) +def _exception_detail(exc: Exception) -> str: + message = " ".join(str(exc).split()) + return type(exc).__name__ if not message else f"{type(exc).__name__}: {message}" + + def _idempotency_key(value: str) -> str: normalized = value.strip() if not 1 <= len(normalized) <= 160: diff --git a/src/k1link/simulation/process_supervisor.py b/src/k1link/simulation/process_supervisor.py index d07e05f..c1f79ab 100644 --- a/src/k1link/simulation/process_supervisor.py +++ b/src/k1link/simulation/process_supervisor.py @@ -30,6 +30,7 @@ class ProcessSpec: ready_log_patterns: tuple[str, ...] = () health_timeout_seconds: float = 0.0 health_poll_interval_seconds: float = 0.05 + keep_stdin_open: bool = False interrupt_timeout_seconds: float = 2.0 terminate_timeout_seconds: float = 1.0 @@ -225,7 +226,7 @@ class PosixProcessSupervisor: spec.argv, cwd=run_root, env=environment, - stdin=subprocess.DEVNULL, + stdin=subprocess.PIPE if spec.keep_stdin_open else subprocess.DEVNULL, stdout=stdout, stderr=stderr, start_new_session=True, @@ -274,31 +275,40 @@ class PosixProcessSupervisor: stderr_path: Path, ) -> None: pending = {pattern: pattern.encode("utf-8") for pattern in spec.ready_log_patterns} + maximum_tail_bytes = max(len(pattern) for pattern in pending.values()) - 1 deadline = time.monotonic() + spec.health_timeout_seconds - while pending: - return_code = process.poll() - if return_code is not None: - raise ProcessSupervisorError( - f"provider {spec.process_id} exited before readiness with {return_code}; " - f"missing markers: {', '.join(pending)}" - ) - for path in (stdout_path, stderr_path): - content = path.read_bytes() - pending = { - pattern: encoded - for pattern, encoded in pending.items() - if encoded not in content - } - if not pending: - return - remaining = deadline - time.monotonic() - if remaining <= 0: - raise ProcessSupervisorError( - f"provider {spec.process_id} did not reach readiness within " - f"{spec.health_timeout_seconds:g} seconds; " - f"missing markers: {', '.join(pending)}" - ) - time.sleep(min(spec.health_poll_interval_seconds, remaining)) + with ( + stdout_path.open("rb", buffering=0) as stdout, + stderr_path.open("rb", buffering=0) as stderr, + ): + streams = (stdout, stderr) + tails = [b"", b""] + while pending: + return_code = process.poll() + if return_code is not None: + raise ProcessSupervisorError( + f"provider {spec.process_id} exited before readiness with {return_code}; " + f"missing markers: {', '.join(pending)}" + ) + for index, stream in enumerate(streams): + content = tails[index] + stream.read() + if content: + pending = { + pattern: encoded + for pattern, encoded in pending.items() + if encoded not in content + } + tails[index] = content[-maximum_tail_bytes:] if maximum_tail_bytes else b"" + if not pending: + return + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ProcessSupervisorError( + f"provider {spec.process_id} did not reach readiness within " + f"{spec.health_timeout_seconds:g} seconds; " + f"missing markers: {', '.join(pending)}" + ) + time.sleep(min(spec.health_poll_interval_seconds, remaining)) def _stop_group( self, @@ -314,6 +324,7 @@ class PosixProcessSupervisor: ): if process.poll() is not None and not _group_exists(pgid): process.wait(timeout=0) + _close_stdin(process) return True try: if group_signal_supported: @@ -321,6 +332,7 @@ class PosixProcessSupervisor: else: process.send_signal(sent_signal) except ProcessLookupError: + _close_stdin(process) return True except PermissionError: # Some macOS application sandboxes deny killpg for an owned @@ -334,11 +346,15 @@ class PosixProcessSupervisor: if process.poll() is not None: process.wait(timeout=0) if not group_signal_supported or not _group_exists(pgid): + _close_stdin(process) return True time.sleep(0.01) - return process.poll() is not None and ( + stopped = process.poll() is not None and ( not group_signal_supported or not _group_exists(pgid) ) + if stopped: + _close_stdin(process) + return stopped def _persist_registry( self, @@ -368,3 +384,8 @@ def _group_exists(pgid: int) -> bool: except PermissionError: return True return True + + +def _close_stdin(process: subprocess.Popen[bytes]) -> None: + if process.stdin is not None and not process.stdin.closed: + process.stdin.close() diff --git a/src/k1link/simulation/stock_rover.py b/src/k1link/simulation/stock_rover.py index 9bf46d5..91fc052 100644 --- a/src/k1link/simulation/stock_rover.py +++ b/src/k1link/simulation/stock_rover.py @@ -171,6 +171,7 @@ def stock_rover_process_specs( ready_log_patterns=profile.px4_ready_log_patterns, health_timeout_seconds=120, health_poll_interval_seconds=0.25, + keep_stdin_open=True, interrupt_timeout_seconds=10, terminate_timeout_seconds=5, ), diff --git a/tests/test_simulation_s1_orchestrator.py b/tests/test_simulation_s1_orchestrator.py index 1685a59..f995704 100644 --- a/tests/test_simulation_s1_orchestrator.py +++ b/tests/test_simulation_s1_orchestrator.py @@ -212,6 +212,10 @@ def test_start_failure_and_profile_drift_fail_terminal(tmp_path: Path) -> None: ) assert failed.state is RunState.FAILED assert failed.terminal_reason == "orchestrator-start-failed" + assert any( + event.payload.get("detail") == "RuntimeError: synthetic start failure" + for event in service.store.list_events("run-s1b-001") + ) store = QualificationRunStore(tmp_path / "drift") store.create(_run()) @@ -343,11 +347,13 @@ def test_process_supervisor_rolls_back_partial_start(tmp_path: Path) -> None: @pytest.mark.skipif(os.name != "posix", reason="target supervisor is POSIX-only") def test_process_supervisor_waits_for_all_declared_ready_markers(tmp_path: Path) -> None: child = ( - "import signal,time;" + "import signal,sys,time;" "signal.signal(signal.SIGINT,lambda *_:exit(0));" "print('provider booting',flush=True);" "time.sleep(0.05);" - "print('transport connected',flush=True);" + "sys.stdout.write('transport ');sys.stdout.flush();" + "time.sleep(0.05);" + "print('connected',flush=True);" "time.sleep(60)" ) supervisor = PosixProcessSupervisor(tmp_path / "runtime") @@ -432,7 +438,10 @@ def test_stock_rover_profile_builds_exact_reviewed_provider_graph() -> None: ) assert specs[0].ready_log_patterns == ("running...",) assert specs[1].argv[-2:] == ("px4_sitl", "gz_rover_ackermann") - assert specs[1].ready_log_patterns[-1] == "Running, connected" + assert specs[1].ready_log_patterns[-1] == ( + "successfully created rt/fmu/out/vehicle_status_v1 data writer" + ) + assert specs[1].keep_stdin_open assert dict(specs[1].environment) == { "HEADLESS": "1", "PX4_SIM_SPEED_FACTOR": "1",