diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 30c4fd2f4..628062a8a 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -2057,15 +2057,6 @@ class DiscordAdapter(BasePlatformAdapter): admitted = await self._dispatch_recovered_message(message) if admitted: dispatched += 1 - channel_id = str( - getattr(getattr(message, "channel", None), "id", "") - ) - if channel_id and message_id: - await asyncio.to_thread( - self._advance_discord_recovery_cursor, - channel_id, - message_id, - ) except asyncio.CancelledError: self._dedup.discard(message_id) self._record_recovery_attempt(message, status="cancelled") @@ -2149,18 +2140,29 @@ class DiscordAdapter(BasePlatformAdapter): continue candidate_channels.append(channel) - yielded = 0 - for channel in candidate_channels: - async for item in self._iter_channel_and_thread_messages( + iterators = [ + self._iter_channel_and_thread_messages( channel, limit=limit, after=after, seen_channels=seen, - ): + ).__aiter__() + for channel in candidate_channels + ] + yielded = 0 + while iterators and yielded < limit: + next_round = [] + for iterator in iterators: + try: + item = await iterator.__anext__() + except StopAsyncIteration: + continue yield item yielded += 1 + next_round.append(iterator) if yielded >= limit: return + iterators = next_round async def _iter_channel_and_thread_messages(self, channel: Any, *, limit: int, after: Any, seen_channels: set[str]): """Yield history from a channel plus active/recent archived child threads.""" @@ -2459,6 +2461,18 @@ class DiscordAdapter(BasePlatformAdapter): ) self._with_discord_recovery_db(_op) + if completed: + def _channel_for_message(conn): + row = conn.execute( + "SELECT COALESCE(thread_id, channel_id) FROM discord_messages " + "WHERE message_id=?", + (reply_to,), + ).fetchone() + return str(row[0]) if row and row[0] else None + + channel_id = self._with_discord_recovery_db(_channel_for_message) + if channel_id: + self._advance_discord_recovery_cursor(channel_id, reply_to) def _discord_message_is_persistently_complete(self, message_id: str) -> bool: if not message_id: diff --git a/tests/gateway/test_discord_missed_message_backfill.py b/tests/gateway/test_discord_missed_message_backfill.py index 63592270b..7d2c5cb9d 100644 --- a/tests/gateway/test_discord_missed_message_backfill.py +++ b/tests/gateway/test_discord_missed_message_backfill.py @@ -824,6 +824,30 @@ async def test_iter_candidates_applies_one_global_scan_limit(adapter, monkeypatc assert set(got).issubset({1, 2, 3, 4}) +@pytest.mark.asyncio +async def test_iter_candidates_round_robins_configured_channels(adapter, monkeypatch): + first = FakeChannel( + channel_id=123, + history_messages=[ + make_message(message_id=1), + make_message(message_id=2), + make_message(message_id=3), + ], + ) + second = FakeChannel( + channel_id=456, + history_messages=[make_message(message_id=4)], + ) + adapter._client.get_channel = lambda channel_id: {123: first, 456: second}[channel_id] + monkeypatch.setattr(adapter, "_missed_message_backfill_limit", lambda: 3) + + got = [] + async for message in adapter._iter_missed_message_backfill_candidates({"123", "456"}): + got.append(message.id) + + assert 4 in got + + @pytest.mark.asyncio async def test_iter_candidates_keeps_latest_messages_when_window_exceeds_limit(adapter, monkeypatch): class RealisticChannel(FakeChannel): @@ -864,6 +888,42 @@ def test_recovery_cursor_round_trip_is_channel_scoped(adapter): assert adapter._discord_recovery_cursor("456") == "2002" +@pytest.mark.asyncio +async def test_cursor_does_not_advance_past_incomplete_dispatched_message(adapter, monkeypatch): + channel = FakeChannel( + channel_id=123, + history_messages=[ + make_message(message_id=1), + make_message(message_id=2), + ], + ) + for message in channel._history_messages: + message.channel = channel + adapter._client.get_channel = lambda _channel_id: channel + monkeypatch.setattr(adapter, "_missed_message_backfill_channels", lambda: {"123"}) + monkeypatch.setattr(adapter, "_should_backfill_discord_message", AsyncMock(return_value=True)) + monkeypatch.setattr(adapter, "_dispatch_recovered_message", AsyncMock(side_effect=[True, True])) + monkeypatch.setattr(adapter, "_missed_message_backfill_max_dispatches", lambda: 10) + + await adapter._run_missed_message_backfill() + + assert adapter._discord_recovery_cursor("123") is None + + +def test_final_delivery_advances_channel_cursor(adapter): + message = make_message(message_id=103, channel=FakeChannel(channel_id=123)) + adapter._record_discord_message_seen(message, status="processing") + + adapter._record_discord_response( + reply_to="103", + result=SimpleNamespace(success=True, message_id="9010"), + content="done", + final=True, + ) + + assert adapter._discord_recovery_cursor("123") == "103" + + @pytest.mark.asyncio async def test_iter_candidates_uses_persisted_channel_cursor(adapter, monkeypatch): class CursorChannel(FakeChannel):