mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-22 16:25:58 +00:00
fix(gateway): take over live platform-lock token holders once
When --replace misses a cross-HERMES_HOME Telegram token holder, platform connect used to retry forever. Terminate a verified gateway holder once (with the takeover marker) and re-acquire the scoped lock (#65176). Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Jaret Bottoms
Teknium
parent
ad7b4c7d2e
commit
155fdd59f8
6 changed files with 615 additions and 37 deletions
|
|
@ -2448,6 +2448,12 @@ class BasePlatformAdapter(ABC):
|
|||
self._fatal_error_message: Optional[str] = None
|
||||
self._fatal_error_retryable = True
|
||||
self._fatal_error_handler: Optional[Callable[["BasePlatformAdapter"], Awaitable[None] | None]] = None
|
||||
# Cross-HERMES_HOME token takeover is armed by GatewayRunner only for
|
||||
# an adapter's initial connect during an explicit ``gateway run
|
||||
# --replace`` startup. Ordinary starts and every reconnect fail safe
|
||||
# through the existing retryable conflict path.
|
||||
self._platform_lock_takeover_allowed = False
|
||||
self._platform_lock_takeover_attempted = False
|
||||
|
||||
# Track active message handlers per session for interrupt support.
|
||||
# _active_sessions stores the per-session interrupt Event; _session_tasks
|
||||
|
|
@ -2847,8 +2853,18 @@ class BasePlatformAdapter(ABC):
|
|||
await result
|
||||
|
||||
def _acquire_platform_lock(self, scope: str, identity: str, resource_desc: str) -> bool:
|
||||
"""Acquire a scoped lock for this adapter. Returns True on success."""
|
||||
from gateway.status import acquire_scoped_lock
|
||||
"""Acquire a scoped lock for this adapter. Returns True on success.
|
||||
|
||||
A live cross-HERMES_HOME holder may be replaced only when the runner
|
||||
explicitly arms this adapter for its initial ``--replace`` connect.
|
||||
The status module validates PID/start-time/home ownership, places the
|
||||
marker in the target's home, and performs the bounded termination.
|
||||
"""
|
||||
from gateway.status import (
|
||||
acquire_scoped_lock,
|
||||
take_over_scoped_lock_holder,
|
||||
)
|
||||
|
||||
self._platform_lock_scope = scope
|
||||
self._platform_lock_identity = identity
|
||||
acquired, existing = acquire_scoped_lock(
|
||||
|
|
@ -2856,6 +2872,42 @@ class BasePlatformAdapter(ABC):
|
|||
)
|
||||
if acquired:
|
||||
return True
|
||||
|
||||
takeover_allowed = bool(
|
||||
getattr(self, "_platform_lock_takeover_allowed", False)
|
||||
)
|
||||
takeover_attempted = bool(
|
||||
getattr(self, "_platform_lock_takeover_attempted", False)
|
||||
)
|
||||
if takeover_allowed and not takeover_attempted and isinstance(existing, dict):
|
||||
# Consume the authority before doing any I/O: one adapter connect
|
||||
# gets at most one termination attempt, even if lock re-acquire or
|
||||
# later initialization fails.
|
||||
self._platform_lock_takeover_allowed = False
|
||||
self._platform_lock_takeover_attempted = True
|
||||
owner_pid = take_over_scoped_lock_holder(existing)
|
||||
if owner_pid is not None:
|
||||
logger.warning(
|
||||
"[%s] %s was held by gateway PID %d — explicit --replace "
|
||||
"handoff completed",
|
||||
self.name,
|
||||
resource_desc,
|
||||
owner_pid,
|
||||
)
|
||||
acquired, existing = acquire_scoped_lock(
|
||||
scope,
|
||||
identity,
|
||||
metadata={"platform": self.platform.value},
|
||||
)
|
||||
if acquired:
|
||||
logger.info(
|
||||
"[%s] Acquired %s after taking over PID %d",
|
||||
self.name,
|
||||
resource_desc,
|
||||
owner_pid,
|
||||
)
|
||||
return True
|
||||
|
||||
owner_pid = existing.get('pid') if isinstance(existing, dict) else None
|
||||
message = (
|
||||
f'{resource_desc} already in use'
|
||||
|
|
|
|||
|
|
@ -3097,6 +3097,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_loop_heartbeat_task: Optional["asyncio.Task"] = None
|
||||
_gateway_started_at: float = 0.0
|
||||
_shutdown_watchdog_done: Optional["threading.Event"] = None
|
||||
_platform_lock_takeover_on_start: bool = False
|
||||
|
||||
def __init__(self, config: Optional[GatewayConfig] = None):
|
||||
global _gateway_runner_ref
|
||||
|
|
@ -3256,6 +3257,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# with the synthetic resume turns for the same session. The queued
|
||||
# events drain only after all startup resume tasks have finished.
|
||||
self._startup_restore_in_progress = False
|
||||
# Set by start_gateway() only for an explicit ``--replace`` launch.
|
||||
# _connect_initial_adapter_with_timeout scopes it to each adapter's
|
||||
# cold-start connect and removes it before any reconnect can run.
|
||||
self._platform_lock_takeover_on_start = False
|
||||
self._startup_restore_queue: List[MessageEvent] = []
|
||||
self._startup_restore_tasks: List[asyncio.Task] = []
|
||||
# LRU cache of live SessionSources keyed by session_key. Used by
|
||||
|
|
@ -3841,6 +3846,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
f"{platform.value} connect timed out after {timeout:g}s"
|
||||
) from exc
|
||||
|
||||
async def _connect_initial_adapter_with_timeout(self, adapter, platform) -> bool:
|
||||
"""Connect one cold-start adapter with tightly scoped replace intent.
|
||||
|
||||
The capability is visible only while this initial connect is awaited.
|
||||
Reconnects call ``_connect_adapter_with_timeout`` directly and adapters
|
||||
also default to deny, so a later network recovery can never evict a
|
||||
healthy token holder.
|
||||
"""
|
||||
adapter._platform_lock_takeover_allowed = bool(
|
||||
self._platform_lock_takeover_on_start
|
||||
)
|
||||
try:
|
||||
return await self._connect_adapter_with_timeout(adapter, platform)
|
||||
finally:
|
||||
adapter._platform_lock_takeover_allowed = False
|
||||
|
||||
@property
|
||||
def should_exit_cleanly(self) -> bool:
|
||||
return self._exit_cleanly
|
||||
|
|
@ -7697,7 +7718,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
error_message=None,
|
||||
)
|
||||
try:
|
||||
success = await self._connect_adapter_with_timeout(adapter, platform)
|
||||
success = await self._connect_initial_adapter_with_timeout(
|
||||
adapter, platform
|
||||
)
|
||||
if await self._abort_startup_if_shutdown_requested(adapter, platform):
|
||||
return True
|
||||
if success:
|
||||
|
|
@ -7805,6 +7828,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
return True
|
||||
except Exception as e:
|
||||
logger.error("Secondary-profile adapter startup failed: %s", e, exc_info=True)
|
||||
finally:
|
||||
# Startup authority is one phase, not a persistent runner mode.
|
||||
# From this point onward every adapter retry is non-evicting.
|
||||
self._platform_lock_takeover_on_start = False
|
||||
|
||||
# A platform we skipped on the primary for a missing credential was
|
||||
# supposed to be picked up by a secondary profile that owns the token.
|
||||
|
|
@ -9503,7 +9530,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
try:
|
||||
with _profile_runtime_scope(profile_home):
|
||||
success = await self._connect_adapter_with_timeout(adapter, platform)
|
||||
success = await self._connect_initial_adapter_with_timeout(
|
||||
adapter, platform
|
||||
)
|
||||
if success:
|
||||
profile_map[platform] = adapter
|
||||
connected += 1
|
||||
|
|
@ -22637,6 +22666,10 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
|
|||
logging.getLogger().setLevel(_stderr_level)
|
||||
|
||||
runner = GatewayRunner(config)
|
||||
# ``--replace`` is explicit startup authority, not a durable reconnect
|
||||
# policy. GatewayRunner scopes this bit to cold adapter connects and clears
|
||||
# it before the background reconnect watcher starts.
|
||||
runner._platform_lock_takeover_on_start = bool(replace)
|
||||
|
||||
# Track whether an unexpected signal initiated the shutdown. When an
|
||||
# unexpected SIGTERM kills the gateway, we exit non-zero so service
|
||||
|
|
|
|||
|
|
@ -142,6 +142,18 @@ def _get_process_hermes_home() -> Path:
|
|||
return _get_platform_default_hermes_home()
|
||||
|
||||
|
||||
def _canonical_hermes_home(path: Path | str) -> Path:
|
||||
"""Return a stable absolute HERMES_HOME path for persisted identity data."""
|
||||
return Path(path).expanduser().resolve(strict=False)
|
||||
|
||||
|
||||
def _same_hermes_home(left: Path | str, right: Path | str) -> bool:
|
||||
"""Compare HERMES_HOME paths with the host platform's case semantics."""
|
||||
return os.path.normcase(str(_canonical_hermes_home(left))) == os.path.normcase(
|
||||
str(_canonical_hermes_home(right))
|
||||
)
|
||||
|
||||
|
||||
def _get_pid_path() -> Path:
|
||||
"""Return the path to the gateway PID file, respecting HERMES_HOME."""
|
||||
home = _get_process_hermes_home()
|
||||
|
|
@ -499,6 +511,11 @@ def _build_pid_record() -> dict:
|
|||
"kind": _GATEWAY_KIND,
|
||||
"argv": list(sys.argv),
|
||||
"start_time": _get_process_start_time(os.getpid()),
|
||||
# Scoped credential locks are machine-global rather than
|
||||
# HERMES_HOME-local. Persist the owning gateway's process home so an
|
||||
# explicit cross-profile --replace can place its planned-takeover
|
||||
# marker where the target process will actually read it.
|
||||
"hermes_home": str(_canonical_hermes_home(_get_process_hermes_home())),
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -1356,10 +1373,14 @@ _PLANNED_STOP_MARKER_FILENAME = ".gateway-planned-stop.json"
|
|||
_PLANNED_STOP_MARKER_TTL_S = 60
|
||||
|
||||
|
||||
def _get_takeover_marker_path() -> Path:
|
||||
"""Return the path to the --replace takeover marker file."""
|
||||
home = _get_process_hermes_home()
|
||||
return home / _TAKEOVER_MARKER_FILENAME
|
||||
def _get_takeover_marker_path(hermes_home: Optional[Path] = None) -> Path:
|
||||
"""Return the path to the --replace takeover marker file.
|
||||
|
||||
``hermes_home`` is supplied only for a verified cross-home handoff. The
|
||||
target process always consumes the marker from its own process-level home.
|
||||
"""
|
||||
home = hermes_home or _get_process_hermes_home()
|
||||
return _canonical_hermes_home(home) / _TAKEOVER_MARKER_FILENAME
|
||||
|
||||
|
||||
def _get_planned_stop_marker_path() -> Path:
|
||||
|
|
@ -1406,21 +1427,24 @@ def _consume_pid_marker_for_self(
|
|||
pass
|
||||
return False
|
||||
|
||||
# Cross-profile guard (#29092): reject markers written by a gateway
|
||||
# running under a different HERMES_HOME. When two profile gateway
|
||||
# services share the same default ~/.hermes (HERMES_HOME not set
|
||||
# distinctly), the marker path resolves to the same file for both. A
|
||||
# --replace from profile B could land in profile A's marker, match on
|
||||
# PID + start_time by coincidence of a shared PID namespace, and make
|
||||
# profile A exit 0 — only to be revived by systemd Restart=always,
|
||||
# which then races the replacer again, flapping indefinitely. The
|
||||
# field is absent in markers written by older Hermes versions; treat
|
||||
# absent as "same home" so old markers and single-profile setups are
|
||||
# unaffected. Leave a mismatched marker in place so the correct
|
||||
# profile can still consume it.
|
||||
replacer_home = record.get("replacer_hermes_home")
|
||||
if replacer_home is not None and replacer_home != str(_get_process_hermes_home()):
|
||||
return False
|
||||
# Cross-profile guard (#29092): new markers explicitly name the verified
|
||||
# TARGET home. That permits a deliberate cross-HERMES_HOME --replace while
|
||||
# ensuring a marker accidentally written into another profile's directory
|
||||
# is ignored. Legacy markers have no target field, so retain the original
|
||||
# same-replacer-home rule for backwards compatibility.
|
||||
our_home = _get_process_hermes_home()
|
||||
target_home = record.get("target_hermes_home")
|
||||
if target_home is not None:
|
||||
if not isinstance(target_home, str) or not _same_hermes_home(
|
||||
target_home, our_home
|
||||
):
|
||||
return False
|
||||
else:
|
||||
replacer_home = record.get("replacer_hermes_home")
|
||||
if replacer_home is not None and not _same_hermes_home(
|
||||
replacer_home, our_home
|
||||
):
|
||||
return False
|
||||
|
||||
our_pid = os.getpid()
|
||||
our_start_time = _get_process_start_time(our_pid)
|
||||
|
|
@ -1451,27 +1475,45 @@ def _consume_pid_marker_for_self(
|
|||
return matches
|
||||
|
||||
|
||||
def write_takeover_marker(target_pid: int) -> bool:
|
||||
def write_takeover_marker(
|
||||
target_pid: int,
|
||||
*,
|
||||
target_home: Optional[Path] = None,
|
||||
target_start_time: Any = _UNSET,
|
||||
) -> bool:
|
||||
"""Record that ``target_pid`` is being replaced by the current process.
|
||||
|
||||
Captures the target's ``start_time`` so that PID reuse after the
|
||||
target exits cannot later match the marker. Also records the
|
||||
replacer's PID and a UTC timestamp for TTL-based staleness checks.
|
||||
|
||||
Returns True on successful write, False on any failure. The caller
|
||||
should proceed with the SIGTERM even if the write fails (the marker
|
||||
is a best-effort signal, not a correctness requirement).
|
||||
A verified scoped-lock handoff supplies ``target_home`` and the already
|
||||
validated ``target_start_time`` so the marker is written into the target
|
||||
gateway's HERMES_HOME rather than the replacer's. Same-home callers omit
|
||||
both arguments and preserve the historical behavior.
|
||||
|
||||
Returns True on successful write, False on any failure. Historical
|
||||
same-home callers may treat the marker as best effort. Cross-home callers
|
||||
must fail closed because the target's supervisor could otherwise revive it
|
||||
without recognizing the handoff.
|
||||
"""
|
||||
try:
|
||||
target_start_time = _get_process_start_time(target_pid)
|
||||
marker_home = _canonical_hermes_home(
|
||||
target_home or _get_process_hermes_home()
|
||||
)
|
||||
if target_start_time is _UNSET:
|
||||
target_start_time = _get_process_start_time(target_pid)
|
||||
record = {
|
||||
"target_pid": target_pid,
|
||||
"target_start_time": target_start_time,
|
||||
"target_hermes_home": str(marker_home),
|
||||
"replacer_pid": os.getpid(),
|
||||
"replacer_hermes_home": str(_get_process_hermes_home()),
|
||||
"replacer_hermes_home": str(
|
||||
_canonical_hermes_home(_get_process_hermes_home())
|
||||
),
|
||||
"written_at": _utc_now_iso(),
|
||||
}
|
||||
_write_json_file(_get_takeover_marker_path(), record)
|
||||
_write_json_file(_get_takeover_marker_path(marker_home), record)
|
||||
return True
|
||||
except (OSError, PermissionError):
|
||||
return False
|
||||
|
|
@ -1496,14 +1538,182 @@ def consume_takeover_marker_for_self() -> bool:
|
|||
)
|
||||
|
||||
|
||||
def clear_takeover_marker() -> None:
|
||||
def clear_takeover_marker(target_home: Optional[Path] = None) -> None:
|
||||
"""Remove the takeover marker unconditionally. Safe to call repeatedly."""
|
||||
try:
|
||||
_get_takeover_marker_path().unlink(missing_ok=True)
|
||||
_get_takeover_marker_path(target_home).unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _validated_scoped_lock_gateway_owner(
|
||||
record: dict[str, Any],
|
||||
) -> Optional[tuple[int, int, Path]]:
|
||||
"""Resolve a live scoped-lock owner to a verified gateway identity.
|
||||
|
||||
A machine-global scoped-lock file is only a claim; it is not sufficient
|
||||
authority to terminate a process or choose a marker destination. Require
|
||||
the lock record, the target HERMES_HOME's gateway PID record, and the live
|
||||
OS process to agree on PID, start-time fingerprint, gateway identity, and
|
||||
process home. Missing legacy metadata fails closed and leaves the normal
|
||||
retryable lock-conflict path in charge.
|
||||
"""
|
||||
if not isinstance(record, dict) or not _record_looks_like_gateway(record):
|
||||
return None
|
||||
|
||||
try:
|
||||
owner_pid = int(record["pid"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
return None
|
||||
if owner_pid <= 0 or owner_pid == os.getpid():
|
||||
return None
|
||||
|
||||
owner_start_time = record.get("start_time")
|
||||
if not isinstance(owner_start_time, int) or isinstance(owner_start_time, bool):
|
||||
return None
|
||||
|
||||
raw_home = record.get("hermes_home")
|
||||
if not isinstance(raw_home, str) or not raw_home.strip():
|
||||
return None
|
||||
if not Path(raw_home).expanduser().is_absolute():
|
||||
return None
|
||||
target_home = _canonical_hermes_home(raw_home)
|
||||
|
||||
if not _pid_exists(owner_pid):
|
||||
return None
|
||||
live_start_time = _get_process_start_time(owner_pid)
|
||||
if live_start_time is None or live_start_time != owner_start_time:
|
||||
return None
|
||||
|
||||
live_cmdline = _read_process_cmdline(owner_pid)
|
||||
if live_cmdline is not None and not looks_like_gateway_runtime_command_line(
|
||||
live_cmdline
|
||||
):
|
||||
return None
|
||||
|
||||
pid_record = _read_json_file(target_home / "gateway.pid")
|
||||
if not isinstance(pid_record, dict) or not _record_looks_like_gateway(pid_record):
|
||||
return None
|
||||
try:
|
||||
pid_record_pid = int(pid_record["pid"])
|
||||
except (KeyError, TypeError, ValueError):
|
||||
return None
|
||||
if pid_record_pid != owner_pid or pid_record.get("start_time") != owner_start_time:
|
||||
return None
|
||||
|
||||
pid_record_home = pid_record.get("hermes_home")
|
||||
if not isinstance(pid_record_home, str) or not _same_hermes_home(
|
||||
pid_record_home, target_home
|
||||
):
|
||||
return None
|
||||
|
||||
return owner_pid, owner_start_time, target_home
|
||||
|
||||
|
||||
def _scoped_lock_owner_state(owner_pid: int, owner_start_time: int) -> str:
|
||||
"""Return ``same``, ``exited``, or ``unknown`` for a validated owner."""
|
||||
if not _pid_exists(owner_pid):
|
||||
return "exited"
|
||||
live_start_time = _get_process_start_time(owner_pid)
|
||||
if live_start_time is None:
|
||||
return "unknown"
|
||||
if live_start_time != owner_start_time:
|
||||
# The original owner exited and the OS recycled its PID. Never signal
|
||||
# the replacement process.
|
||||
return "exited"
|
||||
return "same"
|
||||
|
||||
|
||||
def _wait_for_scoped_lock_owner_exit(
|
||||
owner_pid: int,
|
||||
owner_start_time: int,
|
||||
*,
|
||||
attempts: int,
|
||||
delay: float,
|
||||
) -> tuple[bool, bool]:
|
||||
"""Return ``(exited, safe_to_force)`` after bounded identity-aware waits."""
|
||||
for _ in range(max(0, attempts)):
|
||||
state = _scoped_lock_owner_state(owner_pid, owner_start_time)
|
||||
if state == "exited":
|
||||
return True, False
|
||||
if state == "unknown":
|
||||
return False, False
|
||||
time.sleep(max(0.0, delay))
|
||||
return False, _scoped_lock_owner_state(owner_pid, owner_start_time) == "same"
|
||||
|
||||
|
||||
def take_over_scoped_lock_holder(
|
||||
record: dict[str, Any],
|
||||
*,
|
||||
graceful_attempts: int = 20,
|
||||
force_attempts: int = 20,
|
||||
) -> Optional[int]:
|
||||
"""Terminate one verified scoped-lock holder for explicit ``--replace``.
|
||||
|
||||
Returns the original owner PID only after that exact PID/start-time identity
|
||||
has exited (or ``terminate_pid`` reports it already gone). Validation or
|
||||
marker-write failure returns ``None`` without signalling anything. This is
|
||||
deliberately stricter than the same-home PID-file replacement path: a
|
||||
cross-home handoff must place a consumable marker in the target's home or a
|
||||
service supervisor could revive the target and start a flap loop.
|
||||
"""
|
||||
owner = _validated_scoped_lock_gateway_owner(record)
|
||||
if owner is None:
|
||||
return None
|
||||
owner_pid, owner_start_time, target_home = owner
|
||||
|
||||
if not write_takeover_marker(
|
||||
owner_pid,
|
||||
target_home=target_home,
|
||||
target_start_time=owner_start_time,
|
||||
):
|
||||
return None
|
||||
|
||||
try:
|
||||
state = _scoped_lock_owner_state(owner_pid, owner_start_time)
|
||||
if state == "exited":
|
||||
return owner_pid
|
||||
if state != "same":
|
||||
return None
|
||||
|
||||
try:
|
||||
terminate_pid(owner_pid, force=False)
|
||||
except ProcessLookupError:
|
||||
return owner_pid
|
||||
except (PermissionError, OSError):
|
||||
return None
|
||||
|
||||
exited, safe_to_force = _wait_for_scoped_lock_owner_exit(
|
||||
owner_pid,
|
||||
owner_start_time,
|
||||
attempts=graceful_attempts,
|
||||
delay=0.5,
|
||||
)
|
||||
if exited:
|
||||
return owner_pid
|
||||
if not safe_to_force:
|
||||
return None
|
||||
|
||||
try:
|
||||
terminate_pid(owner_pid, force=True)
|
||||
except ProcessLookupError:
|
||||
return owner_pid
|
||||
except (PermissionError, OSError):
|
||||
return None
|
||||
|
||||
exited, _ = _wait_for_scoped_lock_owner_exit(
|
||||
owner_pid,
|
||||
owner_start_time,
|
||||
attempts=force_attempts,
|
||||
delay=0.25,
|
||||
)
|
||||
return owner_pid if exited else None
|
||||
finally:
|
||||
# The target normally consumes its marker from the signal handler.
|
||||
# Clean up any remainder after an already-gone/forced/failed handoff.
|
||||
clear_takeover_marker(target_home)
|
||||
|
||||
|
||||
def write_planned_stop_marker(target_pid: int) -> bool:
|
||||
"""Record that ``target_pid`` is being stopped intentionally.
|
||||
|
||||
|
|
|
|||
|
|
@ -156,6 +156,7 @@ async def test_start_gateway_verbosity_imports_redacting_formatter(monkeypatch,
|
|||
self.adapters = {}
|
||||
|
||||
async def start(self):
|
||||
assert self._platform_lock_takeover_on_start is False
|
||||
return True
|
||||
|
||||
async def stop(self):
|
||||
|
|
@ -191,6 +192,7 @@ async def test_start_gateway_replace_force_uses_terminate_pid(monkeypatch, tmp_p
|
|||
self.adapters = {}
|
||||
|
||||
async def start(self):
|
||||
assert self._platform_lock_takeover_on_start is True
|
||||
return True
|
||||
|
||||
async def stop(self):
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
"""Regression test for #54167 — stale platform lock must be retryable.
|
||||
"""Regression tests for platform-lock acquire behavior.
|
||||
|
||||
#54167 — stale platform lock must be retryable.
|
||||
When a gateway process is killed (SIGKILL, crash) during Telegram
|
||||
initialization, the scoped lock file survives. On next startup,
|
||||
``acquire_scoped_lock()`` detects the stale lock and deletes it, but may
|
||||
|
|
@ -11,10 +12,11 @@ process grab the lock first).
|
|||
so the reconnect watcher can retry after a delay — not permanently kill
|
||||
the platform.
|
||||
|
||||
Contract asserted here
|
||||
----------------------
|
||||
``_set_fatal_error`` is called with ``retryable=True`` when lock
|
||||
acquisition fails, regardless of the reason.
|
||||
#65176 — a live gateway token conflict may attempt one-shot takeover only
|
||||
during the initial connect of an explicit ``gateway run --replace`` startup.
|
||||
``gateway run --replace`` only kills same-HERMES_HOME PID-file holders.
|
||||
A normal start or reconnect must retain the retryable conflict behavior and
|
||||
must never evict the active holder.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict
|
||||
|
|
@ -23,6 +25,7 @@ from unittest.mock import MagicMock, patch
|
|||
import pytest
|
||||
|
||||
from gateway.platforms.base import BasePlatformAdapter
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
|
||||
class _StubAdapter(BasePlatformAdapter):
|
||||
|
|
@ -54,6 +57,8 @@ def adapter():
|
|||
obj._fatal_error_handler = None
|
||||
obj._platform_lock_scope = None
|
||||
obj._platform_lock_identity = None
|
||||
obj._platform_lock_takeover_allowed = False
|
||||
obj._platform_lock_takeover_attempted = False
|
||||
obj._status_write_logged = None
|
||||
return obj
|
||||
|
||||
|
|
@ -71,3 +76,130 @@ def test_stale_lock_failure_is_retryable(adapter):
|
|||
assert result is False
|
||||
assert adapter._fatal_error_retryable is True
|
||||
assert adapter._fatal_error_code == "telegram-bot-token_lock"
|
||||
|
||||
|
||||
def test_explicit_replace_takeover_reacquires_lock_once(adapter):
|
||||
"""Initial explicit --replace may hand off and re-acquire once (#65176)."""
|
||||
existing = {
|
||||
"pid": 4242,
|
||||
"kind": "hermes-gateway",
|
||||
"argv": ["hermes", "gateway", "run"],
|
||||
"start_time": 123,
|
||||
}
|
||||
acquire = MagicMock(side_effect=[(False, existing), (True, None)])
|
||||
adapter._platform_lock_takeover_allowed = True
|
||||
|
||||
with patch("gateway.status.acquire_scoped_lock", acquire), patch(
|
||||
"gateway.status.take_over_scoped_lock_holder",
|
||||
return_value=4242,
|
||||
) as takeover, patch.object(
|
||||
adapter, "_write_runtime_status_safe"
|
||||
):
|
||||
result = adapter._acquire_platform_lock(
|
||||
"telegram-bot-token", "test-token", "Telegram bot token"
|
||||
)
|
||||
|
||||
assert result is True
|
||||
assert adapter._platform_lock_takeover_allowed is False
|
||||
assert adapter._platform_lock_takeover_attempted is True
|
||||
takeover.assert_called_once_with(existing)
|
||||
assert acquire.call_count == 2
|
||||
|
||||
|
||||
def test_normal_connect_conflict_never_attempts_takeover(adapter):
|
||||
"""A normal start/reconnect cannot evict the current token holder."""
|
||||
existing = {
|
||||
"pid": 5555,
|
||||
"kind": "hermes-gateway",
|
||||
"argv": ["hermes", "gateway", "run"],
|
||||
"start_time": 123,
|
||||
}
|
||||
with patch(
|
||||
"gateway.status.acquire_scoped_lock",
|
||||
return_value=(False, existing),
|
||||
), patch(
|
||||
"gateway.status.take_over_scoped_lock_holder",
|
||||
) as takeover, patch.object(
|
||||
adapter, "_write_runtime_status_safe"
|
||||
):
|
||||
result = adapter._acquire_platform_lock(
|
||||
"telegram-bot-token", "test-token", "Telegram bot token"
|
||||
)
|
||||
|
||||
assert result is False
|
||||
takeover.assert_not_called()
|
||||
assert adapter._platform_lock_takeover_attempted is False
|
||||
assert adapter._fatal_error_retryable is True
|
||||
|
||||
|
||||
def test_failed_explicit_takeover_consumes_authority(adapter):
|
||||
"""A failed handoff is not retried by a later acquire on the same adapter."""
|
||||
existing = {
|
||||
"pid": 7777,
|
||||
"kind": "hermes-gateway",
|
||||
"argv": ["hermes", "gateway", "run"],
|
||||
"start_time": 456,
|
||||
}
|
||||
adapter._platform_lock_takeover_allowed = True
|
||||
|
||||
with patch(
|
||||
"gateway.status.acquire_scoped_lock",
|
||||
return_value=(False, existing),
|
||||
), patch(
|
||||
"gateway.status.take_over_scoped_lock_holder",
|
||||
return_value=None,
|
||||
) as takeover, patch.object(
|
||||
adapter, "_write_runtime_status_safe"
|
||||
):
|
||||
first = adapter._acquire_platform_lock(
|
||||
"telegram-bot-token", "test-token", "Telegram bot token"
|
||||
)
|
||||
second = adapter._acquire_platform_lock(
|
||||
"telegram-bot-token", "test-token", "Telegram bot token"
|
||||
)
|
||||
|
||||
assert first is False
|
||||
assert second is False
|
||||
assert adapter._platform_lock_takeover_allowed is False
|
||||
assert adapter._platform_lock_takeover_attempted is True
|
||||
takeover.assert_called_once_with(existing)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_runner_scopes_replace_intent_to_initial_connect():
|
||||
runner = GatewayRunner.__new__(GatewayRunner)
|
||||
runner._platform_lock_takeover_on_start = True
|
||||
adapter = MagicMock()
|
||||
adapter._platform_lock_takeover_allowed = False
|
||||
seen = []
|
||||
|
||||
async def connect(current_adapter, _platform):
|
||||
seen.append(current_adapter._platform_lock_takeover_allowed)
|
||||
return True
|
||||
|
||||
runner._connect_adapter_with_timeout = connect
|
||||
|
||||
assert await runner._connect_initial_adapter_with_timeout(
|
||||
adapter, MagicMock(value="telegram")
|
||||
) is True
|
||||
assert seen == [True]
|
||||
assert adapter._platform_lock_takeover_allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_runner_clears_replace_intent_when_initial_connect_raises():
|
||||
runner = GatewayRunner.__new__(GatewayRunner)
|
||||
runner._platform_lock_takeover_on_start = True
|
||||
adapter = MagicMock()
|
||||
adapter._platform_lock_takeover_allowed = False
|
||||
|
||||
async def connect(_adapter, _platform):
|
||||
raise RuntimeError("connect failed")
|
||||
|
||||
runner._connect_adapter_with_timeout = connect
|
||||
|
||||
with pytest.raises(RuntimeError, match="connect failed"):
|
||||
await runner._connect_initial_adapter_with_timeout(
|
||||
adapter, MagicMock(value="telegram")
|
||||
)
|
||||
assert adapter._platform_lock_takeover_allowed is False
|
||||
|
|
|
|||
|
|
@ -1421,6 +1421,155 @@ class TestTakeoverMarker:
|
|||
assert not marker_path.exists()
|
||||
|
||||
|
||||
class TestScopedLockTakeover:
|
||||
"""Cross-home takeover requires explicit, corroborated process identity."""
|
||||
|
||||
@staticmethod
|
||||
def _owner_record(target_home: Path, *, pid: int = 4242, start_time: int = 123):
|
||||
target_home.mkdir(parents=True, exist_ok=True)
|
||||
record = {
|
||||
"pid": pid,
|
||||
"kind": "hermes-gateway",
|
||||
"argv": ["python", "-m", "hermes_cli.main", "gateway", "run"],
|
||||
"start_time": start_time,
|
||||
"hermes_home": str(target_home),
|
||||
}
|
||||
(target_home / "gateway.pid").write_text(json.dumps(record))
|
||||
return record
|
||||
|
||||
def test_verified_distinct_home_handoff_marks_target_before_sigterm(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
replacer_home = tmp_path / "replacer"
|
||||
target_home = tmp_path / "target"
|
||||
replacer_home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(replacer_home))
|
||||
record = self._owner_record(target_home)
|
||||
|
||||
alive = iter([True, True, False])
|
||||
monkeypatch.setattr(status, "_pid_exists", lambda _pid: next(alive))
|
||||
monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123)
|
||||
monkeypatch.setattr(
|
||||
status,
|
||||
"_read_process_cmdline",
|
||||
lambda _pid: "python -m hermes_cli.main gateway run",
|
||||
)
|
||||
calls = []
|
||||
|
||||
def terminate(pid, *, force=False):
|
||||
marker_path = target_home / ".gateway-takeover.json"
|
||||
assert marker_path.exists()
|
||||
payload = json.loads(marker_path.read_text())
|
||||
assert payload["target_hermes_home"] == str(target_home)
|
||||
assert payload["replacer_hermes_home"] == str(replacer_home)
|
||||
calls.append((pid, force))
|
||||
|
||||
monkeypatch.setattr(status, "terminate_pid", terminate)
|
||||
|
||||
owner_pid = status.take_over_scoped_lock_holder(
|
||||
record, graceful_attempts=1
|
||||
)
|
||||
|
||||
assert owner_pid == 4242
|
||||
assert calls == [(4242, False)]
|
||||
assert not (target_home / ".gateway-takeover.json").exists()
|
||||
assert not (replacer_home / ".gateway-takeover.json").exists()
|
||||
|
||||
def test_handoff_rejects_uncorroborated_target_home(self, tmp_path, monkeypatch):
|
||||
target_home = tmp_path / "target"
|
||||
record = self._owner_record(target_home)
|
||||
# The lock claims target_home, but that home's PID record names a
|
||||
# different process identity.
|
||||
bad_pid_record = dict(record, pid=9999)
|
||||
(target_home / "gateway.pid").write_text(json.dumps(bad_pid_record))
|
||||
|
||||
monkeypatch.setattr(status, "_pid_exists", lambda _pid: True)
|
||||
monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123)
|
||||
monkeypatch.setattr(
|
||||
status,
|
||||
"_read_process_cmdline",
|
||||
lambda _pid: "python -m hermes_cli.main gateway run",
|
||||
)
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
status, "terminate_pid", lambda *args, **kwargs: calls.append(args)
|
||||
)
|
||||
|
||||
assert status.take_over_scoped_lock_holder(record) is None
|
||||
assert calls == []
|
||||
assert not (target_home / ".gateway-takeover.json").exists()
|
||||
|
||||
def test_handoff_requires_marker_write_before_termination(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
target_home = tmp_path / "target"
|
||||
record = self._owner_record(target_home)
|
||||
monkeypatch.setattr(status, "_pid_exists", lambda _pid: True)
|
||||
monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 123)
|
||||
monkeypatch.setattr(
|
||||
status,
|
||||
"_read_process_cmdline",
|
||||
lambda _pid: "python -m hermes_cli.main gateway run",
|
||||
)
|
||||
monkeypatch.setattr(status, "write_takeover_marker", lambda *a, **k: False)
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
status, "terminate_pid", lambda *args, **kwargs: calls.append(args)
|
||||
)
|
||||
|
||||
assert status.take_over_scoped_lock_holder(record) is None
|
||||
assert calls == []
|
||||
|
||||
def test_pid_reuse_after_sigterm_is_never_force_killed(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
target_home = tmp_path / "target"
|
||||
record = self._owner_record(target_home)
|
||||
monkeypatch.setattr(status, "_pid_exists", lambda _pid: True)
|
||||
starts = iter([123, 123, 999])
|
||||
monkeypatch.setattr(
|
||||
status, "_get_process_start_time", lambda _pid: next(starts)
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
status,
|
||||
"_read_process_cmdline",
|
||||
lambda _pid: "python -m hermes_cli.main gateway run",
|
||||
)
|
||||
calls = []
|
||||
monkeypatch.setattr(
|
||||
status,
|
||||
"terminate_pid",
|
||||
lambda pid, *, force=False: calls.append((pid, force)),
|
||||
)
|
||||
|
||||
assert status.take_over_scoped_lock_holder(
|
||||
record, graceful_attempts=1
|
||||
) == 4242
|
||||
assert calls == [(4242, False)]
|
||||
|
||||
def test_target_accepts_verified_cross_home_marker(self, tmp_path, monkeypatch):
|
||||
replacer_home = tmp_path / "replacer"
|
||||
target_home = tmp_path / "target"
|
||||
replacer_home.mkdir()
|
||||
target_home.mkdir()
|
||||
monkeypatch.setenv("HERMES_HOME", str(replacer_home))
|
||||
monkeypatch.setattr(status, "_get_process_start_time", lambda _pid: 100)
|
||||
|
||||
assert status.write_takeover_marker(
|
||||
os.getpid(),
|
||||
target_home=target_home,
|
||||
target_start_time=100,
|
||||
) is True
|
||||
assert not (replacer_home / ".gateway-takeover.json").exists()
|
||||
assert (target_home / ".gateway-takeover.json").exists()
|
||||
|
||||
# The target process reads its own home. A differing replacer home is
|
||||
# valid only because the marker explicitly names this target home.
|
||||
monkeypatch.setenv("HERMES_HOME", str(target_home))
|
||||
assert status.consume_takeover_marker_for_self() is True
|
||||
assert not (target_home / ".gateway-takeover.json").exists()
|
||||
|
||||
|
||||
class TestPlannedStopMarker:
|
||||
"""Tests for intentional service/manual gateway stop markers."""
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue