diff --git a/src/health-check.py b/src/health-check.py index cba05c0d25..71b6ea82f9 100644 --- a/src/health-check.py +++ b/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", diff --git a/src/watcher_identity.py b/src/watcher_identity.py index 238ea3b57d..0ab7d3ae83 100644 --- a/src/watcher_identity.py +++ b/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 ` -> `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 ", 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 diff --git a/tests/health-check-unread-watcher.test.py b/tests/health-check-unread-watcher.test.py new file mode 100644 index 0000000000..d11f7b20a1 --- /dev/null +++ b/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 `: ` ` 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)