diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 341686a7c6ea..5b8cd7417e88 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1366,10 +1366,19 @@ def _resolve_busy_timeout_ms() -> int: def _sqlite_connect(path: Path) -> sqlite3.Connection: - """Open a Kanban SQLite connection with consistent lock waiting.""" + """Open a Kanban SQLite connection with consistent lock waiting. + + Uses ``connect_tracked`` so the live-connection registry knows this file + is open: while it is, byte-level probes of the same file are refused, + because an ``open()``/``close()`` would cancel this process's POSIX + advisory locks on the database (see ``hermes_cli.sqlite_safe_read``). + The registration is released automatically when the connection closes. + """ + from hermes_cli.sqlite_safe_read import connect_tracked + busy_timeout_ms = _resolve_busy_timeout_ms() - conn = sqlite3.connect( - str(path), + conn = connect_tracked( + path, isolation_level=None, timeout=busy_timeout_ms / 1000.0, ) @@ -1377,15 +1386,6 @@ def _sqlite_connect(path: Path) -> sqlite3.Connection: # the PRAGMA explicitly so it is observable and survives future wrapper # changes. Parameter binding is not supported for PRAGMA assignments. conn.execute(f"PRAGMA busy_timeout={busy_timeout_ms}") - # Register the live connection so byte-level probes of this file are - # refused while it is open (see hermes_cli.sqlite_safe_read): an - # open()/close() would cancel this process's POSIX locks on the DB. - try: - from hermes_cli.sqlite_safe_read import track_connection - - track_connection(path) - except Exception: - pass return conn diff --git a/hermes_cli/sqlite_safe_read.py b/hermes_cli/sqlite_safe_read.py index 5dfd7c2bc89e..d394ef93d354 100644 --- a/hermes_cli/sqlite_safe_read.py +++ b/hermes_cli/sqlite_safe_read.py @@ -72,6 +72,54 @@ def track_connection(path: Path | str) -> None: _live_connections[key] = _live_connections.get(key, 0) + 1 +class TrackedConnection(sqlite3.Connection): + """A ``sqlite3.Connection`` that untracks its path exactly once on close. + + Counting opens is easy; counting closes reliably is not, because callers + close connections in many places (and some hand them to + ``contextlib.closing``). Subclassing the connection puts the decrement on + the one method every close path must go through, so the live-connection + registry cannot drift upward and permanently disable byte-probes. + + Note ``with conn:`` does NOT close a sqlite3 connection (it only commits or + rolls back), so this hook is not fired spuriously by transaction scopes. + """ + + _hermes_tracked_path: str | None = None + + def close(self) -> None: + path = getattr(self, "_hermes_tracked_path", None) + # Untrack before the actual close: even if close() raises, the + # descriptor is going away and holding the path open would wedge + # every future probe. + if path is not None: + self._hermes_tracked_path = None + untrack_connection(path) + super().close() + + +def connect_tracked(path: Path | str, **kwargs) -> sqlite3.Connection: + """``sqlite3.connect`` that registers the connection for the lifetime of the fd. + + Use for any connection to a database whose file might otherwise be + byte-probed (state.db, kanban.db). The registration is released + automatically on ``close()``. + + A caller (or a test double) that supplies its own ``factory`` wins: we + pass ``factory`` positionally-compatible via kwargs only when it is + absent, and simply skip tracking when the resulting object is not a + :class:`TrackedConnection`. Tracking is an optimisation for the probe + guard, never a correctness requirement for opening the database. + """ + if "factory" not in kwargs: + kwargs["factory"] = TrackedConnection + conn = sqlite3.connect(str(path), **kwargs) + if isinstance(conn, TrackedConnection): + conn._hermes_tracked_path = _key(path) + track_connection(path) + return conn + + def untrack_connection(path: Path | str) -> None: """Record that one connection to *path* has been closed.""" key = _key(path) diff --git a/hermes_state.py b/hermes_state.py index b125b36e87c1..343c2b6858dd 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -588,41 +588,6 @@ def apply_wal_with_fallback( return "delete" -def _apply_delete_for_wal_reset_bug( - conn: sqlite3.Connection, - *, - db_label: str, -) -> str: - """Avoid enabling WAL when the linked SQLite has the WAL-reset bug. - - - Already-WAL on disk: leave WAL alone (no live downgrade) and warn. - - Otherwise: set DELETE and warn. - """ - current = "" - try: - row = conn.execute("PRAGMA journal_mode").fetchone() - if row and row[0] is not None: - current = str(row[0]).strip().lower() - except sqlite3.OperationalError: - current = "" - - if current == "wal": - # Do not TRUNCATE / journal_mode=DELETE while other processes may - # still hold this WAL DB open — same safety rule as the NFS path. - _log_wal_reset_bug_once(db_label, kept_wal=True) - _apply_macos_checkpoint_barrier(conn) - _enforce_macos_synchronous_full(conn) - return "wal" - - try: - conn.execute("PRAGMA journal_mode=DELETE") - except sqlite3.OperationalError: - # Best-effort: DELETE is usually already the default for new files. - pass - _log_wal_reset_bug_once(db_label, kept_wal=False) - return "delete" - - def _log_wal_reset_bug_once( db_label: str, *, @@ -1586,19 +1551,21 @@ class CompressionSessionBusyError(RuntimeError): """A non-owner tried to write while compression owns the session.""" -def _track_live_connection(db_path: Path) -> None: - """Register a live connection so byte-probes of this file are refused. +def _connect_tracked_db(path, **kwargs): + """``sqlite3.connect`` that registers the open fd for lock-safety. - See ``hermes_cli.sqlite_safe_read``: once a connection exists, any - ``open()``/``close()`` on the same file cancels this process's POSIX - advisory locks on it -- including a VACUUM's EXCLUSIVE lock. + While a connection is live, byte-level probes of the same file are + refused: an ``open()``/``close()`` cancels every POSIX advisory lock this + process holds on it -- including a running VACUUM's EXCLUSIVE lock. The + registration is released automatically on ``close()``. Falls back to a + plain connect if the helper is unavailable (scaffold/embed installs). """ try: - from hermes_cli.sqlite_safe_read import track_connection + from hermes_cli.sqlite_safe_read import connect_tracked - track_connection(db_path) + return connect_tracked(path, **kwargs) except Exception: - pass + return sqlite3.connect(str(path), **kwargs) def is_zeroed_state_db( @@ -1828,7 +1795,7 @@ class SessionDB: # must already exist + be initialised (callers guard on # db_path.exists()); a SELECT against an empty file raises and # the caller degrades per-profile. - self._conn = sqlite3.connect( + self._conn = _connect_tracked_db( f"file:{self.db_path}?mode=ro", uri=True, check_same_thread=False, @@ -1836,7 +1803,6 @@ class SessionDB: isolation_level=None, ) self._conn.row_factory = sqlite3.Row - _track_live_connection(self.db_path) return self.db_path.parent.mkdir(parents=True, exist_ok=True) @@ -1871,7 +1837,7 @@ class SessionDB: raise sqlite3.DatabaseError(msg) def _connect_and_init(): - self._conn = sqlite3.connect( + self._conn = _connect_tracked_db( str(self.db_path), check_same_thread=False, # Short timeout — application-level retry with random @@ -1884,7 +1850,6 @@ class SessionDB: isolation_level=None, ) self._conn.row_factory = sqlite3.Row - _track_live_connection(self.db_path) apply_wal_with_fallback(self._conn, db_label="state.db") self._conn.execute("PRAGMA foreign_keys=ON") self._fts_cjk_loaded = load_fts5_cjk_extension(self._conn) @@ -2468,12 +2433,6 @@ class SessionDB: logger.debug("WAL checkpoint (TRUNCATE) at close failed: %s", exc) self._conn.close() self._conn = None - try: - from hermes_cli.sqlite_safe_read import untrack_connection - - untrack_connection(self.db_path) - except Exception: - pass # ── Chunked FTS rebuild engine (v23 opt-in optimize) ── # diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 61eab3d9e08e..21c6d857d08f 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -3276,11 +3276,17 @@ def test_connect_falls_back_to_delete_on_locking_protocol(tmp_path, monkeypatch, return super().execute(sql, *args, **kwargs) def wal_blocking_connect(*args, **kwargs): + # connect_tracked supplies its own factory; the caller's wins, and it + # skips tracking for non-TrackedConnection results. + kwargs.pop("factory", None) return real_connect( *args, factory=_WalBlockingConnection, **kwargs ) - with _patch("hermes_cli.kanban_db.sqlite3.connect", side_effect=wal_blocking_connect): + with _patch( + "hermes_cli.sqlite_safe_read.sqlite3.connect", + side_effect=wal_blocking_connect, + ): with caplog.at_level("WARNING", logger="hermes_state"): conn = kb.connect() diff --git a/tests/hermes_cli/test_session_recovery.py b/tests/hermes_cli/test_session_recovery.py index 39ef1430a0d0..6b82635ab169 100644 --- a/tests/hermes_cli/test_session_recovery.py +++ b/tests/hermes_cli/test_session_recovery.py @@ -126,8 +126,9 @@ def test_recovery_rebuilds_canonical_data_without_opening_source( source_hash = _sha256(source) source_stat = source.stat() - # Exercise the vulnerable-runtime fallback: a fresh recovered DB must be - # born in DELETE mode instead of enabling WAL. + # A recovered DB is born in WAL even on a WAL-reset-vulnerable runtime: + # forcing DELETE there was reverted (DELETE is the mode that corrupts + # under Hermes' concurrent writers -- see hermes_cli.sqlite_safe_read). monkeypatch.setattr( hermes_state, "is_sqlite_wal_reset_vulnerable", @@ -144,7 +145,7 @@ def test_recovery_rebuilds_canonical_data_without_opening_source( assert report["complete"] is True assert report["installed"] is False assert report["source_unchanged"] is True - assert report["verification"]["journal_mode"] == "delete" + assert report["verification"]["journal_mode"] == "wal" assert report["verification"]["integrity_check"] == ["ok"] assert report["verification"]["foreign_key_check"] == [] assert report["verification"]["schema_version"] == SCHEMA_VERSION diff --git a/tests/test_sqlite_lock_safe_inspection.py b/tests/test_sqlite_lock_safe_inspection.py index eaa63a47b55b..792a320a8f15 100644 --- a/tests/test_sqlite_lock_safe_inspection.py +++ b/tests/test_sqlite_lock_safe_inspection.py @@ -164,6 +164,83 @@ def test_preopen_read_refused_while_connection_is_live(tmp_path, clean_registry) assert read_header_bytes_preopen(db, length=16) is not None +def test_tracking_registry_does_not_leak_across_close_paths(tmp_path, clean_registry): + """A drifting counter would silently disable the probe guard forever. + + Opens are easy to count; closes happen in many places. If the registry + ever over-counts, ``has_live_connection`` stays true for a path with no + live connection and every later byte-probe is refused — turning the + safety guard into a permanent outage of zeroed-file / header detection. + """ + import contextlib + + from hermes_cli.sqlite_safe_read import connect_tracked + + db = tmp_path / "state.db" + boot = connect_tracked(db, isolation_level=None) + boot.execute("CREATE TABLE t(v TEXT)") + boot.close() + assert not has_live_connection(db) + + # plain close + connect_tracked(db).close() + assert not has_live_connection(db) + + # contextlib.closing + with contextlib.closing(connect_tracked(db)): + assert has_live_connection(db) + assert not has_live_connection(db) + + # `with conn:` is a TRANSACTION scope, not a close — must stay tracked + conn = connect_tracked(db, isolation_level=None) + with conn: + conn.execute("INSERT INTO t(v) VALUES ('x')") + assert has_live_connection(db), "transaction scope must not untrack" + conn.close() + assert not has_live_connection(db) + + # double close is idempotent (must not under-count into negatives) + dup = connect_tracked(db) + dup.close() + dup.close() + assert not has_live_connection(db) + + # nested lifetimes: still live until the last one closes + first = connect_tracked(db) + second = connect_tracked(db) + first.close() + assert has_live_connection(db) + second.close() + assert not has_live_connection(db) + + # churn must not drift + for _ in range(100): + connect_tracked(db).close() + assert not has_live_connection(db) + + +def test_caller_supplied_connection_factory_still_works(tmp_path, clean_registry): + """A caller's own factory wins; tracking is skipped rather than crashing. + + Tracking is an optimisation for the probe guard, never a precondition for + opening the database — passing a custom factory must not raise. + """ + from hermes_cli.sqlite_safe_read import connect_tracked + + class CustomConnection(sqlite3.Connection): + pass + + db = tmp_path / "state.db" + _make_db(db, "WAL") + + conn = connect_tracked(db, factory=CustomConnection) + try: + assert isinstance(conn, CustomConnection) + assert conn.execute("SELECT COUNT(*) FROM t").fetchone()[0] == 200 + finally: + conn.close() + + def test_page_count_bytes_matches_on_disk_size(tmp_path): """The PRAGMA route reports the same size the header field encodes.""" db = tmp_path / "state.db"