From 18b46645437fb44c828458fe02eb4a6aa0bd3076 Mon Sep 17 00:00:00 2001 From: Victor Kyriazakos Date: Wed, 22 Jul 2026 22:08:51 +0000 Subject: [PATCH] fix(monitoring): drain terminal lifecycle events --- agent/monitoring/emitter.py | 33 ++++++++---- agent/monitoring/gateway_health_export.py | 9 ++-- gateway/run.py | 54 ++++++++++++------- .../gateway_health_export_probe.py | 1 + tests/gateway/test_startup_restart_race.py | 10 +++- tests/monitoring/test_emitter.py | 30 +++++++++++ .../monitoring/test_gateway_health_export.py | 35 ++++++++++++ 7 files changed, 139 insertions(+), 33 deletions(-) diff --git a/agent/monitoring/emitter.py b/agent/monitoring/emitter.py index 52bee5e05cf..18a0d4851ae 100644 --- a/agent/monitoring/emitter.py +++ b/agent/monitoring/emitter.py @@ -67,6 +67,7 @@ class MonitoringEmitter: # Drop oldest to make room — bounded memory, newest-wins. try: self._q.get_nowait() + self._q.task_done() self._dropped += 1 self._q.put_nowait(payload) except Exception: @@ -99,7 +100,11 @@ class MonitoringEmitter: batch.append(self._q.get_nowait()) except queue.Empty: break - self._dispatch(batch) + try: + self._dispatch(batch) + finally: + for _ in batch: + self._q.task_done() def _dispatch(self, batch) -> None: # Fan-out to subscribers (OTLP streamers) — fully fail-isolated. @@ -126,15 +131,23 @@ class MonitoringEmitter: # ── introspection / shutdown (tests, CLI) ─────────────────────────────── def flush(self, timeout: float = 2.0) -> None: - """Block until the queue drains (test/CLI helper, NOT the hot path).""" - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if self._q.empty(): - # give the dispatcher a tick to finish the in-flight batch - time.sleep(0.05) - if self._q.empty(): - return - time.sleep(0.02) + """Wait boundedly for queued and in-flight batches to finish dispatch.""" + if timeout <= 0: + return + + finished = threading.Event() + + def _wait_for_completion() -> None: + self._q.join() + finished.set() + + waiter = threading.Thread( + target=_wait_for_completion, + name="hermes-monitoring-flush", + daemon=True, + ) + waiter.start() + finished.wait(timeout=timeout) def stats(self) -> Dict[str, int]: return { diff --git a/agent/monitoring/gateway_health_export.py b/agent/monitoring/gateway_health_export.py index 04fff71a381..efb7575c692 100644 --- a/agent/monitoring/gateway_health_export.py +++ b/agent/monitoring/gateway_health_export.py @@ -118,12 +118,13 @@ class GatewayHealthExportRuntime: except Exception: pass - # Detach synchronously so no new records are collected during a slow - # exporter shutdown. Network flush/close then runs under one bounded - # daemon-thread deadline and can never delay gateway teardown. + # All producers above are now stopped. Drain queued and in-flight + # events before detaching subscribers so the terminal lifecycle event + # cannot race exporter shutdown. The barrier is bounded and fail-open. try: from agent.monitoring.emitter import get_emitter emitter = get_emitter() + emitter.flush(timeout=1.0) if self.streamer is not None: emitter.unsubscribe(self.streamer) if self.log_streamer is not None: @@ -131,6 +132,8 @@ class GatewayHealthExportRuntime: except Exception: pass + # Network flush/close runs under one bounded daemon-thread deadline and + # can never delay gateway teardown indefinitely. closeables = [ item for item in (self.streamer, self.log_streamer, self.metric_provider) if item is not None diff --git a/gateway/run.py b/gateway/run.py index 07d13b21c2e..600c234537b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -9770,12 +9770,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._update_runtime_status("running", self._exit_reason) else: self._update_runtime_status("stopped", self._exit_reason) - try: - _gh_runtime = getattr(self, "_gateway_health_export_runtime", None) - if _gh_runtime is not None: - _gh_runtime.shutdown() - except Exception: - logger.debug("gateway health OTLP export shutdown failed", exc_info=True) + _shutdown_gateway_health_export(self) logger.info("Gateway stopped (total teardown %.2fs)", _phase_elapsed()) self._stop_task = asyncio.create_task(_stop_impl()) @@ -23552,6 +23547,18 @@ async def _await_thread_exit( return not thread.is_alive() +def _shutdown_gateway_health_export(runner: Any) -> None: + """Idempotently drain and detach Gateway Health OTLP export.""" + runtime = getattr(runner, "_gateway_health_export_runtime", None) + if runtime is None: + return + runner._gateway_health_export_runtime = None + try: + runtime.shutdown() + except Exception: + logger.debug("gateway health OTLP export shutdown failed", exc_info=True) + + async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0) -> bool: """ Start the gateway and run until interrupted. @@ -23999,10 +24006,16 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = logger.debug("MCP tool discovery failed: %s", e) # Start the gateway - success = await runner.start() + try: + success = await runner.start() + except BaseException: + _shutdown_gateway_health_export(runner) + raise if not success: + _shutdown_gateway_health_export(runner) return False if runner.should_exit_cleanly: + _shutdown_gateway_health_export(runner) if runner.exit_reason: logger.error("Gateway exiting cleanly: %s", runner.exit_reason) # A clean exit that carries an explicit exit code (e.g. a fatal @@ -24018,19 +24031,22 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = if not runner._running: # Startup was intentionally aborted by restart/shutdown before entering # running mode; preserve that lifecycle path without starting cron. - await runner.wait_for_shutdown() - if runner.should_exit_with_failure: - if runner.exit_reason: - logger.error("Gateway exiting with failure: %s", runner.exit_reason) - return False try: - from tools.mcp_tool import shutdown_mcp_servers - shutdown_mcp_servers() - except Exception: - pass - if runner.exit_code is not None: - raise SystemExit(runner.exit_code) - return True + await runner.wait_for_shutdown() + if runner.should_exit_with_failure: + if runner.exit_reason: + logger.error("Gateway exiting with failure: %s", runner.exit_reason) + return False + try: + from tools.mcp_tool import shutdown_mcp_servers + shutdown_mcp_servers() + except Exception: + pass + if runner.exit_code is not None: + raise SystemExit(runner.exit_code) + return True + finally: + _shutdown_gateway_health_export(runner) # Start the background cron scheduler via the resolved provider so # scheduled jobs fire automatically. The built-in provider is the diff --git a/scripts/observability/gateway_health_export_probe.py b/scripts/observability/gateway_health_export_probe.py index e1b3bc0548d..e1064088e2a 100644 --- a/scripts/observability/gateway_health_export_probe.py +++ b/scripts/observability/gateway_health_export_probe.py @@ -72,6 +72,7 @@ def main() -> None: logging.getLogger("gateway.platforms.slack").warning("Slack token *** rejected for smoke@example.com") emitter.get_emitter().flush(timeout=2.0) time.sleep(args.wait) + write_runtime_status(gateway_state="stopped", active_agents=0) runtime.shutdown() emitter.get_emitter().flush(timeout=2.0) diff --git a/tests/gateway/test_startup_restart_race.py b/tests/gateway/test_startup_restart_race.py index e303cba1c99..bdf5bbc42a4 100644 --- a/tests/gateway/test_startup_restart_race.py +++ b/tests/gateway/test_startup_restart_race.py @@ -250,16 +250,23 @@ async def test_startup_aborts_after_registered_adapter_restart(tmp_path, monkeyp async def test_start_gateway_does_not_start_cron_after_aborted_startup(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) cron_started = False + export_shutdown_calls = 0 + + class ExportRuntime: + def shutdown(self): + nonlocal export_shutdown_calls + export_shutdown_calls += 1 class AbortedStartupRunner: def __init__(self, config): self.config = config self.adapters = {} self._running = False - self.should_exit_cleanly = False + self.should_exit_cleanly = True self.should_exit_with_failure = False self.exit_reason = None self.exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE + self._gateway_health_export_runtime = ExportRuntime() async def start(self): return True @@ -287,3 +294,4 @@ async def test_start_gateway_does_not_start_cron_after_aborted_startup(tmp_path, assert exc.value.code == GATEWAY_SERVICE_RESTART_EXIT_CODE assert cron_started is False + assert export_shutdown_calls == 1 diff --git a/tests/monitoring/test_emitter.py b/tests/monitoring/test_emitter.py index 98ad7b0389b..07fce136f31 100644 --- a/tests/monitoring/test_emitter.py +++ b/tests/monitoring/test_emitter.py @@ -3,6 +3,7 @@ from __future__ import annotations import time +import threading from agent.monitoring.emitter import MonitoringEmitter from agent.monitoring.events import GatewayHealthEvent @@ -65,6 +66,35 @@ def test_subscriber_failure_is_isolated(): assert len(good) == 1 # the raising subscriber did not break fan-out +def test_flush_waits_for_in_flight_subscriber_delivery(): + em = MonitoringEmitter() + subscriber_started = threading.Event() + release_subscriber = threading.Event() + flush_finished = threading.Event() + + def blocking_subscriber(_batch): + subscriber_started.set() + release_subscriber.wait(timeout=2.0) + + em.subscribe(blocking_subscriber) + em.emit({"event": "gateway_health", "name": "gateway.exit"}) + assert subscriber_started.wait(timeout=1.0) + + flush_thread = threading.Thread( + target=lambda: (em.flush(timeout=1.0), flush_finished.set()), + daemon=True, + ) + flush_thread.start() + + try: + assert not flush_finished.wait(timeout=0.1) + finally: + release_subscriber.set() + + assert flush_finished.wait(timeout=1.0) + em.close() + + def test_unsubscribe_stops_delivery(): em = MonitoringEmitter() seen: list = [] diff --git a/tests/monitoring/test_gateway_health_export.py b/tests/monitoring/test_gateway_health_export.py index df50269b3f5..8e144c9abf9 100644 --- a/tests/monitoring/test_gateway_health_export.py +++ b/tests/monitoring/test_gateway_health_export.py @@ -388,6 +388,41 @@ def test_gateway_health_export_start_is_fail_open_when_otlp_missing(monkeypatch) assert runtime.reason == "otlp_unavailable" +def test_gateway_health_export_shutdown_flushes_before_unsubscribe(monkeypatch): + from agent.monitoring import emitter + from agent.monitoring.gateway_health_export import GatewayHealthExportRuntime + + calls = [] + + class FakeEmitter: + def flush(self, timeout): + calls.append(("flush", timeout)) + + def unsubscribe(self, subscriber): + calls.append(("unsubscribe", subscriber)) + + class Streamer: + def shutdown(self): + calls.append(("shutdown", self)) + + streamer = Streamer() + log_streamer = Streamer() + monkeypatch.setattr(emitter, "get_emitter", lambda: FakeEmitter()) + + runtime = GatewayHealthExportRuntime( + enabled=True, + streamer=streamer, + log_streamer=log_streamer, + ) + runtime.shutdown() + + assert calls[0][0] == "flush" + assert calls[1:3] == [ + ("unsubscribe", streamer), + ("unsubscribe", log_streamer), + ] + + def test_gateway_health_export_streams_only_gateway_events(monkeypatch): from agent.monitoring import gateway_health_export