From 095dd671caafff0604982b34210a67e017ef1a1d Mon Sep 17 00:00:00 2001 From: spfcraze Date: Thu, 30 Jul 2026 17:55:57 -0400 Subject: [PATCH] fix(discord): claim dedup ID in missed-message backfill dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The missed-message backfill admitted recovered messages with claim=False, so a successfully backfilled message never entered the dedup cache. Discord replays missed events on a successful resume — and on_ready, which starts the backfill, fires after resumes too — so a live replay of the same message raced or followed the REST-scan dispatch and was admitted again: two agent runs, two replies for one user message. The adapter's own discard() calls on the failure paths ('Release a claimed message ID') were no-ops, proving the claim was intended. Admit with claim=True; whichever path (live or backfill) claims first now wins and the other drops the duplicate. --- plugins/platforms/discord/adapter.py | 15 +++- .../test_discord_missed_message_backfill.py | 78 +++++++++++++++++++ 2 files changed, 89 insertions(+), 4 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 5fab2307c30a8..a968a99f7c0f4 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -2129,9 +2129,11 @@ async def _run_missed_message_backfill(self) -> None: scanned += 1 message_id = str(getattr(message, "id", "")) self._record_discord_message_seen(message, status="discovered") - # A live gateway event may race this REST scan. Check without - # claiming the ID; the shared ingress helper owns the dedup - # write immediately before normal auth/filter dispatch. + # A live gateway event may race this REST scan. Cheap + # pre-filter only: the actual claim happens atomically in + # _dispatch_recovered_message via claim=True, so whichever + # path (live or backfill) claims first wins and the other + # drops the duplicate. if self._dedup.contains(message_id): continue if not await self._should_backfill_discord_message(message): @@ -2215,7 +2217,12 @@ async def _dispatch_recovered_message(self, message: Any) -> bool: ): return False admitted, role_authorized = self._discord_message_admission( - message, claim=False, + # Claim the ID: without this the message never enters the dedup + # cache, so a live gateway replay of the same event (Discord + # replays missed events on resume, racing this REST scan) is + # admitted again and the user gets two runs / two replies. + # Failure paths in the caller release the claim via discard(). + message, claim=True, ) if not admitted: return False diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py index 2e70830a7dc7b..d015c000650fc 100644 --- a/tests/gateway/test_discord_missed_message_backfill.py +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -458,3 +458,81 @@ async def _gen(): assert got == [2, 3, 4] + + +@pytest.mark.asyncio +async def test_recovered_dispatch_claims_dedup_id(adapter): + """Backfilled messages must enter the dedup cache. + + Regression: the backfill path admitted with claim=False, so a recovered + message never entered the dedup cache — a live gateway replay of the same + event (Discord replays missed events on resume) was admitted again and + the user got two runs / two replies. + """ + bot_user = adapter._client.user + message = make_message( + message_id=555, + content=f"<@{bot_user.id}> please ingest", + mentions=[bot_user], + ) + + assert await adapter._dispatch_recovered_message(message) is True + assert adapter._handle_message.call_count == 1 + assert adapter._dedup.contains("555") is True + + # A live replay of the same message must now be dropped. + admitted, _ = adapter._discord_message_admission(message, claim=True) + assert admitted is False + assert adapter._handle_message.call_count == 1 + + +@pytest.mark.asyncio +async def test_recovered_dispatch_second_attempt_dropped(adapter): + """A second backfill dispatch of the same message is a no-op.""" + bot_user = adapter._client.user + message = make_message( + message_id=777, + content=f"<@{bot_user.id}> please ingest", + mentions=[bot_user], + ) + + assert await adapter._dispatch_recovered_message(message) is True + assert await adapter._dispatch_recovered_message(message) is False + assert adapter._handle_message.call_count == 1 + + +@pytest.mark.asyncio +async def test_failed_recovered_dispatch_releases_claim_for_retry(adapter, monkeypatch): + """A dispatch that raises must release the claimed dedup ID. + + The backfill now claims the ID at dispatch time; the scan loop's error + path calls discard() so a transient failure doesn't permanently suppress + the message from a later retry (or the live replay). + """ + bot_user = adapter._client.user + message = make_message( + message_id=888, + content=f"<@{bot_user.id}> please ingest", + mentions=[bot_user], + ) + + async def fake_candidates(_channels): + yield message + + monkeypatch.setenv("DISCORD_MISSED_MESSAGE_BACKFILL_CHANNELS", "123") + monkeypatch.setattr(adapter, "_iter_missed_message_backfill_candidates", fake_candidates) + monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True)) + monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 10) + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"}) + monkeypatch.setattr("asyncio.sleep", AsyncMock()) + + # First pass: dispatch fails -> claim must be released by the error path. + adapter._handle_message = AsyncMock(side_effect=RuntimeError("transient")) + await adapter._run_missed_message_backfill() + assert adapter._dedup.contains("888") is False + + # Retry with a healthy handler: the message dispatches and stays claimed. + adapter._handle_message = AsyncMock(return_value=True) + await adapter._run_missed_message_backfill() + assert adapter._handle_message.call_count == 1 + assert adapter._dedup.contains("888") is True