From cbe062e002f47342a49d411489dc71d86e54e2a7 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 3 Jul 2026 15:19:36 -0700 Subject: [PATCH] [eric] 9router: event-driven death-watcher heals in ~0.2s + two-strike watchdog (busy-probe false negatives) --- backend/apps/nine_router/process.py | 65 +++++++++++++++++-- backend/tests/test_router_watchdog.py | 92 ++++++++++++++++++++++++++- 2 files changed, 148 insertions(+), 9 deletions(-) diff --git a/backend/apps/nine_router/process.py b/backend/apps/nine_router/process.py index 67c330c2..933a8827 100644 --- a/backend/apps/nine_router/process.py +++ b/backend/apps/nine_router/process.py @@ -353,9 +353,10 @@ async def ensure_running(): p_start_lock = asyncio.Lock() async with p_start_lock: await p_ensure_running_impl() - # Arm the watchdog the moment the router becomes a live dependency; users who never route through it never spawn it. + # Arm both healers the moment the router becomes a live dependency; users who never route through it never spawn them. if is_running(): start_watchdog() + start_death_watcher() def has_persisted_connections() -> bool: @@ -381,8 +382,10 @@ p_watchdog_task: "asyncio.Task | None" = None async def p_watchdog_loop() -> None: - """Revive 9Router whenever it dies mid-session (OOM, crash, orphaned by a hard kill) so a - running app never sits on a dead router; only quitting OpenSwarm (stop()) ends it.""" + """Backstop healer for routers we DIDN'T spawn (adopted port-holders have no handle for the + death-watcher). Two-strike confirmation before reviving: the sync is_running probe can + false-negative while a busy router streams, and acting on one bad probe would rotate a LIVE + router's request log and burn a duplicate spawn attempt.""" failures = 0 while True: await asyncio.sleep(P_WATCHDOG_BACKOFF_SECONDS if failures >= 3 else P_WATCHDOG_INTERVAL_SECONDS) @@ -390,7 +393,11 @@ async def p_watchdog_loop() -> None: if is_running(): failures = 0 continue - logger.warning("9Router watchdog: router is down; reviving") + await asyncio.sleep(2) + if is_running(): + failures = 0 + continue + logger.warning("9Router watchdog: router is down (confirmed twice); reviving") await ensure_running() if is_running(): failures = 0 @@ -404,6 +411,49 @@ async def p_watchdog_loop() -> None: logger.exception("9Router watchdog iteration failed") +# Instant healer for the process WE spawned: its exit wakes us the moment it happens (no polling, +# no false positives), so total heal time = just the respawn. Crash-loop guard: 3 deaths inside +# 60s defers to the backed-off watchdog instead of hot-spinning a broken install. +p_death_watcher_task: "asyncio.Task | None" = None +p_recent_death_monos: "list[float]" = [] + + +async def p_death_watch(proc_handle: "subprocess.Popen[Any]") -> None: + global p_is_running_last_ok + loop = asyncio.get_running_loop() + try: + await loop.run_in_executor(None, proc_handle.wait) + except asyncio.CancelledError: + raise + except Exception: + return + # stop() nulls p_process before this continuation can run (it blocks the loop through wait), so a deliberate quit or a superseded handle never triggers a revive. + if proc_handle is not p_process: + return + now = time.monotonic() + p_recent_death_monos.append(now) + del p_recent_death_monos[:-3] + if len(p_recent_death_monos) == 3 and now - p_recent_death_monos[0] < 60: + logger.warning("9Router died 3x in 60s; leaving revival to the backed-off watchdog") + return + logger.warning("9Router process died; instant revive") + p_is_running_last_ok = 0.0 + await ensure_running() + + +def start_death_watcher() -> None: + """Idempotent per spawned handle; no-op for adopted routers (no handle to wait on).""" + global p_death_watcher_task + if p_process is None or p_process.poll() is not None: + return + if p_death_watcher_task is not None and not p_death_watcher_task.done(): + return + try: + p_death_watcher_task = asyncio.get_running_loop().create_task(p_death_watch(p_process)) + except RuntimeError: + logger.warning("9Router death-watcher: no running loop; not armed") + + def start_watchdog() -> None: """Idempotent; armed by ensure_running() on success, cancelled by stop().""" global p_watchdog_task @@ -541,11 +591,14 @@ async def p_ensure_running_impl(): def stop(): """Stop the 9Router subprocess.""" - global p_process, p_watchdog_task - # Cancel the watchdog FIRST or it would revive the router we're about to kill (shutdown = the one sanctioned "down"). + global p_process, p_watchdog_task, p_death_watcher_task + # Cancel the healers FIRST or they would revive the router we're about to kill (shutdown = the one sanctioned "down"). if p_watchdog_task is not None: p_watchdog_task.cancel() p_watchdog_task = None + if p_death_watcher_task is not None: + p_death_watcher_task.cancel() + p_death_watcher_task = None if p_process: try: p_process.terminate() diff --git a/backend/tests/test_router_watchdog.py b/backend/tests/test_router_watchdog.py index 23bae15b..335f03ff 100644 --- a/backend/tests/test_router_watchdog.py +++ b/backend/tests/test_router_watchdog.py @@ -30,16 +30,102 @@ def test_watchdog_revives_then_backs_off(): patch.object(proc, "ensure_running", fake_ensure), \ patch.object(proc.asyncio, "sleep", fake_sleep): task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop()) - while len(sleeps) < 6: + while len(sleeps) < 9: await real_sleep(0) task.cancel() try: await task except asyncio.CancelledError: pass - assert len(ensures) >= 3, "a down router must be revived on every pulse" + assert len(ensures) >= 2, "a confirmed-down router must be revived" assert sleeps[0] == proc.P_WATCHDOG_INTERVAL_SECONDS - assert sleeps[4] == proc.P_WATCHDOG_BACKOFF_SECONDS, "3 straight failures must back off" + assert sleeps[1] == 2, "two-strike: a single failed probe must be re-confirmed before reviving" + assert proc.P_WATCHDOG_BACKOFF_SECONDS in sleeps, "3 straight failures must back off" + + asyncio.run(run()) + + +def test_watchdog_single_false_negative_never_revives(): + async def run(): + sleeps: list = [] + ensures: list = [] + probes: list = [] + + async def fake_sleep(d): + sleeps.append(d) + await real_sleep(0) + + async def fake_ensure(): + ensures.append(1) + + def flaky_is_running(): + # First probe of each pulse fails (busy-router false negative); the confirm succeeds. + probes.append(1) + return len(probes) % 2 == 0 + + real_sleep = asyncio.sleep + with patch.object(proc, "is_running", flaky_is_running), \ + patch.object(proc, "ensure_running", fake_ensure), \ + patch.object(proc.asyncio, "sleep", fake_sleep): + task = asyncio.get_running_loop().create_task(proc.p_watchdog_loop()) + while len(probes) < 8: + await real_sleep(0) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + assert not ensures, "a transient probe failure must never trigger a revive" + + asyncio.run(run()) + + +def test_death_watcher_revives_instantly_and_guards_loops(): + async def run(): + ensures: list = [] + + async def fake_ensure(): + ensures.append(1) + + class FakeProc: + def __init__(self): + self.dead = False + def wait(self): + while not self.dead: + pass + def poll(self): + return 1 if self.dead else None + + fp = FakeProc() + proc.p_recent_death_monos.clear() + with patch.object(proc, "ensure_running", fake_ensure), \ + patch.object(proc, "p_process", fp): + task = asyncio.get_running_loop().create_task(proc.p_death_watch(fp)) + await asyncio.sleep(0.05) + assert not ensures, "no revive while the process lives" + fp.dead = True + for _ in range(200): + if ensures: + break + await asyncio.sleep(0.01) + assert ensures, "process death must trigger an instant revive" + await task + # Crash-loop guard: a 3rd death inside 60s defers to the watchdog. + ensures.clear() + proc.p_recent_death_monos[:] = [proc.time.monotonic() - 5, proc.time.monotonic() - 3] + fp2 = FakeProc(); fp2.dead = True + with patch.object(proc, "ensure_running", fake_ensure), \ + patch.object(proc, "p_process", fp2): + await proc.p_death_watch(fp2) + assert not ensures, "3 deaths in 60s must defer to the backed-off watchdog" + # A superseded/stopped handle never revives. + ensures.clear() + proc.p_recent_death_monos.clear() + fp3 = FakeProc(); fp3.dead = True + with patch.object(proc, "ensure_running", fake_ensure), \ + patch.object(proc, "p_process", None): + await proc.p_death_watch(fp3) + assert not ensures, "a deliberately stopped router must stay down" asyncio.run(run())