From 3be89f1f5bdb0fb743acb200c1434612089e2cef Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 24 Oct 2026 16:41:40 +0530 Subject: [PATCH] fix(process): the spawn ledger's read-modify-write is serialized across processes Review (W2): _append_entry rewrote the machine-wide spawn-ledger.json under a threading.Lock only, so two Hermes processes writing at once each dropped the other's rows. This branch made every spawn_owned a writer, which widened that pre-existing race a lot: three processes registering 15 children each at the same instant kept 18 of 45 rows. Rows that matter can be lost the same way (serve/dashboard rows read by the profile-delete guard and the updater, MCP helper rows read by the orphan reaper). The read-modify-write now also holds an exclusive lock on spawn-ledger.json.lock (the fcntl/msvcrt helper hermes_cli.active_sessions already uses); a lock that cannot be taken falls back to the thread lock, best-effort like the write itself. Profile data cleanup protects the lock file next to the ledger. Test: three processes register 15 live children each at the same instant; all 45 rows are kept. Fails without the file lock (27 of 45 lost). --- hermes_cli/data_cleanup.py | 2 +- hermes_cli/process_identity.py | 37 ++++++++++++++++--- tests/hermes_cli/test_process_identity.py | 43 +++++++++++++++++++++++ 3 files changed, 77 insertions(+), 5 deletions(-) diff --git a/hermes_cli/data_cleanup.py b/hermes_cli/data_cleanup.py index 0aeacb7dcc..7a690d37ae 100644 --- a/hermes_cli/data_cleanup.py +++ b/hermes_cli/data_cleanup.py @@ -28,7 +28,7 @@ def plan_data_removal(home: Path, project: Path, userdata: Path | None = None) - protected = { project.resolve(), base_venv(project).resolve(), store_root(project).resolve(), installs_root().resolve(), machine / "tools", machine / "bin", machine / "cache", - machine / "spawn-ledger.json", home / "gateway.lock", home / ".backup.lock", + machine / "spawn-ledger.json", machine / "spawn-ledger.json.lock", home / "gateway.lock", home / ".backup.lock", home / "runtime" / "active_sessions.lock", home / "cron" / ".tick.lock", store_lock_path(home / "checkpoints"), } diff --git a/hermes_cli/process_identity.py b/hermes_cli/process_identity.py index a7b4a3b5ed..3531633b86 100644 --- a/hermes_cli/process_identity.py +++ b/hermes_cli/process_identity.py @@ -23,7 +23,7 @@ import subprocess import sys import threading import time -from contextlib import contextmanager +from contextlib import contextmanager, suppress from contextvars import ContextVar from dataclasses import asdict, dataclass from pathlib import Path @@ -460,16 +460,45 @@ def _new_entry( ) +@contextmanager +def _ledger_file_lock(lock_path: Path) -> Iterator[None]: + """Hold an exclusive lock on ``lock_path`` for the ledger's read-modify-write. Best-effort like the + write itself: a lock file that cannot be opened or locked leaves the thread lock as the only guard.""" + from hermes_cli.active_sessions import _flock + + try: + from hermes_constants import mkdir_under_hermes_home + mkdir_under_hermes_home(lock_path.parent) + fh = open(lock_path, "a+", encoding="utf-8") + except OSError: + logger.debug("spawn ledger lock unavailable", exc_info=True) + yield + return + with fh: + try: + _flock(fh, lock=True) + except OSError: + logger.debug("spawn ledger lock failed", exc_info=True) + yield + return + try: + yield + finally: + with suppress(OSError): + _flock(fh, lock=False) + + def _append_entry(entry: LedgerEntry) -> bool: """Prune dead entries and append ``entry`` — the ONLY ledger write path. - Serialized under ``_LEDGER_LOCK`` with an atomic tmp+replace; no writer touches the file - outside this function. + Serialized under ``_LEDGER_LOCK`` and, across processes, an exclusive lock on a sidecar file + (every ``spawn_owned`` writes here, so concurrent Hermes processes otherwise dropped each + other's rows), with an atomic tmp+replace; no writer touches the file outside this function. See #91660. """ path = _ledger_path() - with _LEDGER_LOCK: + with _LEDGER_LOCK, _ledger_file_lock(path.with_name(path.name + ".lock")): entries = _read_ledger_or_quarantine(path) or [] # Drop malformed entries, our own stale entry, and provably dead pids. pruned = [ diff --git a/tests/hermes_cli/test_process_identity.py b/tests/hermes_cli/test_process_identity.py index c6f8cf1638..909c2fd235 100644 --- a/tests/hermes_cli/test_process_identity.py +++ b/tests/hermes_cli/test_process_identity.py @@ -448,3 +448,46 @@ def test_reclaim_ignores_unknown_pid_and_never_raises(): assert pi.reap_orphaned_backend_owner(999, 9.0, kill_fn=lambda pid: None) is None with patch.object(pi, "ledger_entries", side_effect=RuntimeError("boom")): assert pi.reap_orphaned_backend_owner(555, 55.0, kill_fn=lambda pid: None) is None + + +_CONCURRENT_WRITER = r""" +import os, subprocess, sys, time +sys.path.insert(0, sys.argv[1]) +from hermes_cli.process_identity import register_child +start_at, n = float(sys.argv[2]), int(sys.argv[3]) +kids = [subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"], stdin=subprocess.DEVNULL) + for _ in range(n)] +while time.time() < start_at: + time.sleep(0.001) +for kid in kids: + register_child(kid.pid, "ledger-race") +print(" ".join(str(k.pid) for k in kids), flush=True) +sys.stdin.read() # keep the children alive until the test has read the ledger +for kid in kids: + kid.kill() +""" + + +def test_concurrent_processes_never_drop_each_others_ledger_rows(tmp_path): + """Every spawn_owned writes the machine-wide ledger; three processes registering at once lost + about two thirds of their rows to each other's read-modify-write before the cross-process lock.""" + import json + import time + from pathlib import Path + + repo = str(Path(__file__).resolve().parents[2]) + env = {**os.environ, "HERMES_HOME": str(tmp_path / "home")} + start_at = time.time() + 2.0 + writers = [subprocess.Popen([sys.executable, "-c", _CONCURRENT_WRITER, repo, str(start_at), "15"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, env=env) + for _ in range(3)] + try: + pids = {int(p) for w in writers for p in w.stdout.readline().split()} + assert len(pids) == 45 + rows = json.loads((tmp_path / "home" / "spawn-ledger.json").read_text(encoding="utf-8")) + kept = {r["pid"] for r in rows if r.get("purpose") == "ledger-race"} + assert kept == pids, f"{len(pids - kept)} of 45 rows lost to a concurrent writer" + finally: + for w in writers: + w.stdin.close() + w.wait(timeout=30)