fix(state): break provably-orphaned repair/FTS-rebuild locks left by dead holders (#100108)

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

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 `<db>.fts_rebuild.lock`, distinct from `<db>.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/<pid>/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

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)