From 94c2a4016bed3309bb33bf60a1c20d707bbfc4a5 Mon Sep 17 00:00:00 2001 From: kenyonxu Date: Fri, 10 Jul 2026 11:22:51 +0800 Subject: [PATCH] fix(gateway): offload both blocking sources in compression-in-flight check (#5) The sync _session_has_compression_in_flight sat on the message hot path and blocked the event loop twice: under session_store._lock during _ensure_loaded_locked (JSON read) and via db.get_compression_lock_holder (SQLite SELECT). Async-ify the method and offload both sources via asyncio.to_thread; await the call site in _handle_active_session_busy_message. --- .claude/settings.json | 15 ++++ gateway/run.py | 35 ++++++--- .../test_compression_in_flight_check.py | 73 +++++++++++++++++++ ...st_compression_interrupt_demotion_56391.py | 15 ++-- 4 files changed, 122 insertions(+), 16 deletions(-) create mode 100644 .claude/settings.json create mode 100644 tests/gateway/test_compression_in_flight_check.py diff --git a/.claude/settings.json b/.claude/settings.json new file mode 100644 index 00000000000..ddee3bfa18f --- /dev/null +++ b/.claude/settings.json @@ -0,0 +1,15 @@ +{ + "permissions": { + "allow": [ + "Bash(systemctl --user status hermes-gateway.service)", + "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"6:00\")", + "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"08:31\")", + "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"08:30\")", + "Bash(journalctl --user -u hermes-gateway.service --no-pager --since \"08:00\")", + "Bash(MAIN_PID=3163287 *)", + "Read(//proc/3163287/**)", + "Bash(sudo -S timeout 3 strace -p 3163287 -e trace=futex -f -o /tmp/strace_main.txt)", + "Read(//tmp/**)" + ] + } +} diff --git a/gateway/run.py b/gateway/run.py index 714949e030e..e6a2c87dcc4 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5174,7 +5174,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: return False - def _session_has_compression_in_flight(self, session_key: str) -> bool: + async def _session_has_compression_in_flight(self, session_key: str) -> bool: """Return True when a compression lock is held for this session's id. Context compression is interrupt-protected (#23975) but gateway @@ -5182,28 +5182,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew the pre-rotation parent while compression is mid-flight, producing orphaned compression siblings (#56391). Callers demote interrupt to queue when this returns True. + + Both blocking sources — the ``session_store`` lock + JSON load, and the + SQLite ``get_compression_lock_holder`` SELECT — are offloaded to a + worker thread so a large state.db never freezes the event loop (#5). """ session_store = getattr(self, "session_store", None) if not session_key or session_store is None: return False try: - with session_store._lock: # noqa: SLF001 — snapshot entry under lock - session_store._ensure_loaded_locked() # noqa: SLF001 - entry = session_store._entries.get(session_key) # noqa: SLF001 - session_id = getattr(entry, "session_id", None) if entry is not None else None - if not session_id: - return False + session_id = await asyncio.to_thread( + self._lookup_session_id_under_store_lock, session_store, session_key + ) except Exception: return False + if not session_id: + return False session_db = getattr(self, "_session_db", None) if session_db is None: return False - db = getattr(session_db, "_db", session_db) + raw_db = getattr(session_db, "_db", session_db) try: - return bool(db.get_compression_lock_holder(str(session_id))) + holder = await asyncio.to_thread( + raw_db.get_compression_lock_holder, str(session_id) + ) + return bool(holder) except Exception: return False + @staticmethod + def _lookup_session_id_under_store_lock(session_store, session_key: str): + """Sync helper run in the thread pool: read session_id under the store lock.""" + # noqa: SLF001 — intentional private access; runs off the event loop. + with session_store._lock: # noqa: SLF001 + session_store._ensure_loaded_locked() # noqa: SLF001 + entry = session_store._entries.get(session_key) # noqa: SLF001 + return getattr(entry, "session_id", None) if entry is not None else None + # Hard cap on per-session pending follow-ups for busy_input_mode=queue # (and the draining/steer-fallback/subagent-demotion paths that share # this entry point). Without a cap, a stuck agent + a rapid-fire user @@ -5426,7 +5441,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew effective_mode = "queue" demoted_for_compression = ( effective_mode == "interrupt" - and self._session_has_compression_in_flight(session_key) + and await self._session_has_compression_in_flight(session_key) ) if demoted_for_compression: logger.info( diff --git a/tests/gateway/test_compression_in_flight_check.py b/tests/gateway/test_compression_in_flight_check.py new file mode 100644 index 00000000000..0a33c4bcb7f --- /dev/null +++ b/tests/gateway/test_compression_in_flight_check.py @@ -0,0 +1,73 @@ +"""#5 regression: _session_has_compression_in_flight must offload both blocking sources to thread pool.""" +import inspect +import threading +from unittest.mock import MagicMock + +import pytest + + +def _make_runner(holder_value=None, record_thread=False, thread_sink=None): + from gateway.run import GatewayRunner + runner = GatewayRunner.__new__(GatewayRunner) + + store = MagicMock() + store._lock = threading.Lock() + store._loaded = True + store._entries = {"k": MagicMock(session_id="sess-123")} + store._ensure_loaded_locked = lambda: None + runner.session_store = store + + raw_db = MagicMock() + if record_thread and thread_sink is not None: + def _holder(sid): + thread_sink["thread"] = threading.get_ident() + return holder_value + raw_db.get_compression_lock_holder = _holder + else: + raw_db.get_compression_lock_holder = MagicMock(return_value=holder_value) + + session_db = MagicMock() + session_db._db = raw_db + runner._session_db = session_db + return runner + + +def test_method_is_coroutine(): + from gateway.run import GatewayRunner + assert inspect.iscoroutinefunction( + GatewayRunner._session_has_compression_in_flight + ), "#5: method must be async, blocking calls offloaded" + + +@pytest.mark.asyncio +async def test_returns_true_when_lock_held(): + runner = _make_runner(holder_value="agent-1") + assert await runner._session_has_compression_in_flight("k") is True + + +@pytest.mark.asyncio +async def test_returns_false_when_no_lock(): + runner = _make_runner(holder_value=None) + assert await runner._session_has_compression_in_flight("k") is False + + +@pytest.mark.asyncio +async def test_returns_false_when_no_session_store(): + from gateway.run import GatewayRunner + runner = GatewayRunner.__new__(GatewayRunner) + runner.session_store = None + runner._session_db = MagicMock() + assert await runner._session_has_compression_in_flight("k") is False + + +@pytest.mark.asyncio +async def test_db_call_runs_off_event_loop(): + """Regression core: get_compression_lock_holder MUST execute in non-event-loop thread.""" + sink = {} + runner = _make_runner(holder_value="agent-1", record_thread=True, thread_sink=sink) + loop_thread = threading.get_ident() + await runner._session_has_compression_in_flight("k") + assert "thread" in sink, "underlying db.get_compression_lock_holder was not called" + assert sink["thread"] != loop_thread, ( + "DB call still on event loop thread — #5 NOT fixed (to_thread not applied)" + ) diff --git a/tests/gateway/test_compression_interrupt_demotion_56391.py b/tests/gateway/test_compression_interrupt_demotion_56391.py index 53b430d5f77..cf9f4287c1f 100644 --- a/tests/gateway/test_compression_interrupt_demotion_56391.py +++ b/tests/gateway/test_compression_interrupt_demotion_56391.py @@ -105,22 +105,25 @@ def _make_parent_no_subagents() -> MagicMock: class TestSessionHasCompressionInFlight: - def test_returns_false_without_session_store(self) -> None: + @pytest.mark.asyncio + async def test_returns_false_without_session_store(self) -> None: runner = _make_runner() runner.session_store = None - assert runner._session_has_compression_in_flight("sk") is False + assert await runner._session_has_compression_in_flight("sk") is False - def test_returns_true_when_lock_held(self) -> None: + @pytest.mark.asyncio + async def test_returns_true_when_lock_held(self) -> None: runner = _make_runner() sk = build_session_key(_make_event().source) runner._session_db._db.get_compression_lock_holder.return_value = "holder-1" - assert runner._session_has_compression_in_flight(sk) is True + assert await runner._session_has_compression_in_flight(sk) is True - def test_returns_false_when_lock_free(self) -> None: + @pytest.mark.asyncio + async def test_returns_false_when_lock_free(self) -> None: runner = _make_runner() sk = build_session_key(_make_event().source) runner._session_db._db.get_compression_lock_holder.return_value = None - assert runner._session_has_compression_in_flight(sk) is False + assert await runner._session_has_compression_in_flight(sk) is False class TestBusyHandlerDemotesInterruptForCompression: