diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index c42b9160737d5..36741ae6d229d 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6066,14 +6066,13 @@ class BasePlatformAdapter(ABC): record_obligation, ) - if await asyncio.to_thread(ledger_enabled): + if ledger_enabled(): _obligation_id = compute_obligation_id( session_key, str(getattr(event, "message_id", "") or ""), text_content, ) - await asyncio.to_thread( - record_obligation, + record_obligation( obligation_id=_obligation_id, session_key=session_key, platform=str( @@ -6084,7 +6083,7 @@ class BasePlatformAdapter(ABC): thread_id=getattr(event.source, "thread_id", None), content=text_content, ) - await asyncio.to_thread(mark_attempting, _obligation_id) + mark_attempting(_obligation_id) except Exception: logger.debug("delivery ledger record failed", exc_info=True) _obligation_id = None @@ -6103,10 +6102,9 @@ class BasePlatformAdapter(ABC): ) if getattr(result, "success", False): - await asyncio.to_thread(mark_delivered, _obligation_id) + mark_delivered(_obligation_id) else: - await asyncio.to_thread( - mark_failed, + mark_failed( _obligation_id, str(getattr(result, "error", "") or ""), ) diff --git a/gateway/run.py b/gateway/run.py index 8c162f7dc9a46..f170f55ddab00 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10231,7 +10231,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew sweep_recoverable, ) - if not await asyncio.to_thread(ledger_enabled): + if not 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): - await asyncio.to_thread(mark_delivered, row["obligation_id"]) + mark_delivered(row["obligation_id"]) redelivered += 1 logger.info( "Redelivered recovered final response to %s:%s " @@ -10292,8 +10292,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew row["obligation_id"], row["attempts"], ) else: - await asyncio.to_thread( - mark_failed, + 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 ac46042c887fd..86584c30ff43f 100644 --- a/tests/gateway/test_delivery_ledger.py +++ b/tests/gateway/test_delivery_ledger.py @@ -196,6 +196,28 @@ class TestGatewayRedeliverySweep: assert sent["content"].startswith(dl.RECOVERED_MARKER) assert sent["content"].endswith("the final answer") + @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 == [] + class TestAttemptsOnlySpentOnRealSends: """``attempts`` is the redelivery budget — it must buy a send. @@ -257,28 +279,6 @@ 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")