feat(health-check): report an inbox held by a watcher nothing is reading

src/health-check.py

@@ -9422,6 +9422,20 @@ def check_task_watcher() -> dict:
                           f"sentinel(s) name no provable live watcher — {'; '.join(faults)}. "
                           f"Each is a separate instance's "
                           "record; a live peer does not clear it"}
+    # A watcher holds its inbox whether or not anything consumes what it
+    # announces; the reader is the only difference visible from outside.
+    unread = []
+    for _p in sorted(live):
+        _sink = watcher_identity.output_sink(_p)
+        if _sink.observed and _sink.read is False:
+            unread.append(f"{_p} -> {_sink.kind} {_sink.target}".rstrip())
+    if unread:
+        return {"name": name, "status": "warn",
+                "detail": f"{len(live)} watcher(s) alive (pids {alive}), but "
+                          f"{len(unread)} announce(s) where nothing is reading: "
+                          f"{'; '.join(unread)}. That inbox is held but not served — "
+                          "a session start will exit naming the holder, so clearing it "
+                          "needs `watch-tasks-stream.sh --force-restart` on the owner's word"}
     if len(live) == 1:
         return {"name": name, "status": "ok", "detail": f"streaming watcher alive (pid {alive})"}
     return {"name": name, "status": "ok",

src/watcher_identity.py

@@ -21,6 +21,7 @@
 
 import os
 import re
+import shutil
 import subprocess
 import sys
 from pathlib import Path
@@ -401,6 +402,88 @@ def standby_present(inbox: str, ps_output: Optional[str] = None,
     return None if saw_undecidable else False
 
 
+class OutputSink(NamedTuple):
+    """Where a watcher's announcements go, and whether anything reads them.
+
+    `read` is None when the question is not worth asking rather than when it
+    failed: see `output_sink`. `observed` False means `lsof` could not be
+    consulted, so nothing here is evidence."""
+    observed: bool
+    kind: str
+    target: str
+    read: Optional[bool]
+
+
+def _lsof(args: List[str], run: Callable) -> Optional[str]:
+    """`lsof` output, or None when it could not be consulted. Exit 1 is lsof's
+    "nothing matched", which is an answer; only a missing or broken lsof is not."""
+    lsof = shutil.which("lsof") or "/usr/sbin/lsof"
+    try:
+        r = run([lsof, *args], capture_output=True, text=True, timeout=10)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    if r.returncode not in (0, 1):
+        return None
+    return r.stdout or ""
+
+
+def _fields(text: str):
+    """lsof -F records as dicts, one per open file, carrying the owning pid."""
+    pid, cur = None, {}
+    for line in text.splitlines():
+        if not line:
+            continue
+        tag, val = line[0], line[1:]
+        if tag == "p":
+            if cur:
+                yield cur
+            pid, cur = val, {}
+        elif tag == "f":
+            if cur:
+                yield cur
+            cur = {"p": pid}
+        else:
+            cur[tag] = val
+    if cur:
+        yield cur
+
+
+def output_sink(pid, run: Callable = subprocess.run) -> OutputSink:
+    """What pid's stdout is, and whether another process is reading it.
+
+    This is how a stray watcher is told from a working one: a watcher holds its
+    inbox whether or not anything consumes what it announces, and the only
+    difference visible from outside is the reader.
+
+    Decided for a regular file (are there other openers with read access) and
+    for /dev/null (nothing can read it). Left at None for a pipe, fifo or
+    socket, which is not evasion: the watcher exits on its first failed write,
+    so those clear themselves at the next announcement. A tty means an operator
+    is attached, which counts as read."""
+    text = _lsof(["-p", str(pid), "-a", "-d", "1", "-F", "ftn"], run)
+    if text is None:
+        return OutputSink(False, "unknown", "", None)
+    rec = next((r for r in _fields(text) if "t" in r), None)
+    if rec is None:
+        return OutputSink(True, "unknown", "", None)
+    kind_raw, target = rec.get("t", ""), rec.get("n", "")
+    if kind_raw == "REG":
+        readers = _lsof(["-F", "pan", "--", target], run)
+        if readers is None:
+            return OutputSink(False, "file", target, None)
+        me = str(pid)
+        found = any(r.get("p") != me and "r" in (r.get("a") or "")
+                    for r in _fields(readers) if r.get("n") == target)
+        return OutputSink(True, "file", target, found)
+    if kind_raw == "CHR":
+        if target.endswith("/null"):
+            return OutputSink(True, "discarded", target, False)
+        return OutputSink(True, "tty", target, True)
+    if kind_raw in ("FIFO", "PIPE", "unix", "IPv4", "IPv6", "sock"):
+        return OutputSink(True, "stream", target, None)
+    return OutputSink(True, kind_raw or "unknown", target, None)
+
+
 def main(argv=None) -> int:
     """`watcher_identity.py <pid>` -> `watcher`, `not-watcher`, `dead` or
     `unknown` on stdout, with `why=` beneath. Exit 0 when decided, 2 when not,
@@ -413,6 +496,17 @@ def main(argv=None) -> int:
     `ps` snapshot itself failed -- `unknown` must never read as `no` to a
     caller deciding whether to start a duplicate watcher."""
     args = list(sys.argv[1:] if argv is None else argv)
+    if args and args[0] == "output-sink":
+        if len(args) != 2 or as_pid(args[1]) is None:
+            print("usage: watcher_identity.py output-sink <pid>", file=sys.stderr)
+            return 64
+        sink = output_sink(args[1])
+        print(f"{sink.kind} {sink.target}".strip())
+        print("read=" + ("unknown" if sink.read is None else ("yes" if sink.read else "no")))
+        if not sink.observed:
+            print("why=lsof could not be consulted", file=sys.stderr)
+            return 2
+        return 0
     if args and args[0] == "standby-present":
         rest = args[1:]
         inbox = None

tests/health-check-unread-watcher.test.py

@@ -0,0 +1,209 @@
+#!/usr/bin/env python3
+"""A watcher nobody reads holds its inbox without serving it — and says so.
+
+A watcher keeps its inbox whether or not anything consumes what it announces,
+and since a second session-role start now exits naming the holder rather than
+doubling it, an unread holder stops tasks reaching the core until someone runs
+`--force-restart`. From outside, the reader is the only difference between that
+and a healthy watcher, so health-check asks for it.
+
+Two halves, tested apart:
+  * `watcher_identity.output_sink` — what fd 1 is and whether anything reads it,
+    driven with recorded `lsof` output so no toolchain or live process is needed.
+  * `check_task_watcher` — a live, sentinel-owning watcher whose output nothing
+    reads must warn, and the same watcher with a reader must stay ok.
+
+Run: python3 tests/health-check-unread-watcher.test.py  (exit 0/1)
+"""
+from __future__ import annotations
+
+import contextlib
+import importlib.util
+import io
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "src"))
+import watcher_identity as wid  # noqa: E402
+
+spec = importlib.util.spec_from_file_location("health_check", REPO / "src" / "health-check.py")
+hc = importlib.util.module_from_spec(spec)
+assert spec and spec.loader
+spec.loader.exec_module(hc)
+
+WATCHER_PID = "7100"
+# Two tokens: the argv shape classify_argv can decide from a flattened string.
+INBOX_ARGV = "bash src/watch-tasks-stream.sh"
+
+
+def fake_lsof(fd1: str, readers: str = "", rc: int = 0):
+    """Answer the two lsof calls output_sink makes: fd 1, then that path's openers."""
+    def run(cmd, **kw):
+        text = readers if "--" in cmd else fd1
+        return mock.Mock(returncode=rc, stdout=text)
+    return run
+
+
+FD1_FILE = "p7100\nf1\ntREG\nn/ws/logs/watcher.log\n"
+FD1_SOCKET = "p7100\nf1\ntunix\nn->0xdeadbeef\n"
+FD1_NULL = "p7100\nf1\ntCHR\nn/dev/null\n"
+FD1_TTY = "p7100\nf1\ntCHR\nn/dev/ttys004\n"
+READER = "p9001\nf3\nar\nn/ws/logs/watcher.log\n"
+WRITER_ONLY = "p9002\nf3\naw\nn/ws/logs/watcher.log\n"
+SELF_ONLY = "p7100\nf1\naw\nn/ws/logs/watcher.log\n"
+
+
+class OutputSink(unittest.TestCase):
+    def test_a_file_with_a_reader_is_read(self):
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_FILE, SELF_ONLY + READER))
+        self.assertEqual((s.observed, s.kind, s.read), (True, "file", True))
+
+    def test_a_file_with_only_writers_is_unread(self):
+        """THE STRAY: the watcher and an fswatch hold it for writing, nobody reads."""
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_FILE, SELF_ONLY + WRITER_ONLY))
+        self.assertEqual((s.observed, s.kind, s.read), (True, "file", False))
+        self.assertEqual(s.target, "/ws/logs/watcher.log")
+
+    def test_the_watchers_own_write_handle_is_not_a_reader(self):
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_FILE, SELF_ONLY))
+        self.assertIs(s.read, False)
+
+    def test_dev_null_is_decided_unread_without_asking_who_reads_it(self):
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_NULL))
+        self.assertEqual((s.kind, s.read), ("discarded", False))
+
+    def test_a_terminal_counts_as_read(self):
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_TTY))
+        self.assertEqual((s.kind, s.read), ("tty", True))
+
+    def test_a_socket_is_left_undecided_because_it_self_clears(self):
+        """The watcher exits on its first failed write, so a dead peer resolves itself."""
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_SOCKET))
+        self.assertEqual((s.kind, s.read), ("stream", None))
+
+    def test_an_unconsultable_lsof_is_not_observed(self):
+        def boom(cmd, **kw):
+            raise OSError("no lsof")
+        s = wid.output_sink(WATCHER_PID, run=boom)
+        self.assertEqual((s.observed, s.read), (False, None))
+
+    def test_an_lsof_that_errors_is_not_observed(self):
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_FILE, rc=2))
+        self.assertIs(s.observed, False)
+
+    def test_lsof_exit_1_is_an_answer_not_a_failure(self):
+        """Exit 1 is lsof's "nothing matched": the path has no other opener."""
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(FD1_FILE, SELF_ONLY, rc=1))
+        self.assertEqual((s.observed, s.read), (True, False))
+
+
+def probe_with_sink(sink) -> dict:
+    """check_task_watcher over one live, sentinel-owning watcher with this sink."""
+    with tempfile.TemporaryDirectory() as td:
+        ws = Path(td)
+        (ws / "state" / "cores").mkdir(parents=True)
+        (ws / "state" / "cores" / f"{hc._host_label()}.alive").write_text("{}")
+        (ws / "state" / "watch-tasks-stream.pid").write_text(WATCHER_PID)
+        saved = (hc.WORKSPACE_DIR, hc._proc_argv, hc._watcher_trees, hc._ps_snapshot,
+                 hc._pid_parent, hc._pid_instance_id, hc._pid_actor_id)
+        try:
+            hc.WORKSPACE_DIR = ws
+            hc._proc_argv = lambda pid: INBOX_ARGV
+            hc._watcher_trees = lambda *a, **k: {WATCHER_PID: {WATCHER_PID}}
+            hc._ps_snapshot = lambda *a, **k: ""
+            hc._pid_parent = lambda pid, ps=None: None
+            hc._pid_instance_id = lambda pid: ""
+            hc._pid_actor_id = lambda pid: ""
+            with mock.patch.object(hc.watcher_identity, "output_sink", lambda p: sink):
+                return hc.check_task_watcher()
+        finally:
+            (hc.WORKSPACE_DIR, hc._proc_argv, hc._watcher_trees, hc._ps_snapshot,
+             hc._pid_parent, hc._pid_instance_id, hc._pid_actor_id) = saved
+
+
+class Probe(unittest.TestCase):
+    def test_an_unread_watcher_warns_and_names_what_to_run(self):
+        r = probe_with_sink(wid.OutputSink(True, "file", "/ws/logs/watcher.log", False))
+        self.assertEqual(r["status"], "warn")
+        self.assertIn("nothing is reading", r["detail"])
+        self.assertIn("/ws/logs/watcher.log", r["detail"])
+        self.assertIn("--force-restart", r["detail"])
+
+    def test_a_read_watcher_stays_ok(self):
+        r = probe_with_sink(wid.OutputSink(True, "file", "/ws/logs/watcher.log", True))
+        self.assertEqual(r["status"], "ok")
+
+    def test_an_undecided_sink_stays_ok(self):
+        """A socket says nothing either way; silence must not become a warning."""
+        r = probe_with_sink(wid.OutputSink(True, "stream", "->0xdeadbeef", None))
+        self.assertEqual(r["status"], "ok")
+
+    def test_an_unobserved_sink_stays_ok(self):
+        """No lsof is not evidence of a stray."""
+        r = probe_with_sink(wid.OutputSink(False, "unknown", "", None))
+        self.assertEqual(r["status"], "ok")
+
+
+class OutputSinkEdges(unittest.TestCase):
+    """lsof output shapes the happy paths above never produce."""
+
+    def test_blank_lines_and_a_typeless_record_before_the_typed_one_are_skipped(self):
+        # `p` opens the block; each `f` starts a record; a blank line is skipped;
+        # a record with no `t` field says nothing and the next one is read.
+        fd1 = "p7100\nf1\nn/dev/null\n\nf1\ntREG\nn/ws/logs/watcher.log\n"
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof(fd1, SELF_ONLY + READER))
+        self.assertEqual((s.kind, s.target, s.read), ("file", "/ws/logs/watcher.log", True))
+
+    def test_an_unconsultable_reader_lookup_is_unobserved_but_keeps_the_target(self):
+        def run(cmd, **kw):
+            if "--" in cmd:
+                return mock.Mock(returncode=2, stdout="")
+            return mock.Mock(returncode=0, stdout=FD1_FILE)
+        s = wid.output_sink(WATCHER_PID, run=run)
+        self.assertEqual((s.observed, s.kind, s.target, s.read), (False, "file", "/ws/logs/watcher.log", None))
+
+    def test_an_unfamiliar_kind_is_reported_as_itself_and_left_undecided(self):
+        s = wid.output_sink(WATCHER_PID, run=fake_lsof("p7100\nf1\ntDIR\nn/ws\n"))
+        self.assertEqual((s.observed, s.kind, s.read), (True, "DIR", None))
+
+
+class OutputSinkCli(unittest.TestCase):
+    """`output-sink <pid>`: `<kind> <target>` then `read=yes|no|unknown`; rc 2 when
+    lsof could not be consulted, rc 64 on a bad pid."""
+
+    def _run(self, argv, sink=None):
+        out, err = io.StringIO(), io.StringIO()
+        ctx = mock.patch.object(wid, "output_sink", lambda p: sink) if sink else contextlib.nullcontext()
+        with ctx, contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+            rc = wid.main(argv)
+        return rc, out.getvalue().splitlines(), err.getvalue()
+
+    def test_an_unread_file_prints_its_target_and_read_no(self):
+        rc, out, _ = self._run(["output-sink", WATCHER_PID],
+                               wid.OutputSink(True, "file", "/ws/logs/watcher.log", False))
+        self.assertEqual((rc, out), (0, ["file /ws/logs/watcher.log", "read=no"]))
+
+    def test_a_read_tty_prints_read_yes(self):
+        rc, out, _ = self._run(["output-sink", WATCHER_PID], wid.OutputSink(True, "tty", "/dev/ttys004", True))
+        self.assertEqual((rc, out), (0, ["tty /dev/ttys004", "read=yes"]))
+
+    def test_an_undecided_stream_prints_read_unknown(self):
+        rc, out, _ = self._run(["output-sink", WATCHER_PID], wid.OutputSink(True, "stream", "", None))
+        self.assertEqual((rc, out), (0, ["stream", "read=unknown"]))
+
+    def test_unobserved_is_rc_2_and_says_why(self):
+        rc, out, err = self._run(["output-sink", WATCHER_PID], wid.OutputSink(False, "unknown", "", None))
+        self.assertEqual((rc, out), (2, ["unknown", "read=unknown"]))
+        self.assertIn("why=", err)
+
+    def test_a_missing_or_non_numeric_pid_is_a_usage_error(self):
+        self.assertEqual(self._run(["output-sink"])[0], 64)
+        self.assertEqual(self._run(["output-sink", "abc"])[0], 64)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)