diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index 1f947937c..a4a5a936f 100644 --- a/backend/app/channels/discord.py +++ b/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 diff --git a/backend/tests/blocking_io/test_discord_channel_state.py b/backend/tests/blocking_io/test_discord_channel_state.py new file mode 100644 index 000000000..8c2d7d59c --- /dev/null +++ b/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"}