diff --git a/solstone/think/runner.py b/solstone/think/runner.py index 9c8b5409a..5e5d8cca5 100644 --- a/solstone/think/runner.py +++ b/solstone/think/runner.py @@ -33,6 +33,7 @@ from solstone.think.callosum import CallosumConnection from solstone.think.utils import CHRONICLE_DIR, get_journal, now_ms logger = logging.getLogger(__name__) +KILL_REAP_GRACE_S = 0.5 def _get_journal_path() -> Path: @@ -406,8 +407,9 @@ class ManagedProcess: Sends SIGTERM to the immediate child and the process group. Waits up to `timeout` seconds for graceful exit. If the process is still alive after - `timeout`, escalates to SIGKILL on the group and child, then re-raises - `subprocess.TimeoutExpired` after the kill completes. + `timeout`, escalates to SIGKILL on the group and child, waits up to + `KILL_REAP_GRACE_S` for the kill to reap, then abandons the wait and + re-raises `subprocess.TimeoutExpired`. Args: timeout: Seconds to wait after SIGTERM before SIGKILL (default: 15). @@ -452,8 +454,17 @@ class ManagedProcess: self.process.kill() except (ProcessLookupError, OSError): pass - self.process.wait() - logger.debug(f"{self.name} killed with code {self.process.returncode}") + try: + self.process.wait(timeout=KILL_REAP_GRACE_S) + logger.debug(f"{self.name} killed with code {self.process.returncode}") + except subprocess.TimeoutExpired: + logger.warning( + "%s remained unreaped %.1fs after SIGKILL " + "(likely D-state; supervisor exiting; SIGKILL guarantees " + "eventual death)", + self.name, + KILL_REAP_GRACE_S, + ) raise def cleanup(self) -> None: diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index d1c679107..7a2a9be26 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -101,10 +101,25 @@ _parent_death_sigterm_sent = threading.Event() def app_supervised_graceful_budget_s() -> float: - """Worst-case non-wedge graceful time after watcher self-SIGTERM.""" + """Sum of configured app-supervised graceful shutdown step budgets. + + After the parent-death watcher self-SIGTERMs, the graceful path runs these + configured caps in finally-block order: handle_shutdown's managed-child + reap, task-queue drain, managed child-stop, and Callosum join. + + The assertion that this budget stays below APP_SUPERVISED_SHUTDOWN_CEILING_S + guards that configured step budgets leave room under the hard parent-death + backstop. It is not a guarantee that wall time can never exceed this sum. + In the common non-D-state case, bounded terminate calls (timeout plus + KILL_REAP_GRACE_S) keep the graceful path well under the ceiling. For + pathological slow-to-reap children, a step may exceed its nominal cap; the + parent-death backstop remains the hard guarantee by sleeping to the ceiling, + SIGKILLing every child's process group, then calling os._exit(1). + """ return ( HANDLE_SHUTDOWN_REAP_S + APP_SUPERVISED_TASK_DRAIN_S + + APP_SUPERVISED_CHILD_STOP_S + APP_SUPERVISED_CALLOSUM_JOIN_S ) @@ -155,34 +170,84 @@ def enforce_parent_death_shutdown_deadline( *, ceiling: float = APP_SUPERVISED_SHUTDOWN_CEILING_S, managed_procs: Iterable[RunnerManagedProcess] | None = None, + task_procs: Iterable[RunnerManagedProcess] | None = None, sent_event: "threading.Event | None" = None, kill: Callable[[int, int], None] | None = None, + killpg: Callable[[int, int], None] | None = None, + getpgid: Callable[[int], int] | None = None, exit_now: Callable[[int], NoReturn] | None = None, sleep: Callable[[float], None] | None = None, ) -> None: del reason + own_pid = os.getpid() + own_pgid = os.getpgrp() kill_fn = kill or os.kill + killpg_fn = killpg or os.killpg + getpgid_fn = getpgid or os.getpgid exit_now_fn = exit_now or os._exit sleep_fn = sleep or time.sleep sigterm_sent = sent_event if sent_event is not None else _parent_death_sigterm_sent if not sigterm_sent.is_set(): sigterm_sent.set() - kill_fn(os.getpid(), signal.SIGTERM) + kill_fn(own_pid, signal.SIGTERM) sleep_fn(ceiling) procs = managed_procs if managed_procs is not None else _managed_procs - for managed in procs: + if task_procs is not None: + task_snapshot = task_procs + elif _task_queue is not None: + with _task_queue._lock: + task_snapshot = list(_task_queue._active.values()) + else: + task_snapshot = [] + + def _kill_group(managed: RunnerManagedProcess) -> None: if not managed.is_running(): - continue + return try: - managed.process.kill() + pgid = getpgid_fn(managed.process.pid) + except (ProcessLookupError, OSError): + logger.exception( + "parent-death backstop: could not resolve pgid for %s", + managed.name, + ) + return + + if pgid == own_pgid or pgid == own_pid: + logger.warning( + "parent-death backstop: refusing to signal supervisor's own " + "group (pgid=%s) for %s", + pgid, + managed.name, + ) + return + + try: + killpg_fn(pgid, signal.SIGKILL) except Exception: logger.exception( "parent-death backstop: SIGKILL failed for %s", managed.name ) + for managed in procs: + try: + _kill_group(managed) + except Exception: + logger.exception( + "parent-death backstop: unexpected failure for %s", + getattr(managed, "name", managed), + ) + for managed in task_snapshot: + try: + _kill_group(managed) + except Exception: + logger.exception( + "parent-death backstop: unexpected failure for %s", + getattr(managed, "name", managed), + ) + exit_now_fn(1) @@ -237,12 +302,12 @@ def _candidate_journal(proc: "psutil.Process") -> Path | None: return None -# The long-lived managed-service proctitles set by setproctitle at -# sol_cli.py (f"{binary}:{cmd}"). setproctitle is in-process and persists -# until the process exits, so an orphaned service still reports its title -# via proc.name() after the supervisor dies — which is what lets the sweep -# find it. The supervisor-owned `llama-server` reports its own bare binary -# name (no colon prefix) and is included here so the sweep reaps it too. +# The long-lived journal proctitles set by setproctitle at sol_cli.py +# (f"{binary}:{cmd}"). setproctitle is in-process and persists until the +# process exits, so an orphaned service or task child still reports its title +# via proc.name() after the supervisor dies, which is what lets the sweep find +# it. The supervisor-owned `llama-server` reports its own bare binary name (no +# colon prefix) and is included here so the sweep reaps it too. # The mlx-vlm server is a Python process, but our launcher sets the same # managed proctitle so proc.name() is stable for orphan sweeping. _LOCAL_SERVER_PROCTITLES = frozenset( @@ -251,17 +316,18 @@ _LOCAL_SERVER_PROCTITLES = frozenset( MLX_SERVER_PROCESS_NAME, } ) -_MANAGED_SERVICE_PROCTITLES = ( - frozenset( - { - "journal:sense", - "journal:cortex", - "journal:convey", - "journal:spl", - } - ) - | _LOCAL_SERVER_PROCTITLES -) + + +def _is_sweepable_orphan_name(name: str) -> bool: + """True if proc.name() identifies a sweepable orphan of this install. + + Any `journal:*` proctitle - managed service or task-queue child - plus the + bare local-server binary names. A PPID-1, same-journal `journal:*` process + is by definition an orphan of a dead supervisor. `solstone:*`/`sol:*` and a + bare `journal` (no colon) are deliberately not matched because they cannot + be positively classified as a sub-command of this install. + """ + return name.startswith("journal:") or name in _LOCAL_SERVER_PROCTITLES def _sweep_orphaned_sol_processes(journal: Path, grace: float = 5.0) -> int: @@ -271,7 +337,7 @@ def _sweep_orphaned_sol_processes(journal: Path, grace: float = 5.0) -> int: targets: list[int] = [] for proc in psutil.process_iter(["name", "ppid", "username", "pid"]): try: - if proc.name() not in _MANAGED_SERVICE_PROCTITLES: + if not _is_sweepable_orphan_name(proc.name()): continue if proc.ppid() != 1: continue diff --git a/tests/test_runner.py b/tests/test_runner.py index 7615c04b2..23dbe2944 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -12,7 +12,7 @@ from unittest.mock import Mock, call import pytest -from solstone.think.runner import ManagedProcess, run_task +from solstone.think.runner import KILL_REAP_GRACE_S, ManagedProcess, run_task @pytest.fixture @@ -69,7 +69,41 @@ def test_terminate_uses_process_group(monkeypatch): timeout_process.terminate.assert_called_once_with() timeout_process.kill.assert_called_once_with() - assert timeout_process.wait.call_args_list == [call(timeout=2), call()] + assert timeout_process.wait.call_args_list == [ + call(timeout=2), + call(timeout=KILL_REAP_GRACE_S), + ] + assert killpg.call_args_list == [ + call(456, signal.SIGTERM), + call(456, signal.SIGKILL), + ] + + +def test_terminate_abandons_unreaped_child_after_sigkill(monkeypatch): + killpg = Mock() + monkeypatch.setattr("solstone.think.runner.os.getpgid", lambda pid: 456) + monkeypatch.setattr("solstone.think.runner.os.killpg", killpg) + + process = Mock() + process.pid = 125 + graceful_timeout = subprocess.TimeoutExpired(cmd=["test"], timeout=2) + process.wait.side_effect = [ + graceful_timeout, + subprocess.TimeoutExpired(cmd=["test"], timeout=KILL_REAP_GRACE_S), + ] + process.returncode = None + managed = _managed_for_process(process) + + with pytest.raises(subprocess.TimeoutExpired) as exc_info: + managed.terminate(timeout=2) + + assert exc_info.value is graceful_timeout + process.terminate.assert_called_once_with() + process.kill.assert_called_once_with() + assert process.wait.call_args_list == [ + call(timeout=2), + call(timeout=KILL_REAP_GRACE_S), + ] assert killpg.call_args_list == [ call(456, signal.SIGTERM), call(456, signal.SIGKILL), diff --git a/tests/test_supervisor_orphan_sweep.py b/tests/test_supervisor_orphan_sweep.py index 9b38888ee..2507d0a17 100644 --- a/tests/test_supervisor_orphan_sweep.py +++ b/tests/test_supervisor_orphan_sweep.py @@ -7,7 +7,7 @@ from pathlib import Path import pytest from solstone.think import supervisor -from solstone.think.supervisor import _MANAGED_SERVICE_PROCTITLES +from solstone.think.providers.mlx_server import MLX_SERVER_PROCESS_NAME TEST_JOURNAL = Path("/journal/test") @@ -60,8 +60,23 @@ class TestOrphanSweep: ) return kills - @pytest.mark.parametrize("proctitle", sorted(_MANAGED_SERVICE_PROCTITLES)) - def test_managed_service_proctitles_are_sigtermed(self, monkeypatch, proctitle): + @pytest.mark.parametrize( + "proctitle", + [ + "journal:sense", + "journal:cortex", + "journal:convey", + "journal:spl", + "journal:think", + "journal:heartbeat", + "journal:identity", + "journal:providers", + "journal:facet-candidates", + "llama-server", + MLX_SERVER_PROCESS_NAME, + ], + ) + def test_sweepable_orphan_proctitles_are_sigtermed(self, monkeypatch, proctitle): procs = [_FakeProcess(pid=111, name=proctitle)] kills = self._patch_common(monkeypatch, procs) monkeypatch.setattr(supervisor.psutil, "pid_exists", lambda _pid: False) @@ -71,9 +86,9 @@ class TestOrphanSweep: @pytest.mark.parametrize( "proctitle", - ["journal:think", "journal:heartbeat", "sol:call"], + ["sol:call", "solstone:convey", "journal", "python"], ) - def test_non_managed_orphan_proctitles_are_ignored(self, monkeypatch, proctitle): + def test_non_sweepable_orphan_proctitles_are_ignored(self, monkeypatch, proctitle): procs = [_FakeProcess(pid=111, name=proctitle)] kills = self._patch_common(monkeypatch, procs) diff --git a/tests/test_supervisor_shutdown.py b/tests/test_supervisor_shutdown.py index a0df84d1d..790e0f9df 100644 --- a/tests/test_supervisor_shutdown.py +++ b/tests/test_supervisor_shutdown.py @@ -14,10 +14,10 @@ import solstone.think.supervisor as supervisor class FakeManaged: - def __init__(self, name, exits_after_terminate=True): + def __init__(self, name, exits_after_terminate=True, pid=12345): self.name = name self.process = MagicMock() - self.process.pid = 12345 + self.process.pid = pid self._running = True self._exits_after_terminate = exits_after_terminate self.process.terminate.side_effect = self._on_terminate @@ -121,20 +121,26 @@ def test_parent_watcher_shutdown_deadline_sends_sigterm_once_kills_children(): sent_event = threading.Event() managed = FakeManaged("stuck", exits_after_terminate=False) kill = MagicMock() + killpg = MagicMock() + getpgid = MagicMock(return_value=54321) exit_now = MagicMock() supervisor.enforce_parent_death_shutdown_deadline( "eof", ceiling=0, managed_procs=[managed], + task_procs=[], sent_event=sent_event, kill=kill, + killpg=killpg, + getpgid=getpgid, exit_now=exit_now, sleep=lambda _seconds: None, ) kill.assert_called_once_with(os.getpid(), signal.SIGTERM) - assert managed.process.kill.called + getpgid.assert_called_once_with(12345) + killpg.assert_called_once_with(54321, signal.SIGKILL) exit_now.assert_called_once_with(1) @@ -143,24 +149,114 @@ def test_parent_watcher_shutdown_deadline_does_not_resend_sigterm(): sent_event.set() managed = FakeManaged("stuck", exits_after_terminate=False) kill = MagicMock() + killpg = MagicMock() + getpgid = MagicMock(return_value=54321) exit_now = MagicMock() supervisor.enforce_parent_death_shutdown_deadline( "eof", ceiling=0, managed_procs=[managed], + task_procs=[], sent_event=sent_event, kill=kill, + killpg=killpg, + getpgid=getpgid, exit_now=exit_now, sleep=lambda _seconds: None, ) kill.assert_not_called() - assert managed.process.kill.called + getpgid.assert_called_once_with(12345) + killpg.assert_called_once_with(54321, signal.SIGKILL) + exit_now.assert_called_once_with(1) + + +def test_backstop_kills_managed_and_task_children_by_pgid(): + sent_event = threading.Event() + managed = FakeManaged("managed", exits_after_terminate=False, pid=100) + task = FakeManaged("task", exits_after_terminate=False, pid=200) + killpg_calls: list[tuple[int, signal.Signals]] = [] + exit_now = MagicMock() + + supervisor.enforce_parent_death_shutdown_deadline( + "eof", + ceiling=0, + managed_procs=[managed], + task_procs=[task], + sent_event=sent_event, + kill=MagicMock(), + killpg=lambda pgid, sig: killpg_calls.append((pgid, sig)), + getpgid=lambda pid: pid, + exit_now=exit_now, + sleep=lambda _seconds: None, + ) + + assert killpg_calls == [(100, signal.SIGKILL), (200, signal.SIGKILL)] + exit_now.assert_called_once_with(1) + + +def test_backstop_never_signals_supervisor_own_group(): + sent_event = threading.Event() + own_group = FakeManaged("own-group", exits_after_terminate=False, pid=100) + sibling = FakeManaged("sibling", exits_after_terminate=False, pid=200) + sibling_pgid = os.getpid() + 100000 + killpg = MagicMock() + exit_now = MagicMock() + + supervisor.enforce_parent_death_shutdown_deadline( + "eof", + ceiling=0, + managed_procs=[own_group, sibling], + task_procs=[], + sent_event=sent_event, + kill=MagicMock(), + killpg=killpg, + getpgid=lambda pid: os.getpgrp() if pid == 100 else sibling_pgid, + exit_now=exit_now, + sleep=lambda _seconds: None, + ) + + killpg.assert_called_once_with(sibling_pgid, signal.SIGKILL) + exit_now.assert_called_once_with(1) + + +def test_backstop_one_kill_failure_does_not_skip_others_or_block_exit(): + sent_event = threading.Event() + first = FakeManaged("first", exits_after_terminate=False, pid=100) + second = FakeManaged("second", exits_after_terminate=False, pid=200) + attempts: list[tuple[int, signal.Signals]] = [] + exit_now = MagicMock() + + def killpg(pgid, sig): + attempts.append((pgid, sig)) + if pgid == 100: + raise OSError("permission denied") + + supervisor.enforce_parent_death_shutdown_deadline( + "eof", + ceiling=0, + managed_procs=[first, second], + task_procs=[], + sent_event=sent_event, + kill=MagicMock(), + killpg=killpg, + getpgid=lambda pid: pid, + exit_now=exit_now, + sleep=lambda _seconds: None, + ) + + assert attempts == [(100, signal.SIGKILL), (200, signal.SIGKILL)] exit_now.assert_called_once_with(1) def test_app_supervised_graceful_budget_stays_under_hard_ceiling(): + assert supervisor.app_supervised_graceful_budget_s() == ( + supervisor.HANDLE_SHUTDOWN_REAP_S + + supervisor.APP_SUPERVISED_TASK_DRAIN_S + + supervisor.APP_SUPERVISED_CHILD_STOP_S + + supervisor.APP_SUPERVISED_CALLOSUM_JOIN_S + ) assert ( supervisor.app_supervised_graceful_budget_s() < supervisor.APP_SUPERVISED_SHUTDOWN_CEILING_S