mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-26 17:38:36 +00:00
fix(gateway): prevent reconnect watcher wedge after network-loss fatal error (#70344)
Three-part fix for the gateway going silently deaf after a retryable fatal adapter error (e.g. httpx.ConnectError on Telegram): 1. **Detach-on-timeout in _connect_adapter_with_timeout** — Replaced plain asyncio.wait_for with the task-detach pattern used by _await_adapter_cleanup_with_timeout. asyncio.wait_for cancels the overdue task but then waits for it to exit, so a connect() that catches CancelledError can block recovery forever. The detach pattern releases the runner at the deadline via consume_detached_task_result. 2. **Ensure reconnect watcher always runs after escalation** — Added _ensure_reconnect_watcher_running(), called after queueing a retryable fatal error. If the reconnect watcher task has died (exhausted restart budget, terminal exception), it is respawned so queued platforms are never permanently stranded. 3. **Faulthandler at gateway startup** — Enabled faulthandler + SIGUSR2 dump to a rotating file under HERMES_HOME/logs/ for post-mortem diagnosis of future event-loop freezes. Tests added for _ensure_reconnect_watcher_running (alive, dead, not-started, not-running), fatal-error integration (retryable calls ensure, non-retryable does not), and _connect_adapter_with_timeout (timeout raises, success returns).
This commit is contained in:
webtecnica
Teknium
parent
9c65cdb043
commit
83fe362e80
2 changed files with 305 additions and 8 deletions
|
|
@ -27,6 +27,7 @@ except ModuleNotFoundError:
|
|||
import asyncio
|
||||
import concurrent.futures
|
||||
import dataclasses
|
||||
import faulthandler
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
|
|
@ -3292,6 +3293,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_gateway_started_at: float = 0.0
|
||||
_shutdown_watchdog_done: Optional["threading.Event"] = None
|
||||
_platform_lock_takeover_on_start: bool = False
|
||||
_reconnect_watcher_task: Optional["asyncio.Task"] = None
|
||||
|
||||
def __init__(self, config: Optional[GatewayConfig] = None):
|
||||
global _gateway_runner_ref
|
||||
|
|
@ -4047,14 +4049,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
timeout = self._platform_connect_timeout_secs()
|
||||
if timeout <= 0:
|
||||
return await adapter.connect(is_reconnect=is_reconnect)
|
||||
# Use the detach-on-timeout pattern instead of plain asyncio.wait_for:
|
||||
# asyncio.wait_for cancels the overdue task but then waits for it to
|
||||
# exit. An adapter connect() that catches CancelledError can therefore
|
||||
# block recovery forever (the watcher never reaches the next retry).
|
||||
# Keep ownership of the old task through its done callback, but
|
||||
# release the runner at the deadline (#70344).
|
||||
task = asyncio.ensure_future(
|
||||
adapter.connect(is_reconnect=is_reconnect)
|
||||
)
|
||||
try:
|
||||
return await asyncio.wait_for(
|
||||
adapter.connect(is_reconnect=is_reconnect), timeout=timeout
|
||||
)
|
||||
except asyncio.TimeoutError as exc:
|
||||
raise TimeoutError(
|
||||
f"{platform.value} connect timed out after {timeout:g}s"
|
||||
) from exc
|
||||
done, _pending = await asyncio.wait({task}, timeout=timeout)
|
||||
except asyncio.CancelledError:
|
||||
task.cancel()
|
||||
task.add_done_callback(consume_detached_task_result)
|
||||
raise
|
||||
if task in done:
|
||||
result = await task
|
||||
return bool(result)
|
||||
task.cancel()
|
||||
task.add_done_callback(consume_detached_task_result)
|
||||
raise TimeoutError(
|
||||
f"{platform.value} connect timed out after {timeout:g}s"
|
||||
)
|
||||
|
||||
async def _connect_initial_adapter_with_timeout(self, adapter, platform) -> bool:
|
||||
"""Connect one cold-start adapter with tightly scoped replace intent.
|
||||
|
|
@ -4731,6 +4748,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"%s queued for background reconnection",
|
||||
adapter.platform.value,
|
||||
)
|
||||
# Ensure the reconnect watcher is alive — if it died (e.g. from
|
||||
# exhausting its restart budget), respawn it so queued platforms
|
||||
# are not permanently stranded (#70344).
|
||||
self._ensure_reconnect_watcher_running()
|
||||
|
||||
if not self.adapters and not self._failed_platforms:
|
||||
self._exit_reason = adapter.fatal_error_message or "All messaging adapters disconnected"
|
||||
|
|
@ -7742,6 +7763,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
Returns True if at least one adapter connected successfully.
|
||||
"""
|
||||
logger.info("Starting Hermes Gateway...")
|
||||
# Enable faulthandler at gateway start so that SIGUSR2 (or an
|
||||
# internal watchdog) can dump all thread and task stacks to stderr
|
||||
# for post-mortem diagnosis of event-loop freezes (#70344).
|
||||
faulthandler.enable()
|
||||
# Also dump stacks to a rotating file for off-line analysis when
|
||||
# the gateway is running under a service manager that doesn't
|
||||
# capture stderr.
|
||||
try:
|
||||
_log_dir = getattr(self.config, "log_dir", None) or os.path.join(
|
||||
os.environ.get("HERMES_HOME", str(Path.home() / ".hermes")),
|
||||
"logs",
|
||||
)
|
||||
_faulthandler_path = os.path.join(_log_dir, "gateway_faulthandler.log")
|
||||
os.makedirs(_log_dir, exist_ok=True)
|
||||
faulthandler.register(
|
||||
signal.SIGUSR2,
|
||||
file=open(_faulthandler_path, "a"),
|
||||
all_threads=True,
|
||||
chain=True,
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Could not set up faulthandler file logging", exc_info=True)
|
||||
|
||||
try:
|
||||
self._gateway_loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
|
|
@ -8461,7 +8505,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
len(self._failed_platforms),
|
||||
", ".join(p.value for p in self._failed_platforms),
|
||||
)
|
||||
self._spawn_supervised(self._platform_reconnect_watcher, "platform_reconnect_watcher")
|
||||
# Track the reconnect watcher task so _ensure_reconnect_watcher_running
|
||||
# can detect if it dies and respawn it (#70344).
|
||||
self._reconnect_watcher_task = asyncio.create_task(
|
||||
self._platform_reconnect_watcher()
|
||||
)
|
||||
self._background_tasks.add(self._reconnect_watcher_task)
|
||||
|
||||
# Start background handoff watcher — picks up CLI sessions marked
|
||||
# handoff_state='pending' in state.db and re-binds them to the
|
||||
|
|
@ -9017,6 +9066,30 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# self state, so inheriting the mixin keeps every self._kanban_* call site
|
||||
# working unchanged while lifting ~1,000 LOC out of this file.
|
||||
|
||||
def _ensure_reconnect_watcher_running(self) -> None:
|
||||
"""Ensure the platform reconnect watcher background task is alive.
|
||||
|
||||
If the tracked reconnect watcher task has died (e.g. from exhausting
|
||||
its restart budget, or a terminal exception that _spawn_supervised
|
||||
could not recover), respawns it so platforms queued for reconnection
|
||||
are not permanently stranded. Called after queueing a retryable fatal
|
||||
error in _handle_adapter_fatal_error (#70344).
|
||||
"""
|
||||
if not getattr(self, "_running", False):
|
||||
return
|
||||
task = getattr(self, "_reconnect_watcher_task", None)
|
||||
if task is not None and not task.done():
|
||||
return # already alive
|
||||
logger.warning(
|
||||
"Reconnect watcher task is dead (done=%s) — respawning",
|
||||
task.done() if task is not None else "N/A",
|
||||
)
|
||||
self._reconnect_watcher_task = asyncio.create_task(
|
||||
self._platform_reconnect_watcher()
|
||||
)
|
||||
if getattr(self, "_background_tasks", None) is not None:
|
||||
self._background_tasks.add(self._reconnect_watcher_task)
|
||||
|
||||
async def _platform_reconnect_watcher(self) -> None:
|
||||
"""Background task that periodically retries connecting failed platforms.
|
||||
|
||||
|
|
|
|||
|
|
@ -1017,3 +1017,227 @@ class TestFatalHandoffCancellationProof:
|
|||
|
||||
assert runner._exit_with_failure is True
|
||||
assert runner.stop.await_count == 1
|
||||
|
||||
|
||||
# ── _ensure_reconnect_watcher_running ──────────────────────────────────
|
||||
|
||||
|
||||
class TestEnsureReconnectWatcherRunning:
|
||||
"""Verify _ensure_reconnect_watcher_running respawns the watcher when dead."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reconnect_watcher_alive_does_nothing(self):
|
||||
"""Task is alive => no-op."""
|
||||
runner = _make_runner()
|
||||
runner._running = True
|
||||
runner._background_tasks = set()
|
||||
|
||||
async def _dummy():
|
||||
await asyncio.sleep(3600)
|
||||
|
||||
runner._reconnect_watcher_task = asyncio.create_task(_dummy())
|
||||
runner._background_tasks.add(runner._reconnect_watcher_task)
|
||||
|
||||
old_task = runner._reconnect_watcher_task
|
||||
runner._ensure_reconnect_watcher_running()
|
||||
|
||||
# Same task, not replaced
|
||||
assert runner._reconnect_watcher_task is old_task
|
||||
assert not runner._reconnect_watcher_task.done()
|
||||
|
||||
old_task.cancel()
|
||||
try:
|
||||
await old_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reconnect_watcher_dead_respawns(self):
|
||||
"""Watcher is done => respawn."""
|
||||
runner = _make_runner()
|
||||
runner._running = True
|
||||
runner._background_tasks = set()
|
||||
runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0))
|
||||
await runner._reconnect_watcher_task # let it finish
|
||||
|
||||
assert runner._reconnect_watcher_task.done()
|
||||
|
||||
runner._ensure_reconnect_watcher_running()
|
||||
|
||||
assert runner._reconnect_watcher_task is not None
|
||||
assert not runner._reconnect_watcher_task.done()
|
||||
assert runner._reconnect_watcher_task in runner._background_tasks
|
||||
|
||||
runner._reconnect_watcher_task.cancel()
|
||||
try:
|
||||
await runner._reconnect_watcher_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reconnect_watcher_not_running_respawns(self):
|
||||
"""No task at all => creates one."""
|
||||
runner = _make_runner()
|
||||
runner._running = True
|
||||
runner._background_tasks = set()
|
||||
runner._reconnect_watcher_task = None
|
||||
|
||||
runner._ensure_reconnect_watcher_running()
|
||||
|
||||
assert runner._reconnect_watcher_task is not None
|
||||
assert not runner._reconnect_watcher_task.done()
|
||||
assert runner._reconnect_watcher_task in runner._background_tasks
|
||||
|
||||
runner._reconnect_watcher_task.cancel()
|
||||
try:
|
||||
await runner._reconnect_watcher_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_not_running_noop(self):
|
||||
"""_running is False => no-op."""
|
||||
runner = _make_runner()
|
||||
runner._running = False
|
||||
runner._reconnect_watcher_task = None
|
||||
runner._ensure_reconnect_watcher_running()
|
||||
assert runner._reconnect_watcher_task is None
|
||||
|
||||
|
||||
# ── _handle_adapter_fatal_error calls _ensure_reconnect_watcher ────────
|
||||
|
||||
|
||||
class TestFatalErrorCallsEnsureWatcher:
|
||||
"""Verify _handle_adapter_fatal_error calls _ensure_reconnect_watcher_running."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_retryable_fatal_error_calls_ensure_watcher(self):
|
||||
"""A retryable fatal error queues the platform AND ensures watcher is alive."""
|
||||
runner = _make_runner()
|
||||
runner._running = True
|
||||
runner._background_tasks = set()
|
||||
runner._failed_platforms = {}
|
||||
runner._fatal_handler_tasks = set()
|
||||
runner._reconnect_watcher_task = asyncio.create_task(asyncio.sleep(0))
|
||||
# Let the dummy watcher finish so _ensure_reconnect_watcher_running
|
||||
# detects it's dead and respawns.
|
||||
await runner._reconnect_watcher_task
|
||||
|
||||
platform_config = PlatformConfig(enabled=True, token="test")
|
||||
runner.config = GatewayConfig(
|
||||
platforms={Platform.TELEGRAM: platform_config}
|
||||
)
|
||||
|
||||
adapter = StubAdapter(
|
||||
platform=Platform.TELEGRAM,
|
||||
succeed=False,
|
||||
fatal_error="network outage",
|
||||
fatal_retryable=True,
|
||||
)
|
||||
# Pre-set fatal error attributes so the handler can read them
|
||||
# without going through connect() (#70344).
|
||||
adapter._set_fatal_error(
|
||||
"NETWORK_ERROR", "network outage", retryable=True
|
||||
)
|
||||
# Populate adapters so the impl pops it and queues for reconnect
|
||||
runner.adapters[Platform.TELEGRAM] = adapter
|
||||
|
||||
call_count = {"ensure": 0}
|
||||
|
||||
def tracking_ensure():
|
||||
call_count["ensure"] += 1
|
||||
|
||||
with patch.object(
|
||||
runner,
|
||||
"_ensure_reconnect_watcher_running",
|
||||
side_effect=tracking_ensure,
|
||||
):
|
||||
await runner._handle_adapter_fatal_error(adapter)
|
||||
|
||||
assert Platform.TELEGRAM in runner._failed_platforms
|
||||
assert call_count["ensure"] >= 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nonretryable_fatal_error_does_not_call_ensure(self):
|
||||
"""A non-retryable error must NOT queue the platform or call the watcher."""
|
||||
runner = _make_runner()
|
||||
runner._running = True
|
||||
runner._background_tasks = set()
|
||||
runner._failed_platforms = {}
|
||||
runner._fatal_handler_tasks = set()
|
||||
runner._reconnect_watcher_task = None
|
||||
|
||||
platform_config = PlatformConfig(enabled=True, token="test")
|
||||
runner.config = GatewayConfig(
|
||||
platforms={Platform.TELEGRAM: platform_config}
|
||||
)
|
||||
|
||||
adapter = StubAdapter(
|
||||
platform=Platform.TELEGRAM,
|
||||
succeed=False,
|
||||
fatal_error="bad token",
|
||||
fatal_retryable=False,
|
||||
)
|
||||
# Pre-set fatal error attributes so the handler can read them
|
||||
# without going through connect() (#70344).
|
||||
adapter._set_fatal_error(
|
||||
"AUTH_FAILED", "bad token", retryable=False
|
||||
)
|
||||
runner.adapters[Platform.TELEGRAM] = adapter
|
||||
|
||||
ensure_called = False
|
||||
|
||||
def noop_ensure():
|
||||
nonlocal ensure_called
|
||||
ensure_called = True
|
||||
|
||||
with patch.object(runner, "_ensure_reconnect_watcher_running", side_effect=noop_ensure):
|
||||
await runner._handle_adapter_fatal_error(adapter)
|
||||
|
||||
assert Platform.TELEGRAM not in runner._failed_platforms
|
||||
assert not ensure_called
|
||||
|
||||
|
||||
# ── _connect_adapter_with_timeout detach-on-timeout ────────────────────
|
||||
|
||||
|
||||
class TestConnectAdapterDetachOnTimeout:
|
||||
"""Verify _connect_adapter_with_timeout uses the detach pattern."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect_timed_out_raises_timeouterror(self):
|
||||
"""A connect() that never finishes must raise TimeoutError."""
|
||||
runner = _make_runner()
|
||||
|
||||
adapter = StubAdapter(succeed=True)
|
||||
|
||||
async def _slow_connect(**kwargs):
|
||||
await asyncio.sleep(3600) # never finishes
|
||||
|
||||
with patch.object(adapter, "connect", side_effect=_slow_connect):
|
||||
with patch.object(
|
||||
runner, "_platform_connect_timeout_secs", return_value=0.01
|
||||
):
|
||||
with pytest.raises(TimeoutError, match="timed out"):
|
||||
await runner._connect_adapter_with_timeout(
|
||||
adapter, Platform.TELEGRAM
|
||||
)
|
||||
|
||||
# After the TimeoutError, the slow connect coroutine should have been
|
||||
# cancelled and detached, so the event loop can move on.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect_success_returns_true(self):
|
||||
"""A successful connect returns True."""
|
||||
runner = _make_runner()
|
||||
adapter = StubAdapter(succeed=True)
|
||||
|
||||
with patch.object(
|
||||
runner, "_platform_connect_timeout_secs", return_value=30.0
|
||||
):
|
||||
result = await runner._connect_adapter_with_timeout(
|
||||
adapter, Platform.TELEGRAM
|
||||
)
|
||||
|
||||
assert result is True
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue