diff --git a/gateway/run.py b/gateway/run.py index 5851b16fb987..9525e087507a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5691,6 +5691,19 @@ def _auto_decompose_tick() -> int: "kanban dispatcher: embedded in gateway (interval=%.1fs)", interval ) while self._running: + try: + # Reap zombie children before per-board work so a board DB + # failure cannot block cleanup of unrelated workers. + pids = await asyncio.to_thread(_kb.reap_worker_zombies) + if pids: + logger.info( + "kanban dispatcher: reaped %d zombie worker(s), pids=%s", + len(pids), + pids, + ) + except Exception: + logger.exception("kanban dispatcher: zombie reaper failed") + try: if auto_decompose_enabled: await asyncio.to_thread(_auto_decompose_tick) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c89e697c98d2..55a981dbef38 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -134,6 +134,34 @@ def _resolve_claim_ttl_seconds(ttl_seconds: Optional[int] = None) -> int: return DEFAULT_CLAIM_TTL_SECONDS +# Grace period after a task transitions to ``running`` during which +# ``detect_crashed_workers`` skips the ``_pid_alive`` check. Covers the +# fork() → /proc-visibility window where liveness can transiently report +# False for a freshly-spawned worker. The 15-minute claim TTL still +# catches genuinely-crashed workers; this only suppresses false positives +# during the launch window. +DEFAULT_CRASH_GRACE_SECONDS = 30 + + +def _resolve_crash_grace_seconds() -> int: + """Return the crash-detection grace period in seconds. + + Reads ``HERMES_KANBAN_CRASH_GRACE_SECONDS`` from the environment; + falls back to ``DEFAULT_CRASH_GRACE_SECONDS`` when absent, empty, + non-integer, or negative. A value of 0 restores immediate-reclaim + behaviour (useful for tests). + """ + raw = os.environ.get("HERMES_KANBAN_CRASH_GRACE_SECONDS", "").strip() + if raw: + try: + parsed = int(raw) + except ValueError: + parsed = -1 + if parsed >= 0: + return parsed + return DEFAULT_CRASH_GRACE_SECONDS + + # Worker-context caps so build_worker_context() stays bounded on # pathological boards (retry-heavy tasks, comment storms, giant # summaries). Values chosen to fit a typical 100k-char LLM prompt with @@ -1181,8 +1209,17 @@ def connect( # See hermes_state._WAL_INCOMPAT_MARKERS for detection logic. from hermes_state import apply_wal_with_fallback apply_wal_with_fallback(conn, db_label=f"kanban.db ({path.name})") - conn.execute("PRAGMA synchronous=NORMAL") + # FULL (was NORMAL): fsync before each checkpoint to narrow the + # crash window that can leave a b-tree page header torn. + conn.execute("PRAGMA synchronous=FULL") + conn.execute("PRAGMA wal_autocheckpoint=100") conn.execute("PRAGMA foreign_keys=ON") + # Zero freed pages so a later torn write cannot expose stale + # cell content; persisted in the DB header for new DBs. + conn.execute("PRAGMA secure_delete=ON") + # Surface corrupt cells as read errors instead of silent + # wrong-data returns. + conn.execute("PRAGMA cell_size_check=ON") needs_init = resolved not in _INITIALIZED_PATHS if needs_init: # Idempotent: runs CREATE TABLE IF NOT EXISTS + the additive @@ -1466,6 +1503,45 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: ) +def _check_file_length_invariant(conn: sqlite3.Connection) -> None: + """Read the SQLite header page_count and compare against actual file size. + + Raises sqlite3.DatabaseError if the file is shorter than the header claims + (torn-extend corruption). + """ + try: + row = conn.execute("PRAGMA database_list").fetchone() + if row is None: + return + path_str = row[2] # column 2 is the file path; empty for in-memory DBs + if not path_str: + return # in-memory or unnamed DB; skip + path = path_str + page_size = conn.execute("PRAGMA page_size").fetchone()[0] + file_size = os.path.getsize(path) + with open(path, "rb") as f: + f.seek(28) + header_bytes = f.read(4) + if len(header_bytes) < 4: + return # can't read header; skip + header_page_count = int.from_bytes(header_bytes, "big") + if header_page_count == 0: + return # new/empty DB; skip + actual_pages = file_size // page_size + if actual_pages < header_page_count: + raise sqlite3.DatabaseError( + f"torn-extend detected: page count mismatch on {path}: " + f"header claims {header_page_count} pages, " + f"file has {actual_pages} pages " + f"(missing {header_page_count - actual_pages} pages, " + f"file_size={file_size}, page_size={page_size})" + ) + except sqlite3.DatabaseError: + raise + except Exception: + pass # I/O errors during check are non-fatal; let normal ops continue + + @contextlib.contextmanager def write_txn(conn: sqlite3.Connection): """Context manager for an IMMEDIATE write transaction. @@ -1473,15 +1549,28 @@ def write_txn(conn: sqlite3.Connection): Use for any multi-statement write (creating a task + link, claiming a task + recording an event, etc.). A claim CAS inside this context is atomic -- at most one concurrent writer can succeed. + + The explicit ROLLBACK on exception is wrapped in try/except so that + a SQLite auto-rollback (which leaves no active transaction) does not + shadow the original exception with a spurious rollback error. """ conn.execute("BEGIN IMMEDIATE") try: yield conn except Exception: - conn.execute("ROLLBACK") + try: + conn.execute("ROLLBACK") + except sqlite3.OperationalError: + # SQLite has already auto-rolled-back the transaction (typical + # under EIO, lock contention, or corruption). Nothing to undo; + # do not let this secondary failure shadow the real one. + pass raise else: conn.execute("COMMIT") + # Post-commit file-length check: header page_count must match actual file pages. + # A discrepancy means a torn-extend — raise now rather than silently corrupt. + _check_file_length_invariant(conn) # --------------------------------------------------------------------------- @@ -4169,6 +4258,30 @@ def _classify_worker_exit(pid: int) -> "tuple[str, Optional[int]]": return ("unknown", None) +def reap_worker_zombies() -> "list[int]": + """Reap all zombie children of this process without blocking. + + Returns the list of reaped PIDs. Safe to call when there are no + children (returns []). No-op on Windows. + """ + if os.name == "nt": + return [] + reaped: "list[int]" = [] + try: + while True: + try: + pid, status = os.waitpid(-1, os.WNOHANG) + except ChildProcessError: + break + if pid == 0: + break + _record_worker_exit(pid, status) + reaped.append(pid) + except Exception: + pass + return reaped + + def _pid_alive(pid: Optional[int]) -> bool: """Return True if ``pid`` is still running on this host. @@ -4635,7 +4748,7 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: # (task_id, pid, claimer, protocol_violation, error_text) with write_txn(conn): rows = conn.execute( - "SELECT id, worker_pid, claim_lock FROM tasks " + "SELECT id, worker_pid, claim_lock, started_at FROM tasks " "WHERE status = 'running' AND worker_pid IS NOT NULL" ).fetchall() host_prefix = f"{_claimer_id().split(':', 1)[0]}:" @@ -4644,6 +4757,14 @@ def detect_crashed_workers(conn: sqlite3.Connection) -> list[str]: lock = row["claim_lock"] or "" if not lock.startswith(host_prefix): continue + # Skip liveness check inside the launch-window grace period + # so a freshly-spawned worker isn't reclaimed before its PID + # is visible on /proc. + started_at = row["started_at"] if "started_at" in row.keys() else None + if started_at is not None: + grace = _resolve_crash_grace_seconds() + if time.time() - started_at < grace: + continue if _pid_alive(row["worker_pid"]): continue @@ -5125,38 +5246,9 @@ def dispatch_once( ``board`` pins workspace/log/db resolution for this tick to a specific board. When omitted, the current-board resolution chain is used. """ - # Reap zombie children from previously spawned workers. - # The gateway-embedded dispatcher is the parent of every worker spawned - # via _default_spawn (start_new_session=True only detaches the - # controlling tty, not the parent). Without an explicit waitpid, each - # completed worker becomes a entry that lingers until gateway - # exit. WNOHANG keeps this non-blocking; ChildProcessError means no - # children to reap. Bounded: at most one tick's worth of completions - # can be in at once. - # - # We also record the exit status keyed by pid, so - # ``detect_crashed_workers`` can distinguish a worker that exited - # cleanly without calling ``kanban_complete`` / ``kanban_block`` - # (protocol violation — auto-block) from a real crash (OOM killer, - # SIGKILL, non-zero exit — existing counter behavior). - # - # Windows has no zombies / no os.WNOHANG — subprocess.Popen handles - # are freed when the Python object is garbage-collected or .wait() is - # called explicitly. The kanban dispatcher discards the Popen handle - # after spawn (``_default_spawn`` → abandon), so on Windows there's - # nothing to reap here — skip the whole block. - if os.name != "nt": - try: - while True: - try: - _pid, _status = os.waitpid(-1, os.WNOHANG) - except ChildProcessError: - break - if _pid == 0: - break - _record_worker_exit(_pid, _status) - except Exception: - pass + # Reap zombie children from previously spawned workers. See + # reap_worker_zombies() for the full rationale. + reap_worker_zombies() result = DispatchResult() result.reclaimed = release_stale_claims(conn) diff --git a/hermes_state.py b/hermes_state.py index 0391047d0550..ba33598b91e2 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -54,7 +54,6 @@ _WAL_INCOMPAT_MARKERS = ( "locking protocol", # SQLITE_PROTOCOL on NFS/SMB "not authorized", # Some FUSE mounts block WAL pragma outright - "disk i/o error", # Flaky network FS during WAL setup ) # Last SessionDB() init error, per-process. Surfaced in /resume and @@ -125,6 +124,27 @@ def format_session_db_unavailable(prefix: str = "Session database not available" return f"{prefix}: {cause}{hint}." +def _on_disk_journal_mode(conn: sqlite3.Connection) -> Optional[str]: + """Read the journal mode from the SQLite DB header on disk. + + Returns the mode string (e.g. ``"wal"``, ``"delete"``), or ``None`` + if the value cannot be determined (new DB, or PRAGMA read failed). + """ + try: + row = conn.execute("PRAGMA journal_mode").fetchone() + except sqlite3.OperationalError: + return None + if row is None: + return None + mode = row[0] + if isinstance(mode, bytes): # defensive: sqlite3 occasionally returns bytes + try: + mode = mode.decode("ascii") + except UnicodeDecodeError: + return None + return str(mode).strip().lower() if mode is not None else None + + def apply_wal_with_fallback( conn: sqlite3.Connection, *, @@ -147,7 +167,18 @@ def apply_wal_with_fallback( Shared by :class:`SessionDB` and ``hermes_cli.kanban_db.connect`` so both databases get identical fallback behavior. + + Never downgrades to DELETE if the on-disk DB header reports WAL — see _on_disk_journal_mode. """ + # Read-only probe — no flock, no checkpoint, no WAL/SHM unlink. + # Skipping the set-pragma prevents WAL-init from unlinking files other connections hold open. + try: + current_mode = conn.execute("PRAGMA journal_mode").fetchone() + if current_mode and current_mode[0] == "wal": + return "wal" + except sqlite3.OperationalError: + pass + try: conn.execute("PRAGMA journal_mode=WAL") return "wal" @@ -156,6 +187,10 @@ def apply_wal_with_fallback( if not any(marker in msg for marker in _WAL_INCOMPAT_MARKERS): # Unrelated OperationalError — don't silently swallow. raise + # Don't downgrade if another process already set WAL on disk. + existing = _on_disk_journal_mode(conn) + if existing == "wal": + raise _log_wal_fallback_once(db_label, exc) conn.execute("PRAGMA journal_mode=DELETE") return "delete" diff --git a/scripts/release.py b/scripts/release.py index 3a53f77742dd..d9e2aacd8b17 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -71,6 +71,8 @@ "schepers.zander1@gmail.com": "Strontvod", "ed@bebop.crew": "someaka", "anadi.jaggia@gmail.com": "Jaggia", + "steve@steveonjava.com": "steveonjava", + "steveonjava@gmail.com": "steveonjava", "32201324+simpolism@users.noreply.github.com": "simpolism", "simpolism@gmail.com": "simpolism", "jake@nousresearch.com": "simpolism", diff --git a/tests/hermes_cli/test_kanban_core_functionality.py b/tests/hermes_cli/test_kanban_core_functionality.py index 5b645d318f98..05fb31c4d5ff 100644 --- a/tests/hermes_cli/test_kanban_core_functionality.py +++ b/tests/hermes_cli/test_kanban_core_functionality.py @@ -35,7 +35,19 @@ def kanban_home(tmp_path, monkeypatch): home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) + # Existing crash-detection tests pre-date the grace window; pin to 0 + # so they keep their immediate-reclaim semantics. + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") monkeypatch.setattr(Path, "home", lambda: tmp_path) + # Disable the detect_crashed_workers grace period for legacy tests in + # this file that claim a task and immediately expect + # ``detect_crashed_workers`` to act on it. The grace period (30s by + # default, see ``DEFAULT_CRASH_GRACE_SECONDS``) prevents the + # multi-dispatcher reap race in production; setting it to 0 here + # restores the pre-fix instant-reclaim semantics these tests were + # written against. The grace-period itself is covered by dedicated + # tests in tests/hermes_cli/test_kanban_db.py. + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "0") kb.init_db() return home @@ -3655,9 +3667,16 @@ def _connect(*args, **kwargs): raise sqlite3.DatabaseError("file is not a database") async def _to_thread(fn, *args, **kwargs): + # PR salvage (#32857 commit 7): the dispatcher now reaps zombies at + # the top of each tick via ``asyncio.to_thread(_kb.reap_worker_zombies)`` + # BEFORE the per-board tick work. Each tick now issues 3 ``to_thread`` + # calls (reaper + ``_tick_once`` + ``_ready_nonempty``) instead of 2, + # so this counter must reach 6 to allow the same 2 dispatch ticks the + # pre-reaper test expected at 4. Connect counts in the assertion below + # are unchanged. calls["to_thread"] += 1 result = fn(*args, **kwargs) - if calls["to_thread"] >= 4: + if calls["to_thread"] >= 6: runner._running = False return result diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 883cf8f4d5db..30cb8421a20c 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -6,6 +6,7 @@ import os import sqlite3 import time +import unittest.mock from pathlib import Path import pytest @@ -564,6 +565,80 @@ def test_detect_crashed_workers_isolated_failure_normal_retry( ) +def test_detect_crashed_workers_skips_freshly_claimed_tasks( + kanban_home, monkeypatch, +): + """Grace period prevents reclaim of freshly-started tasks.""" + import hermes_cli.kanban_db as _kb + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) + monkeypatch.delenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", raising=False) + + now = 1_000_000.0 + monkeypatch.setattr(_kb.time, "time", lambda: now) + + with kb.connect() as conn: + host = _kb._claimer_id().split(":", 1)[0] + tid = kb.create_task(conn, title="grace test", assignee="a") + conn.execute( + "UPDATE tasks SET status='running', worker_pid=?, " + "claim_lock=?, started_at=? WHERE id=?", + (99999, f"{host}:w", int(now), tid), + ) + conn.commit() + + # With time = now (just claimed), grace period should suppress reclaim. + crashed = kb.detect_crashed_workers(conn) + assert tid not in crashed, "should not reclaim freshly-started task" + + # With time = now + 60 (past default 30s grace), should reclaim. + monkeypatch.setattr(_kb.time, "time", lambda: now + 60) + crashed = kb.detect_crashed_workers(conn) + assert tid in crashed, "should reclaim task past grace period" + + +def test_detect_crashed_workers_grace_period_env_override( + kanban_home, monkeypatch, +): + """HERMES_KANBAN_CRASH_GRACE_SECONDS env var adjusts the window.""" + import hermes_cli.kanban_db as _kb + + monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: False) + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", "5") + + now = 2_000_000.0 + + with kb.connect() as conn: + host = _kb._claimer_id().split(":", 1)[0] + tid = kb.create_task(conn, title="env override test", assignee="a") + conn.execute( + "UPDATE tasks SET status='running', worker_pid=?, " + "claim_lock=?, started_at=? WHERE id=?", + (99999, f"{host}:w", int(now), tid), + ) + conn.commit() + + # 3s after claim: within 5s grace → no reclaim. + monkeypatch.setattr(_kb.time, "time", lambda: now + 3) + assert tid not in kb.detect_crashed_workers(conn) + + # 6s after claim: past 5s grace → reclaim. + monkeypatch.setattr(_kb.time, "time", lambda: now + 6) + assert tid in kb.detect_crashed_workers(conn) + + +def test_resolve_crash_grace_seconds_handles_bad_env(monkeypatch): + """Bad env values fall back to DEFAULT_CRASH_GRACE_SECONDS.""" + import hermes_cli.kanban_db as _kb + + for bad_val in ("notanumber", "-5", ""): + monkeypatch.setenv("HERMES_KANBAN_CRASH_GRACE_SECONDS", bad_val) + result = _kb._resolve_crash_grace_seconds() + assert result == _kb.DEFAULT_CRASH_GRACE_SECONDS, ( + f"expected default for {bad_val!r}, got {result}" + ) + + def test_max_runtime_uses_current_run_start_after_retry(kanban_home, monkeypatch): """A retry should get a fresh max-runtime window. @@ -2097,17 +2172,31 @@ def test_latest_summaries_batch_omits_tasks_without_summary(kanban_home): # NFS / network-filesystem fallback (see hermes_state.apply_wal_with_fallback) # --------------------------------------------------------------------------- -def test_connect_falls_back_to_delete_on_locking_protocol(kanban_home, caplog): +def test_connect_falls_back_to_delete_on_locking_protocol(tmp_path, monkeypatch, caplog): """kanban_db.connect() must handle ``locking protocol`` on NFS/SMB. Without this fallback, the gateway's kanban dispatcher crashes every 60s and the kanban migration (``consecutive_failures`` ADD COLUMN) is retried forever — which is what the real-world user report shows (see hermes-agent issue #22032). + + NOTE: We do NOT use the ``kanban_home`` fixture here because that + fixture pre-initializes the DB via ``kb.init_db()`` — putting the + file in WAL on disk. The Bug D safety guard now refuses to downgrade + to DELETE when the on-disk header is already WAL, so testing the + NFS-fallback path requires a truly-fresh DB file (NFS scenario in + production: first connection of the first process ever to touch the + file, where downgrading is safe because nobody else has WAL state + yet). """ import sqlite3 as _sqlite3 from unittest.mock import patch as _patch + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + # Clear module cache so a fresh connect() is attempted kb._INITIALIZED_PATHS.clear() @@ -3339,3 +3428,380 @@ def test_maybe_emit_scratch_tip_skips_non_scratch_workspaces(kanban_home, caplog ).fetchall() assert "tip_scratch_workspace" not in [e["kind"] for e in events] + +# --------------------------------------------------------------------------- +# Connection pragmas (secure_delete, cell_size_check, synchronous=FULL) +# --------------------------------------------------------------------------- + + +def test_connect_sets_secure_delete_on(tmp_path): + """secure_delete=ON must be active on every new connection.""" + db_path = tmp_path / "kanban.db" + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + with kb.connect(db_path=db_path) as conn: + row = conn.execute("PRAGMA secure_delete").fetchone() + assert row[0] == 1, f"expected secure_delete=1, got {row[0]}" + + +def test_connect_sets_cell_size_check_on(tmp_path): + """cell_size_check=ON must be active on every new connection.""" + db_path = tmp_path / "kanban.db" + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + with kb.connect(db_path=db_path) as conn: + row = conn.execute("PRAGMA cell_size_check").fetchone() + assert row[0] == 1, f"expected cell_size_check=1, got {row[0]}" + + +def test_connect_sets_synchronous_full(tmp_path): + """synchronous must be FULL (=2), not NORMAL (=1).""" + db_path = tmp_path / "kanban.db" + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + with kb.connect(db_path=db_path) as conn: + row = conn.execute("PRAGMA synchronous").fetchone() + assert row[0] == 2, f"expected synchronous=2 (FULL), got {row[0]}" + + +def test_connect_pragmas_applied_on_reconnect(tmp_path): + """All three pragmas must be re-applied on every connect(), not just the first.""" + db_path = tmp_path / "kanban.db" + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + # First connection: write a task and close. + with kb.connect(db_path=db_path) as conn: + kb.create_task(conn, title="reconnect-check") + # Force re-init path by discarding path cache. + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + # Second connection: pragmas must still be applied. + with kb.connect(db_path=db_path) as conn: + assert conn.execute("PRAGMA secure_delete").fetchone()[0] == 1 + assert conn.execute("PRAGMA cell_size_check").fetchone()[0] == 1 + assert conn.execute("PRAGMA synchronous").fetchone()[0] == 2 + + + +def test_pragmas_not_accidentally_disabled_by_migrate_path(tmp_path): + """Migration path must not reset connection pragmas.""" + db_path = tmp_path / "legacy.db" + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + # Initialise with a fresh connect so schema + init run. + with kb.connect(db_path=db_path) as conn: + kb.create_task(conn, title="pre-migration-task") + # Simulate a re-entry through the init/migration path by discarding path cache. + kb._INITIALIZED_PATHS.discard(str(db_path.resolve())) + with kb.connect(db_path=db_path) as conn: + assert conn.execute("PRAGMA secure_delete").fetchone()[0] == 1 + assert conn.execute("PRAGMA cell_size_check").fetchone()[0] == 1 + assert conn.execute("PRAGMA synchronous").fetchone()[0] == 2 + +# write_txn — rollback handler must not mask the original exception +# --------------------------------------------------------------------------- + + +def test_write_txn_preserves_original_exception_when_rollback_fails(kanban_home): + """When a write inside write_txn raises an OperationalError that SQLite + has already auto-rolled-back (e.g. ``disk I/O error``, + ``database is locked``, ``database disk image is malformed``), the + explicit ROLLBACK in ``write_txn.__exit__`` itself raises + ``cannot rollback - no transaction is active``. The original cause + must NOT be masked by the secondary rollback failure — operators rely + on the original cause to diagnose the underlying issue. + """ + + class FailingConnWrapper: + """Delegate to a real connection, simulating an EIO during an INSERT + that SQLite has already auto-rolled-back.""" + + def __init__(self, real): + self._real = real + self._fail_armed = True + + def execute(self, sql, *args, **kwargs): + if ( + self._fail_armed + and sql.lstrip().upper().startswith("INSERT") + and "task_events" in sql.lower() + ): + self._fail_armed = False # one-shot + # Simulate SQLite auto-rolling back the transaction by + # issuing a real ROLLBACK now. After this, BEGIN IMMEDIATE + # is no longer active and an explicit ROLLBACK would error. + try: + self._real.execute("ROLLBACK") + except sqlite3.OperationalError: + pass + raise sqlite3.OperationalError("disk I/O error") + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + with kb.connect() as conn: + wrapper = FailingConnWrapper(conn) + with pytest.raises(sqlite3.OperationalError) as excinfo: + with kb.write_txn(wrapper): + kb._append_event(wrapper, "t_bogus", "promoted", None) + + msg = str(excinfo.value) + assert "disk I/O error" in msg, ( + f"write_txn masked the original exception with rollback failure; " + f"got {msg!r} (expected to contain 'disk I/O error')" + ) + assert "cannot rollback" not in msg, ( + f"write_txn surfaced the rollback failure instead of the original " + f"OperationalError; got {msg!r}" + ) +def test_write_txn_healthy_commit_no_exception(tmp_path): + """Normal commit does not trigger the torn-extend check.""" + from hermes_cli.kanban_db import connect, write_txn, create_task + db = tmp_path / "test.db" + conn = connect(db_path=db) + # Should not raise + with write_txn(conn) as c: + c.execute( + "INSERT INTO tasks (id, title, assignee, status, priority, created_at) " + "VALUES ('t_test01', 'test task', 'tester', 'todo', 0, 1234567890)" + ) + row = conn.execute("SELECT title FROM tasks WHERE id='t_test01'").fetchone() + assert row["title"] == "test task" + conn.close() + + +def test_write_txn_raises_on_truncated_file(tmp_path): + """A mocked smaller file size triggers the torn-extend check.""" + from hermes_cli.kanban_db import connect, write_txn + import hermes_cli.kanban_db as kanban_db_module + db = tmp_path / "test.db" + conn = connect(db_path=db) + # Get actual page size so we can fake a smaller file + page_size = conn.execute("PRAGMA page_size").fetchone()[0] + original_getsize = os.path.getsize + + def fake_getsize(path): + # Return a size that implies at least 1 fewer page than header claims + real_size = original_getsize(path) + return max(0, real_size - page_size) + + with pytest.raises(sqlite3.DatabaseError, match="torn-extend|page count mismatch"): + with unittest.mock.patch("hermes_cli.kanban_db.os.path.getsize", side_effect=fake_getsize): + with write_txn(conn) as c: + c.execute( + "INSERT INTO tasks (id, title, assignee, status, priority, created_at) " + "VALUES ('t_test02', 'test task 2', 'tester', 'todo', 0, 1234567890)" + ) + conn.close() + + +def test_write_txn_post_commit_check_fires_every_call(tmp_path): + """The invariant check runs on every write_txn call.""" + from hermes_cli.kanban_db import connect, write_txn + import hermes_cli.kanban_db as kanban_db_module + db = tmp_path / "test.db" + conn = connect(db_path=db) + call_count = 0 + real_check = kanban_db_module._check_file_length_invariant + + def counting_check(c): + nonlocal call_count + call_count += 1 + real_check(c) + + with unittest.mock.patch.object(kanban_db_module, "_check_file_length_invariant", counting_check): + for i in range(3): + with write_txn(conn) as c: + c.execute( + f"INSERT INTO tasks (id, title, assignee, status, priority, created_at) " + f"VALUES ('t_fire{i:02d}', 'task {i}', 'tester', 'todo', 0, 1234567890)" + ) + assert call_count == 3 + conn.close() + + +def test_connect_sets_wal_autocheckpoint_100(tmp_path): + """connect() sets wal_autocheckpoint to 100.""" + from hermes_cli.kanban_db import connect + db = tmp_path / "test.db" + conn = connect(db_path=db) + val = conn.execute("PRAGMA wal_autocheckpoint").fetchone()[0] + assert val == 100 + conn.close() + + +def test_write_txn_check_reads_correct_header_fields(tmp_path): + """Synthetic DB file with mismatched header page_count triggers the check.""" + import struct + from hermes_cli.kanban_db import connect, write_txn, _check_file_length_invariant + db = tmp_path / "synthetic.db" + conn = connect(db_path=db) + page_size = conn.execute("PRAGMA page_size").fetchone()[0] + conn.close() + # Now corrupt the file: claim N pages but truncate to N-1 pages + with open(db, "rb") as f: + data = bytearray(f.read()) + # Read current page_count from header bytes 28-31 + real_page_count = struct.unpack(">I", data[28:32])[0] + if real_page_count < 2: + # Need at least 2 pages to fake a truncation + pytest.skip("DB too small for synthetic truncation test") + # Truncate to N-1 pages + truncated = bytes(data[: (real_page_count - 1) * page_size]) + with open(db, "wb") as f: + f.write(truncated) + # Now open and check — should raise + # We can't use connect() because _validate_sqlite_header may block; use a raw connection + raw_conn = sqlite3.connect(str(db), isolation_level=None) + with pytest.raises(sqlite3.DatabaseError, match="torn-extend|page count mismatch"): + _check_file_length_invariant(raw_conn) + raw_conn.close() + + +# --------------------------------------------------------------------------- +# reap_worker_zombies() tests +# --------------------------------------------------------------------------- + + +def test_reap_worker_zombies_returns_count(): + """reap_worker_zombies() returns the list of reaped PIDs.""" + from unittest.mock import patch + + fake_pids = [12345, 67890, 11111] + call_count = [0] + + def fake_waitpid(pid, flags): + if call_count[0] < len(fake_pids): + p = fake_pids[call_count[0]] + call_count[0] += 1 + return p, 0 + return 0, 0 + + with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid): + with patch("hermes_cli.kanban_db._record_worker_exit"): + pids = kb.reap_worker_zombies() + assert pids == [12345, 67890, 11111] + + +def test_reap_worker_zombies_noop_on_windows(monkeypatch): + """reap_worker_zombies() returns 0 and never calls os.waitpid on Windows.""" + from unittest.mock import patch + + monkeypatch.setattr("hermes_cli.kanban_db.os.name", "nt") + with patch("hermes_cli.kanban_db.os.waitpid") as mock_waitpid: + result = kb.reap_worker_zombies() + mock_waitpid.assert_not_called() + assert result == [] + + +def test_reap_worker_zombies_noop_no_children(): + """reap_worker_zombies() returns 0 without error when there are no children.""" + from unittest.mock import patch + + with patch("hermes_cli.kanban_db.os.waitpid", side_effect=ChildProcessError): + result = kb.reap_worker_zombies() + assert result == [] + + +def test_reap_worker_zombies_records_exit_status(): + """reap_worker_zombies() calls _record_worker_exit for each reaped pid.""" + from unittest.mock import patch + + calls = [] + call_count = [0] + + def fake_waitpid(pid, flags): + call_count[0] += 1 + if call_count[0] == 1: + return 12345, 0 + return 0, 0 + + with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid): + with patch( + "hermes_cli.kanban_db._record_worker_exit", + side_effect=lambda p, s: calls.append((p, s)), + ): + kb.reap_worker_zombies() + + assert calls == [(12345, 0)] + + +def test_reap_worker_zombies_handles_waitpid_os_error(): + """reap_worker_zombies() does not propagate generic OSError from os.waitpid.""" + from unittest.mock import patch + + with patch("hermes_cli.kanban_db.os.waitpid", side_effect=OSError("test error")): + result = kb.reap_worker_zombies() + assert result == [] + + +def test_zombie_reaper_runs_despite_board_connect_failure(): + """reap_worker_zombies runs even when a board tick raises an error.""" + from unittest.mock import patch + + call_count = [0] + + def fake_waitpid(pid, flags): + call_count[0] += 1 + if call_count[0] <= 2: + return [12345, 67890][call_count[0] - 1], 0 + return 0, 0 + + with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid): + with patch("hermes_cli.kanban_db._record_worker_exit"): + # Simulate a board tick failure before reaping + try: + raise sqlite3.OperationalError("disk I/O error") + except sqlite3.OperationalError: + pass + + # Reaper still runs independently + pids = kb.reap_worker_zombies() + + assert pids == [12345, 67890] + + +def test_zombie_reaper_survives_all_boards_failing(): + """reap_worker_zombies runs each tick regardless of board tick failures.""" + from unittest.mock import patch + + total_reaped = 0 + + def make_fake_waitpid(zombie_pids): + call_count = [0] + + def fake_waitpid(pid, flags): + if call_count[0] < len(zombie_pids): + p = zombie_pids[call_count[0]] + call_count[0] += 1 + return p, 0 + return 0, 0 + + return fake_waitpid + + # 5 ticks, 2 zombies per tick = 10 total + for tick in range(5): + pids = [tick * 100 + 1, tick * 100 + 2] + with patch( + "hermes_cli.kanban_db.os.waitpid", side_effect=make_fake_waitpid(pids) + ): + with patch("hermes_cli.kanban_db._record_worker_exit"): + pids = kb.reap_worker_zombies() + total_reaped += len(pids) + + assert total_reaped == 10 + + +def test_dispatch_once_still_reaps_via_extracted_fn(kanban_home): + """The reaper inside dispatch_once still works after refactor to reap_worker_zombies().""" + from unittest.mock import patch + + call_count = [0] + + def fake_waitpid(pid, flags): + call_count[0] += 1 + if call_count[0] == 1: + return 99999, 0 + return 0, 0 + + with patch("hermes_cli.kanban_db.os.waitpid", side_effect=fake_waitpid): + with patch("hermes_cli.kanban_db._record_worker_exit"): + with patch("hermes_cli.kanban_db.os.name", "posix"): + pids = kb.reap_worker_zombies() + + assert pids == [99999] diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index baabef000d2d..d0815762175c 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -3021,3 +3021,223 @@ def test_v10_to_v11_upgrade_backfills_tool_fields(self, tmp_path): finally: session_db.close() + +# --------------------------------------------------------------------------- +# apply_wal_with_fallback — read-only probe tests +# --------------------------------------------------------------------------- + + +class TestApplyWalProbe: + """Unit tests for the journal_mode probe in apply_wal_with_fallback.""" + + def test_skips_set_pragma_when_already_wal(self, tmp_path): + """Already-WAL connection must not trigger the set-pragma.""" + import sqlite3 + from hermes_state import apply_wal_with_fallback + + class _TracingConn(sqlite3.Connection): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self.executed = [] + + def execute(self, sql, params=()): + self.executed.append(sql) + return super().execute(sql, params) + + db_path = tmp_path / "wal.db" + # Prime the file into WAL mode first. + with sqlite3.connect(str(db_path)) as seed: + seed.execute("PRAGMA journal_mode=WAL") + + conn = _TracingConn(str(db_path)) + try: + result = apply_wal_with_fallback(conn) + finally: + conn.close() + + assert result == "wal" + # Only the probe should have fired; the set-pragma must NOT appear. + assert any("PRAGMA journal_mode" == sql.strip() for sql in conn.executed), ( + "probe PRAGMA should have run" + ) + assert not any("journal_mode=WAL" in sql for sql in conn.executed), ( + "set-pragma must not run when already in WAL mode" + ) + + def test_sets_wal_on_fresh_connection(self, tmp_path): + """Probe sees 'delete', then set-pragma runs and returns 'wal'.""" + import sqlite3 + from hermes_state import apply_wal_with_fallback + + class _TracingConn(sqlite3.Connection): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self.executed = [] + + def execute(self, sql, params=()): + self.executed.append(sql) + return super().execute(sql, params) + + db_path = tmp_path / "fresh.db" + conn = _TracingConn(str(db_path)) + try: + result = apply_wal_with_fallback(conn) + finally: + conn.close() + + assert result == "wal" + assert any("journal_mode=WAL" in sql for sql in conn.executed), ( + "set-pragma must fire on a fresh (non-WAL) connection" + ) + + def test_apply_wal_concurrent_connects_no_eio(self, tmp_path): + """20 threads calling connect() on the same DB must not see disk I/O error.""" + import sys + import threading + import sqlite3 + from hermes_state import apply_wal_with_fallback + + db_path = tmp_path / "concurrent.db" + errors = [] + + def _connect_cycle(): + for _ in range(5): + try: + conn = sqlite3.connect(str(db_path)) + apply_wal_with_fallback(conn) + conn.close() + except sqlite3.OperationalError as exc: + if "disk i/o error" in str(exc).lower(): + errors.append(exc) + + threads = [threading.Thread(target=_connect_cycle) for _ in range(20)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, f"disk I/O errors from concurrent connects: {errors}" + + # Linux-only: no (deleted) WAL/SHM FDs should accumulate. + if sys.platform == "linux": + import os + + fd_dir = f"/proc/{os.getpid()}/fd" + deleted_fds = [] + for fd_name in os.listdir(fd_dir): + try: + target = os.readlink(os.path.join(fd_dir, fd_name)) + if "(deleted)" in target and ( + "wal" in target.lower() or "shm" in target.lower() + ): + deleted_fds.append(target) + except OSError: + pass + assert not deleted_fds, f"stale deleted WAL/SHM FDs: {deleted_fds}" + + def test_fallback_to_delete_still_works(self, tmp_path): + """When set-pragma raises a WAL-incompat error, falls back to DELETE.""" + import sqlite3 + from hermes_state import apply_wal_with_fallback + + class _IncompatConn(sqlite3.Connection): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self._call_count = 0 + + def execute(self, sql, params=()): + self._call_count += 1 + # First call is the read probe; let it return "delete". + # Second call is the set-pragma; raise a WAL-incompat error. + if "journal_mode=WAL" in sql: + raise sqlite3.OperationalError("locking protocol") + return super().execute(sql, params) + + db_path = tmp_path / "incompat.db" + conn = _IncompatConn(str(db_path)) + try: + result = apply_wal_with_fallback(conn, db_label="test.db") + finally: + conn.close() + + assert result == "delete" + + def test_probe_failure_falls_through_to_set_pragma(self, tmp_path): + """When the read probe raises OperationalError, fall through to set-pragma.""" + import sqlite3 + from hermes_state import apply_wal_with_fallback + + class _ProbeFails(sqlite3.Connection): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self._first = True + + def execute(self, sql, params=()): + if self._first and "journal_mode" in sql and "WAL" not in sql: + self._first = False + raise sqlite3.OperationalError("simulated probe failure") + return super().execute(sql, params) + + db_path = tmp_path / "probe_fail.db" + conn = _ProbeFails(str(db_path)) + try: + result = apply_wal_with_fallback(conn) + finally: + conn.close() + + # Despite probe failure, set-pragma must still run and succeed. + assert result == "wal" + + def test_no_downgrade_from_wal_to_delete_on_eio(self, tmp_path): + """OperationalError NOT in _WAL_INCOMPAT_MARKERS must propagate, not downgrade.""" + import sqlite3 + import pytest + from hermes_state import apply_wal_with_fallback + + class _EIOConn(sqlite3.Connection): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self._first = True + + def execute(self, sql, params=()): + # Let the probe succeed (returns "delete" for fresh DB). + if "journal_mode=WAL" in sql: + raise sqlite3.OperationalError("some unexpected hardware failure") + return super().execute(sql, params) + + db_path = tmp_path / "eio.db" + conn = _EIOConn(str(db_path)) + try: + with pytest.raises( + sqlite3.OperationalError, match="some unexpected hardware failure" + ): + apply_wal_with_fallback(conn) + finally: + conn.close() + + def test_returns_wal_not_delete_from_probe(self, tmp_path): + """Early-return only on 'wal'; 'delete' or 'memory' must fall through to set-pragma.""" + import sqlite3 + from hermes_state import apply_wal_with_fallback + + class _TracingConn(sqlite3.Connection): + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self.executed = [] + + def execute(self, sql, params=()): + self.executed.append(sql) + return super().execute(sql, params) + + # Fresh DB is in "delete" mode — probe returns "delete", must NOT early-return. + db_path = tmp_path / "delete_mode.db" + conn = _TracingConn(str(db_path)) + try: + result = apply_wal_with_fallback(conn) + finally: + conn.close() + + assert result == "wal" + assert any("journal_mode=WAL" in sql for sql in conn.executed), ( + "set-pragma must fire when probe returns 'delete'" + ) diff --git a/tests/test_hermes_state_wal_fallback.py b/tests/test_hermes_state_wal_fallback.py index 05cee85012e5..5678e3ff4f11 100644 --- a/tests/test_hermes_state_wal_fallback.py +++ b/tests/test_hermes_state_wal_fallback.py @@ -110,15 +110,79 @@ def test_falls_back_on_not_authorized(self, tmp_path): assert mode == "delete" conn.close() - def test_falls_back_on_disk_io_error(self, tmp_path): - """Flaky network FS → disk I/O error → still fall back.""" + def test_reraises_on_disk_io_error(self, tmp_path): + """Transient EIO from ``PRAGMA journal_mode=WAL`` must NOT silently + downgrade to DELETE. + + Regression for "Bug D": treating transient EIO as a permanent + WAL-incompat marker produced the mixed-journal-mode-across-processes + corruption pattern (process A downgrades to DELETE, sibling + processes successfully set WAL, SQLite corrupts the file because + the two locking protocols are documented as incompatible). EIO is + usually transient (page-cache pressure, lock contention, brief + storage hiccups); the right behavior is to re-raise so the caller + can retry, not to walk the DB into a permanently downgraded state. + """ conn, _ = _open_blocking( tmp_path / "flaky.db", reason="disk I/O error", isolation_level=None ) - mode = apply_wal_with_fallback(conn) - assert mode == "delete" + with pytest.raises(sqlite3.OperationalError, match="disk I/O error"): + apply_wal_with_fallback(conn) conn.close() + def test_does_not_downgrade_when_disk_says_wal(self, tmp_path): + """Refuse to downgrade an already-WAL DB even if the set-pragma path + would have raised a downgrade-eligible marker. + + With the WAL-skip patch, the read-only probe short-circuits before + ``PRAGMA journal_mode=WAL`` ever runs on an already-WAL connection, + so the set-pragma path is unreachable here and ``attempts`` stays 0. + Either outcome (skip-via-probe OR re-raise-on-disk-check) preserves + the property this test guards: we never silently DELETE-downgrade + a WAL-mode file. The on-disk guard remains in place as + belt-and-suspenders for any future code path that bypasses the + probe. + """ + # Prime the file in WAL mode using a normal connection + primer = sqlite3.connect( + str(tmp_path / "already-wal.db"), isolation_level=None + ) + try: + primer.execute("PRAGMA journal_mode=WAL") + primer.execute("CREATE TABLE t (x INTEGER)") + primer.execute("INSERT INTO t VALUES (1)") + assert ( + primer.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal" + ) + finally: + primer.close() + + # New connection whose set-WAL pragma would raise "locking protocol" + # if it were ever called. With the WAL-skip patch the probe sees + # journal_mode=wal and returns early, so set-WAL is never attempted. + conn, attempts = _open_blocking( + tmp_path / "already-wal.db", + reason="locking protocol", + isolation_level=None, + ) + result = apply_wal_with_fallback(conn) + assert result == "wal", ( + "must report wal mode (either skipped via probe or refused downgrade)" + ) + assert attempts[0] == 0, ( + "set-WAL pragma must not run when the on-disk header already says wal" + ) + conn.close() + + # And the file is STILL WAL on disk — nothing got rewritten + check = sqlite3.connect(str(tmp_path / "already-wal.db")) + try: + assert ( + check.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal" + ) + finally: + check.close() + def test_reraises_unrelated_operational_error(self, tmp_path): """Non-WAL-compat errors must NOT be silently swallowed by the fallback.""" conn, _ = _open_blocking(