fix: retry transient agent gateway restart failures
This commit is contained in:
1 parent
83a2471990
commit
885e267213
4 files changed
+218
-1
No files matched your search
@@ -251,6 +251,24 @@ def _gateway_running_pid(gateway_status: Any, pid_path: Path | None) -> int | No
|
||||
return get_running_pid()
|
||||
|
||||
|
||||
def get_active_profile_gateway_running_pid() -> int | None:
|
||||
"""Return the confirmed active-profile PID without state fallbacks.
|
||||
|
||||
Update recovery needs process identity, not just a recent ``running`` state:
|
||||
an old gateway can remain alive when the restart CLI fails before executing.
|
||||
Use the same active-profile home targeted by the restart helper rather than
|
||||
the default-root health path. Keep this fail-closed when status is unavailable.
|
||||
"""
|
||||
try:
|
||||
from api.profiles import get_active_hermes_home
|
||||
|
||||
gateway_status = _gateway_status_module()
|
||||
gateway_pid_path = Path(get_active_hermes_home()) / _GATEWAY_PID_FILE
|
||||
return _gateway_running_pid(gateway_status, gateway_pid_path)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _runtime_detail_subset(runtime_status: dict[str, Any] | None) -> dict[str, Any]:
|
||||
"""Return only non-sensitive runtime fields for the browser.
|
||||
|
||||
|
||||
+53
-1
@@ -24,6 +24,7 @@ from collections import OrderedDict
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from api.agent_health import get_active_profile_gateway_running_pid
|
||||
from api.gateway_restart import restart_active_profile_gateway
|
||||
from api.config import REPO_ROOT, STREAMS, STREAMS_LOCK
|
||||
|
||||
@@ -42,6 +43,7 @@ _cache_lock = threading.Lock()
|
||||
_check_in_progress = False
|
||||
_apply_lock = threading.Lock() # prevents concurrent stash/pull/pop on same repo
|
||||
CACHE_TTL = 1800 # 30 minutes
|
||||
_AGENT_GATEWAY_RESTART_RETRY_DELAY_S = 1.0
|
||||
_GIT_DIAGNOSTIC_MAX_CHARS = 300
|
||||
_CREDENTIAL_IN_URL_RE = re.compile(r"([a-zA-Z][a-zA-Z0-9+.-]*://)([^/@\s'\"]+)@")
|
||||
_GITHUB_TOKEN_RE = re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b")
|
||||
@@ -1814,11 +1816,61 @@ def _ensure_gateway_restart_for_agent_update() -> tuple[bool, dict]:
|
||||
- ok is False when restart did not complete and callers must abort success.
|
||||
- restart_payload contains helper status fields for response shaping.
|
||||
"""
|
||||
gateway_pid_before_restart = get_active_profile_gateway_running_pid()
|
||||
restart_result = restart_active_profile_gateway()
|
||||
status = str(restart_result.get("status") or "")
|
||||
if status in {"completed", "in_progress"}:
|
||||
return True, restart_result
|
||||
return False, restart_result
|
||||
if status != "failed":
|
||||
return False, restart_result
|
||||
|
||||
# launchd can briefly fail to spawn the replacement gateway while it is
|
||||
# rotating the supervised process (#6045). Retry exactly once after a
|
||||
# bounded delay so an already-applied Agent update is not reported as a
|
||||
# complete failure because of that transient process handoff.
|
||||
time.sleep(_AGENT_GATEWAY_RESTART_RETRY_DELAY_S)
|
||||
retry_result = restart_active_profile_gateway()
|
||||
retry_status = str(retry_result.get("status") or "")
|
||||
if retry_status in {"completed", "in_progress"}:
|
||||
return True, {
|
||||
**retry_result,
|
||||
"retry_attempted": True,
|
||||
"initial_failure": restart_result.get("message"),
|
||||
}
|
||||
if retry_status != "failed":
|
||||
return False, {
|
||||
**retry_result,
|
||||
"retry_attempted": True,
|
||||
"initial_failure": restart_result.get("message"),
|
||||
}
|
||||
|
||||
# A restart command can still exit non-zero after launchd has recovered the
|
||||
# service. Only accept that recovery when the confirmed local PID changed;
|
||||
# a merely-alive old gateway has not loaded the updated Agent checkout.
|
||||
time.sleep(_AGENT_GATEWAY_RESTART_RETRY_DELAY_S)
|
||||
gateway_pid_after_retry = get_active_profile_gateway_running_pid()
|
||||
if (
|
||||
gateway_pid_before_restart is not None
|
||||
and gateway_pid_after_retry is not None
|
||||
and gateway_pid_after_retry != gateway_pid_before_restart
|
||||
):
|
||||
return True, {
|
||||
"status": "completed",
|
||||
"message": "Gateway service recovered after a transient restart failure",
|
||||
"retry_attempted": True,
|
||||
"process_replaced": True,
|
||||
"initial_failure": restart_result.get("message"),
|
||||
"retry_failure": retry_result.get("message"),
|
||||
}
|
||||
|
||||
initial_message = str(restart_result.get("message") or "Restart failed")
|
||||
retry_message = str(retry_result.get("message") or "retry did not complete")
|
||||
return False, {
|
||||
**retry_result,
|
||||
"message": f"{initial_message}; recovery retry did not complete: {retry_message}",
|
||||
"retry_attempted": True,
|
||||
"initial_failure": restart_result.get("message"),
|
||||
}
|
||||
|
||||
|
||||
def _agent_gateway_restart_failure_message(target: str, restart_result: dict) -> str:
|
||||
|
||||
@@ -131,6 +131,30 @@ def test_agent_health_payload_alive_uses_safe_runtime_details(monkeypatch):
|
||||
assert "pid" not in payload["details"]
|
||||
|
||||
|
||||
def test_active_profile_gateway_running_pid_uses_active_profile_path(monkeypatch, tmp_path):
|
||||
from api import agent_health, profiles
|
||||
|
||||
active_home = tmp_path / "profiles" / "active"
|
||||
gateway_status = _PathSensitiveGatewayStatus(active_home)
|
||||
monkeypatch.setattr(agent_health, "_gateway_status_module", lambda: gateway_status)
|
||||
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: active_home)
|
||||
|
||||
assert agent_health.get_active_profile_gateway_running_pid() == 98765
|
||||
assert gateway_status.running_pid_path == active_home / "gateway.pid"
|
||||
|
||||
|
||||
def test_active_profile_gateway_running_pid_fails_closed_when_status_is_unavailable(monkeypatch):
|
||||
from api import agent_health
|
||||
|
||||
monkeypatch.setattr(
|
||||
agent_health,
|
||||
"_gateway_status_module",
|
||||
lambda: (_ for _ in ()).throw(ModuleNotFoundError("gateway.status")),
|
||||
)
|
||||
|
||||
assert agent_health.get_active_profile_gateway_running_pid() is None
|
||||
|
||||
|
||||
def test_agent_health_payload_down_when_gateway_metadata_exists_but_no_process(monkeypatch):
|
||||
from api import agent_health
|
||||
|
||||
|
||||
@@ -829,6 +829,129 @@ class TestApplyForceUpdate:
|
||||
class TestAgentUpdateRequiresGatewayRestart:
|
||||
"""Agent updates must prove gateway restart before returning ok=True."""
|
||||
|
||||
def test_agent_gateway_restart_retries_one_transient_failure(self, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
restart_results = iter([
|
||||
{'status': 'failed', 'message': 'Restart failed: bad file descriptor'},
|
||||
{'status': 'completed', 'message': 'Gateway service restarted successfully'},
|
||||
])
|
||||
restart_calls = []
|
||||
sleeps = []
|
||||
|
||||
def fake_restart():
|
||||
restart_calls.append('called')
|
||||
return next(restart_results)
|
||||
|
||||
monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
|
||||
monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
|
||||
monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', lambda: 101)
|
||||
|
||||
ok, result = upd._ensure_gateway_restart_for_agent_update()
|
||||
|
||||
assert ok is True
|
||||
assert result['status'] == 'completed'
|
||||
assert result['retry_attempted'] is True
|
||||
assert 'bad file descriptor' in result['initial_failure']
|
||||
assert restart_calls == ['called', 'called']
|
||||
assert sleeps == [upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S]
|
||||
|
||||
def test_agent_gateway_restart_retry_busy_stays_fail_closed(self, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
restart_results = iter([
|
||||
{'status': 'failed', 'message': 'Restart failed: first'},
|
||||
{'status': 'busy', 'message': 'Restart already in progress'},
|
||||
])
|
||||
sleeps = []
|
||||
gateway_pid_calls = []
|
||||
|
||||
monkeypatch.setattr(upd, 'restart_active_profile_gateway', lambda: next(restart_results))
|
||||
monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
|
||||
monkeypatch.setattr(
|
||||
upd,
|
||||
'get_active_profile_gateway_running_pid',
|
||||
lambda: gateway_pid_calls.append(101) or 101,
|
||||
)
|
||||
|
||||
ok, result = upd._ensure_gateway_restart_for_agent_update()
|
||||
|
||||
assert ok is False
|
||||
assert result['status'] == 'busy'
|
||||
assert result['retry_attempted'] is True
|
||||
assert 'first' in result['initial_failure']
|
||||
assert sleeps == [upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S]
|
||||
assert gateway_pid_calls == [101]
|
||||
|
||||
def test_agent_gateway_restart_accepts_verified_process_replacement_after_retry_failure(self, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
timeline = []
|
||||
restart_results = iter([
|
||||
{'status': 'failed', 'message': 'Restart failed: first'},
|
||||
{'status': 'failed', 'message': 'Restart failed: retry'},
|
||||
])
|
||||
sleeps = []
|
||||
gateway_pids = iter([101, 202])
|
||||
|
||||
def fake_restart():
|
||||
timeline.append('restart')
|
||||
return next(restart_results)
|
||||
|
||||
def fake_gateway_pid():
|
||||
pid = next(gateway_pids)
|
||||
timeline.append(f'pid:{pid}')
|
||||
return pid
|
||||
|
||||
monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
|
||||
monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
|
||||
monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', fake_gateway_pid)
|
||||
|
||||
ok, result = upd._ensure_gateway_restart_for_agent_update()
|
||||
|
||||
assert ok is True
|
||||
assert result['status'] == 'completed'
|
||||
assert result['retry_attempted'] is True
|
||||
assert result['process_replaced'] is True
|
||||
assert 'first' in result['initial_failure']
|
||||
assert 'retry' in result['retry_failure']
|
||||
assert timeline == ['pid:101', 'restart', 'restart', 'pid:202']
|
||||
assert sleeps == [
|
||||
upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
|
||||
upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
|
||||
]
|
||||
|
||||
def test_agent_gateway_restart_fails_closed_after_retry_and_health_check(self, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
restart_results = iter([
|
||||
{'status': 'failed', 'message': 'Restart failed: first'},
|
||||
{'status': 'failed', 'message': 'Restart failed: retry'},
|
||||
])
|
||||
restart_calls = []
|
||||
sleeps = []
|
||||
|
||||
def fake_restart():
|
||||
restart_calls.append('called')
|
||||
return next(restart_results)
|
||||
|
||||
monkeypatch.setattr(upd, 'restart_active_profile_gateway', fake_restart)
|
||||
monkeypatch.setattr(upd.time, 'sleep', sleeps.append)
|
||||
monkeypatch.setattr(upd, 'get_active_profile_gateway_running_pid', lambda: 101)
|
||||
|
||||
ok, result = upd._ensure_gateway_restart_for_agent_update()
|
||||
|
||||
assert ok is False
|
||||
assert result['status'] == 'failed'
|
||||
assert result['retry_attempted'] is True
|
||||
assert 'Restart failed: first' in result['message']
|
||||
assert 'Restart failed: retry' in result['message']
|
||||
assert restart_calls == ['called', 'called']
|
||||
assert sleeps == [
|
||||
upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
|
||||
upd._AGENT_GATEWAY_RESTART_RETRY_DELAY_S,
|
||||
]
|
||||
|
||||
def test_apply_update_agent_requires_gateway_restart(self, tmp_path, monkeypatch):
|
||||
import api.updates as upd
|
||||
|
||||
|
||||
Reference in new issue
Block a user