From 69fa594a07725022088e9efdd2a2ef5fe91738c4 Mon Sep 17 00:00:00 2001 From: David Beyer Date: Sat, 11 Jul 2026 22:16:24 -0700 Subject: [PATCH] fix(gateway): single-owner gate + zero-sub early exit for kanban notifier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The kanban notifier watcher had two gaps its own comment claimed were already handled: - Every gateway process on the machine polled every board DB each tick. dispatch_in_gateway defaults to true, so N gateways meant N concurrent pollers per board — exactly the -shm/-wal contention the dispatcher's singleton lock exists to prevent, but the notifier had no equivalent. - Boards with zero subscriptions were still opened writable every tick (connect() runs schema init/migration on first open per process), churning WAL state on DBs with nothing to deliver. Three changes in gateway/kanban_watchers.py: - Notifier singleton lock: acquire .notifier.lock (beside the dispatcher's .dispatcher.lock at the machine-global kanban root) via the existing _acquire_singleton_lock helper, after the env + dispatch_in_gateway short-circuits so those still win. Contended → this gateway polls nothing; unavailable → config-only fallback, mirroring the dispatcher's branches. Released on cancellation and on loop exit (the mid-sleep return became break so the single post-loop release covers it). - Zero-sub early exit: probe each board with the new read-only count_notify_subs before connect(); zero subscriptions skips the writable open entirely. A failed probe falls back to the writable open — delivery is never lost to the cheap check. - The stale gate comment (which described a dispatch-owner gate that never existed) now documents the real gate order. hermes_cli/kanban_db.py gains count_notify_subs: mode=ro URI open, missing DB or legacy table-less DB counts 0 without creating/migrating anything, WAL-uncheckpointed rows visible, sqlite3.Error propagated so callers pick the fallback. The singleton-lock helper docstrings now cover both loops. Co-Authored-By: Claude Fable 5 --- gateway/kanban_watchers.py | 78 ++++++-- hermes_cli/kanban_db.py | 38 ++++ .../test_kanban_notifier_owner_gate.py | 182 ++++++++++++++++++ .../test_kanban_count_notify_subs.py | 98 ++++++++++ 4 files changed, 383 insertions(+), 13 deletions(-) create mode 100644 tests/gateway/test_kanban_notifier_owner_gate.py create mode 100644 tests/hermes_cli/test_kanban_count_notify_subs.py diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index eb1c68ffd66b..1d6ecfb46a32 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -58,15 +58,17 @@ def _resolve_auto_decompose_settings( def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]": - """Take an exclusive, non-blocking advisory lock for the sole dispatcher. + """Take an exclusive, non-blocking advisory lock for a singleton loop. Only one gateway process machine-wide may run the embedded kanban - dispatcher: concurrent dispatchers double the reclaim frequency (each - runs its own ``release_stale_claims`` → promote → dispatch loop), double - claim-attempt events in the event log, and — with ``wal_autocheckpoint=0`` — - concurrent manual WAL checkpoints can corrupt index pages. The - ``dispatch_in_gateway`` config flag is the primary control; this lock is the - backstop that survives config drift and same-profile restart races. + dispatcher (``.dispatcher.lock``): concurrent dispatchers double the reclaim + frequency (each runs its own ``release_stale_claims`` → promote → dispatch + loop), double claim-attempt events in the event log, and — with + ``wal_autocheckpoint=0`` — concurrent manual WAL checkpoints can corrupt + index pages. The notifier uses the same helper with its own + ``.notifier.lock`` so only one gateway polls board DBs for notifications. + The ``dispatch_in_gateway`` config flag is the primary control; the lock is + the backstop that survives config drift and same-profile restart races. Delegates to :func:`gateway.status._try_acquire_file_lock` (``fcntl`` on POSIX, ``msvcrt`` on Windows) so the guard is cross-platform. @@ -95,7 +97,7 @@ def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]": def _release_singleton_lock(handle) -> None: - """Release a dispatcher singleton lock acquired via :func:`_acquire_singleton_lock`.""" + """Release a singleton lock acquired via :func:`_acquire_singleton_lock`.""" if handle is None: return try: @@ -131,10 +133,13 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: cross boards, so delivery semantics are unchanged — this is purely a fan-out of the single-DB poll. """ - # Gate: only the dispatch-owning gateway opens kanban DBs for notifier polling. - # Non-dispatch gateways have no subscriptions to deliver — all kanban state lives - # in the dispatch owner's per-board DBs. This prevents N-gateway -shm contention. - # TODO: gate per-board when per-board dispatcher_owner tracking lands. + # Gate order mirrors `_kanban_dispatcher_watcher`: the + # HERMES_KANBAN_DISPATCH_IN_GATEWAY env override, then the + # `kanban.dispatch_in_gateway` config flag, then the machine-global + # singleton `.notifier.lock` below — so only ONE gateway machine-wide + # polls kanban DBs for notifications. Per-board work is further gated + # by a read-only subscription probe, so boards with zero subscriptions + # are never opened writable (no schema migration, no checkpoints). try: from hermes_cli.config import load_config as _load_config except Exception: @@ -162,6 +167,32 @@ async def _kanban_notifier_watcher(self, interval: float = 5.0) -> None: logger.warning("kanban notifier: kanban_db not importable; notifier disabled") return + # Single-notifier backstop, mirroring the dispatcher's singleton lock: + # `dispatch_in_gateway` defaults to true, so without it every gateway + # polls every board DB each tick — the exact N-gateway -shm/-wal + # contention this gate exists to prevent. The lock lives at the + # machine-global kanban root (shared across profiles by design), so it + # serialises ALL gateways; a `.notifier.lock` separate from + # `.dispatcher.lock` lets a gateway notify without dispatching. + self._kanban_notifier_lock_handle = None + _lock_path = _kb.kanban_home() / "kanban" / ".notifier.lock" + _lock_handle, _lock_state = _acquire_singleton_lock(_lock_path) + if _lock_state == "contended": + logger.info( + "kanban notifier: another gateway already holds the notifier " + "lock (%s); this gateway will NOT poll for notifications.", + _lock_path, + ) + return + if _lock_state == "held": + self._kanban_notifier_lock_handle = _lock_handle # hold for process lifetime + logger.info("kanban notifier: holding singleton notifier lock (%s)", _lock_path) + else: + logger.warning( + "kanban notifier: advisory lock unavailable at %s; proceeding " + "on config control alone.", _lock_path, + ) + # "status" covers dashboard drag-drop and `_set_status_direct()` # writes — surface those transitions to subscribers too. TERMINAL_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked") @@ -230,6 +261,19 @@ def _collect(): ) continue seen_db_paths.add(resolved_db_path) + try: + if _kb.count_notify_subs(board=slug) == 0: + logger.debug( + "kanban notifier: board %s has no subscriptions; skipping open", + slug, + ) + continue + except Exception as exc: + logger.debug( + "kanban notifier: read-only subscription probe failed " + "for board %s (%s); falling back to writable open", + slug, exc, + ) try: conn = _kb.connect(board=slug) except Exception as exc: @@ -565,14 +609,22 @@ def _collect(): await asyncio.to_thread( self._kanban_unsub, sub, board_slug, ) + except asyncio.CancelledError: + logger.debug("kanban notifier: cancelled") + _release_singleton_lock(self._kanban_notifier_lock_handle) + self._kanban_notifier_lock_handle = None + raise except Exception as exc: logger.warning("kanban notifier tick failed: %s", exc) # Sleep with cancellation checks. for _ in range(int(max(1, interval))): if not self._running: - return + break await asyncio.sleep(1) + _release_singleton_lock(self._kanban_notifier_lock_handle) + self._kanban_notifier_lock_handle = None + def _kanban_advance( self, sub: dict, cursor: int, board: Optional[str] = None, ) -> None: diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 518e74eb0647..14f6b72364c3 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -8308,6 +8308,44 @@ def list_notify_subs( return [dict(r) for r in rows] +def count_notify_subs( + db_path: Optional[Path] = None, + *, + board: Optional[str] = None, +) -> int: + """Count ``kanban_notify_subs`` rows via a read-only connection. + + Cheap probe for the gateway notifier's zero-subscription early exit: + unlike :func:`connect`, this never creates the DB file, never runs + schema init/migration, and never opens the database writable (no + write locks, no checkpoints — though a read-only open of a WAL + database may still create the ``-shm``/``-wal`` sidecars, it cannot + write table content). Rows in a not-yet-checkpointed WAL are + visible, so a freshly added subscription is never missed. A missing + DB, or a legacy DB that predates the subscriptions table, counts as + zero. Path resolution matches :func:`connect` (explicit ``db_path``, + else ``board`` via :func:`kanban_db_path`). Raises + :class:`sqlite3.Error` when the DB exists but cannot be read + (locked, corrupt); callers choose their own fallback. + """ + path = db_path if db_path is not None else kanban_db_path(board=board) + if not path.exists(): + return 0 + conn = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True) + try: + try: + row = conn.execute( + "SELECT COUNT(*) FROM kanban_notify_subs" + ).fetchone() + except sqlite3.OperationalError as exc: + if "no such table" in str(exc).lower(): + return 0 + raise + return int(row[0]) if row else 0 + finally: + conn.close() + + def remove_notify_sub( conn: sqlite3.Connection, *, diff --git a/tests/gateway/test_kanban_notifier_owner_gate.py b/tests/gateway/test_kanban_notifier_owner_gate.py new file mode 100644 index 000000000000..5e8411adf154 --- /dev/null +++ b/tests/gateway/test_kanban_notifier_owner_gate.py @@ -0,0 +1,182 @@ +"""Tests for the kanban notifier single-owner gate + zero-subscription skip. + +The notifier used to writable-open EVERY board DB from EVERY gateway whose +config left ``kanban.dispatch_in_gateway`` at the default (true) — the exact +N-gateway ``-shm``/``-wal`` contention its own top-of-function comment +claimed to prevent. It now mirrors the dispatcher's machine-global singleton +advisory lock with its own ``/kanban/.notifier.lock``: the +lock-losing gateway polls zero boards and opens zero connections. Per-board +work is further gated by a read-only subscription probe +(``kanban_db.count_notify_subs``), so boards with zero subscriptions are +never opened writable. +""" + +import asyncio + +from unittest.mock import patch + +from gateway.config import Platform +from gateway.kanban_watchers import _acquire_singleton_lock, _release_singleton_lock +from gateway.run import GatewayRunner +from hermes_cli import kanban_db as kb + + +class RecordingAdapter: + def __init__(self): + self.sent = [] + + async def send(self, chat_id, text, metadata=None): + self.sent.append({"chat_id": chat_id, "text": text, "metadata": metadata or {}}) + + +def _make_runner(adapter): + runner = GatewayRunner.__new__(GatewayRunner) + runner._running = True + runner.adapters = {Platform.TELEGRAM: adapter} + runner._kanban_sub_fail_counts = {} + return runner + + +async def _run_one_notifier_tick(monkeypatch, runner): + real_sleep = asyncio.sleep + + async def fake_sleep(delay): + if delay == 5: + return None + runner._running = False + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + await runner._kanban_notifier_watcher(interval=1) + + +def _create_completed_task(*, subscribe: bool) -> str: + conn = kb.connect() + try: + tid = kb.create_task(conn, title="owner gate", assignee="worker") + if subscribe: + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat-1") + kb.complete_task(conn, tid, summary="done") + return tid + finally: + conn.close() + + +def test_zero_sub_board_is_never_opened_writable(tmp_path, monkeypatch): + """A board with zero subscriptions must be skipped BEFORE `_kb.connect`.""" + db_path = tmp_path / "zero-subs.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + _create_completed_task(subscribe=False) + + adapter = RecordingAdapter() + runner = _make_runner(adapter) + + with patch.object(kb, "connect", wraps=kb.connect) as spy_connect: + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + spy_connect.assert_not_called() + assert adapter.sent == [] + + +def test_subscribed_board_still_delivers_through_the_gate(tmp_path, monkeypatch): + """Regression: the lock + probe must not change delivery for the owner.""" + db_path = tmp_path / "subscribed.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + tid = _create_completed_task(subscribe=True) + + adapter = RecordingAdapter() + runner = _make_runner(adapter) + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + assert len(adapter.sent) == 1 + assert tid in adapter.sent[0]["text"] + + +def test_lock_losing_instance_polls_zero_boards(tmp_path, monkeypatch): + """While another holder owns `.notifier.lock`, the watcher must return at + the gate: zero board enumerations, zero probes, zero DB opens, zero + deliveries.""" + db_path = tmp_path / "locked-out.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + _create_completed_task(subscribe=True) + + lock_path = kb.kanban_home() / "kanban" / ".notifier.lock" + handle, state = _acquire_singleton_lock(lock_path) + assert state == "held", "test precondition: we hold the notifier lock" + try: + adapter = RecordingAdapter() + runner = _make_runner(adapter) + with patch.object(kb, "connect", wraps=kb.connect) as spy_connect, \ + patch.object(kb, "list_boards", wraps=kb.list_boards) as spy_boards, \ + patch.object(kb, "count_notify_subs", wraps=kb.count_notify_subs) as spy_probe: + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + finally: + _release_singleton_lock(handle) + + spy_boards.assert_not_called() + spy_probe.assert_not_called() + spy_connect.assert_not_called() + assert adapter.sent == [] + + +def test_two_instances_exactly_one_polls(tmp_path, monkeypatch): + """Two concurrent notifier instances against one kanban root: the first + acquires `.notifier.lock` and delivers; the second returns at the lock — + no second poller, no double delivery.""" + db_path = tmp_path / "two-instances.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + _create_completed_task(subscribe=True) + + adapter1 = RecordingAdapter() + adapter2 = RecordingAdapter() + runner1 = _make_runner(adapter1) + runner2 = _make_runner(adapter2) + + real_sleep = asyncio.sleep + + async def fake_sleep(delay): + if delay == 5: + return None + runner1._running = False + runner2._running = False + await real_sleep(0) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + + async def run_both(): + await asyncio.gather( + runner1._kanban_notifier_watcher(interval=1), + runner2._kanban_notifier_watcher(interval=1), + ) + + asyncio.run(run_both()) + + deliveries = adapter1.sent + adapter2.sent + assert len(deliveries) == 1, f"exactly one delivery expected, got {deliveries!r}" + # gather() starts runner1 first, so it wins the lock deterministically; + # runner2 must be the locked-out instance. + assert adapter2.sent == [] + + +def test_notifier_lock_released_on_return(tmp_path, monkeypatch): + """A finished watcher must release `.notifier.lock` so a successor in the + same process (gateway restart-in-place, next test) can acquire it.""" + db_path = tmp_path / "release.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + _create_completed_task(subscribe=True) + + adapter = RecordingAdapter() + asyncio.run(_run_one_notifier_tick(monkeypatch, _make_runner(adapter))) + assert len(adapter.sent) == 1 + + lock_path = kb.kanban_home() / "kanban" / ".notifier.lock" + handle, state = _acquire_singleton_lock(lock_path) + try: + assert state == "held", "watcher exit must release the notifier lock" + finally: + _release_singleton_lock(handle) diff --git a/tests/hermes_cli/test_kanban_count_notify_subs.py b/tests/hermes_cli/test_kanban_count_notify_subs.py new file mode 100644 index 000000000000..f1361e29c5ee --- /dev/null +++ b/tests/hermes_cli/test_kanban_count_notify_subs.py @@ -0,0 +1,98 @@ +"""Tests for ``kanban_db.count_notify_subs`` — the read-only subscription probe. + +The gateway notifier uses it to skip boards with zero subscriptions BEFORE +any writable ``connect()``: the probe must never create the DB file, never +run schema init/migration, and never write — that first-open cost on every +tick is exactly what the zero-sub early exit avoids. It must also never +UNDER-count: rows sitting in a not-yet-checkpointed WAL still count, or the +notifier would skip a board that has a live subscription. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setenv("HERMES_KANBAN_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + return home + + +def test_missing_db_counts_zero_and_creates_nothing(kanban_home): + db_path = kb.kanban_db_path(board="default") + assert not db_path.exists() + assert kb.count_notify_subs(board="default") == 0 + assert not db_path.exists(), "read-only probe must not create the DB" + + +def test_counts_rows_via_board_resolution(kanban_home): + conn = kb.connect(board="default") + try: + tid = kb.create_task(conn, title="t", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c2") + finally: + conn.close() + assert kb.count_notify_subs(board="default") == 2 + + +def test_probe_is_read_only_and_sees_uncheckpointed_wal_rows(kanban_home): + """A sub committed by a still-open writer (rows only in the WAL, not yet + checkpointed into the main DB file) must be counted — under-counting + would make the notifier skip a board that has a live subscription. And + the probe itself must be read-only: the writer's connection stays the + only writer.""" + conn = kb.connect(board="default") + try: + tid = kb.create_task(conn, title="t", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + # Writer still open: the row lives in the -wal, not the main file. + assert kb.count_notify_subs(board="default") == 1 + finally: + conn.close() + + +def test_legacy_db_without_subs_table_counts_zero_and_stays_unmigrated(tmp_path): + legacy = tmp_path / "legacy.db" + conn = sqlite3.connect(legacy) + try: + conn.execute("CREATE TABLE something_else (id INTEGER)") + conn.commit() + finally: + conn.close() + assert kb.count_notify_subs(db_path=legacy) == 0 + # The probe must not have run schema init on the foreign/legacy DB. + conn = sqlite3.connect(legacy) + try: + tables = { + r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + } + finally: + conn.close() + assert "kanban_notify_subs" not in tables, ( + "read-only probe must never create schema" + ) + + +def test_explicit_db_path_overrides_board(kanban_home, tmp_path): + pinned = tmp_path / "pinned.db" + conn = kb.connect(db_path=pinned) + try: + tid = kb.create_task(conn, title="t", assignee="w") + kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="c1") + finally: + conn.close() + assert kb.count_notify_subs(pinned) == 1 + assert kb.count_notify_subs(board="default") == 0