diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index d3ae4651e..35cf22840 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -71,6 +71,12 @@ class DiscordChannel(Channel): # Lock protecting _active_threads and the JSON file from concurrent access. # _run_client (Discord loop thread) and the main thread both read/write. self._thread_store_lock = threading.Lock() + # Set once _load_active_threads() has run: the in-memory map is then a + # faithful view of the file, so stop() may flush it. It stays False when + # start() bails before the load (missing bot_token, discord import + # error) — ChannelService then stops the instance, and an ungated flush + # would overwrite the persisted mappings with an empty snapshot (#2897). + self._thread_store_loaded = False store = config.get("channel_store") if store is not None: self._thread_store_path = store._path.parent / "discord_threads.json" @@ -140,17 +146,24 @@ class DiscordChannel(Channel): try: if not self._thread_store_path.exists(): logger.debug("[Discord] no thread mappings file at %s", self._thread_store_path) - return - data = json.loads(self._thread_store_path.read_text()) - self._active_threads.clear() - self._active_thread_ids.clear() - for channel_id, thread_id in data.items(): - self._active_threads[channel_id] = thread_id - self._active_thread_ids.add(thread_id) - if self._active_threads: - logger.info("[Discord] restored %d thread mappings from %s", len(self._active_threads), self._thread_store_path) + else: + data = json.loads(self._thread_store_path.read_text()) + self._active_threads.clear() + self._active_thread_ids.clear() + for channel_id, thread_id in data.items(): + self._active_threads[channel_id] = thread_id + self._active_thread_ids.add(thread_id) + if self._active_threads: + logger.info("[Discord] restored %d thread mappings from %s", len(self._active_threads), self._thread_store_path) except Exception: logger.exception("[Discord] failed to load thread mappings") + return + # The in-memory map is now a superset of the on-disk store (empty + # when the file is absent), so stop() may safely flush it. Until + # this point — a start() that bailed on a missing bot_token or a + # discord import error, which ChannelService then stops — the map is + # empty and flushing would clobber the file with {}. + self._thread_store_loaded = True 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. @@ -196,6 +209,20 @@ class DiscordChannel(Channel): self._running = False self.bus.unsubscribe_outbound(self._on_outbound) + # Best-effort durability: flush in-memory thread mappings so the most + # recent channel->thread mapping survives a hard shutdown (process + # killed between a thread creation and its background persistence + # write). The create path already persists off the event loop after + # each new thread, so this is a safety net, not the primary write path. + # Gated on _thread_store_loaded: a stop() taken before the initial load + # has an empty in-memory map, and flushing it would overwrite the file + # with {} — the #2897 data loss this PR exists to prevent. + if self._thread_store_loaded: + try: + await asyncio.to_thread(self._persist_thread_mappings) + except Exception: + logger.warning("[Discord] failed to flush thread mappings during shutdown") + discord_loop = self._discord_loop current_loop = asyncio.get_running_loop() if discord_loop is None or discord_loop is current_loop: diff --git a/backend/tests/test_discord_channel.py b/backend/tests/test_discord_channel.py index ec9208029..1f999ebae 100644 --- a/backend/tests/test_discord_channel.py +++ b/backend/tests/test_discord_channel.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import builtins import gc +import json import threading import weakref from types import SimpleNamespace @@ -33,6 +34,87 @@ def test_discord_channel_init() -> None: assert channel.name == "discord" +# --------------------------------------------------------------------------- +# thread-mapping persistence across restart (#2897) +# --------------------------------------------------------------------------- + + +def test_discord_thread_mapping_persists_across_restart(tmp_path) -> None: + """A channel->thread mapping written before shutdown is restored on a + subsequent start, so conversations are not lost across restarts (#2897).""" + store_path = tmp_path / "discord_threads.json" + + # First process lifetime: record and persist a mapping. + first = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + first._thread_store_path = store_path + first._record_thread_mapping("chan-1", "thread-1") + first._persist_thread_mappings() + assert store_path.exists() + assert json.loads(store_path.read_text()) == {"chan-1": "thread-1"} + + # Restart: a brand-new channel instance reads the same file. + second = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + second._thread_store_path = store_path + second._active_threads.clear() + second._active_thread_ids.clear() + second._load_active_threads() + + assert second._active_threads == {"chan-1": "thread-1"} + assert "thread-1" in second._active_thread_ids + + +@pytest.mark.asyncio +async def test_discord_stop_flushes_thread_mappings(tmp_path) -> None: + """stop() best-effort flushes in-memory thread mappings to disk so the + most recent mapping survives a hard shutdown.""" + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._thread_store_path = tmp_path / "discord_threads.json" + # stop() only flushes once the load has marked the in-memory map + # authoritative (see DiscordChannel._thread_store_loaded); simulate the + # normal post-start() state so this exercises the flush, not the guard. + channel._thread_store_loaded = True + # Minimal shutdown context: no live client/loop/thread to tear down. + channel._discord_loop = None + channel._client = None + channel._thread = None + channel._cancel_ephemeral_tasks = AsyncMock() + + channel._record_thread_mapping("chan-2", "thread-2") + await channel.stop() + + assert channel._thread_store_path.exists() + assert json.loads(channel._thread_store_path.read_text()) == {"chan-2": "thread-2"} + + +@pytest.mark.asyncio +async def test_discord_stop_does_not_clobber_store_before_load(tmp_path) -> None: + """stop() before the initial load must not overwrite the persisted file. + + ``ChannelService`` deliberately stops a channel whose ``start()`` bailed + before ``_load_active_threads()`` ran (missing bot_token / discord import + error), when the in-memory map is still empty. An ungated flush would write + ``{}`` over the persisted mappings — the #2897 data loss this PR exists to + prevent. + """ + store_path = tmp_path / "discord_threads.json" + store_path.write_text(json.dumps({"chan-9": "thread-9"})) + + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._thread_store_path = store_path + # start() bailed before the load, so the flag is still False. + assert channel._thread_store_loaded is False + # Minimal shutdown context: no live client/loop/thread to tear down. + channel._discord_loop = None + channel._client = None + channel._thread = None + channel._cancel_ephemeral_tasks = AsyncMock() + + await channel.stop() + + # The pre-existing mapping survives intact: nothing was flushed over it. + assert json.loads(store_path.read_text()) == {"chan-9": "thread-9"} + + def _make_discord_message(text: str): return SimpleNamespace( id=111,