fix(channels): offload blocking filesystem IO in Discord channel (#3927)

DiscordChannel ran synchronous filesystem IO on the event loop: thread-mapping
persistence/restore and outbound attachment reads. Offload all of it via
asyncio.to_thread:
- start() -> _load_active_threads (restore mappings on startup)
- _on_message -> _persist_thread_mappings (flush mappings to disk)
- send_file -> _read_attachment_bytes (read bytes; handed to discord.File as
  an in-memory BytesIO buffer)

Thread-mapping state is split to avoid a race surfaced in review (#3927):
_record_thread_mapping updates the in-memory _active_threads dict and
_active_thread_ids set synchronously on the event loop, so a follow-up message
in a newly created thread is recognized immediately — before the offloaded
persistence write completes. Deferring that update into the worker thread
opened a window where _on_message's membership check misclassified the message
as orphaned and created a duplicate thread.

__init__ only computes paths, so construction stays IO-free.

Blockbuster regression tests cover the IO-free constructor, the
record-then-persist split (memory visible before persistence), discard of a
replaced thread id, and the load path.

Co-authored-by: Claude <noreply@anthropic.com>

backend/app/channels/discord.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import io
 import json
 import logging
 import threading
@@ -108,7 +109,7 @@ class DiscordChannel(Channel):
 
         self._thread = threading.Thread(target=self._run_client, daemon=True)
         self._thread.start()
-        self._load_active_threads()
+        await asyncio.to_thread(self._load_active_threads)
         logger.info("Discord channel started")
 
     def _load_active_threads(self) -> None:
@@ -129,23 +130,45 @@ class DiscordChannel(Channel):
             except Exception:
                 logger.exception("[Discord] failed to load thread mappings")
 
-    def _save_thread(self, channel_id: str, thread_id: str) -> None:
-        """Persist a Discord thread mapping to the dedicated JSON file."""
+    def _record_thread_mapping(self, channel_id: str, thread_id: str) -> None:
+        """Synchronously update the in-memory channel->thread mapping and its reverse-lookup set.
+
+        Runs on the event loop (no IO, no await) so a follow-up message in the
+        newly created thread is recognized immediately, before the offloaded
+        persistence write completes. Deferring this update into the worker
+        thread opened a window where ``_active_thread_ids`` had not yet been
+        updated and an inbound message was misclassified as orphaned (see the
+        #3927 review). Persistence is handled separately by
+        ``_persist_thread_mappings``.
+        """
+        old_id = self._active_threads.get(channel_id)
+        self._active_threads[channel_id] = thread_id
+        if old_id:
+            self._active_thread_ids.discard(old_id)
+        self._active_thread_ids.add(thread_id)
+
+    def _persist_thread_mappings(self) -> None:
+        """Flush the current in-memory thread mappings to disk.
+
+        Intended for ``asyncio.to_thread``: this is pure filesystem IO. The
+        in-memory state is updated synchronously by ``_record_thread_mapping``,
+        so persistence latency never delays visibility of a new mapping to
+        inbound-message handling. The mapping is snapshotted under the store
+        lock so a concurrent record cannot mutate the dict mid-serialization.
+        """
         with self._thread_store_lock:
             try:
-                data: dict[str, str] = {}
-                if self._thread_store_path.exists():
-                    data = json.loads(self._thread_store_path.read_text())
-                old_id = data.get(channel_id)
-                data[channel_id] = thread_id
-                # Update reverse-lookup set
-                if old_id:
-                    self._active_thread_ids.discard(old_id)
-                self._active_thread_ids.add(thread_id)
+                snapshot = dict(self._active_threads)
                 self._thread_store_path.parent.mkdir(parents=True, exist_ok=True)
-                self._thread_store_path.write_text(json.dumps(data, indent=2))
+                self._thread_store_path.write_text(json.dumps(snapshot, indent=2))
             except Exception:
-                logger.exception("[Discord] failed to save thread mapping for channel %s", channel_id)
+                logger.exception("[Discord] failed to persist thread mappings")
+
+    @staticmethod
+    def _read_attachment_bytes(path: str) -> bytes:
+        """Read an attachment file synchronously (intended for ``asyncio.to_thread``)."""
+        with open(path, "rb") as fp:
+            return fp.read()
 
     async def stop(self) -> None:
         self._running = False
@@ -204,14 +227,15 @@ class DiscordChannel(Channel):
             return False
 
         try:
-            # Keep the file handle open only for the duration of the upload: discord.py
-            # reads ``fp`` while ``target.send`` runs on ``_discord_loop``; once that
-            # future resolves the bytes are consumed, so closing here is safe and avoids
-            # leaking the handle on both the success and failure paths.
-            with open(str(attachment.actual_path), "rb") as fp:
-                file = self._discord_module.File(fp, filename=attachment.filename)
-                send_future = asyncio.run_coroutine_threadsafe(target.send(file=file), self._discord_loop)
-                await asyncio.wrap_future(send_future)
+            # Read the attachment off the event loop (open + read are blocking IO),
+            # then hand discord.py an in-memory buffer. The bytes are consumed while
+            # ``target.send`` runs on ``_discord_loop``; once that future resolves the
+            # buffer can be reclaimed, so this avoids leaking a file handle on both the
+            # success and failure paths.
+            data = await asyncio.to_thread(self._read_attachment_bytes, str(attachment.actual_path))
+            file = self._discord_module.File(io.BytesIO(data), filename=attachment.filename)
+            send_future = asyncio.run_coroutine_threadsafe(target.send(file=file), self._discord_loop)
+            await asyncio.wrap_future(send_future)
             logger.info("[Discord] file uploaded: %s", attachment.filename)
             return True
         except Exception:
@@ -357,8 +381,8 @@ class DiscordChannel(Channel):
                 thread_obj = await self._create_thread(message)
                 if thread_obj is not None:
                     target_thread_id = str(thread_obj.id)
-                    self._active_threads[channel_id] = target_thread_id
-                    self._save_thread(channel_id, target_thread_id)
+                    self._record_thread_mapping(channel_id, target_thread_id)
+                    await asyncio.to_thread(self._persist_thread_mappings)
                     thread_id = target_thread_id
                     chat_id = channel_id
                     typing_target = thread_obj
@@ -384,8 +408,8 @@ class DiscordChannel(Channel):
             thread_obj = await self._create_thread(message)
             if thread_obj is not None:
                 target_thread_id = str(thread_obj.id)
-                self._active_threads[channel_id] = target_thread_id
-                self._save_thread(channel_id, target_thread_id)
+                self._record_thread_mapping(channel_id, target_thread_id)
+                await asyncio.to_thread(self._persist_thread_mappings)
                 thread_id = target_thread_id
                 chat_id = channel_id
                 typing_target = thread_obj  # Type into the new thread
@@ -407,8 +431,8 @@ class DiscordChannel(Channel):
                 typing_target = message.channel  # Type into the channel
             else:
                 target_thread_id = str(thread_obj.id)
-                self._active_threads[channel_id] = target_thread_id
-                self._save_thread(channel_id, target_thread_id)
+                self._record_thread_mapping(channel_id, target_thread_id)
+                await asyncio.to_thread(self._persist_thread_mappings)
                 thread_id = target_thread_id
                 chat_id = channel_id
                 typing_target = thread_obj  # Type into the new thread

backend/tests/blocking_io/test_discord_channel_state.py

@@ -0,0 +1,122 @@
+"""Regression anchor: Discord channel filesystem IO must not block the event loop.
+
+``DiscordChannel`` persists channel->thread mappings to a dedicated JSON file
+(``discord_threads.json``) via synchronous filesystem calls, and reads outbound
+attachments from disk before uploading. The async entry points offload all of
+that IO via ``asyncio.to_thread``:
+
+- ``start()`` -> ``_load_active_threads`` (``exists`` + ``read_text`` to restore
+  mappings on startup)
+- ``_on_message`` -> ``_record_thread_mapping`` updates the in-memory mapping
+  synchronously (no IO), then ``_persist_thread_mappings`` flushes it to disk
+  (``mkdir`` + ``write_text``) via ``asyncio.to_thread``. Memory is updated
+  before persistence so a follow-up message in the new thread is recognized
+  immediately — see the race noted in the #3927 review.
+- ``send_file`` -> ``_read_attachment_bytes`` (``open`` + ``read`` for outbound
+  attachments; bytes are handed to ``discord.File`` as an in-memory buffer)
+
+If any of it regresses back onto the event loop, the strict Blockbuster gate
+raises ``BlockingError`` and these tests fail.
+
+``__init__`` only computes paths (``Path.home()`` / ``store._path.parent``), so
+construction is IO-free; ``ChannelService._start_channel()`` instantiates the
+channel directly on the async path without blocking. The IO-bearing helpers
+(``_persist_thread_mappings`` / ``_load_active_threads``) are wrapped in
+``asyncio.to_thread``; ``_record_thread_mapping`` is pure memory and runs
+inline on the event loop so its update is visible before persistence completes.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+import pytest
+
+from app.channels.discord import DiscordChannel
+from app.channels.message_bus import MessageBus
+
+pytestmark = pytest.mark.asyncio
+
+
+class _FakeStore:
+    """Stand-in for ChannelStore so the thread-mapping file lands under tmp_path."""
+
+    def __init__(self, tmp_path: Path) -> None:
+        self._path = tmp_path / "channel_store.json"
+
+
+async def test_discord_constructor_is_io_free_on_async_path(tmp_path: Path) -> None:
+    """``__init__`` must not touch the filesystem — ``_start_channel`` constructs inline."""
+    # Direct construction on the event loop, no asyncio.to_thread wrapper —
+    # mirrors the production _start_channel path. __init__ only resolves paths
+    # (Path.home / store._path.parent); if it regresses to doing exists/read_text
+    # the Blockbuster gate raises BlockingError here.
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "t", "channel_store": _FakeStore(tmp_path)})
+    assert channel._bot_token == "t"
+    assert channel._thread_store_path == tmp_path / "discord_threads.json"
+
+
+async def test_discord_record_then_persist_does_not_block_event_loop(tmp_path: Path) -> None:
+    """``_record_thread_mapping`` updates memory synchronously; ``_persist_thread_mappings``
+    writes the thread-mapping JSON through ``asyncio.to_thread``."""
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "test-token", "channel_store": _FakeStore(tmp_path)})
+
+    # _on_message does: _record_thread_mapping(...) then await asyncio.to_thread(_persist_thread_mappings)
+    channel._record_thread_mapping("chan-1", "thread-1")
+    assert channel._active_threads == {"chan-1": "thread-1"}
+    assert "thread-1" in channel._active_thread_ids
+
+    await asyncio.to_thread(channel._persist_thread_mappings)
+
+    data = json.loads(await asyncio.to_thread(channel._thread_store_path.read_text))
+    assert data == {"chan-1": "thread-1"}
+
+
+async def test_discord_record_thread_mapping_visible_before_persist(tmp_path: Path) -> None:
+    """Race regression (#3927 review): ``_record_thread_mapping`` must update
+    ``_active_thread_ids`` synchronously so an inbound message in the new
+    thread is recognized BEFORE the offloaded persistence write completes. If
+    the memory update were deferred to the worker thread, ``_on_message``'s
+    membership check would misclassify the message as orphaned and create a
+    duplicate thread.
+    """
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "test-token", "channel_store": _FakeStore(tmp_path)})
+
+    # Record WITHOUT awaiting any persistence — memory must already reflect it.
+    channel._record_thread_mapping("chan-1", "thread-1")
+    assert "thread-1" in channel._active_thread_ids
+    assert channel._active_threads["chan-1"] == "thread-1"
+    # Persistence is a separate offloaded step; the file is not written yet.
+    assert not await asyncio.to_thread(channel._thread_store_path.exists)
+
+
+async def test_discord_record_thread_mapping_discards_replaced_thread(tmp_path: Path) -> None:
+    """Recording a new thread for a channel that already had one drops the old
+    thread id from the reverse-lookup set, so messages in the stale thread are
+    no longer treated as active (mirrors the discard the old ``_save_thread``
+    performed against the on-disk record).
+    """
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "test-token", "channel_store": _FakeStore(tmp_path)})
+
+    channel._record_thread_mapping("chan-1", "thread-1")
+    channel._record_thread_mapping("chan-1", "thread-2")  # replace
+
+    assert channel._active_threads == {"chan-1": "thread-2"}
+    assert "thread-1" not in channel._active_thread_ids
+    assert "thread-2" in channel._active_thread_ids
+
+
+async def test_discord_load_active_threads_does_not_block_event_loop(tmp_path: Path) -> None:
+    """``_load_active_threads`` restores mappings through ``asyncio.to_thread`` (from ``start()``)."""
+    path = tmp_path / "discord_threads.json"
+    await asyncio.to_thread(path.write_text, json.dumps({"chan-1": "thread-1", "chan-2": "thread-2"}))
+
+    channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "test-token", "channel_store": _FakeStore(tmp_path)})
+
+    # start() does: await asyncio.to_thread(self._load_active_threads)
+    await asyncio.to_thread(channel._load_active_threads)
+
+    assert channel._active_threads == {"chan-1": "thread-1", "chan-2": "thread-2"}
+    assert channel._active_thread_ids == {"thread-1", "thread-2"}