fix(watcher): getDifference needs a date, and only a completed drain counts

backend/migrations/versions/a7f3d2b81c04_retry_bot_backlog.py

@@ -0,0 +1,25 @@
+"""retry bot backlog
+
+Revision ID: a7f3d2b81c04
+Revises: e2c9a41f7b68
+Create Date: 2026-09-18 17:30:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "a7f3d2b81c04"
+down_revision: str | None = "e2c9a41f7b68"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """No drain has succeeded yet, so let every bot try the queue once more."""
+    op.execute("UPDATE watched_bots SET backfilled_at = NULL")
+
+
+def downgrade() -> None:
+    pass

backend/src/watcher/drain.py

@@ -6,6 +6,8 @@ same queue Bot API ``getUpdates`` drains is reachable over MTProto through
 ``updates.getState``, which throws that queue away - this walks it instead.
 """
 
+import time
+
 from pyrogram import raw, types
 from pyrogram.errors import RPCError
 
@@ -48,12 +50,16 @@ async def _parse_all(
     return parsed
 
 
-async def drain_backlog(client: PyroClient, account_id: int) -> int:
+async def drain_backlog(client: PyroClient, account_id: int) -> int | None:
-    """Walk the bot's pending update queue into the archive. Returns stored count."""
+    """Walk the bot's pending queue into the archive.
+
+    Returns the number of messages stored, or ``None`` when the walk could not
+    be completed - the caller keeps the bot marked as still needing a drain.
+    """
     ctx = client.capture
     if ctx is None:
-        return 0
+        return None
-    pts, qts, date = EARLIEST_PTS, EARLIEST_QTS, 0
+    pts, qts, date = EARLIEST_PTS, EARLIEST_QTS, int(time.time())
     stored = 0
     for _ in range(MAX_SLICES):
         try:
@@ -64,7 +70,7 @@ async def drain_backlog(client: PyroClient, account_id: int) -> int:
             logger.warning(
                 f"[yellow]Account {account_id}: getDifference failed:[/] {exc}"
             )
-            return stored
+            return None
 
         if isinstance(diff, raw.types.updates.DifferenceEmpty):
             logger.info(f"Account {account_id}: backlog empty at pts={pts}")

backend/src/watcher/registry.py

@@ -185,6 +185,8 @@ class BotRegistry:
 
     async def _backlog(self, client: PyroClient, bot_id: int, account_id: int) -> None:
         stored = await drain_backlog(client, account_id)
+        if stored is None:
+            return
         await bots.mark_backfilled(self._pool, bot_id)
         if stored:
             await accounts.notify_accounts_changed(self._pool)