From 897651abe9d850cfb9e467f28d5fede813dda9f0 Mon Sep 17 00:00:00 2001 From: Artemii Peretiachenko Date: Thu, 30 Jul 2026 13:41:38 +0200 Subject: [PATCH] Retry Telegram flood waits so client messages are not dropped. Also fix reply-target error detection so fallback without reply_to works when Telegram says the replied message is gone. Co-authored-by: Cursor --- bot/services/relay.py | 81 +++++++++++++++++++++++++++++++++++-------- 1 file changed, 66 insertions(+), 15 deletions(-) diff --git a/bot/services/relay.py b/bot/services/relay.py index 1b0a635..a087398 100644 --- a/bot/services/relay.py +++ b/bot/services/relay.py @@ -7,7 +7,12 @@ from typing import Any, Dict, Optional from aiogram import Bot from aiogram.enums import ChatType, ParseMode -from aiogram.exceptions import TelegramAPIError, TelegramBadRequest, TelegramForbiddenError +from aiogram.exceptions import ( + TelegramAPIError, + TelegramBadRequest, + TelegramForbiddenError, + TelegramRetryAfter, +) from aiogram.types import Message, MessageReactionUpdated, ReactionTypeCustomEmoji, ReactionTypeEmoji, User from aiogram.utils.text_decorations import html_decoration @@ -27,6 +32,34 @@ GENERAL_TOPIC_ID = 1 TOPIC_NAME_MAX_LEN = 128 CAPTION_MAX_LEN = 1024 +# How many times to wait out Telegram flood control before giving up. +FLOOD_RETRY_ATTEMPTS = 5 + + +async def _with_flood_retry(api_call, *args, **kwargs): + """Call Telegram API; on 429 wait retry_after and try again.""" + for attempt in range(1, FLOOD_RETRY_ATTEMPTS + 1): + try: + return await api_call(*args, **kwargs) + except TelegramRetryAfter as exc: + wait = max(int(exc.retry_after), 1) + method = getattr(api_call, "__name__", type(api_call).__name__) + if attempt >= FLOOD_RETRY_ATTEMPTS: + logger.error( + "Rate limited on %s after %s attempt(s); giving up (retry_after=%ss)", + method, + attempt, + wait, + ) + raise + logger.warning( + "Rate limited on %s; waiting %ss (attempt %s/%s)", + method, + wait, + attempt, + FLOOD_RETRY_ATTEMPTS, + ) + await asyncio.sleep(wait) def build_topic_name(user: User) -> str: @@ -129,9 +162,15 @@ def _is_message_gone_error(exc: BaseException) -> bool: def _is_reply_error(exc: BaseException) -> bool: text = str(exc).lower() - return "reply" in text and ( - "not found" in text or "message to be replied" in text or "replied message" in text - ) + # Telegram uses "replied" (y→i), so "reply" is not a substring of "replied". + return any( + marker in text + for marker in ( + "message to be replied", + "replied message", + "message to reply not found", + ) + ) or ("reply" in text and "not found" in text) def _is_topic_missing_error(exc: BaseException) -> bool: @@ -224,7 +263,8 @@ class RelayService: return fresh.topic_id await self._repo.clear_topic_id(user.id) - topic = await bot.create_forum_topic( + topic = await _with_flood_retry( + bot.create_forum_topic, chat_id=self._manager_group_id, name=build_topic_name(user), ) @@ -277,12 +317,12 @@ class RelayService: async def _call_with_optional_reply(self, api_call, kwargs: Dict[str, Any]): """Call Telegram API; retry once without reply_to if the reply target is gone.""" try: - return await api_call(**kwargs) + return await _with_flood_retry(api_call, **kwargs) except TelegramBadRequest as exc: if kwargs.get("reply_to_message_id") is not None and _is_reply_error(exc): retry = dict(kwargs) retry.pop("reply_to_message_id", None) - return await api_call(**retry) + return await _with_flood_retry(api_call, **retry) raise async def _copy_client_message( @@ -460,7 +500,8 @@ class RelayService: async def _notify_managers(self, bot: Bot, topic_id: int, key: str) -> None: try: - await bot.send_message( + await _with_flood_retry( + bot.send_message, chat_id=self._manager_group_id, message_thread_id=topic_id, text=t(DEFAULT_LANGUAGE, key), @@ -513,7 +554,8 @@ class RelayService: try: if message.text is not None and not _is_captionable(message): body = _html_body(message.text, message.entities) - await bot.edit_message_text( + await _with_flood_retry( + bot.edit_message_text, chat_id=self._manager_group_id, message_id=link.manager_message_id, text=format_labeled_html(f"{client_display_name(user)}:", body), @@ -527,7 +569,8 @@ class RelayService: message.caption, message.caption_entities, ) - await bot.edit_message_caption( + await _with_flood_retry( + bot.edit_message_caption, chat_id=self._manager_group_id, message_id=link.manager_message_id, caption=caption, @@ -566,7 +609,8 @@ class RelayService: try: if message.text is not None and not _is_captionable(message): body = _html_body(message.text, message.entities) - await bot.edit_message_text( + await _with_flood_retry( + bot.edit_message_text, chat_id=client.telegram_id, message_id=link.client_message_id, text=format_manager_html(client.language, body), @@ -580,7 +624,8 @@ class RelayService: message.caption, message.caption_entities, ) - await bot.edit_message_caption( + await _with_flood_retry( + bot.edit_message_caption, chat_id=client.telegram_id, message_id=link.client_message_id, caption=caption, @@ -666,7 +711,11 @@ class RelayService: async def _safe_delete(self, bot: Bot, chat_id: int, message_id: int) -> None: try: - await bot.delete_message(chat_id=chat_id, message_id=message_id) + await _with_flood_retry( + bot.delete_message, + chat_id=chat_id, + message_id=message_id, + ) except TelegramAPIError: logger.info("Could not delete message %s in chat %s", message_id, chat_id) @@ -684,7 +733,8 @@ class RelayService: if link is None: return try: - await bot.set_message_reaction( + await _with_flood_retry( + bot.set_message_reaction, chat_id=self._manager_group_id, message_id=link.manager_message_id, reaction=reactions, @@ -703,7 +753,8 @@ class RelayService: if link is None: return try: - await bot.set_message_reaction( + await _with_flood_retry( + bot.set_message_reaction, chat_id=link.client_telegram_id, message_id=link.client_message_id, reaction=reactions,