From 49828a3fd6de657b444efc185b307ca08cb83d9f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:58:36 -0700 Subject: [PATCH] feat(kanban): periodic WAL checkpoint (TRUNCATE) on the dispatcher tick MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kanban connections set wal_autocheckpoint=100, but SQLite's passive autocheckpoint backs off whenever any reader holds an open snapshot — on a busy multi-process board the -wal file can grow without bound between gateway restarts. After each successful dispatch tick, while still holding the board's single-writer dispatch flock, run PRAGMA wal_checkpoint(TRUNCATE) best-effort at a coarse interval (>=5 min since this process last checkpointed that board; module-level per-path monotonic timestamp, so multi-board dispatchers checkpoint each board on its own clock). Success and busy/locked skips are both logged at DEBUG; a failing checkpoint can never fail the tick. --- hermes_cli/kanban_db.py | 52 ++++++++++- tests/hermes_cli/test_kanban_db_repair.py | 102 ++++++++++++++++++++++ 2 files changed, 153 insertions(+), 1 deletion(-) diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index accd57c8f9ec..a45c385db695 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1505,6 +1505,52 @@ def _dispatch_tick_lock(db_path: Path): handle.close() +# Periodic WAL checkpoint state for the dispatcher tick path. The kanban +# connections run with ``wal_autocheckpoint=100``, but a passive +# autocheckpoint can be starved forever on a busy multi-process board (any +# reader with an open snapshot blocks the WAL reset), letting the -wal file +# grow without bound between gateway restarts. Once per coarse interval the +# dispatcher — the board's single writer during a tick, and holding the +# dispatch flock — issues an explicit ``wal_checkpoint(TRUNCATE)``. +# Best-effort: a busy/locked checkpoint is logged at DEBUG and retried next +# interval. Keyed per resolved DB path so multi-board dispatchers checkpoint +# each board on its own clock. +_WAL_CHECKPOINT_INTERVAL_SECONDS = 300.0 +_LAST_WAL_CHECKPOINT: dict[str, float] = {} +_WAL_CHECKPOINT_LOCK = threading.Lock() + + +def _maybe_checkpoint_wal(conn: sqlite3.Connection, db_path: Path) -> None: + """Run ``PRAGMA wal_checkpoint(TRUNCATE)`` at a coarse interval. + + Called from the dispatcher tick while the board's dispatch lock is + held. No-ops (cheaply) until ``_WAL_CHECKPOINT_INTERVAL_SECONDS`` has + elapsed since this process last checkpointed this board. Never raises: + the checkpoint is pure hygiene and must not fail a dispatch tick. + """ + try: + key = str(db_path.resolve()) + except OSError: + key = str(db_path) + now = time.monotonic() + with _WAL_CHECKPOINT_LOCK: + last = _LAST_WAL_CHECKPOINT.get(key) + if last is not None and (now - last) < _WAL_CHECKPOINT_INTERVAL_SECONDS: + return + # Claim the slot before doing the work so concurrent ticks (other + # threads in this process) don't double-checkpoint on the boundary. + _LAST_WAL_CHECKPOINT[key] = now + try: + row = conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + _log.debug( + "kanban WAL checkpoint (TRUNCATE) on %s -> %s " + "(busy, wal_frames, checkpointed_frames)", + key, tuple(row) if row is not None else None, + ) + except sqlite3.Error as exc: + _log.debug("kanban WAL checkpoint on %s skipped: %s", key, exc) + + def _looks_like_tls_record_at(data: bytes, offset: int) -> bool: """Return True for a TLS record header at ``data[offset:]``.""" if len(data) < offset + 5: @@ -7669,7 +7715,7 @@ def dispatch_once( with _dispatch_tick_lock(db_path) as held: if not held: return DispatchResult(skipped_locked=True) - return _dispatch_once_locked( + result = _dispatch_once_locked( conn, spawn_fn=spawn_fn, ttl_seconds=ttl_seconds, @@ -7682,6 +7728,10 @@ def dispatch_once( default_assignee=default_assignee, max_in_progress_per_profile=max_in_progress_per_profile, ) + # Still under the dispatch lock: opportunistically truncate the WAL + # at a coarse interval so it cannot grow unbounded between restarts. + _maybe_checkpoint_wal(conn, db_path) + return result def _dispatch_once_locked( diff --git a/tests/hermes_cli/test_kanban_db_repair.py b/tests/hermes_cli/test_kanban_db_repair.py index 112d0112642e..2b7147dc4eae 100644 --- a/tests/hermes_cli/test_kanban_db_repair.py +++ b/tests/hermes_cli/test_kanban_db_repair.py @@ -274,3 +274,105 @@ def test_identical_corrupt_bytes_still_reuse_one_backup(tmp_path): backups.add(excinfo.value.backup_path) assert len(backups) == 1 assert len(list(tmp_path.glob("kanban.db.corrupt.*.bak"))) == 1 + + +# --------------------------------------------------------------------------- +# Periodic WAL checkpoint on the dispatcher tick path +# --------------------------------------------------------------------------- + +class _ConnProxy: + """Delegating wrapper so tests can observe/deny wal_checkpoint PRAGMAs. + + ``sqlite3.Connection`` is an immutable C type — its methods cannot be + monkeypatched — so the spy wraps the connection object instead. + """ + + def __init__(self, conn, recorded, fail_checkpoint=False): + self._conn = conn + self._recorded = recorded + self._fail_checkpoint = fail_checkpoint + + def execute(self, sql, *args, **kwargs): + if "wal_checkpoint" in str(sql).lower(): + self._recorded.append(str(sql)) + if self._fail_checkpoint: + raise sqlite3.OperationalError("database is locked") + return self._conn.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._conn, name) + + +def test_dispatch_tick_runs_wal_checkpoint_at_interval(tmp_path, monkeypatch): + """First tick checkpoints; ticks inside the interval don't; after the + interval elapses the next tick checkpoints again.""" + db_path = tmp_path / "kanban.db" + _build_board_db(db_path, tasks=1) + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + # Fresh per-path clock so previous tests can't have claimed the slot. + monkeypatch.setattr(kb, "_LAST_WAL_CHECKPOINT", {}) + + executed: list[str] = [] + conn = kb.connect(db_path=db_path) + proxy = _ConnProxy(conn, executed) + try: + kb.dispatch_once(proxy, spawn_fn=lambda *a, **k: None, dry_run=True) + assert len(executed) == 1, "first tick should checkpoint" + + kb.dispatch_once(proxy, spawn_fn=lambda *a, **k: None, dry_run=True) + kb.dispatch_once(proxy, spawn_fn=lambda *a, **k: None, dry_run=True) + assert len(executed) == 1, "ticks inside the interval must not checkpoint" + + # Age the per-path timestamp past the interval → next tick fires. + key = str(db_path.resolve()) + kb._LAST_WAL_CHECKPOINT[key] -= ( + kb._WAL_CHECKPOINT_INTERVAL_SECONDS + 1.0 + ) + kb.dispatch_once(proxy, spawn_fn=lambda *a, **k: None, dry_run=True) + assert len(executed) == 2, "tick after the interval should checkpoint" + assert all("TRUNCATE" in sql.upper() for sql in executed) + finally: + conn.close() + + +def test_wal_checkpoint_failure_never_fails_the_tick(tmp_path, monkeypatch): + """A busy/erroring checkpoint is best-effort: logged, never raised.""" + db_path = tmp_path / "kanban.db" + _build_board_db(db_path, tasks=1) + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + monkeypatch.setattr(kb, "_LAST_WAL_CHECKPOINT", {}) + + executed: list[str] = [] + conn = kb.connect(db_path=db_path) + proxy = _ConnProxy(conn, executed, fail_checkpoint=True) + try: + result = kb.dispatch_once( + proxy, spawn_fn=lambda *a, **k: None, dry_run=True, + ) + assert not result.skipped_locked + assert executed, "checkpoint was attempted (and failed) this tick" + finally: + conn.close() + + +def test_wal_checkpoint_truncates_wal_file(tmp_path, monkeypatch): + """End-to-end: the checkpoint actually truncates the -wal sidecar.""" + db_path = tmp_path / "kanban.db" + _build_board_db(db_path, tasks=1) + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + monkeypatch.setattr(kb, "_LAST_WAL_CHECKPOINT", {}) + + conn = kb.connect(db_path=db_path) + try: + # Generate WAL frames. + for i in range(30): + kb.create_task(conn, title=f"wal-{i}") + wal = tmp_path / "kanban.db-wal" + assert wal.exists() and wal.stat().st_size > 0 + + kb.dispatch_once(conn, spawn_fn=lambda *a, **k: None, dry_run=True) + assert wal.stat().st_size == 0, ( + "wal_checkpoint(TRUNCATE) should reset the -wal file to 0 bytes" + ) + finally: + conn.close()