diff --git a/.env.example b/.env.example index 758ec3b..95f94fb 100644 --- a/.env.example +++ b/.env.example @@ -2,7 +2,9 @@ BOT_TOKEN= # Прокси для Telegram API. Пусто — без прокси. -# В Docker на Linux: http://host.docker.internal:20171 +# В Docker на Linux рекомендуется SOCKS5: +# socks5h://host.docker.internal:20170 +# HTTP-вариант: http://host.docker.internal:20171 TELEGRAM_PROXY= # 0 — повторять до успеха; иначе число попыток TELEGRAM_RETRY_ATTEMPTS=0 diff --git a/README.md b/README.md index 5379139..ac46f1a 100644 --- a/README.md +++ b/README.md @@ -48,14 +48,17 @@ Copy-Item .env.example .env Если сервер не имеет прямого доступа к Telegram, укажи прокси V2RayA/HTTP/SOCKS5: ```dotenv -# Docker на Linux (V2RayA слушает на хосте) -TELEGRAM_PROXY=http://host.docker.internal:20171 +# Docker на Linux, рекомендуется SOCKS5 +TELEGRAM_PROXY=socks5h://host.docker.internal:20170 + +# HTTP-вариант +# TELEGRAM_PROXY=http://host.docker.internal:20171 # Запуск без Docker, на том же хосте -# TELEGRAM_PROXY=http://127.0.0.1:20171 +# TELEGRAM_PROXY=socks5h://127.0.0.1:20170 ``` -Для SOCKS5 используй `socks5h://host.docker.internal:20170`; зависимость `socksio` уже включена в `requirements.txt`. +SOCKS5 стабильнее держит long polling; зависимость `socksio` уже включена в `requirements.txt`. ## Запуск @@ -177,6 +180,7 @@ TELEGRAM_HEALTH_INTERVAL=60 - Backoff растёт до `TELEGRAM_RETRY_MAX_DELAY`. - Критичные отправки (результат, счёт, подтверждение) повторяются бесконечно; обновления статуса — ограниченно, чтобы не плодить зависшие задачи. - Раз в `TELEGRAM_HEALTH_INTERVAL` выполняется `getMe`; при сбое проверка повторяется с backoff, пока Telegram не ответит. +- Watchdog проверяет pending updates: если они не обрабатываются 5 минут, polling перезапускается, а при неудаче контейнер перезапускается автоматически. ## Подтверждение возраста diff --git a/bot.py b/bot.py index fc1e7e6..58ee623 100644 --- a/bot.py +++ b/bot.py @@ -2,12 +2,14 @@ from __future__ import annotations import asyncio import logging +import os import time import uuid from datetime import datetime, timedelta, timezone from datetime import time as dt_time from pathlib import Path +import httpx from telegram import ( CopyTextButton, InlineKeyboardButton, @@ -25,6 +27,7 @@ from telegram.ext import ( ContextTypes, MessageHandler, PreCheckoutQueryHandler, + TypeHandler, filters, ) @@ -38,6 +41,21 @@ from workflows import WorkflowBuilder logger = logging.getLogger(__name__) MSK = timezone(timedelta(hours=3)) +POLLING_WATCHDOG_INTERVAL = 60.0 +POLLING_WATCHDOG_STUCK_AFTER = 300.0 + + +def polling_is_stuck( + *, + now: float, + last_update_at: float, + pending_update_count: int, + stuck_after: float, +) -> bool: + return ( + pending_update_count > 0 + and now - last_update_at >= stuck_after + ) START_TEXT = ( "Пришли текст — сгенерирую картинку.\n" @@ -71,17 +89,24 @@ class MpakoBot: self._last_edit: dict[int, tuple[float, int]] = {} self._daily_task: asyncio.Task[None] | None = None self._health_task: asyncio.Task[None] | None = None + self._watchdog_task: asyncio.Task[None] | None = None + self._last_update_at = time.monotonic() self._application: Application | None = None def run(self) -> None: builder = ( Application.builder() .token(self._settings.bot_token) - .connect_timeout(20) - .read_timeout(60) - .write_timeout(60) - .get_updates_connect_timeout(20) - .get_updates_read_timeout(60) + .connect_timeout(10) + .read_timeout(30) + .write_timeout(30) + .pool_timeout(5) + .connection_pool_size(10) + .get_updates_connect_timeout(10) + .get_updates_read_timeout(15) + .get_updates_write_timeout(15) + .get_updates_pool_timeout(5) + .get_updates_connection_pool_size(10) .post_init(self._post_init) .post_shutdown(self._post_shutdown) ) @@ -91,6 +116,9 @@ class MpakoBot: ) application = builder.build() self._application = application + application.add_handler( + TypeHandler(Update, self._track_update), group=-1 + ) application.add_handler(CommandHandler("start", self._start)) application.add_handler(CommandHandler("help", self._help)) application.add_handler(CommandHandler("status", self._status)) @@ -136,6 +164,9 @@ class MpakoBot: self._health_task = asyncio.create_task( self._telegram_health_loop(), name="telegram-health" ) + self._watchdog_task = asyncio.create_task( + self._polling_watchdog_loop(), name="polling-watchdog" + ) enhancer = ( PromptEnhancer(self._comfy, self._settings) if self._settings.prompt_enhance_enabled @@ -153,6 +184,10 @@ class MpakoBot: logger.info("Бот запущен, ComfyUI: %s", self._settings.comfy_base_url) async def _post_shutdown(self, application: Application) -> None: + if self._watchdog_task is not None: + self._watchdog_task.cancel() + await asyncio.gather(self._watchdog_task, return_exceptions=True) + self._watchdog_task = None if self._health_task is not None: self._health_task.cancel() await asyncio.gather(self._health_task, return_exceptions=True) @@ -168,6 +203,79 @@ class MpakoBot: await self._db.close() logger.info("Бот остановлен") + async def _track_update( + self, update: Update, context: ContextTypes.DEFAULT_TYPE + ) -> None: + self._last_update_at = time.monotonic() + + async def _polling_watchdog_loop(self) -> None: + while True: + await asyncio.sleep(POLLING_WATCHDOG_INTERVAL) + pending = await self._pending_updates_count() + if pending is None: + continue + if not polling_is_stuck( + now=time.monotonic(), + last_update_at=self._last_update_at, + pending_update_count=pending, + stuck_after=POLLING_WATCHDOG_STUCK_AFTER, + ): + continue + logger.warning( + "Polling watchdog: %s обновлений не обрабатываются," + " перезапускаю updater", + pending, + ) + await self._restart_polling() + + async def _pending_updates_count(self) -> int | None: + url = ( + "https://api.telegram.org/bot" + f"{self._settings.bot_token}/getWebhookInfo" + ) + proxy = self._settings.telegram_proxy or None + try: + async with httpx.AsyncClient( + proxy=proxy, + timeout=httpx.Timeout(10.0, connect=5.0), + ) as client: + response = await client.get(url) + data = response.json() + except (httpx.HTTPError, ValueError) as exc: + logger.warning( + "Polling watchdog: проверка не удалась (%s)", + type(exc).__name__, + ) + return None + if not data.get("ok"): + return None + result = data.get("result") or {} + return int(result.get("pending_update_count") or 0) + + async def _restart_polling(self) -> None: + application = self._application + if application is None or application.updater is None: + return + try: + await asyncio.wait_for( + application.updater.stop(), timeout=15 + ) + await asyncio.wait_for( + application.updater.start_polling( + allowed_updates=Update.ALL_TYPES, + drop_pending_updates=False, + ), + timeout=30, + ) + except Exception: + logger.exception( + "Polling watchdog: не удалось перезапустить updater," + " выхожу для рестарта контейнера" + ) + os._exit(1) + self._last_update_at = time.monotonic() + logger.warning("Polling watchdog: updater перезапущен") + async def _daily_credits_loop(self) -> None: while True: now = datetime.now(MSK) diff --git a/checklist.md b/checklist.md index e7999ea..b4d7c9b 100644 --- a/checklist.md +++ b/checklist.md @@ -8,7 +8,7 @@ - Реализованы блоки 2–7, базовое хранилище (9), файлы Docker (10), тесты (11), README (12), веб-история (13), улучшение промптов (14), модерация (15), оплата Stars (16), баланс/рефералы (17), управление пользователями (18), возрастной gate (19) и надёжность Telegram (20). - Вживую проверены: T2I (Z-Image), I2I (Krea2), WebSocket-прогресс, улучшение промптов, Docker, CI и веб-история. -- 56 тестов: 34 проходят локально без БД, 22 теста БД выполняются в CI на PostgreSQL через `TEST_DATABASE_URL`. Покрыты workflows, очередь, веб-история, enhancer, модерация, платежи, премиум-очередь, баланс/рефералы, управление пользователями, retry, миграция. +- 59 тестов: 37 проходят локально без БД, 22 теста БД выполняются в CI на PostgreSQL через `TEST_DATABASE_URL`. Покрыты workflows, очередь, веб-история, enhancer, модерация, платежи, премиум-очередь, баланс/рефералы, управление пользователями, retry, миграция, watchdog. - Не проверено: реальная оплата Stars и реферальная выплата в Telegram, очередь из 3+ пользователей, падение ComfyUI, рестарт с `running`-задачей, детекция лица на реальном фото. PostgreSQL-миграция и тесты БД ещё не запускались вживую — это сделает CI и первый деплой. ## 0. Допущения и решения по умолчанию @@ -376,7 +376,8 @@ ON requests(user_id) WHERE status IN ('queued', 'running'); - [x] `_retry` с настраиваемыми попытками, backoff и режимом «до успеха». - [x] Health-check `getMe` с повторами. - [x] Ретраи критичных отправок и загрузок файлов. -- [x] Тесты retry. +- [x] Watchdog polling: если pending updates не убывают 5 минут, updater перезапускается, при неудаче перезапускается контейнер. +- [x] Тесты retry и watchdog. **DoD:** при таймаутах прокси операции повторяются, health-check фиксирует восстановление связи. diff --git a/requirements.txt b/requirements.txt index a93f530..b450652 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ python-telegram-bot>=21.6,<23 -httpx[socks]>=0.27,<1 +httpx[socks]>=0.28,<1 psycopg[binary]>=3.2,<4 psycopg-pool>=3.2,<4 websockets>=13,<16 diff --git a/tests/test_watchdog.py b/tests/test_watchdog.py new file mode 100644 index 0000000..f8fa439 --- /dev/null +++ b/tests/test_watchdog.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from bot import polling_is_stuck + + +def test_polling_is_stuck_for_unprocessed_updates() -> None: + assert ( + polling_is_stuck( + now=1000.0, + last_update_at=100.0, + pending_update_count=2, + stuck_after=300.0, + ) + is True + ) + + +def test_polling_is_not_stuck_without_pending_updates() -> None: + assert ( + polling_is_stuck( + now=1000.0, + last_update_at=100.0, + pending_update_count=0, + stuck_after=300.0, + ) + is False + ) + + +def test_polling_is_not_stuck_after_recent_update() -> None: + assert ( + polling_is_stuck( + now=1000.0, + last_update_at=900.0, + pending_update_count=2, + stuck_after=300.0, + ) + is False + )