From 498800a22ef606323ffe14b515d77f0b87999232 Mon Sep 17 00:00:00 2001 From: ibaldr89 <163388887+ibaldr89@users.noreply.github.com> Date: Sun, 19 Jul 2026 12:31:15 +0000 Subject: [PATCH] fix(gateway): offload delivery ledger I/O --- gateway/platforms/base.py | 12 +++-- gateway/run.py | 7 +-- tests/gateway/test_delivery_ledger.py | 47 ++++++++++++++++ .../gateway/test_delivery_ledger_producer.py | 53 +++++++++++++++++++ 4 files changed, 111 insertions(+), 8 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 36741ae6d2..c42b916073 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6066,13 +6066,14 @@ class BasePlatformAdapter(ABC): record_obligation, ) - if ledger_enabled(): + if await asyncio.to_thread(ledger_enabled): _obligation_id = compute_obligation_id( session_key, str(getattr(event, "message_id", "") or ""), text_content, ) - record_obligation( + await asyncio.to_thread( + record_obligation, obligation_id=_obligation_id, session_key=session_key, platform=str( @@ -6083,7 +6084,7 @@ class BasePlatformAdapter(ABC): thread_id=getattr(event.source, "thread_id", None), content=text_content, ) - mark_attempting(_obligation_id) + await asyncio.to_thread(mark_attempting, _obligation_id) except Exception: logger.debug("delivery ledger record failed", exc_info=True) _obligation_id = None @@ -6102,9 +6103,10 @@ class BasePlatformAdapter(ABC): ) if getattr(result, "success", False): - mark_delivered(_obligation_id) + await asyncio.to_thread(mark_delivered, _obligation_id) else: - mark_failed( + await asyncio.to_thread( + mark_failed, _obligation_id, str(getattr(result, "error", "") or ""), ) diff --git a/gateway/run.py b/gateway/run.py index f170f55dda..8c162f7dc9 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10231,7 +10231,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew sweep_recoverable, ) - if not ledger_enabled(): + if not await asyncio.to_thread(ledger_enabled): return 0 # Only claim rows we can actually send this boot: self.adapters # holds a platform only after its connect() succeeded, and each @@ -10283,7 +10283,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew result = None try: if result is not None and getattr(result, "success", False): - mark_delivered(row["obligation_id"]) + await asyncio.to_thread(mark_delivered, row["obligation_id"]) redelivered += 1 logger.info( "Redelivered recovered final response to %s:%s " @@ -10292,7 +10292,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew row["obligation_id"], row["attempts"], ) else: - mark_failed( + await asyncio.to_thread( + mark_failed, row["obligation_id"], str(getattr(result, "error", "") or "send failed"), ) diff --git a/tests/gateway/test_delivery_ledger.py b/tests/gateway/test_delivery_ledger.py index cec3b8339e..ac46042c88 100644 --- a/tests/gateway/test_delivery_ledger.py +++ b/tests/gateway/test_delivery_ledger.py @@ -10,6 +10,7 @@ id stability, and the startup redelivery sweep's contract: """ import time +import threading from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -50,6 +51,30 @@ def _row(oid): } +def _blocking_probe(): + """Return a blocking ledger call and an event-loop progress witness.""" + ledger_started = threading.Event() + event_loop_progressed = threading.Event() + blocked_event_loop = [] + + def _slow_ledger_call(*args, **kwargs): + ledger_started.set() + if not event_loop_progressed.wait(timeout=0.5): + blocked_event_loop.append(True) + + async def _event_loop_witness(): + import asyncio + + deadline = asyncio.get_running_loop().time() + 1 + while not ledger_started.is_set(): + if asyncio.get_running_loop().time() >= deadline: + raise AssertionError("ledger call never started") + await asyncio.sleep(0) + event_loop_progressed.set() + + return _slow_ledger_call, _event_loop_witness, blocked_event_loop + + def _orphan(oid): """Make the row look like it belongs to a dead process.""" with dl._connect() as conn: @@ -232,6 +257,28 @@ class TestUnconnectedPlatformKeepsItsBudget: runner._async_session_store = _store return runner + @pytest.mark.parametrize( + ("send_success", "ledger_method"), + [(True, "mark_delivered"), (False, "mark_failed")], + ) + @pytest.mark.asyncio + async def test_slow_state_update_does_not_block_event_loop( + self, send_success, ledger_method + ): + import asyncio + + _record() + _orphan("ob-1") + runner = self._runner(self._adapter(success=send_success)) + slow_update, event_loop_witness, blocked_event_loop = _blocking_probe() + + with patch.object(dl, ledger_method, side_effect=slow_update): + await asyncio.gather( + runner._redeliver_pending_obligations(), event_loop_witness() + ) + + assert blocked_event_loop == [] + @pytest.mark.asyncio async def test_row_survives_boots_where_its_platform_is_down(self): _record(platform="slack") diff --git a/tests/gateway/test_delivery_ledger_producer.py b/tests/gateway/test_delivery_ledger_producer.py index 8b99ea1a7f..c02f3d345a 100644 --- a/tests/gateway/test_delivery_ledger_producer.py +++ b/tests/gateway/test_delivery_ledger_producer.py @@ -8,6 +8,7 @@ block the send. """ import asyncio +import threading from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -65,6 +66,28 @@ def _rows(): ).fetchall() +def _blocking_probe(): + """Return a blocking ledger call and an event-loop progress witness.""" + ledger_started = threading.Event() + event_loop_progressed = threading.Event() + blocked_event_loop = [] + + def _slow_ledger_call(*args, **kwargs): + ledger_started.set() + if not event_loop_progressed.wait(timeout=0.5): + blocked_event_loop.append(True) + + async def _event_loop_witness(): + deadline = asyncio.get_running_loop().time() + 1 + while not ledger_started.is_set(): + if asyncio.get_running_loop().time() >= deadline: + raise AssertionError("ledger call never started") + await asyncio.sleep(0) + event_loop_progressed.set() + + return _slow_ledger_call, _event_loop_witness, blocked_event_loop + + async def _run(adapter, event, response="final answer"): adapter._message_handler = AsyncMock(return_value=response) session_key = "agent:main:slack:channel:C1" @@ -97,6 +120,36 @@ class TestProducerHook: assert rows[0][1] == "failed" + @pytest.mark.asyncio + async def test_slow_ledger_record_does_not_block_event_loop(self): + adapter = _Adapter() + slow_record, event_loop_witness, blocked_event_loop = _blocking_probe() + + with patch( + "gateway.delivery_ledger.record_obligation", + side_effect=slow_record, + ), patch("gateway.delivery_ledger.mark_attempting"): + await asyncio.gather(_run(adapter, _event()), event_loop_witness()) + + assert blocked_event_loop == [] + assert adapter.sent == ["final answer"] + + @pytest.mark.asyncio + async def test_slow_ledger_update_does_not_block_event_loop(self): + adapter = _Adapter() + slow_delivered, event_loop_witness, blocked_event_loop = _blocking_probe() + + with patch("gateway.delivery_ledger.record_obligation"), patch( + "gateway.delivery_ledger.mark_attempting" + ), patch( + "gateway.delivery_ledger.mark_delivered", + side_effect=slow_delivered, + ): + await asyncio.gather(_run(adapter, _event()), event_loop_witness()) + + assert blocked_event_loop == [] + assert adapter.sent == ["final answer"] + @pytest.mark.asyncio async def test_crash_between_attempting_and_ack_is_recoverable(self): """The core scenario (#58818): process dies mid-send. The row must