mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
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).
This commit is contained in:
1 parent
ae95f297bf
commit
3be89f1f5b
3 files changed
+77
-5
No files matched your search
@@ -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"),
|
||||
}
|
||||
|
||||
@@ -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 = [
|
||||
|
||||
@@ -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)
|
||||
Reference in new issue
Block a user