diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index ca81157decb06..1c3bb9e7e42e1 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1347,6 +1347,7 @@ class Event: # lock (the in-process _INIT_LOCK + idempotent init remain the backstop). _INIT_LOCK_TIMEOUT_SECONDS = 10.0 _INIT_LOCK_POLL_SECONDS = 0.05 +_WRITE_LOCK_LOCAL = threading.local() def _resolve_busy_timeout_ms() -> int: @@ -1539,11 +1540,18 @@ def _dispatch_tick_lock(db_path: Path): acquired = True handle = None try: + if acquired: + held = getattr(_WRITE_LOCK_LOCAL, "held", set()) + held.add(str(db_path.resolve())) + _WRITE_LOCK_LOCAL.held = held yield acquired finally: if handle is not None: try: if acquired: + held = getattr(_WRITE_LOCK_LOCAL, "held", set()) + held.discard(str(db_path.resolve())) + _WRITE_LOCK_LOCAL.held = held if _IS_WINDOWS: import msvcrt @@ -2736,6 +2744,38 @@ def _execute_boundary_with_retry(conn: sqlite3.Connection, sql: str) -> None: time.sleep(random.uniform(_BUSY_RETRY_MIN_S, _BUSY_RETRY_MAX_S)) +@contextlib.contextmanager +def _common_write_lock(conn: sqlite3.Connection): + """Serialize every board write with the dispatcher's cross-process lock.""" + db_path = Path(conn.execute("PRAGMA database_list").fetchone()[2]).resolve() + if str(db_path) in getattr(_WRITE_LOCK_LOCAL, "held", set()): + yield + return + lock_path = db_path.with_name(db_path.name + ".dispatch.lock") + lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = lock_path.open("a+b") + try: + if _IS_WINDOWS: + import msvcrt + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1) + else: + import fcntl + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + yield + finally: + try: + if _IS_WINDOWS: + import msvcrt + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + finally: + handle.close() + + @contextlib.contextmanager def write_txn(conn: sqlite3.Connection): """Context manager for an IMMEDIATE write transaction. @@ -2749,32 +2789,26 @@ def write_txn(conn: sqlite3.Connection): shadow the original exception with a spurious rollback error. """ _assert_not_delegated_child_mutation() - _execute_boundary_with_retry(conn, "BEGIN IMMEDIATE") - try: - yield conn - except Exception: - 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: + with _common_write_lock(conn): + _execute_boundary_with_retry(conn, "BEGIN IMMEDIATE") try: - _execute_boundary_with_retry(conn, "COMMIT") + yield conn except Exception: - # COMMIT exhausted retries with the txn still open; roll back so the - # connection isn't poisoned for the next BEGIN IMMEDIATE. try: conn.execute("ROLLBACK") except sqlite3.OperationalError: pass raise - # 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) + else: + try: + _execute_boundary_with_retry(conn, "COMMIT") + except Exception: + try: + conn.execute("ROLLBACK") + except sqlite3.OperationalError: + pass + raise + _check_file_length_invariant(conn) # --------------------------------------------------------------------------- diff --git a/tests/hermes_cli/test_kanban_dispatch_lock.py b/tests/hermes_cli/test_kanban_dispatch_lock.py index 6acbf2ac216a8..fae1029626114 100644 --- a/tests/hermes_cli/test_kanban_dispatch_lock.py +++ b/tests/hermes_cli/test_kanban_dispatch_lock.py @@ -12,6 +12,8 @@ from __future__ import annotations +import multiprocessing +import time from pathlib import Path import pytest @@ -101,3 +103,91 @@ def test_reentrant_same_path_lock_is_exclusive(conn): assert held_a is True with kb._dispatch_tick_lock(db_path) as held_b: assert held_b is False, "same-board lock must be exclusive" + + +def test_write_txn_shares_the_dispatch_tick_lock_in_process(conn): + """``write_txn`` must take the *same* lock file as ``_dispatch_tick_lock``. + + Regression for the gap described in FASE2_escopo_correcao.md: dashboard/ + CLI writes (which go through ``write_txn``) previously had no cross- + process serialization against the dispatcher tick. This does not + exercise real OS-level blocking (see the multiprocessing test below for + that); it pins the invariant that both code paths resolve to the exact + same lock file for a given board, which is what makes them mutually + exclusive across processes. + """ + db_path = kb.kanban_db_path(board="default") + dispatch_lock_path = db_path.with_name(db_path.name + ".dispatch.lock") + resolved = Path(conn.execute("PRAGMA database_list").fetchone()[2]).resolve() + write_lock_path = resolved.with_name(resolved.name + ".dispatch.lock") + assert write_lock_path == dispatch_lock_path.resolve() + + kb.create_task(conn, title="t", assignee="w") + with kb.write_txn(conn): + conn.execute( + "INSERT INTO task_events (task_id, kind, payload, created_at) " + "VALUES (?, ?, ?, ?)", + ("lock-test", "test", None, 0), + ) + + +def _hold_dispatch_lock_then_signal(db_path_str, ready_evt, release_evt): + """Child process: acquire the board's dispatch lock and hold it.""" + import fcntl + + db_path = Path(db_path_str) + lock_path = db_path.with_name(db_path.name + ".dispatch.lock") + lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = lock_path.open("a+b") + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + ready_evt.set() + release_evt.wait(timeout=10) + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + handle.close() + + +def test_write_txn_blocks_behind_a_cross_process_dispatch_lock_holder(conn): + """A ``write_txn`` must block while another OS process holds the lock.""" + db_path = kb.kanban_db_path(board="default") + kb.create_task(conn, title="t", assignee="w") + + ready_evt = multiprocessing.Event() + release_evt = multiprocessing.Event() + proc = multiprocessing.Process( + target=_hold_dispatch_lock_then_signal, + args=(str(db_path), ready_evt, release_evt), + ) + proc.start() + try: + assert ready_evt.wait(timeout=5), "child never acquired the dispatch lock" + started = time.monotonic() + + import threading + + def _release_after_delay(): + time.sleep(0.3) + release_evt.set() + + releaser = threading.Thread(target=_release_after_delay) + releaser.start() + + with kb.write_txn(conn): + conn.execute( + "UPDATE tasks SET priority = 5 WHERE id = " + "(SELECT id FROM tasks LIMIT 1)" + ) + elapsed = time.monotonic() - started + releaser.join() + assert elapsed >= 0.25, ( + "write_txn returned before the external dispatch-lock holder " + "released it — the cross-process lock is not being honoured" + ) + finally: + release_evt.set() + proc.join(timeout=5) + if proc.is_alive(): + proc.terminate() + proc.join(timeout=5) + + row = conn.execute("PRAGMA integrity_check").fetchone() + assert row[0] == "ok"