From 894fc35337f3380897fe1a67d42aeb6403b359ef Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 1 Sep 2026 09:49:45 -0700 Subject: [PATCH] fix(state): break provably-orphaned repair/FTS-rebuild locks left by dead holders (#100108) --- hermes_state.py | 45 ++-- hermes_state_common.py | 256 ++++++++++++++++++++-- tests/state/test_fts_rebuild_admission.py | 139 ++++++++++++ 3 files changed, 408 insertions(+), 32 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index 43413ccc19..21807841bc 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -96,6 +96,10 @@ from hermes_state_common import ( # noqa: F401 (re-exported for back-compat) _PREVIEW_MAX_CHARS, _PREVIEW_SCAFFOLD_WINDOW, _PREVIEW_SCAFFOLDED_SQL, + _acquire_db_flock, + _clear_lock_holder_record, + _describe_lock_holder, + _read_lock_holder_record, ) from hermes_state_portability import SessionPortabilityMixin from hermes_state_schema import SessionSchemaMixin @@ -2279,7 +2283,11 @@ def _cross_process_repair_lock(db_path: Path): ``flock`` is the right primitive for this: the kernel drops the lock when the holding process dies, so a crashed repairer cannot leave a stale lock - that wedges every future repair (a pidfile would). The acquire is still + that wedges every future repair (a pidfile would). One exception exists + (issue #100108): a forked child that inherited the lock fd keeps the + flock alive after the acquirer dies, so the acquire path records the + holder's pid + start time and breaks the lock when that holder is + provably dead (see ``_acquire_db_flock``). The acquire is still bounded because a *live* repairer can legitimately sit in ``VACUUM`` for minutes on a large DB, and an unbounded wait would hang the caller's open with no traceback (the failure shape of #36644). @@ -2301,30 +2309,36 @@ def _cross_process_repair_lock(db_path: Path): acquired = False try: - deadline = time.monotonic() + _REPAIR_LOCK_TIMEOUT_SECONDS - while True: - try: - if _IS_WINDOWS: + if _IS_WINDOWS: + deadline = time.monotonic() + _REPAIR_LOCK_TIMEOUT_SECONDS + while True: + try: import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) - else: - import fcntl - - fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - acquired = True - break - except (BlockingIOError, OSError): - if time.monotonic() >= deadline: + acquired = True break - time.sleep(_REPAIR_LOCK_POLL_SECONDS) + except (BlockingIOError, OSError): + if time.monotonic() >= deadline: + break + time.sleep(_REPAIR_LOCK_POLL_SECONDS) + else: + acquired, handle = _acquire_db_flock( + str(lock_path), + handle, + _REPAIR_LOCK_TIMEOUT_SECONDS, + _REPAIR_LOCK_POLL_SECONDS, + "state.db repair lock", + ) if not acquired: + record = None if _IS_WINDOWS else _read_lock_holder_record(handle) logger.warning( "state.db repair lock %s held by another process for more " "than %.0fs — skipping schema surgery in this process to " - "avoid racing the repairer.", + "avoid racing the repairer. Recorded holder: %s.", lock_path, _REPAIR_LOCK_TIMEOUT_SECONDS, + _describe_lock_holder(record), ) yield acquired finally: @@ -2338,6 +2352,7 @@ def _cross_process_repair_lock(db_path: Path): else: import fcntl + _clear_lock_holder_record(handle) fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except OSError: # pragma: no cover - best effort release pass diff --git a/hermes_state_common.py b/hermes_state_common.py index 2d2793bcf6..c35b6135a1 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -7,6 +7,7 @@ hermes_state re-imports every name here for backward compatibility. """ import contextlib +import json import logging import os import sys @@ -897,9 +898,15 @@ END; # Semantics mirror `hermes_state._cross_process_repair_lock` (the schema- # surgery authority): portable (msvcrt on Windows, flock elsewhere), bounded # wait, and FAIL CLOSED — a caller that cannot acquire the lock must NOT -# rebuild. The kernel drops both lock types when the holder dies, so a crashed -# rebuilder cannot wedge future rebuilds. It lives here (not hermes_state) -# because the search/schema mixins cannot import hermes_state (cycle). +# rebuild. The kernel drops both lock types when the holder dies — UNLESS a +# forked child inherited the lock fd (flock rides the open file description, +# which fork() duplicates), in which case the orphaned descriptor holds the +# lock forever (issue #100108). `_acquire_db_flock` therefore records the +# holder's pid + start time under the lock and, when the recorded holder is +# provably dead, breaks the orphaned lock by unlinking and retaking it on a +# fresh inode; indeterminate liveness still defers. It lives here (not +# hermes_state) because the search/schema mixins cannot import hermes_state +# (cycle). # # The lock file is `.fts_rebuild.lock`, distinct from `.repair.lock`: # schema surgery runs on an EXCLUSIVE offline connection and can legitimately @@ -912,6 +919,213 @@ _FTS_REBUILD_LOCK_TIMEOUT_SECONDS = 120.0 _FTS_REBUILD_LOCK_POLL_SECONDS = 0.1 _IS_WINDOWS = sys.platform == "win32" +# Post-break re-acquire budget: once a provably-orphaned lock has been broken +# the fresh inode is uncontended (or contended only by live processes), so a +# short bounded wait suffices — never re-enter the full timeout. +_LOCK_BREAK_REACQUIRE_SECONDS = 5.0 + + +def _proc_start_ticks(pid: int): + """Kernel start time of *pid* in clock ticks, or None when unknowable. + + Field 22 of ``/proc//stat`` (``starttime``) uniquely identifies a + process together with its PID: a recycled PID gets a different start + time. Returns None off Linux or on any read/parse failure — callers must + treat None as "unknowable" and FAIL CLOSED. + """ + try: + with open(f"/proc/{pid}/stat", "rb") as fh: + stat = fh.read() + # comm (field 2) may contain spaces/parens; split after the LAST ')'. + return int(stat.rsplit(b")", 1)[1].split()[19]) + except (OSError, ValueError, IndexError): + return None + + +def _read_lock_holder_record(handle): + """Best-effort parse of the holder metadata JSON in a lock file.""" + try: + handle.seek(0) + raw = handle.read(4096) + except (OSError, ValueError): + return None + if not raw: + return None + try: + record = json.loads(raw.decode("utf-8", "replace")) + except (ValueError, UnicodeDecodeError): + return None + return record if isinstance(record, dict) else None + + +def _write_lock_holder_record(handle) -> None: + """Record this process as the lock holder (advisory, best effort). + + Written under the flock so contenders that time out can tell an + orphaned-fd holder (recorded process dead, flock inherited by a forked + child — issue #100108) from a live wedged holder. + """ + try: + record = { + "pid": os.getpid(), + "start_ticks": _proc_start_ticks(os.getpid()), + "acquired_at": time.time(), + } + handle.seek(0) + handle.truncate() + handle.write(json.dumps(record, sort_keys=True).encode("utf-8")) + handle.flush() + except (OSError, ValueError): + pass + + +def _clear_lock_holder_record(handle) -> None: + """Erase holder metadata before a normal release. + + Guarantees that a surviving record always describes an ABNORMAL exit + (holder died without releasing), which is the only condition under which + a contender may break the lock. + """ + try: + handle.seek(0) + handle.truncate() + handle.flush() + except (OSError, ValueError): + pass + + +def _lock_holder_provably_dead(record) -> bool: + """True ONLY when the recorded holder is provably dead or PID-recycled. + + Any indeterminate state (no record, malformed record, PID owned by + another user, /proc unavailable, start-time unknowable) returns False — + the caller must FAIL CLOSED and defer, never break a possibly-live + holder's lock. + """ + if not isinstance(record, dict): + return False + try: + pid = int(record["pid"]) + except (KeyError, TypeError, ValueError): + return False + if pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return True + except OSError: + # PermissionError et al.: the PID exists (or is unknowable) — closed. + return False + recorded_ticks = record.get("start_ticks") + if recorded_ticks is None: + return False + current_ticks = _proc_start_ticks(pid) + if current_ticks is None: + return False + # Same PID, different kernel start time: the recorded holder is dead and + # its PID was recycled by an unrelated process. + return current_ticks != recorded_ticks + + +def _acquire_db_flock(lock_path, handle, timeout_seconds, poll_seconds, description): + """Bounded POSIX flock acquire with orphaned-holder staleness break. + + Returns ``(acquired, handle)``; *handle* may have been re-opened (the + caller owns closing whichever handle comes back). + + Why breaking exists at all (issue #100108): ``flock`` belongs to the open + file DESCRIPTION, which ``fork()`` duplicates into every child. A holder + that forks (multiprocessing worker, daemonized helper) and then dies + leaves the flock held by a child that will never release it — the + kernel's holder-death release never triggers, and every contender defers + forever. The recorded-holder liveness check distinguishes exactly that + case: the process that ACQUIRED is provably dead (so its critical section + died with it), yet the flock is still held. Only then is the lock file + unlinked and retaken on a fresh inode; the orphan's flock stays on the + old unlinked inode where it blocks nobody. Every successful acquire + verifies its inode still names *lock_path*, so a racer that locked a dead + inode retries instead of running concurrently with the breaker. + Indeterminate liveness always defers (fail closed). + """ + import fcntl + + deadline = time.monotonic() + timeout_seconds + broke_lock = False + while True: + try: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except (BlockingIOError, OSError): + if time.monotonic() < deadline: + time.sleep(poll_seconds) + continue + if broke_lock: + return False, handle + record = _read_lock_holder_record(handle) + if not _lock_holder_provably_dead(record): + return False, handle + logger.warning( + "%s %s is held by an orphaned file descriptor (recorded " + "holder pid %s is dead — a forked child inherited the lock " + "fd); breaking the stale lock and retaking it on a fresh " + "file.", + description, + lock_path, + (record or {}).get("pid"), + ) + try: + os.unlink(lock_path) + handle.close() + handle = open(lock_path, "a+b") + except OSError as exc: + logger.warning( + "Could not break stale %s %s (%s) — deferring.", + description, + lock_path, + exc, + ) + return False, handle + broke_lock = True + deadline = time.monotonic() + _LOCK_BREAK_REACQUIRE_SECONDS + continue + # flock acquired — verify the path still names our inode: a breaker + # may have unlinked/replaced the file while we waited, and a lock on + # a dead inode excludes nobody. + try: + fd_stat = os.fstat(handle.fileno()) + path_stat = os.stat(lock_path) + same_file = ( + fd_stat.st_dev == path_stat.st_dev + and fd_stat.st_ino == path_stat.st_ino + ) + except OSError: + same_file = False + if same_file: + _write_lock_holder_record(handle) + return True, handle + try: + handle.close() + handle = open(lock_path, "a+b") + except OSError: + return False, handle + if time.monotonic() >= deadline: + return False, handle + + +def _describe_lock_holder(record) -> str: + """Human-readable holder identity for deferral warnings.""" + if not isinstance(record, dict) or "pid" not in record: + return "unknown (no holder record; pre-fix writer or non-Hermes)" + pid = record.get("pid") + acquired_at = record.get("acquired_at") + age = "" + try: + if acquired_at is not None: + age = f", acquired {time.time() - float(acquired_at):.0f}s ago" + except (TypeError, ValueError): + pass + return f"pid {pid}{age}" + @contextlib.contextmanager def fts_rebuild_admission(db_path): @@ -945,30 +1159,37 @@ def fts_rebuild_admission(db_path): acquired = False try: - deadline = time.monotonic() + _FTS_REBUILD_LOCK_TIMEOUT_SECONDS - while True: - try: - if _IS_WINDOWS: + if _IS_WINDOWS: + deadline = time.monotonic() + _FTS_REBUILD_LOCK_TIMEOUT_SECONDS + while True: + try: import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) - else: - import fcntl - - fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - acquired = True - break - except (BlockingIOError, OSError): - if time.monotonic() >= deadline: + acquired = True break - time.sleep(_FTS_REBUILD_LOCK_POLL_SECONDS) + except (BlockingIOError, OSError): + if time.monotonic() >= deadline: + break + time.sleep(_FTS_REBUILD_LOCK_POLL_SECONDS) + else: + acquired, handle = _acquire_db_flock( + lock_path, + handle, + _FTS_REBUILD_LOCK_TIMEOUT_SECONDS, + _FTS_REBUILD_LOCK_POLL_SECONDS, + "FTS rebuild lock", + ) if not acquired: + record = None if _IS_WINDOWS else _read_lock_holder_record(handle) logger.warning( "FTS rebuild lock %s held by another process for more than " "%.0fs — deferring this rebuild to avoid racing the holder " - "(the stale-FTS breadcrumb keeps it retryable).", + "(the stale-FTS breadcrumb keeps it retryable). " + "Recorded holder: %s.", lock_path, _FTS_REBUILD_LOCK_TIMEOUT_SECONDS, + _describe_lock_holder(record), ) yield acquired finally: @@ -982,6 +1203,7 @@ def fts_rebuild_admission(db_path): else: import fcntl + _clear_lock_holder_record(handle) fcntl.flock(handle.fileno(), fcntl.LOCK_UN) except OSError: # pragma: no cover - best effort release pass diff --git a/tests/state/test_fts_rebuild_admission.py b/tests/state/test_fts_rebuild_admission.py index b6baf087a0..96f7cd6523 100644 --- a/tests/state/test_fts_rebuild_admission.py +++ b/tests/state/test_fts_rebuild_admission.py @@ -224,3 +224,142 @@ class TestSchemaPathAdmission: # Recovered: breadcrumb cleared, triggers restored. assert _meta_value(db_path, FTS_STALE_KEY) is None assert _base_fts_triggers(db_path) == set(_FTS_TRIGGERS) + + +# --------------------------------------------------------------------------- +# Orphaned-fd staleness break (issue #100108). +# +# flock belongs to the open file DESCRIPTION, which fork() duplicates into +# children. A holder that forks (multiprocessing worker, daemonized helper) +# and then crashes leaves the flock held by the child forever — the kernel's +# holder-death release never fires, and every contender deferred forever +# ("FTS rebuild lock ... held by another process for more than 120s"). +# The fix records the acquirer's pid + start time under the lock; a contender +# that times out breaks the lock ONLY when that recorded holder is provably +# dead, and fails closed on any indeterminate state. +# --------------------------------------------------------------------------- + +_ORPHANING_HOLDER_SCRIPT = """ +import os, sys, time +sys.path.insert(0, {repo!r}) +import hermes_state_common + +admission = hermes_state_common.fts_rebuild_admission({db!r}) +admitted = admission.__enter__() +assert admitted is True +pid = os.fork() +if pid == 0: + # Forked child: shares the lock fd's open file description. Sleep far + # beyond the test, never releasing. + time.sleep(600) + os._exit(0) +print("child", pid, flush=True) +# Crash WITHOUT releasing (no __exit__): simulates the production holder +# dying mid-rebuild after having forked. +os._exit(1) +""" + + +@contextlib.contextmanager +def _orphaned_fork_holder(db_path: Path): + """Real #100108 shape: acquirer records itself, forks, dies.""" + import os + import signal + + script = _ORPHANING_HOLDER_SCRIPT.format( + repo=str(Path(hermes_state_common.__file__).parent), db=str(db_path) + ) + proc = subprocess.Popen( + [sys.executable, "-c", script], stdout=subprocess.PIPE, text=True + ) + line = proc.stdout.readline().strip() + assert line.startswith("child ") + grandchild = int(line.split()[1]) + proc.wait(timeout=10) # the acquirer is now dead; grandchild holds the fd + try: + yield grandchild + finally: + with contextlib.suppress(OSError): + os.kill(grandchild, signal.SIGKILL) + + +class TestOrphanedHolderStalenessBreak: + @pytest.mark.live_system_guard_bypass + def test_rebuild_breaks_lock_of_dead_forker(self, db, fast_timeout): + """The #100108 repro: recorded holder dead, forked child holds the + flock. The contender must break the orphaned lock and rebuild.""" + with _orphaned_fork_holder(db.db_path): + assert db.rebuild_fts() >= 1 + + def test_admission_still_fails_closed_for_live_unrecorded_holder( + self, db, fast_timeout + ): + """A live holder that wrote no record (pre-fix build, non-Hermes + tool) is indeterminate — must defer, never break.""" + with _rebuild_lock_held_by_other_process(db.db_path): + assert db.rebuild_fts() == 0 + + def test_admission_fails_closed_for_live_recorded_holder( + self, db, fast_timeout, monkeypatch + ): + """A record naming a live pid must defer even after timeout.""" + import json + import os + + lock = _lock_file(db.db_path) + with _rebuild_lock_held_by_other_process(db.db_path) as proc: + record = { + "pid": proc.pid, + "start_ticks": hermes_state_common._proc_start_ticks(proc.pid), + "acquired_at": 0, + } + lock.write_bytes(json.dumps(record).encode()) + assert db.rebuild_fts() == 0 + + def test_holder_record_cleared_on_normal_release(self, tmp_path): + lock = tmp_path / "x.db.fts_rebuild.lock" + with hermes_state_common.fts_rebuild_admission(tmp_path / "x.db") as ok: + assert ok is True + assert b"pid" in lock.read_bytes() + assert lock.read_bytes() == b"" + + @pytest.mark.live_system_guard_bypass + def test_repair_lock_breaks_orphaned_holder(self, tmp_path, monkeypatch): + """_cross_process_repair_lock shares the same staleness break.""" + import hermes_state + + monkeypatch.setattr(hermes_state, "_REPAIR_LOCK_TIMEOUT_SECONDS", 0.5) + db_path = tmp_path / "state.db" + db_path.touch() + + script = """ +import os, sys, time +sys.path.insert(0, {repo!r}) +from pathlib import Path +import hermes_state + +lock_cm = hermes_state._cross_process_repair_lock(Path({db!r})) +assert lock_cm.__enter__() is True +pid = os.fork() +if pid == 0: + time.sleep(600) + os._exit(0) +print("child", pid, flush=True) +os._exit(1) +""".format(repo=str(Path(hermes_state_common.__file__).parent), db=str(db_path)) + import os + import signal + + proc = subprocess.Popen( + [sys.executable, "-c", script], stdout=subprocess.PIPE, text=True + ) + grandchild = int(proc.stdout.readline().strip().split()[1]) + proc.wait(timeout=10) + try: + import hermes_state as hs + + with hs._cross_process_repair_lock(db_path) as holding: + assert holding is True + finally: + with contextlib.suppress(OSError): + os.kill(grandchild, signal.SIGKILL)