From 274fd56dca3025fd511cc4d50335a9139a774c2f Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Mon, 14 Sep 2026 06:17:39 -0700 Subject: [PATCH] fix(state): WAL lock guard follows the handle's lifecycle Three gaps in the #110544 guard, all reported in its review and reproduced: - A writer reopened by _reopen_after_close_locked (teardown/worker race, #94736) came back with no guard: the next stray close + foreign close deleted its WAL again. - _try_wal_checkpoint refreshed the guard outside self._lock; landing after close() it pinned an OFD lock with no connection behind it, so a foreign `PRAGMA journal_mode=DELETE` saw `database is locked` forever. - Refcounts keyed on (fd, inode) treated a recycled fd number as a surviving lock: A+B live, close A, C reuses A's fd, close B left C recorded as guarded while a foreign EXCLUSIVE succeeded. The guard now counts handles per inode, re-locks every matching descriptor on each hold (OFD re-lock is idempotent), and unlocks on the last handle only; the reopen path holds it; the checkpoint refresh runs under self._lock and skips a closed handle. The macOS holder scan folds case so a case-only alias of the sidecar path on APFS still matches. --- hermes_state.py | 8 +- hermes_state_dbfile.py | 6 +- hermes_state_lockguard.py | 100 ++++++++++-------- .../test_wal_lock_guard_lifecycle.py | 73 +++++++++++++ 4 files changed, 139 insertions(+), 48 deletions(-) create mode 100644 tests/hermes_state/test_wal_lock_guard_lifecycle.py diff --git a/hermes_state.py b/hermes_state.py index 391a951106..8070e6ae3a 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -865,6 +865,8 @@ class SessionDB( f"in flight (a session-teardown path called close() before " f"this worker finished — #94736) and the automatic reopen failed: {exc}" ) from exc + if self._wal_active: # a reopened writer is a live generation holder like the first open + self._wal_lock_guard = _lockguard.hold(self.db_path) def _execute_write( self, fn: Callable[[sqlite3.Connection], T], patience_s: Optional[float] = None, @@ -1330,10 +1332,12 @@ class SessionDB( """ if self._quarantine_reason() is not None: return - if self._wal_lock_guard: - _lockguard.hold(self.db_path, self._wal_lock_guard) # a -shm minted after open try: with self._lock: + if self._conn is None: + return # closed underneath the timer: nothing to checkpoint, nothing to re-guard + if self._wal_lock_guard: + _lockguard.hold(self.db_path, self._wal_lock_guard) # a -shm minted after open result = self._conn.execute("PRAGMA wal_checkpoint(PASSIVE)").fetchone() if result and result[1] > 0: logger.debug("WAL checkpoint: %d/%d pages checkpointed", result[2], result[1]) diff --git a/hermes_state_dbfile.py b/hermes_state_dbfile.py index cf38326bec..ac07cccf68 100644 --- a/hermes_state_dbfile.py +++ b/hermes_state_dbfile.py @@ -295,10 +295,12 @@ def _iter_darwin_sidecar_holders(db_path) -> List[Tuple[int, str]]: path for the vnode, while ``os.path.abspath`` does not resolve symlinks -- a textual compare of the two silently misses every sidecar under a symlinked prefix (on macOS ``/var`` itself).""" base = os.path.realpath(os.path.abspath(os.fspath(db_path))) - watched = {os.path.normcase(path): path for path in (base + "-wal", base + "-shm")} + # APFS/HFS+ are case-insensitive by default and libproc reports the pathname as the opener + # spelled it; ``os.path.normcase`` is the identity on darwin, so fold case here. + watched = {path.casefold(): path for path in (base + "-wal", base + "-shm")} holders: List[Tuple[int, str]] = [] for pid, _fd, target, identity in _iter_darwin_fd_targets(): - literal = watched.get(os.path.normcase(target)) + literal = watched.get(target.casefold()) if literal is not None and _identity_is_truly_unlinked(identity, literal): holders.append((pid, target)) return holders diff --git a/hermes_state_lockguard.py b/hermes_state_lockguard.py index f5adf5a49a..de2532b934 100644 --- a/hermes_state_lockguard.py +++ b/hermes_state_lockguard.py @@ -14,6 +14,12 @@ extra to track or retire), and they conflict with a foreign EXCLUSIVE exactly li so the sibling's close-time unlink is refused while a guarded handle is open. The guard is lifted before the handle's own close so a true last close still ends the generation normally. No-op on Windows and on runtimes without OFD locks. + +Ownership model: the guard is a property of the *descriptor*, and a descriptor number is +reusable. Each ``hold()`` therefore locks every matching descriptor unconditionally (an OFD +re-lock on an already-locked description is idempotent) and ``release()`` unlocks only while +another handle in this process still needs the range — tracked by handle count per INODE, not +per fd, so a recycled fd number can never be mistaken for a surviving lock. """ from __future__ import annotations @@ -23,7 +29,7 @@ import os import struct import sys import threading -from typing import Dict, Optional, Tuple +from typing import Dict, Optional, Set, Tuple logger = logging.getLogger("hermes_state") @@ -50,12 +56,12 @@ except ImportError: # Windows _FLOCK_FORMAT = "@qqihh" if sys.platform == "darwin" or "bsd" in sys.platform else "@hhqqi" Identity = Tuple[int, int] -Held = Dict[int, Identity] # fd -> (st_dev, st_ino) it referenced when locked +Held = Dict[Identity, Tuple[int, int]] # inode this handle guards -> its (start, length) range -# Several handles in one process share the same inodes (and see each other's descriptors), so a -# lock on a given (fd, inode) is reference-counted: only the last holder unlocks it. +# Handles per guarded inode in this process. Several SessionDB handles on one file share the +# same descriptors' locks (hold() locks every matching descriptor), so the LAST handle unlocks. _LOCK = threading.Lock() -_REFS: Dict[Tuple[int, Identity], int] = {} +_HANDLES: Dict[Identity, int] = {} def supported() -> bool: @@ -86,9 +92,9 @@ def _identity(path: str) -> Optional[Identity]: return (st.st_dev, st.st_ino) -def _own_fds_for(identities: Dict[Identity, Tuple[int, int]]): - """Yield ``(fd, identity, (start, length))`` for every descriptor of this process on one of - *identities* (SQLite's own connection descriptors; the cached header-probe fd too, harmless).""" +def _own_fds_for(identities: Set[Identity]): + """Yield ``(fd, identity)`` for every descriptor of this process on one of *identities* + (SQLite's own connection descriptors; the cached header-probe fd too, harmless).""" for fd_dir in ("/proc/self/fd", "/dev/fd"): try: names = os.listdir(fd_dir) @@ -103,58 +109,64 @@ def _own_fds_for(identities: Dict[Identity, Tuple[int, int]]): except OSError: continue ident = (st.st_dev, st.st_ino) - rng = identities.get(ident) - if rng is not None: - yield fd, ident, rng + if ident in identities: + yield fd, ident return +def _guard_ranges(db_path) -> Held: + base = os.fspath(db_path) + ranges: Held = {} + for path, rng in ((base, (_SHARED_FIRST, _SHARED_SIZE)), (base + "-shm", (_SHM_DMS_BYTE, 1))): + ident = _identity(path) + if ident is not None: + ranges[ident] = rng + return ranges + + def hold(db_path, held: Optional[Held] = None) -> Held: """Lock the guard ranges on every descriptor this process has open on ``state.db`` and its - ``-shm``; returns the record :func:`release` needs (pass it back to extend an existing one: - a ``-shm`` minted after open, a reopened connection). Safe to repeat.""" + ``-shm``. Returns the record :func:`release` needs; pass it back to extend an existing one + (a ``-shm`` minted after open, a reopened connection). Idempotent per handle: an inode already + in *held* is re-locked (cheap, covers a new descriptor) without a second handle count.""" held = {} if held is None else held if not supported(): return held - base = os.fspath(db_path) - wanted: Dict[Identity, Tuple[int, int]] = {} - for path, rng in ((base, (_SHARED_FIRST, _SHARED_SIZE)), (base + "-shm", (_SHM_DMS_BYTE, 1))): - ident = _identity(path) - if ident is not None: - wanted[ident] = rng + ranges = _guard_ranges(db_path) try: with _LOCK: - for fd, ident, (start, length) in _own_fds_for(wanted): - if held.get(fd) == ident: - continue - if _REFS.get((fd, ident)) or _ofd_lock(fd, _F_RDLCK, start, length): - held[fd] = ident - _REFS[(fd, ident)] = _REFS.get((fd, ident), 0) + 1 + for fd, ident in _own_fds_for(set(ranges)): + start, length = ranges[ident] + if _ofd_lock(fd, _F_RDLCK, start, length) and ident not in held: + held[ident] = ranges[ident] + _HANDLES[ident] = _HANDLES.get(ident, 0) + 1 except OSError: - logger.debug("WAL lock guard unavailable for %s", base, exc_info=True) + logger.debug("WAL lock guard unavailable for %s", os.fspath(db_path), exc_info=True) return held def release(held: Held) -> None: - """Unlock what :func:`hold` locked, on descriptors that still reference the same inode (a - number recycled onto another file is left alone). Call BEFORE the handle's own close so - SQLite's close-time reset sees only real holders: a sibling process's intact locks still - refuse the unlink, and a true last close ends the generation, so a later ``state.db`` - replace never pairs with a stale WAL.""" - if not supported(): + """Drop this handle's claim. The last handle on an inode unlocks the range on every descriptor + still referencing it. Call BEFORE the handle's own close so SQLite's close-time reset sees only + real holders: a sibling process's intact locks still refuse the unlink, and a true last close + ends the generation, so a later ``state.db`` replace never pairs with a stale WAL.""" + if not supported() or not held: return with _LOCK: - for fd, ident in list(held.items()): - remaining = _REFS.get((fd, ident), 1) - 1 + to_unlock: Held = {} + for ident, rng in held.items(): + remaining = _HANDLES.get(ident, 1) - 1 if remaining > 0: - _REFS[(fd, ident)] = remaining - continue - _REFS.pop((fd, ident), None) - try: - st = os.fstat(fd) - if (st.st_dev, st.st_ino) == ident: - _ofd_lock(fd, _F_UNLCK, _SHARED_FIRST, _SHARED_SIZE) - _ofd_lock(fd, _F_UNLCK, _SHM_DMS_BYTE, 1) - except OSError: - pass + _HANDLES[ident] = remaining + else: + _HANDLES.pop(ident, None) + to_unlock[ident] = rng held.clear() + if not to_unlock: + return + try: + for fd, ident in _own_fds_for(set(to_unlock)): + start, length = to_unlock[ident] + _ofd_lock(fd, _F_UNLCK, start, length) + except OSError: + pass diff --git a/tests/hermes_state/test_wal_lock_guard_lifecycle.py b/tests/hermes_state/test_wal_lock_guard_lifecycle.py new file mode 100644 index 0000000000..66e861fb8e --- /dev/null +++ b/tests/hermes_state/test_wal_lock_guard_lifecycle.py @@ -0,0 +1,73 @@ +"""The WAL lock guard tracks the lifecycle of the handle it protects. + +Three shapes the first cut got wrong (review on #110544): a writer reopened after ``close()`` +raced a live caller came back unguarded; the periodic checkpoint's guard refresh could land after +``close()`` and pin an OFD lock with no connection behind it; and refcounts keyed on a reusable +descriptor NUMBER mistook a recycled fd for a surviving lock. Linux-only: OFD locks + ``/proc``. +""" + +import os +import sqlite3 +import subprocess +import sys + +import pytest + +import hermes_state_lockguard as lg +from hermes_state import SessionDB +from tests.hermes_state._wal_generation_harness import make_db, pin_wal, require_wal + +pytestmark = pytest.mark.linux_only + + +def _foreign_exclusive_ok(path: str) -> bool: + """Another process tries the EXCLUSIVE a close-time WAL reset needs; True = nothing guards.""" + code = ( + "import fcntl, os, struct, sys\n" + f"fd = os.open({path!r}, os.O_RDWR)\n" + "lk = struct.pack('@hhqqi', fcntl.F_WRLCK, 0, 0x40000002, 510, 0)\n" + "try:\n fcntl.fcntl(fd, 37, lk); print('EXCLUSIVE_ACQUIRED')\n" + "except BlockingIOError:\n print('REFUSED')\n" + ) + return "EXCLUSIVE_ACQUIRED" in subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=True).stdout + + +def test_reopened_writer_is_guarded_again(tmp_path, monkeypatch): + pin_wal(monkeypatch) + db = make_db(tmp_path / "state.db", "s", "seed") + require_wal(db) + db.close() + db.append_message("s", role="user", content="after close") # #94736 teardown/worker reopen + try: + assert db._conn is not None and db._wal_lock_guard, "reopen returned an unguarded writer" + assert not _foreign_exclusive_ok(str(db.db_path)) + finally: + db.close() + assert _foreign_exclusive_ok(str(db.db_path)) # a true last close lifts the guard + + +def test_guard_never_outlives_the_handle_under_fd_reuse(tmp_path, monkeypatch): + """A+B live -> close A (its fd number is recycled by C) -> close B: C must still be guarded, + and once C closes nothing may be left locked.""" + pin_wal(monkeypatch) + path = tmp_path / "state.db" + a = make_db(path, "s", "seed") + require_wal(a) + b = SessionDB(db_path=path) + a.close() + c = SessionDB(db_path=path) + b.close() + try: + assert not _foreign_exclusive_ok(str(path)), "C recorded as guarded while nothing locks" + for name in ("state.db", "state.db-shm"): # the stray close the guard exists for + os.close(os.open(tmp_path / name, os.O_RDONLY)) + subprocess.run([sys.executable, "-c", + f"import sqlite3; c = sqlite3.connect({str(path)!r}); " + "c.execute('select count(*) from messages').fetchone(); c.close()"], check=True) + c.append_message("s", role="user", content="still writes") + finally: + c.close() + assert _foreign_exclusive_ok(str(path)), "a lock survived the last handle's close" + assert not lg._HANDLES + assert sqlite3.connect(path).execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 2