From 9600414002146432fb84c351d8fc900df7e976ca Mon Sep 17 00:00:00 2001 From: Sunshine <135408348+sunshine-lang@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:56:56 +0800 Subject: [PATCH] fix(channels): retain strong reference to Discord ack-reaction tasks (#5049) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(channels): retain strong reference to Discord ack-reaction tasks The event loop keeps only weak references to scheduled tasks, so the bare asyncio.create_task(self._add_reaction(message)) at both message-acceptance sites could be garbage-collected mid-flight and silently drop the ✅ acknowledgment. The module-level retention set holds each task until completion and logs failures, mirroring the deferred subagent cleanup fix in #4928 and the stream bridge cleanup fix in #4931. * fix(channels): drain in-flight ack reactions on Discord channel shutdown Adversarial follow-up to the retention fix: stop() cancelled typing tasks but left in-flight ack-reaction tasks in the module retention set when the channel stopped mid-HTTP-call, pinning the channel instance and the discord Message object graph across restart cycles. _cancel_ephemeral_tasks / _discard_ephemeral_tasks now cover both families at every stop() cleanup point and the _run_client finally. Also tests the done-callback error-log branch and documents the GC test as a contract pin. * test(channels): pin Discord ack-reaction drain wiring and sharpen wording Second adversarial pass found the drain test cancelled a never-started task (no yield between schedule and cancel), nothing exercised the real stop() cross-loop wiring, and the cleanup log messages still said "typing-task" while now covering both families. The drain test now yields so the task genuinely suspends first, a new test drives stop() through a real background loop and fails if the wiring reverts to typing-only cleanup, wording covers ephemeral tasks, and the discard path snapshots the set before iterating. * refactor(channels): scope Discord ack-reaction retention to the channel instance The retention set was module-level, so one DiscordChannel's stop() would cancel every other instance's in-flight ack reactions. Instance-level matches the typing-task bookkeeping and keeps shutdown scoped; pinned by a two-channel isolation test. * test(channels): drive two-channel ack isolation through full stop() Independent review of the instance-scoping refactor confirmed the P2 fix but suggested matching the review's letter exactly: the isolation test now shuts the first channel down via stop() instead of the drain helper, and one stale docstring reference to the former module-level set is corrected. --- backend/app/channels/discord.py | 81 ++++++++++-- backend/tests/test_discord_channel.py | 170 ++++++++++++++++++++++++++ 2 files changed, 240 insertions(+), 11 deletions(-) diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index 977cebf1c..4a7a52ad2 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -67,6 +67,15 @@ class DiscordChannel(Channel): # Typing indicator management self._typing_tasks: dict[str, asyncio.Task] = {} + # Strong references for this channel's in-flight ack-reaction tasks. + # The event loop keeps only weak references to scheduled tasks, so a + # bare ``asyncio.create_task`` could be garbage-collected mid-flight + # and silently drop the acknowledgment reaction (same retention + # pattern as the deferred subagent cleanup in #4928). Instance-level + # on purpose, matching ``_typing_tasks``: one channel's shutdown must + # not cancel another instance's in-flight reactions. + self._ack_reaction_tasks: set[asyncio.Task[None]] = set() + self._client = None self._thread: threading.Thread | None = None self._discord_loop: asyncio.AbstractEventLoop | None = None @@ -177,25 +186,26 @@ class DiscordChannel(Channel): discord_loop = self._discord_loop current_loop = asyncio.get_running_loop() if discord_loop is None or discord_loop is current_loop: - await self._cancel_typing_tasks() + await self._cancel_ephemeral_tasks() elif discord_loop.is_running(): - # Serialize cleanup with _start_typing() on the owning loop. The + # Serialize cleanup with _start_typing()/_schedule_ack_reaction() + # on the owning loop. The # loop may exit between is_running() and scheduling, so neither # scheduling nor waiting is allowed to block the rest of stop(). - cleanup_coro = self._cancel_typing_tasks() + cleanup_coro = self._cancel_ephemeral_tasks() try: cleanup_future = asyncio.run_coroutine_threadsafe(cleanup_coro, discord_loop) except RuntimeError: cleanup_coro.close() - logger.warning("[Discord] event loop stopped before typing-task cleanup could be scheduled") + logger.warning("[Discord] event loop stopped before ephemeral-task cleanup could be scheduled") else: try: await asyncio.wait_for(asyncio.wrap_future(cleanup_future), timeout=10) except TimeoutError: cleanup_future.cancel() - logger.warning("[Discord] typing-task cleanup timed out after 10s") + logger.warning("[Discord] ephemeral-task cleanup timed out after 10s") except Exception: - logger.exception("[Discord] error while cleaning up typing tasks") + logger.exception("[Discord] error while cleaning up ephemeral tasks") if self._client and discord_loop and discord_loop.is_running(): close_coro = self._client.close() @@ -222,7 +232,7 @@ class DiscordChannel(Channel): # discard the stale references here; awaiting them from this loop would # raise a cross-loop RuntimeError. if discord_loop and discord_loop is not current_loop and not discord_loop.is_running(): - self._discard_typing_tasks() + self._discard_ephemeral_tasks() self._client = None self._discord_loop = None @@ -315,6 +325,36 @@ class DiscordChannel(Channel): ) self._typing_tasks.clear() + async def _cancel_ack_reaction_tasks(self) -> None: + """Cancel in-flight ack reactions so stop() does not strand them. + + A task interrupted mid-HTTP-call would otherwise stay in the + retention set forever, pinning this channel instance and the discord + Message object graph after shutdown. + """ + pending = [task for task in self._ack_reaction_tasks if not task.done()] + for task in pending: + task.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + def _discard_ack_reaction_tasks(self) -> None: + """Forget stale ack-reaction tasks after the owning loop stopped.""" + for task in list(self._ack_reaction_tasks): + if not task.done(): + logger.warning("[Discord] discarding pending ack-reaction task for stopped loop") + self._ack_reaction_tasks.clear() + + async def _cancel_ephemeral_tasks(self) -> None: + """Cancel typing indicators and in-flight ack reactions together.""" + await self._cancel_typing_tasks() + await self._cancel_ack_reaction_tasks() + + def _discard_ephemeral_tasks(self) -> None: + """Forget stale typing and ack-reaction tasks (stopped-loop fallback).""" + self._discard_typing_tasks() + self._discard_ack_reaction_tasks() + async def _stop_typing(self, chat_id: str, thread_ts: str | None = None) -> None: """Stops the typing loop for a specific target.""" target_id = thread_ts or chat_id @@ -330,6 +370,25 @@ class DiscordChannel(Channel): except Exception: logger.debug("[Discord] failed to add reaction to message %s", message.id, exc_info=True) + def _on_ack_reaction_task_done(self, task: asyncio.Task[None]) -> None: + self._ack_reaction_tasks.discard(task) + if task.cancelled(): + return + exc = task.exception() + if exc is not None: + logger.error("[Discord] ack reaction task failed: %s", exc) + + def _schedule_ack_reaction(self, message) -> asyncio.Task[None]: + """Schedule the ack reaction with a strong task reference. + + The delayed HTTP call must survive until completion; see + ``_ack_reaction_tasks`` for why a bare ``create_task`` is unsafe here. + """ + task = asyncio.create_task(self._add_reaction(message)) + self._ack_reaction_tasks.add(task) + task.add_done_callback(self._on_ack_reaction_task_done) + return task + async def _on_message(self, message) -> None: if not self._running or not self._client: return @@ -450,7 +509,7 @@ class DiscordChannel(Channel): # Start typing indicator in the thread if typing_target: await self._start_typing(typing_target, chat_id, thread_id) - asyncio.create_task(self._add_reaction(message)) + self._schedule_ack_reaction(message) return # Thread not tracked (orphaned) — create new thread and handle below @@ -561,7 +620,7 @@ class DiscordChannel(Channel): # Start typing/reaction only after bounded admission succeeds. if typing_target: await self._start_typing(typing_target, chat_id, thread_id) - asyncio.create_task(self._add_reaction(message)) + self._schedule_ack_reaction(message) finally: if not reservation_transferred: reservation.release() @@ -642,9 +701,9 @@ class DiscordChannel(Channel): logger.exception("Discord client error") finally: try: - self._discord_loop.run_until_complete(self._cancel_typing_tasks()) + self._discord_loop.run_until_complete(self._cancel_ephemeral_tasks()) except Exception: - logger.exception("Error while cleaning up Discord typing tasks") + logger.exception("Error while cleaning up Discord ephemeral tasks") try: if self._client and not self._client.is_closed(): self._discord_loop.run_until_complete(self._client.close()) diff --git a/backend/tests/test_discord_channel.py b/backend/tests/test_discord_channel.py index 16fd9bfc0..dbeb45f09 100644 --- a/backend/tests/test_discord_channel.py +++ b/backend/tests/test_discord_channel.py @@ -4,7 +4,9 @@ from __future__ import annotations import asyncio import builtins +import gc import threading +import weakref from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -263,3 +265,171 @@ async def test_send_file_closes_handle_when_send_fails(tmp_path) -> None: assert handles[0].closed is True finally: _stop_bg_loop(bg_loop, bg_thread) + + +@pytest.mark.asyncio +async def test_ack_reaction_task_retained_under_gc() -> None: + """The ack-reaction task is retained and runs to completion. + + This pins the retention contract (scheduled → in the set, done → + discarded): the fake coroutine never suspends on an unrooted future, so + actual mid-flight GC cannot be reproduced deterministically here — the + same limitation the #4928 precedent test has. + + A bare ``asyncio.create_task`` holds only a weak loop reference, so the + ✅ acknowledgment could be garbage-collected mid-flight. The instance-level + retention set keeps the task strongly referenced until completion (same + pattern as #4928 / #4931). + """ + bus = MessageBus() + channel = DiscordChannel(bus=bus, config={"bot_token": "token"}) + + reacted = asyncio.Event() + + async def _add_reaction(_emoji: str) -> None: + reacted.set() + + message = SimpleNamespace(id=111, add_reaction=_add_reaction) + + task = channel._schedule_ack_reaction(message) + weak_task = weakref.ref(task) + del task + gc.collect() + + retained = weak_task() + assert retained is not None, "ack reaction task was garbage-collected mid-flight" + assert retained in channel._ack_reaction_tasks + + await asyncio.wait_for(retained, timeout=1.0) + assert reacted.is_set() + # Completed tasks are discarded so the retention set cannot grow unboundedly. + assert retained not in channel._ack_reaction_tasks + + +@pytest.mark.asyncio +async def test_ack_reaction_retention_is_isolated_per_channel() -> None: + """One channel's shutdown must not cancel another instance's in-flight reactions. + + The retention set is instance-level (matching ``_typing_tasks``): a + module-level set would let ``stop()`` on one channel drain every other + channel's pending ack tasks. + """ + first = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + second = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + + release = asyncio.Event() + + async def _hang(_emoji: str) -> None: + await release.wait() + + task = second._schedule_ack_reaction(SimpleNamespace(id=666, add_reaction=_hang)) + await asyncio.sleep(0) # let the task start and suspend + assert task in second._ack_reaction_tasks + assert task not in first._ack_reaction_tasks + + # Full shutdown of the first channel must leave the second instance's + # in-flight reaction alone (the review asked for independent shutdown). + await first.stop() + + assert not task.done() + assert task in second._ack_reaction_tasks + + release.set() + await asyncio.wait_for(task, timeout=1.0) + + +@pytest.mark.asyncio +async def test_ack_reaction_task_survives_reaction_failure() -> None: + """A failing add_reaction completes quietly and is discarded from the set.""" + + bus = MessageBus() + channel = DiscordChannel(bus=bus, config={"bot_token": "token"}) + + async def _boom(_emoji: str) -> None: + raise RuntimeError("discord api down") + + message = SimpleNamespace(id=222, add_reaction=_boom) + + task = channel._schedule_ack_reaction(message) + await asyncio.wait_for(task, timeout=1.0) + assert task not in channel._ack_reaction_tasks + + +@pytest.mark.asyncio +async def test_stop_drains_in_flight_ack_reaction_tasks() -> None: + """stop()'s cleanup path cancels in-flight ack reactions on the owning loop. + + Without the drain, a task interrupted mid-HTTP-call would sit in the + module retention set forever, pinning the channel and Message graph. + """ + release = asyncio.Event() + + async def _hang(_emoji: str) -> None: + await release.wait() + + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + message = SimpleNamespace(id=333, add_reaction=_hang) + + task = channel._schedule_ack_reaction(message) + assert task in channel._ack_reaction_tasks + # Yield once so the task actually starts and suspends inside _hang — + # cancelling a never-started task would not exercise the mid-flight path. + await asyncio.sleep(0) + assert not task.done() + + await channel._cancel_ephemeral_tasks() + + assert task.cancelled() + assert task not in channel._ack_reaction_tasks + + +@pytest.mark.asyncio +async def test_ack_reaction_task_failure_is_logged_and_discarded(caplog) -> None: + """An exception escaping _add_reaction is logged at error and discarded.""" + bus = MessageBus() + channel = DiscordChannel(bus=bus, config={"bot_token": "token"}) + + async def _explode(_self, _message) -> None: + raise RuntimeError("unexpected boom") + + with patch.object(DiscordChannel, "_add_reaction", _explode), caplog.at_level("ERROR", logger="app.channels.discord"): + task = channel._schedule_ack_reaction(SimpleNamespace(id=444)) + with pytest.raises(RuntimeError, match="unexpected boom"): + await asyncio.wait_for(task, timeout=1.0) + + assert task not in channel._ack_reaction_tasks + assert any("ack reaction task failed" in record.message for record in caplog.records) + + +@pytest.mark.asyncio +async def test_stop_wiring_drains_ack_tasks_across_loops() -> None: + """The real stop() path (cross-loop branch) drains in-flight ack reactions. + + Guards the wiring itself: reverting stop() to typing-only cleanup must + fail here, not just at the helper level. + """ + bg_loop, bg_thread = _start_bg_loop() + try: + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._discord_loop = bg_loop + + release = asyncio.Event() + + async def _hang(_emoji: str) -> None: + await release.wait() + + async def _schedule_on_bg_loop() -> asyncio.Task: + task = channel._schedule_ack_reaction(SimpleNamespace(id=555, add_reaction=_hang)) + await asyncio.sleep(0.05) # let the task start and suspend on the event + return task + + schedule_future = asyncio.run_coroutine_threadsafe(_schedule_on_bg_loop(), bg_loop) + task = await asyncio.wrap_future(schedule_future) + assert task in channel._ack_reaction_tasks + + await channel.stop() + + assert task.cancelled() + assert not channel._ack_reaction_tasks + finally: + _stop_bg_loop(bg_loop, bg_thread)