diff --git a/backend/migrations/versions/a7f3d2b81c04_retry_bot_backlog.py b/backend/migrations/versions/a7f3d2b81c04_retry_bot_backlog.py new file mode 100644 index 0000000..bb65d91 --- /dev/null +++ b/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 diff --git a/backend/src/watcher/drain.py b/backend/src/watcher/drain.py index e61b0a0..0f49e07 100644 --- a/backend/src/watcher/drain.py +++ b/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: - """Walk the bot's pending update queue into the archive. Returns stored count.""" +async def drain_backlog(client: PyroClient, account_id: int) -> int | None: + """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 - pts, qts, date = EARLIEST_PTS, EARLIEST_QTS, 0 + return None + 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}") diff --git a/backend/src/watcher/registry.py b/backend/src/watcher/registry.py index 462f929..8a342db 100644 --- a/backend/src/watcher/registry.py +++ b/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)