fix(gateway): reclaim own orphaned session leases

This commit is contained in:
fangliquanflq
2026-09-03 00:04:00 +08:00
committed by kshitij
parent 60ee3790e0
commit 2c4474cc69
4 changed files with 194 additions and 32 deletions
+63 -24
View File
@@ -18,7 +18,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterator, Optional
from hermes_constants import get_hermes_home
from hermes_constants import get_default_hermes_root, get_hermes_home
logger = logging.getLogger(__name__)
@@ -757,6 +757,42 @@ def transfer_active_session(
return updated
def _drop_self_orphans(
entries: list[dict[str, Any]], own_live_lease_ids: set[str] | None
) -> list[dict[str, Any]]:
"""Drop this process's leases only when its caller can vouch for owners."""
if own_live_lease_ids is None:
return entries
pid = os.getpid()
return [
entry
for entry in entries
if entry.get("pid") != pid
or str(entry.get("lease_id") or "") in own_live_lease_ids
]
def _release_orphaned_leases_in_home(
registry_home: Path, live_lease_ids: set[str]
) -> int:
state_path = _state_path(registry_home)
if not state_path.exists():
return 0
with _FileLock(_lock_path(registry_home)):
try:
entries = _prune_dead(_read_entries(state_path, strict=True))
except ActiveSessionRegistryError:
logger.warning(
"Active-session registry is unavailable; skipping orphaned-lease sweep"
)
return 0
kept = _drop_self_orphans(entries, live_lease_ids)
dropped = len(entries) - len(kept)
if dropped:
_write_entries(state_path, kept)
return dropped
def release_orphaned_leases(live_lease_ids: set[str]) -> int:
"""Drop this process's registry entries that no live session owns.
@@ -767,30 +803,26 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int:
real, so it drops the rest itself — exact, with no heartbeat write on the
turn path and no staleness threshold to tune.
"""
pid = os.getpid()
state_path = _state_path()
# No registry file yet means no leases have ever been written under this
# home — don't take a lock (or create its file) on the idle-reaper tick.
if not state_path.exists():
return 0
with _FileLock(_lock_path()):
root = get_default_hermes_root()
homes = [root]
profiles_root = root / "profiles"
try:
homes.extend(
profile
for profile in profiles_root.iterdir()
if profile.is_dir() and not profile.name.startswith(".")
)
except OSError:
pass
dropped = 0
for home in homes:
try:
raw_entries = _read_entries(state_path, strict=True)
entries = _prune_dead(raw_entries)
except ActiveSessionRegistryError:
logger.warning(
"Active-session registry is unavailable; skipping orphaned-lease sweep"
dropped += _release_orphaned_leases_in_home(home, live_lease_ids)
except OSError as exc:
logger.debug(
"orphaned-lease sweep failed for %s: %s", home, exc
)
return 0
kept = [
entry
for entry in entries
if entry.get("pid") != pid
or str(entry.get("lease_id") or "") in live_lease_ids
]
dropped = len(entries) - len(kept)
if dropped:
_write_entries(state_path, kept)
return dropped
@@ -812,6 +844,7 @@ def active_session_liveness_guard(
session_id: str,
*,
registry_home: str | Path | None = None,
own_live_lease_ids: set[str] | None = None,
) -> Iterator[bool]:
"""Hold the registry lock while reporting whether ``session_id`` is leased.
@@ -823,6 +856,7 @@ def active_session_liveness_guard(
state_path, lock_path = _lease_paths(registry_home=registry_home)
with _FileLock(lock_path):
entries = _prune_dead(_read_entries(state_path, strict=True), strict=True)
entries = _drop_self_orphans(entries, own_live_lease_ids)
_write_entries(state_path, entries)
yield bool(target) and any(
str(entry.get("session_id") or "") == target for entry in entries
@@ -833,6 +867,8 @@ def active_session_liveness_guard(
def release_active_session_liveness_guard(
lease: ActiveSessionLease,
session_id: str,
*,
own_live_lease_ids: set[str] | None = None,
) -> Iterator[bool]:
"""Remove ``lease`` and hold its registry lock through a lifecycle write.
@@ -842,7 +878,9 @@ def release_active_session_liveness_guard(
"""
if not lease.enabled or lease.released:
with active_session_liveness_guard(
session_id, registry_home=_registry_home_for_lease(lease)
session_id,
registry_home=_registry_home_for_lease(lease),
own_live_lease_ids=own_live_lease_ids,
) as active:
yield active
return
@@ -857,6 +895,7 @@ def release_active_session_liveness_guard(
for entry in entries
if str(entry.get("lease_id") or "") != lease.lease_id
]
kept = _drop_self_orphans(kept, own_live_lease_ids)
if len(kept) != len(entries):
_write_entries(state_path, kept)
lease.released = True
+37
View File
@@ -172,6 +172,43 @@ def test_release_orphaned_leases_reclaims_only_unowned_own_pid_entries(tmp_path,
assert orphan is not None
def test_release_orphaned_leases_sweeps_profile_runtime_registries(
tmp_path, monkeypatch
):
root = tmp_path / "hermes"
profile = root / "profiles" / "worker"
profile.mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(root))
root_lease, root_error = active_sessions.try_acquire_active_session(
session_id="root-orphan", surface="desktop", config={}, registry_home=root
)
profile_lease, profile_error = active_sessions.try_acquire_active_session(
session_id="profile-orphan",
surface="desktop",
config={},
registry_home=profile,
)
assert root_lease is not None and root_error is None
assert profile_lease is not None and profile_error is None
assert active_sessions.release_orphaned_leases(set()) == 2
assert active_sessions.active_session_registry_snapshot(root) == []
assert active_sessions.active_session_registry_snapshot(profile) == []
def test_drop_self_orphans_spares_foreign_and_vouched_leases():
own = os.getpid()
entries = [
{"lease_id": "orphan", "pid": own},
{"lease_id": "live", "pid": own},
{"lease_id": "foreign", "pid": own + 1},
]
assert active_sessions._drop_self_orphans(entries, None) == entries
assert active_sessions._drop_self_orphans(entries, {"live"}) == entries[1:]
def test_release_under_profile_home_override_targets_acquisition_registry(
tmp_path, monkeypatch
):
@@ -270,6 +270,79 @@ def test_automatic_cleanup_preserves_corrupt_registry_without_overwrite(
assert state_path.read_text(encoding="utf-8") == corrupt
def test_own_live_lease_ids_reports_live_owners_and_skips_the_excluded(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _Lease:
def __init__(self, lease_id: str) -> None:
self.lease_id = lease_id
first = _Lease("first")
second = _Lease("second")
monkeypatch.setattr(
server,
"_sessions",
{
"one": {"active_session_lease": first},
"two": {"active_session_lease": second},
"three": {"active_session_lease": None},
},
)
assert server._own_live_lease_ids() == {"first", "second"}
assert server._own_live_lease_ids(exclude=first) == {"second"}
def test_automatic_cleanup_reclaims_own_orphan_lease_not_treated_as_sibling(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
profile_home = tmp_path / "profile-home"
session_id = "own-orphan-session"
owner_lease, message = server._claim_active_session_slot(
session_id,
live_session_id="vanished-runtime",
surface="desktop",
profile_home=profile_home,
)
assert owner_lease is not None and message is None
ended: list[tuple[str, str]] = []
class _FakeDB:
def get_session(self, target: str) -> dict[str, str]:
return {"id": target, "source": "desktop"}
def end_session(self, target: str, reason: str) -> None:
ended.append((target, reason))
@contextlib.contextmanager
def _profile_db(_session: dict):
yield _FakeDB()
monkeypatch.setattr(server, "_sessions", {})
monkeypatch.setattr(server, "_session_db", _profile_db)
monkeypatch.setattr(
server, "_notify_session_boundary", lambda *args, **kwargs: None
)
monkeypatch.setattr(
"tools.async_delegation.interrupt_for_session", lambda *args, **kwargs: None
)
session = {
"active_session_lease": None,
"agent": None,
"history": [],
"history_lock": threading.Lock(),
"profile_home": str(profile_home),
"session_key": session_id,
"slash_worker": None,
"source": "desktop",
}
server._finalize_session(session, end_reason="ws_orphan_reap")
assert ended == [(session_id, "ws_orphan_reap")]
assert active_session_registry_snapshot(registry_home=profile_home) == []
def test_liveness_guard_serializes_cross_process_acquire(tmp_path: Path) -> None:
home = tmp_path / "guard-home"
waiting_file = tmp_path / "child-waiting"
+21 -8
View File
@@ -764,6 +764,17 @@ def _release_active_session_slot(session: dict | None) -> bool:
return False
def _own_live_lease_ids(*, exclude=None) -> set[str]:
"""Snapshot leases still backed by this process's live session records."""
with _sessions_lock:
return {
str(lease.lease_id)
for session in _sessions.values()
if (lease := session.get("active_session_lease")) is not None
and lease is not exclude
}
@contextlib.contextmanager
def _other_runtime_lease_guard(session_id: str, session: dict):
"""Release this runtime and lock sibling ownership through the DB write."""
@@ -784,13 +795,20 @@ def _other_runtime_lease_guard(session_id: str, session: dict):
last_error: Exception | None = None
stack = contextlib.ExitStack()
own_live_lease_ids = _own_live_lease_ids(exclude=lease)
for attempt in range(3):
try:
if lease is not None and getattr(lease, "enabled", False):
guard = release_active_session_liveness_guard(lease, session_id)
guard = release_active_session_liveness_guard(
lease,
session_id,
own_live_lease_ids=own_live_lease_ids,
)
else:
guard = active_session_liveness_guard(
session_id, registry_home=session.get("profile_home")
session_id,
registry_home=session.get("profile_home"),
own_live_lease_ids=own_live_lease_ids,
)
active = stack.enter_context(guard)
break
@@ -1930,12 +1948,7 @@ def _reclaim_orphaned_leases() -> None:
try:
from hermes_cli.active_sessions import release_orphaned_leases
with _sessions_lock:
live = {
lease.lease_id
for session in _sessions.values()
if (lease := session.get("active_session_lease")) is not None
}
live = _own_live_lease_ids()
if dropped := release_orphaned_leases(live):
logger.info("Reclaimed %d orphaned active-session lease(s)", dropped)
except Exception: