diff --git a/gateway/status.py b/gateway/status.py index 7b8e9ff57583..5ef44a9d15ec 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -13,6 +13,7 @@ concurrently under distinct configurations). import hashlib import json +import logging import os import shlex import signal @@ -23,7 +24,7 @@ import time from datetime import datetime, timezone from pathlib import Path from hermes_constants import get_hermes_home, _get_platform_default_hermes_home -from typing import Any, Optional +from typing import Any, NamedTuple, Optional from utils import atomic_json_write if sys.platform == "win32": @@ -46,6 +47,84 @@ _GATEWAY_RUNNING_PID_CACHE_TTL_SECONDS = 1.0 _gateway_running_pid_cache_lock = threading.Lock() _gateway_running_pid_cache: dict[tuple[str, bool, bool], tuple[float, tuple[Any, ...], Optional[int]]] = {} +logger = logging.getLogger(__name__) + + +class StormInfo(NamedTuple): + """Result of a respawn-storm check: how many starts, over what window, and + the backoff the caller should sleep to break the storm.""" + + count: int + window_s: float + backoff_s: float + + +def _get_starts_log_path() -> Path: + """Path to the append-only gateway-start ledger used by the respawn-storm + breaker. Distinct from ``restart_loop.json`` (the auto-resume guard) — no + collision.""" + return get_hermes_home() / "gateway-starts.log" + + +def record_start_and_check_storm( + max_starts: int = 5, window_s: float = 120.0, *, backoff_cap_s: float = 300.0 +) -> Optional[StormInfo]: + """Record this gateway start and report whether a respawn storm is underway. + + Appends the current UTC timestamp to the starts-log, prunes entries older + than ``window_s``, and ring-buffers the file so it can't grow unbounded. + Returns a :class:`StormInfo` when more than ``max_starts`` starts landed in + the window (with an exponential backoff capped at ``backoff_cap_s``), else + ``None``. + + Best-effort: any bookkeeping failure is logged and swallowed so a broken + ledger can never crash gateway startup. + """ + try: + path = _get_starts_log_path() + path.parent.mkdir(parents=True, exist_ok=True) + + now = datetime.now(timezone.utc).timestamp() + + existing: list[float] = [] + if path.exists(): + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + existing.append(float(line)) + except ValueError: + continue + + existing.append(now) + + # Keep only starts within the sliding window for the storm decision. + recent = [ts for ts in existing if now - ts <= window_s] + + # Ring-buffer what we persist so the file stays bounded even if the + # window is wide or starts are frequent. + keep = max(max_starts * 4, 40) + to_write = existing[-keep:] + + tmp = path.with_suffix(".tmp") + tmp.write_text( + "\n".join(repr(ts) for ts in to_write) + "\n", encoding="utf-8" + ) + os.replace(tmp, path) + + if len(recent) > max_starts: + backoff = min( + backoff_cap_s, 5.0 * (2 ** min(len(recent) - max_starts, 6)) + ) + return StormInfo(count=len(recent), window_s=window_s, backoff_s=backoff) + return None + except Exception as _e: + logger.debug( + "respawn-storm breaker bookkeeping failed (non-fatal): %s", _e + ) + return None + def _get_process_hermes_home() -> Path: """Return the process-level HERMES_HOME, skipping context-local overrides. diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 7024cf32e76f..0984f88a8c86 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -3982,7 +3982,17 @@ def generate_launchd_plist() -> str: KeepAlive - + + + ThrottleInterval + 30 + + ExitTimeOut + 25 + StandardOutPath {log_dir}/gateway.log @@ -4856,6 +4866,39 @@ def run_gateway(verbose: int = 0, quiet: bool = False, replace: bool = False, fo _atexit.register(_atexit_hook) + # Portable, app-level respawn-storm circuit breaker. launchd/systemd have + # their own throttles, but this backstop works on every platform (and covers + # supervisors that lack a respawn floor). Set HERMES_GATEWAY_MAX_STARTS<=0 to + # disable. Kept best-effort: a bookkeeping failure must never block startup. + try: + import time as _time + + from gateway.status import record_start_and_check_storm + + try: + _max_starts = int(os.getenv("HERMES_GATEWAY_MAX_STARTS", "5")) + except ValueError: + _max_starts = 5 + try: + _win = float(os.getenv("HERMES_GATEWAY_START_WINDOW_S", "120")) + except ValueError: + _win = 120.0 + _storm = ( + record_start_and_check_storm(max_starts=_max_starts, window_s=_win) + if _max_starts > 0 + else None + ) + if _storm is not None: + logger.warning( + "Gateway (re)started %d times in %.0fs — backing off %.0fs to break a respawn storm.", + _storm.count, + _storm.window_s, + _storm.backoff_s, + ) + _time.sleep(_storm.backoff_s) + except Exception as _be: + logger.debug("respawn-storm breaker check failed (non-fatal): %s", _be) + success = False try: success = asyncio.run(start_gateway(replace=replace, verbosity=verbosity)) diff --git a/tests/gateway/test_status.py b/tests/gateway/test_status.py index 16ce08207731..26405808f006 100644 --- a/tests/gateway/test_status.py +++ b/tests/gateway/test_status.py @@ -3,6 +3,7 @@ import json import os import sys +import time from pathlib import Path from types import SimpleNamespace @@ -1660,3 +1661,63 @@ class TestGatewayBusyDerivation: assert status.derive_gateway_drainable( gateway_running=True, gateway_state=state ) is False, state + + +class TestRespawnStormBreaker: + def test_no_storm_under_threshold(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + for _ in range(5): + result = status.record_start_and_check_storm( + max_starts=5, window_s=120.0 + ) + assert result is None + + def test_storm_detected_over_threshold(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + results = [ + status.record_start_and_check_storm(max_starts=5, window_s=120.0) + for _ in range(7) + ] + last = results[-1] + assert last is not None + assert last.count >= 6 + assert last.backoff_s > 0 + + def test_old_starts_pruned_outside_window(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + log_path = status._get_starts_log_path() + old = time.time() - 10000 + log_path.write_text( + "\n".join(repr(old) for _ in range(10)) + "\n", encoding="utf-8" + ) + # All seeded entries are far outside the window, so a single new start + # cannot exceed the threshold. + result = status.record_start_and_check_storm(max_starts=5, window_s=120.0) + assert result is None + + def test_starts_log_written_atomically(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + for _ in range(6): + status.record_start_and_check_storm(max_starts=5, window_s=120.0) + + # No leftover temp file from the atomic write. + assert not list(tmp_path.glob("*.tmp")) + + log_path = status._get_starts_log_path() + assert log_path.exists() + for line in log_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + float(line) # every persisted line must parse as a float + + +class TestLaunchdPlistRespawnGovernance: + def test_plist_has_throttle_interval(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from hermes_cli.gateway import generate_launchd_plist + + plist = generate_launchd_plist() + assert "ThrottleInterval" in plist + assert "ExitTimeOut" in plist + assert "KeepAlive" in plist