From eb0cc27201666b7e1f8c6fc9f406e31d718e86e3 Mon Sep 17 00:00:00 2001 From: lEWFkRAD <186512915+lEWFkRAD@users.noreply.github.com> Date: Sat, 4 Jul 2026 10:20:08 -0400 Subject: [PATCH] fix(logging): drive rotating file handlers through an async QueueListener MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On Windows every Hermes process (gateway, serve, TUI/slash workers, MCP servers, CLI commands) writes the shared rotating logs through concurrent-log-handler's cross-process rotation lock. When the emitting thread is an asyncio event loop, a lock wait blocks the loop — stalling it for seconds and dropping WebSocket clients (the 'gateway keeps going down' symptom seen in #58265). Route every file handler through a single QueueListener on a dedicated thread: loggers only enqueue (non-blocking); the listener does the file I/O and rotation-lock wait off the hot path. The QueueHandler funnels via the root logger; per-handler levels and component filters are preserved by respect_handler_level + handler.handle on the listener thread. An atexit hook stops the listener before logging.shutdown closes the file handlers. - _NonFormattingQueueHandler passes the raw record (in-process queue) so target handlers apply their own RedactingFormatter/filters. - flush_log_queue() drains synchronously (shutdown + tests). - rotating_file_handlers() exposes the handlers now behind the listener; tests updated to use it. Extends the #58265 fix: the provider-key warn-storm was one amplifier of this contention; this takes the contention off the event loop entirely. --- hermes_logging.py | 120 ++++++++++++++++++++++++++++- tests/test_hermes_logging.py | 145 +++++++++++++++++++---------------- 2 files changed, 199 insertions(+), 66 deletions(-) diff --git a/hermes_logging.py b/hermes_logging.py index 247a6f62334..d5c58e6aca3 100644 --- a/hermes_logging.py +++ b/hermes_logging.py @@ -27,11 +27,14 @@ Session context: that thread will include ``[session_id]`` for filtering/correlation. """ +import atexit import io import logging import os +import queue import sys import threading +from logging.handlers import QueueHandler, QueueListener from pathlib import Path from typing import Optional, Sequence @@ -543,6 +546,117 @@ class _ManagedRotatingFileHandler(RotatingFileHandler): self._record_stream_stat() +# --------------------------------------------------------------------------- +# Asynchronous file logging — keep the cross-process rotation lock off the loop +# +# The rotating file handlers serialize rollover with a cross-process lock (see +# the module header): when several Hermes processes log to the same file, an +# ``emit`` can block while another process holds that lock. When the emitting +# thread is an asyncio event loop, that block stalls the loop and drops +# WebSocket clients. To keep file I/O off the hot path, every file handler is +# driven by a single ``QueueListener`` on a dedicated thread; loggers only touch +# an in-memory queue (a non-blocking enqueue). +# --------------------------------------------------------------------------- + +_log_queue: "Optional[queue.SimpleQueue]" = None +_queue_listener: Optional[QueueListener] = None +_queued_file_handlers: list = [] +_queue_atexit_registered = False + + +class _NonFormattingQueueHandler(QueueHandler): + """``QueueHandler`` for an in-process queue. + + Stdlib ``prepare()`` formats the record and drops ``args``/``exc_info`` so it + can be pickled to another process. Our queue is in-process, so we skip that + and pass the raw record through — the target file handlers must apply their + own ``RedactingFormatter`` and component filters on the listener thread. + """ + + def prepare(self, record: logging.LogRecord) -> logging.LogRecord: + return record + + +def _stop_queue_listener() -> None: + """Flush and stop the background log listener (idempotent).""" + global _queue_listener + listener, _queue_listener = _queue_listener, None + if listener is not None: + try: + listener.stop() + except Exception: + pass + + +def _register_queued_handler(handler: logging.Handler) -> None: + """Route *handler* through the shared async queue instead of attaching it to + *root* directly, so emitting threads never block on file I/O or the + cross-process rotation lock. The ``QueueListener`` applies each handler's + own level and filters on its worker thread.""" + global _log_queue, _queue_listener, _queue_atexit_registered + if _log_queue is None: + _log_queue = queue.SimpleQueue() + qh = _NonFormattingQueueHandler(_log_queue) + qh._hermes_queue = True # type: ignore[attr-defined] + # Always funnel through the root logger so records from any logger + # (production passes root here; callers may pass a child) reach the + # queue via propagation. + logging.getLogger().addHandler(qh) + _queued_file_handlers.append(handler) + # Rebuild the listener with the full target set. This only happens while + # init_logging() adds handlers (2-3 times, queue empty), so stop() returns + # immediately. + if _queue_listener is not None: + _queue_listener.stop() + _queue_listener = QueueListener( + _log_queue, *_queued_file_handlers, respect_handler_level=True + ) + _queue_listener.start() + if not _queue_atexit_registered: + # Runs before logging.shutdown (registered earlier at import time), so + # the listener stops before its file handlers are closed. + atexit.register(_stop_queue_listener) + _queue_atexit_registered = True + + +def flush_log_queue() -> None: + """Block until all queued records have been written, then resume. + + Draining is done by stopping the listener (which processes every pending + record before joining) and restarting it. Used at shutdown and by tests + that read a log file right after emitting to it.""" + listener = _queue_listener + if listener is not None: + listener.stop() + listener.start() + + +def rotating_file_handlers() -> list: + """Return the live rotating file handlers. + + They are attached to the async ``QueueListener`` rather than the root + logger, so callers/tests must use this instead of scanning + ``logging.getLogger().handlers``.""" + return list(_queued_file_handlers) + + +def _reset_queued_handlers() -> None: + """Tear down the async logging queue + listener (test-isolation helper).""" + global _log_queue + _stop_queue_listener() + root = logging.getLogger() + for h in list(root.handlers): + if getattr(h, "_hermes_queue", False): + root.removeHandler(h) + for h in list(_queued_file_handlers): + try: + h.close() + except Exception: + pass + _queued_file_handlers.clear() + _log_queue = None + + def _add_rotating_handler( logger: logging.Logger, path: Path, @@ -563,7 +677,7 @@ def _add_rotating_handler( for gateway.log). """ resolved = path.resolve() - for existing in logger.handlers: + for existing in _queued_file_handlers: if ( isinstance(existing, RotatingFileHandler) and Path(getattr(existing, "baseFilename", "")).resolve() == resolved @@ -579,7 +693,9 @@ def _add_rotating_handler( handler.setFormatter(formatter) if log_filter is not None: handler.addFilter(log_filter) - logger.addHandler(handler) + # Route through the async queue instead of ``logger.addHandler(handler)`` so + # the rotation-lock wait never runs on the caller's (often event-loop) thread. + _register_queued_handler(handler) def _read_logging_config(): diff --git a/tests/test_hermes_logging.py b/tests/test_hermes_logging.py index ca0b13ff5da..6d05def051f 100644 --- a/tests/test_hermes_logging.py +++ b/tests/test_hermes_logging.py @@ -31,23 +31,20 @@ def _reset_logging_state(): assertions are stable regardless of test ordering. """ hermes_logging._logging_initialized = False + # File handlers now live behind the async QueueListener, not on the root + # logger; tear down any leaked from other xdist tests in this worker. + hermes_logging._reset_queued_handlers() root = logging.getLogger() prev_root_level = root.level root.setLevel(logging.NOTSET) - # Strip ALL RotatingFileHandlers — not just the ones we added — so that - # handlers leaked from other test modules in the same xdist worker don't - # pollute our counts. - pre_existing = [] - for h in list(root.handlers): - if isinstance(h, RotatingFileHandler): - root.removeHandler(h) - h.close() - else: - pre_existing.append(h) + # Snapshot the remaining (non-file) handlers so we can strip whatever the + # test adds. + pre_existing = list(root.handlers) # Ensure the record factory is installed (it's idempotent). hermes_logging._install_session_record_factory() yield - # Restore — remove any handlers added during the test. + # Restore — tear down async file logging + remove handlers added by the test. + hermes_logging._reset_queued_handlers() for h in list(root.handlers): if h not in pre_existing: root.removeHandler(h) @@ -81,7 +78,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -93,7 +90,7 @@ class TestSetupLogging: root = logging.getLogger() error_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "errors.log" in getattr(h, "baseFilename", "") ] @@ -106,7 +103,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -120,7 +117,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -131,7 +128,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -144,7 +141,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -165,8 +162,7 @@ class TestSetupLogging: test_logger.info("test message for agent.log") # Flush handlers - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() agent_log = hermes_home / "logs" / "agent.log" assert agent_log.exists() @@ -179,8 +175,7 @@ class TestSetupLogging: test_logger = logging.getLogger("test_hermes_logging.warning_test") test_logger.warning("this is a warning") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() agent_log = hermes_home / "logs" / "agent.log" errors_log = hermes_home / "logs" / "errors.log" @@ -193,8 +188,7 @@ class TestSetupLogging: test_logger = logging.getLogger("test_hermes_logging.info_test") test_logger.info("info only message") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() errors_log = hermes_home / "logs" / "errors.log" if errors_log.exists(): @@ -210,7 +204,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -228,7 +222,7 @@ class TestSetupLogging: root = logging.getLogger() agent_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "agent.log" in getattr(h, "baseFilename", "") ] @@ -254,7 +248,7 @@ class TestGatewayMode: root = logging.getLogger() gw_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "gateway.log" in getattr(h, "baseFilename", "") ] @@ -265,7 +259,7 @@ class TestGatewayMode: root = logging.getLogger() gw_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "gateway.log" in getattr(h, "baseFilename", "") ] @@ -278,7 +272,7 @@ class TestGatewayMode: root = logging.getLogger() gw_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "gateway.log" in getattr(h, "baseFilename", "") ] @@ -286,8 +280,7 @@ class TestGatewayMode: logging.getLogger("gateway.run").info("gateway connected after cli init") - for h in root.handlers: - h.flush() + hermes_logging.flush_log_queue() gw_log = hermes_home / "logs" / "gateway.log" assert gw_log.exists() @@ -301,7 +294,7 @@ class TestGatewayMode: root = logging.getLogger() gw_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "gateway.log" in getattr(h, "baseFilename", "") ] @@ -314,8 +307,7 @@ class TestGatewayMode: gw_logger = logging.getLogger("plugins.platforms.telegram.adapter") gw_logger.info("telegram connected") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() gw_log = hermes_home / "logs" / "gateway.log" assert gw_log.exists() @@ -331,8 +323,7 @@ class TestGatewayMode: agent_logger = logging.getLogger("agent.context_compressor") agent_logger.info("compressing context") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() gw_log = hermes_home / "logs" / "gateway.log" if gw_log.exists(): @@ -356,8 +347,7 @@ class TestGatewayMode: gw_logger.info("gateway msg") file_logger.info("file msg") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() agent_log = hermes_home / "logs" / "agent.log" content = agent_log.read_text() @@ -373,7 +363,7 @@ class TestGuiMode: root = logging.getLogger() gui_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "gui.log" in getattr(h, "baseFilename", "") ] @@ -385,7 +375,7 @@ class TestGuiMode: root = logging.getLogger() gui_handlers = [ - h for h in root.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) and "gui.log" in getattr(h, "baseFilename", "") ] @@ -398,8 +388,7 @@ class TestGuiMode: logging.getLogger("tui_gateway.ws").info("ws connected") logging.getLogger("gateway.run").info("gateway event") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() gui_log = hermes_home / "logs" / "gui.log" assert gui_log.exists() @@ -420,8 +409,7 @@ class TestSessionContext: test_logger = logging.getLogger("test.session_tag") test_logger.info("tagged message") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() agent_log = hermes_home / "logs" / "agent.log" content = agent_log.read_text() @@ -436,8 +424,7 @@ class TestSessionContext: test_logger = logging.getLogger("test.no_session") test_logger.info("untagged message") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() agent_log = hermes_home / "logs" / "agent.log" content = agent_log.read_text() @@ -457,8 +444,7 @@ class TestSessionContext: test_logger = logging.getLogger("test.cleared") test_logger.info("after clear") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() agent_log = hermes_home / "logs" / "agent.log" content = agent_log.read_text() @@ -473,14 +459,12 @@ class TestSessionContext: def thread_a(): hermes_logging.set_session_context("thread_a_session") logging.getLogger("test.thread_a").info("from thread A") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() def thread_b(): hermes_logging.set_session_context("thread_b_session") logging.getLogger("test.thread_b").info("from thread B") - for h in logging.getLogger().handlers: - h.flush() + hermes_logging.flush_log_queue() ta = threading.Thread(target=thread_a) tb = threading.Thread(target=thread_b) @@ -701,7 +685,7 @@ class TestAddRotatingHandler: ) rotating_handlers = [ - h for h in logger.handlers + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) ] assert len(rotating_handlers) == 1 @@ -725,7 +709,7 @@ class TestAddRotatingHandler: log_filter=component_filter, ) - handlers = [h for h in logger.handlers if isinstance(h, RotatingFileHandler)] + handlers = [h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler)] assert len(handlers) == 1 assert component_filter in handlers[0].filters # Clean up @@ -746,7 +730,7 @@ class TestAddRotatingHandler: formatter=formatter, ) - handlers = [h for h in logger.handlers if isinstance(h, RotatingFileHandler)] + handlers = [h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler)] assert len(handlers) == 1 # No _SessionFilter on the handler — record factory handles it assert len(handlers[0].filters) == 0 @@ -754,7 +738,7 @@ class TestAddRotatingHandler: # But session_tag still works (via record factory) hermes_logging.set_session_context("factory_test") logger.info("test msg") - handlers[0].flush() + hermes_logging.flush_log_queue() content = log_path.read_text() assert "[factory_test]" in content @@ -802,10 +786,10 @@ class TestAddRotatingHandler: formatter=formatter, ) handler = next( - h for h in logger.handlers if isinstance(h, RotatingFileHandler) + h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler) ) logger.info("a" * 256) - handler.flush() + hermes_logging.flush_log_queue() finally: os.umask(old_umask) @@ -953,7 +937,7 @@ class TestExternalRotationRecovery: # Match the record factory that hermes_logging installs at import time. record.session_tag = "" handler.emit(record) - handler.flush() + hermes_logging.flush_log_queue() def test_recovers_after_external_rename(self, tmp_path): """logrotate-style external rename: ``mv gateway.log gateway.log.1``. @@ -1069,9 +1053,7 @@ class TestExternalRotationRecovery: rotated = hermes_home / "logs" / "gateway.log.1" logging.getLogger("gateway.run").info("line BEFORE rotation") - for h in logging.getLogger().handlers: - try: h.flush() - except Exception: pass + hermes_logging.flush_log_queue() assert "BEFORE rotation" in gw_path.read_text() # External actor renames the file out from under us. @@ -1084,9 +1066,7 @@ class TestExternalRotationRecovery: hermes_logging.setup_logging(hermes_home=hermes_home, mode="gateway") logging.getLogger("gateway.run").info("line AFTER rotation") - for h in logging.getLogger().handlers: - try: h.flush() - except Exception: pass + hermes_logging.flush_log_queue() # The new record must reach the live gateway.log, not the rotated # backup. Allen's logs had everything past the rotation point @@ -1165,3 +1145,40 @@ class TestSafeStderr: logger.info("Session hygiene: 400 messages — auto-compressing") finally: logger.removeHandler(handler) + + +class TestAsyncQueueLogging: + """File logging runs through a QueueListener so emits never block on the + cross-process rotation lock (Windows event-loop-stall fix).""" + + def test_file_handlers_not_on_root(self, hermes_home): + hermes_logging.setup_logging(hermes_home=hermes_home) + root = logging.getLogger() + # Rotating file handlers live on the async listener, never on root. + assert not any(isinstance(h, RotatingFileHandler) for h in root.handlers) + # Exactly one queue handler funnels records to the listener. + queue_handlers = [ + h for h in root.handlers if getattr(h, "_hermes_queue", False) + ] + assert len(queue_handlers) == 1 + # The real file handlers are discoverable via the accessor. + assert any( + "agent.log" in getattr(h, "baseFilename", "") + for h in hermes_logging.rotating_file_handlers() + ) + + def test_records_reach_file_through_queue(self, hermes_home): + hermes_logging.setup_logging(hermes_home=hermes_home) + logging.getLogger("test_async.queue").info("through the queue") + hermes_logging.flush_log_queue() + agent_log = hermes_home / "logs" / "agent.log" + assert "through the queue" in agent_log.read_text() + + def test_queue_preserves_per_handler_levels(self, hermes_home): + hermes_logging.setup_logging(hermes_home=hermes_home) + logging.getLogger("test_async.levels").info("info-level line") + hermes_logging.flush_log_queue() + errors_log = hermes_home / "logs" / "errors.log" + # INFO must not reach the WARNING+ errors.log even through the queue. + if errors_log.exists(): + assert "info-level line" not in errors_log.read_text()