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.
This commit is contained in:
+6
-2
@@ -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])
|
||||
|
||||
@@ -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
|
||||
|
||||
+56
-44
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user