From 38b39b87efcfef6b8cb60d519e0fee2dc09a8349 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 17 Jul 2026 04:19:08 -0700 Subject: [PATCH] fix(discord): keep recovery ledger I/O off event loop Offload scan bookkeeping and final-delivery writes so SQLite contention cannot stall Discord heartbeats or message delivery. --- plugins/platforms/discord/adapter.py | 54 +++++++++++++++---- .../test_discord_missed_message_backfill.py | 42 +++++++++++++++ 2 files changed, 87 insertions(+), 9 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 628062a8a66f..215ae0d46d18 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -2023,10 +2023,20 @@ class DiscordAdapter(BasePlatformAdapter): self.name, ) return - scan_id = self._record_recovery_scan_start(channels) + scan_id = await asyncio.to_thread( + self._record_recovery_scan_start, + channels, + ) if not channels: logger.info("[%s] Missed-message backfill enabled but no channels configured", self.name) - self._record_recovery_scan_complete(scan_id, status="skipped", scanned=0, missed=0, dispatched=0) + await asyncio.to_thread( + self._record_recovery_scan_complete, + scan_id, + status="skipped", + scanned=0, + missed=0, + dispatched=0, + ) return max_dispatches = self._missed_message_backfill_max_dispatches() @@ -2067,7 +2077,14 @@ class DiscordAdapter(BasePlatformAdapter): raise if dispatched >= max_dispatches: break - self._record_recovery_scan_complete(scan_id, status="success", scanned=scanned, missed=missed, dispatched=dispatched) + await asyncio.to_thread( + self._record_recovery_scan_complete, + scan_id, + status="success", + scanned=scanned, + missed=missed, + dispatched=dispatched, + ) logger.info( "[%s] Missed-message backfill complete: scanned=%d missed=%d dispatched=%d", self.name, @@ -2076,10 +2093,25 @@ class DiscordAdapter(BasePlatformAdapter): dispatched, ) except asyncio.CancelledError: - self._record_recovery_scan_complete(scan_id, status="cancelled", scanned=scanned, missed=missed, dispatched=dispatched) + await asyncio.to_thread( + self._record_recovery_scan_complete, + scan_id, + status="cancelled", + scanned=scanned, + missed=missed, + dispatched=dispatched, + ) raise except Exception as exc: # pragma: no cover - defensive logging - self._record_recovery_scan_complete(scan_id, status="failed", scanned=scanned, missed=missed, dispatched=dispatched, error=str(exc)) + await asyncio.to_thread( + self._record_recovery_scan_complete, + scan_id, + status="failed", + scanned=scanned, + missed=missed, + dispatched=dispatched, + error=str(exc), + ) logger.warning("[%s] Missed-message backfill failed: %s", self.name, exc, exc_info=True) async def _dispatch_recovered_message(self, message: Any) -> bool: @@ -2825,7 +2857,8 @@ class DiscordAdapter(BasePlatformAdapter): # Forum channels reject channel.send() — create a thread post instead. if self._is_forum_parent(channel): result = await self._send_to_forum(channel, content) - self._record_discord_response( + await asyncio.to_thread( + self._record_discord_response, reply_to=reply_to, result=result, content=content, @@ -2900,7 +2933,8 @@ class DiscordAdapter(BasePlatformAdapter): message_id=message_ids[0] if message_ids else None, raw_response={"message_ids": message_ids} ) - self._record_discord_response( + await asyncio.to_thread( + self._record_discord_response, reply_to=reply_to, result=result, content=content, @@ -2911,7 +2945,8 @@ class DiscordAdapter(BasePlatformAdapter): except Exception as e: # pragma: no cover - defensive logging logger.error("[%s] Failed to send Discord message: %s", self.name, e, exc_info=True) result = SendResult(success=False, error=str(e)) - self._record_discord_response( + await asyncio.to_thread( + self._record_discord_response, reply_to=reply_to, result=result, content=content, @@ -3124,7 +3159,8 @@ class DiscordAdapter(BasePlatformAdapter): raise result = SendResult(success=True, message_id=message_id) if finalize: - self._record_discord_response( + await asyncio.to_thread( + self._record_discord_response, reply_to=(metadata or {}).get("reply_to_message_id"), result=result, content=content, diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py index 7d2c5cb9db23..d33e5e38755a 100644 --- a/tests/gateway/test_discord_missed_message_backfill.py +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -659,6 +659,48 @@ async def test_processing_hook_offloads_contended_ledger(adapter, monkeypatch): await processing +@pytest.mark.asyncio +async def test_recovery_scan_offloads_ledger_writes(adapter, monkeypatch): + def slow_scan_start(_channels): + import time + time.sleep(0.1) + return "scan" + + monkeypatch.setattr(adapter, "_record_recovery_scan_start", slow_scan_start) + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: set()) + scan = asyncio.create_task(adapter._run_missed_message_backfill()) + await asyncio.sleep(0.01) + + assert scan.done() is False + await scan + + +@pytest.mark.asyncio +async def test_send_offloads_final_delivery_ledger_write(adapter, monkeypatch): + channel = FakeChannel(channel_id=123) + channel.send = AsyncMock(return_value=SimpleNamespace(id=9011)) + channel.fetch_message = AsyncMock() + adapter._client.get_channel = lambda _channel_id: channel + + def slow_record(**_kwargs): + import time + time.sleep(0.1) + + monkeypatch.setattr(adapter, "_record_discord_response", slow_record) + sending = asyncio.create_task( + adapter.send( + "123", + "done", + reply_to="104", + metadata={"notify": True}, + ) + ) + await asyncio.sleep(0.01) + + assert sending.done() is False + assert (await sending).success is True + + def test_final_delivery_remains_complete_after_processing_hook(adapter): message = make_message(message_id=91) event = MessageEvent(