diff --git a/codec_alerts.py b/codec_alerts.py index 20b80d9..66844ea 100644 --- a/codec_alerts.py +++ b/codec_alerts.py @@ -27,6 +27,7 @@ import subprocess import time import urllib.error +import urllib.parse import urllib.request from datetime import datetime from email.mime.text import MIMEText @@ -177,6 +178,28 @@ def _check_service(url: str, timeout: int = 5) -> bool: return False +def _is_listening(url: str, timeout: int = 2) -> bool: + """Is something still accepting TCP connections on this URL's port? + + Distinguishes BUSY from DEAD. A single-threaded model server doing a large + prefill (an 11k-token prompt at ~800 tok/s blocks for ~14s) cannot answer an + HTTP probe inside `_check_service`'s 5s timeout — but its socket is still + bound and accepting. A crashed process is not. + + Without this, one slow probe made the heartbeat auto-restart the model + MID-GENERATION, which killed the user's in-flight chat request: the reply + came back as reasoning-with-no-answer. Long prompts were self-destructing. + """ + try: + parsed = urllib.parse.urlparse(url if "//" in url else "//" + url) + host = parsed.hostname or "127.0.0.1" + port = parsed.port or (443 if parsed.scheme == "https" else 80) + with socket.create_connection((host, int(port)), timeout=timeout): + return True + except Exception: + return False + + def _try_restart(service_pm2_name: str) -> bool: """Attempt to restart a service via PM2. Returns True if restart command succeeded.""" try: @@ -261,6 +284,22 @@ def check_services_and_alert(): if failures[name] == 1: state[f"down_since_{name}"] = now + # BUSY IS NOT DOWN. If the port still accepts connections, the + # process is alive and merely too busy to answer inside the probe + # timeout — the normal state for a model server mid-prefill on a + # long prompt. Restarting it there killed the user's in-flight + # request and returned an empty answer. Never restart on that; + # a genuinely crashed process stops listening and still recovers. + if _is_listening(url): + failures[name] = 0 + if f"down_since_{name}" in state: + del state[f"down_since_{name}"] + log.info( + "%s slow to answer but still listening — treating as busy, " + "not restarting", name, + ) + continue + if failures[name] == 1: # First failure — try auto-restart (with cooldown to prevent # restart loops). extra_services never appear in _PM2_NAMES, diff --git a/tests/test_watchdog.py b/tests/test_watchdog.py index 52a5676..1a283d7 100644 --- a/tests/test_watchdog.py +++ b/tests/test_watchdog.py @@ -105,15 +105,74 @@ def probe(url, timeout=5): def test_builtin_down_triggers_single_restart(alert_harness, monkeypatch): - """LLM+Vision share qwen3.6 — one down pass must restart it at most once.""" + """LLM+Vision share qwen3.6 — one down pass must restart it at most once. + + `_is_listening` is forced False so this exercises a genuinely DEAD service. + Without the override the real probe would reach a live :8083 on a dev box and + the busy-guard would skip the restart, passing for the wrong reason. + """ cfg, state_holder, calls = alert_harness def probe(url, timeout=5): return "8083" not in url # qwen URL down, everything else up monkeypatch.setattr(codec_alerts, "_check_service", probe) + monkeypatch.setattr(codec_alerts, "_is_listening", lambda url, timeout=2: False) codec_alerts.check_services_and_alert() - assert calls["restarts"].count("qwen3.6") <= 1, calls["restarts"] + assert calls["restarts"].count("qwen3.6") == 1, calls["restarts"] + + +# ── busy-is-not-down guard ─────────────────────────────────────────────────── +# A model server mid-prefill on a long prompt cannot answer the 5s HTTP probe, +# but its socket is still bound. Restarting it there killed the user's in-flight +# chat request and returned an empty answer ("reasoning ran on without reaching +# a final answer"). Long prompts were self-destructing. + + +def test_slow_but_listening_service_is_not_restarted(alert_harness, monkeypatch): + """HTTP probe times out while the port still accepts → BUSY, never restart.""" + cfg, state_holder, calls = alert_harness + + monkeypatch.setattr(codec_alerts, "_check_service", + lambda url, timeout=5: "8083" not in url) + monkeypatch.setattr(codec_alerts, "_is_listening", lambda url, timeout=2: True) + + codec_alerts.check_services_and_alert() + + assert "qwen3.6" not in calls["restarts"], ( + "a busy-but-listening model must never be restarted mid-generation: " + f"{calls['restarts']}" + ) + assert not [a for a in calls["alerts"] if a[0] == "critical"], calls["alerts"] + + +def test_busy_service_does_not_accumulate_failures(alert_harness, monkeypatch): + """Repeated slow-but-listening passes must not escalate to a critical alert.""" + cfg, state_holder, calls = alert_harness + + monkeypatch.setattr(codec_alerts, "_check_service", + lambda url, timeout=5: "8083" not in url) + monkeypatch.setattr(codec_alerts, "_is_listening", lambda url, timeout=2: True) + + for _ in range(4): + codec_alerts.check_services_and_alert() + + assert not calls["restarts"], calls["restarts"] + assert not [a for a in calls["alerts"] if a[0] == "critical"], calls["alerts"] + + +def test_is_listening_distinguishes_busy_from_dead(): + """The guard's whole job: a bound socket is 'busy', a closed port is 'dead'.""" + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.bind(("127.0.0.1", 0)) + srv.listen(1) + port = srv.getsockname()[1] + try: + # bound but never answers HTTP — exactly a model mid-prefill + assert codec_alerts._is_listening(f"http://127.0.0.1:{port}/v1/models") is True + finally: + srv.close() + assert codec_alerts._is_listening(f"http://127.0.0.1:{port}/v1/models") is False # ── check_pm2_restart_storms ─────────────────────────────────────────────────