fix(state): stop the on-write identity probe cancelling our own POSIX locks (#100368)
This commit is contained in:
Teknium
+71
-4
@@ -4224,13 +4224,80 @@ def divert_session_transcript_jsonl(session_id: str, messages) -> "Optional[Path
|
||||
return path
|
||||
|
||||
|
||||
def _read_sqlite_application_id(db_path: Path) -> "Optional[int]":
|
||||
"""Read application_id from the SQLite header without opening a connection."""
|
||||
# _read_sqlite_application_id runs on EVERY write via _raise_if_db_replaced,
|
||||
# against the LIVE state.db. A bare open()/read()/close() there is the
|
||||
# howtocorrupt §2.2 bug: close() cancels every POSIX advisory lock this
|
||||
# process holds on the file — measured on Linux/SQLite 3.53.1, one probe call
|
||||
# drops the WAL-mode DMS shared lock the writer connection holds on state.db
|
||||
# (see hermes_cli/sqlite_safe_read.py for the module built around this rule).
|
||||
# With the DMS lock gone, a fresh opener in another process can treat this
|
||||
# writer as dead and rerun WAL-index recovery underneath it.
|
||||
#
|
||||
# The probe therefore reads through a per-path fd cached for the life of the
|
||||
# process: opening an fd never cancels locks (only close() does), and
|
||||
# os.pread takes no shared file position. When the path is re-pointed at a
|
||||
# new inode (the very replacement this probe exists to detect), the stale fd
|
||||
# is RETIRED, never closed — closing it would cancel the live connection's
|
||||
# locks on the old file, the exact bug being avoided. Replacement events are
|
||||
# rare and halt writes anyway, so the leak is bounded.
|
||||
_HEADER_PROBE_LOCK = threading.Lock()
|
||||
_HEADER_PROBE_FDS: "dict[str, tuple[int, int, int]]" = {} # key -> (fd, dev, ino)
|
||||
_RETIRED_HEADER_PROBE_FDS: "list[int]" = [] # intentionally never closed
|
||||
|
||||
|
||||
def _pread_db_header(db_path: Path, length: int) -> "Optional[bytes]":
|
||||
"""Lock-safe raw header read of a possibly-live SQLite database.
|
||||
|
||||
POSIX: pread from a cached, never-closed fd (rebound when the path names
|
||||
a new inode). Windows: plain read — advisory-lock cancellation is a
|
||||
POSIX-only hazard and msvcrt locks do not share the failure mode.
|
||||
"""
|
||||
if _IS_WINDOWS:
|
||||
try:
|
||||
with db_path.open("rb") as handle:
|
||||
return handle.read(length)
|
||||
except OSError:
|
||||
return None
|
||||
key = str(db_path)
|
||||
try:
|
||||
with db_path.open("rb") as handle:
|
||||
header = handle.read(_STATE_DB_APPLICATION_ID_OFFSET + 4)
|
||||
st = os.stat(db_path)
|
||||
except OSError:
|
||||
return None
|
||||
with _HEADER_PROBE_LOCK:
|
||||
cached = _HEADER_PROBE_FDS.get(key)
|
||||
if cached is not None and (cached[1], cached[2]) != (st.st_dev, st.st_ino):
|
||||
# Path re-pointed at a new file. Retire (never close) the old fd.
|
||||
_RETIRED_HEADER_PROBE_FDS.append(cached[0])
|
||||
cached = None
|
||||
del _HEADER_PROBE_FDS[key]
|
||||
if cached is None:
|
||||
try:
|
||||
fd = os.open(db_path, os.O_RDONLY)
|
||||
except OSError:
|
||||
return None
|
||||
try:
|
||||
fst = os.fstat(fd)
|
||||
except OSError:
|
||||
_RETIRED_HEADER_PROBE_FDS.append(fd)
|
||||
return None
|
||||
cached = (fd, fst.st_dev, fst.st_ino)
|
||||
_HEADER_PROBE_FDS[key] = cached
|
||||
try:
|
||||
return os.pread(cached[0], length, 0)
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def _read_sqlite_application_id(db_path: Path) -> "Optional[int]":
|
||||
"""Read application_id from the SQLite header without opening a connection.
|
||||
|
||||
Safe against live databases: routed through :func:`_pread_db_header`,
|
||||
which never issues a ``close()`` that would cancel this process's POSIX
|
||||
locks on the file (howtocorrupt §2.2).
|
||||
"""
|
||||
header = _pread_db_header(db_path, _STATE_DB_APPLICATION_ID_OFFSET + 4)
|
||||
if header is None:
|
||||
return None
|
||||
if len(header) < _STATE_DB_APPLICATION_ID_OFFSET + 4:
|
||||
return None
|
||||
if header[:16] != b"SQLite format 3\x00":
|
||||
|
||||
@@ -190,3 +190,110 @@ def test_divert_session_transcript_jsonl_appends(tmp_path, monkeypatch):
|
||||
def _stat_changed(path: Path, recorded) -> bool:
|
||||
st = os.stat(path)
|
||||
return (st.st_dev, st.st_ino) != recorded
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lock safety of the identity probe itself (#100368 / howtocorrupt §2.2).
|
||||
#
|
||||
# _read_sqlite_application_id runs on EVERY write against the LIVE state.db.
|
||||
# Before the _pread_db_header fix it did open("rb")/read/close, and that
|
||||
# close() cancelled every POSIX advisory lock this process held on the file
|
||||
# — including the WAL-mode DMS shared lock of the writer connection. These
|
||||
# tests measure the actual kernel lock table (/proc/locks), so they are
|
||||
# Linux-only; the hazard itself is POSIX-only.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _posix_locks_on(paths):
|
||||
"""Set of (inode, type, mode, start, end) locks held by this pid."""
|
||||
import sys as _sys
|
||||
if not _sys.platform.startswith("linux"):
|
||||
pytest.skip("lock-table probe requires /proc/locks (Linux)")
|
||||
inodes = {}
|
||||
for p in paths:
|
||||
try:
|
||||
inodes[os.stat(p).st_ino] = str(p)
|
||||
except OSError:
|
||||
continue
|
||||
pid = os.getpid()
|
||||
held = set()
|
||||
for line in Path("/proc/locks").read_text().splitlines():
|
||||
parts = line.split()
|
||||
try:
|
||||
lpid = int(parts[4])
|
||||
ino = int(parts[5].split(":")[2])
|
||||
except (IndexError, ValueError):
|
||||
continue
|
||||
if lpid == pid and ino in inodes:
|
||||
held.add((ino, parts[1], parts[3], parts[6], parts[7]))
|
||||
return held
|
||||
|
||||
|
||||
def test_identity_probe_does_not_cancel_live_posix_locks(tmp_path):
|
||||
"""The on-write header probe must not drop the writer's DMS lock."""
|
||||
from hermes_state import _read_sqlite_application_id
|
||||
|
||||
live = tmp_path / "state.db"
|
||||
db = _make_db(live, "probe-sess", "seed")
|
||||
try:
|
||||
sidecars = [live, Path(str(live) + "-shm")]
|
||||
# Hold an open write transaction: that is when the connection holds
|
||||
# POSIX range locks on the main db file, and exactly the state a
|
||||
# concurrent _raise_if_db_replaced probe (another thread, same
|
||||
# process) can destroy.
|
||||
db._conn.execute("BEGIN IMMEDIATE")
|
||||
db._conn.execute(
|
||||
"UPDATE sessions SET source = source WHERE id = 'probe-sess'"
|
||||
)
|
||||
before = _posix_locks_on(sidecars)
|
||||
assert before, "expected in-transaction WAL connection to hold POSIX locks"
|
||||
|
||||
for _ in range(3):
|
||||
_read_sqlite_application_id(live)
|
||||
|
||||
after = _posix_locks_on(sidecars)
|
||||
db._conn.rollback()
|
||||
lost = before - after
|
||||
assert not lost, (
|
||||
"identity probe cancelled POSIX locks held by the live "
|
||||
f"connection (howtocorrupt §2.2): {lost}"
|
||||
)
|
||||
# The decisive check: the WAL DMS shared lock on the MAIN db file
|
||||
# must survive. With the pre-fix open/read/close probe the close()
|
||||
# cancels it (it is already gone by the time the connection has run
|
||||
# its first identity check in __init__), leaving other processes
|
||||
# free to treat this writer as dead and rerun WAL-index recovery
|
||||
# underneath it.
|
||||
db_ino = os.stat(live).st_ino
|
||||
main_db_locks = {lk for lk in after if lk[0] == db_ino}
|
||||
assert main_db_locks, (
|
||||
"live writer connection holds no POSIX lock on state.db itself — "
|
||||
"the WAL DMS lock was cancelled by a raw open/close probe "
|
||||
"(howtocorrupt §2.2)"
|
||||
)
|
||||
# The connection must still be able to commit.
|
||||
db.append_message("probe-sess", role="user", content="post-probe")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_identity_probe_still_detects_replacement_after_fd_cache(tmp_path):
|
||||
"""The cached-fd probe rebinds when the path names a new inode."""
|
||||
from hermes_state import _read_sqlite_application_id
|
||||
|
||||
live = tmp_path / "state.db"
|
||||
other = tmp_path / "other.db"
|
||||
db = _make_db(live, "live-sess", "original")
|
||||
_require_identity(db)
|
||||
first = _read_sqlite_application_id(live) # populates the fd cache
|
||||
db.close()
|
||||
|
||||
alt = _make_db(other, "other-sess", "replacement")
|
||||
alt.close()
|
||||
os.replace(other, live)
|
||||
|
||||
second = _read_sqlite_application_id(live)
|
||||
assert second is not None
|
||||
assert second != first, (
|
||||
"probe kept reading the retired inode instead of rebinding to the "
|
||||
"replacement file"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user