From e3b458d77478e20ab8e06634dc2a93bb263e2064 Mon Sep 17 00:00:00 2001 From: vvzvlad Date: Sun, 5 Jul 2026 04:00:20 +0300 Subject: [PATCH] docs: add staged stability fix plan (static serving + hangs) Detailed 7-stage plan covering: RPC timeouts under the global throttle gate, background worker task_done/queue fixes, atomic large-video downloads, FloodWait-to-429 mapping, FileResponse migration, event-loop hygiene for feed rendering, SQLite write batching, and a lightweight /ping healthcheck. Co-Authored-By: Claude Fable 5 --- plans/2026-07-05-stability-fix-plan.md | 347 +++++++++++++++++++++++++ 1 file changed, 347 insertions(+) create mode 100644 plans/2026-07-05-stability-fix-plan.md diff --git a/plans/2026-07-05-stability-fix-plan.md b/plans/2026-07-05-stability-fix-plan.md new file mode 100644 index 0000000..75d4540 --- /dev/null +++ b/plans/2026-07-05-stability-fix-plan.md @@ -0,0 +1,347 @@ +# План стабилизации pyrogram-bridge: статика + зависания + +Дата: 2026-07-05. База: аудит api_server.py / telegram_client.py / tg_throttle.py / tg_cache.py / file_io.py / rss_generator.py / post_parser.py. + +Проверенные факты, на которых строится план: +- `FloodWait` — подкласс `RPCError` (проверено на установленном Kurigram). +- Starlette 0.45.3 поддерживает Range в `FileResponse` из коробки (проверено по исходникам в venv). +- Дефолтный executor `asyncio.to_thread` = min(32, cpu+4) потоков; в контейнере на 1–2 CPU это 5–6. +- В venv стоит Kurigram 2.2.4, в requirements закреплён 2.2.22 — локальное окружение не соответствует прод-образу. + +## Целевые инварианты (что должно стать верным после ремонта) + +1. Любой файл в кэше с финальным именем (`{file_unique_id}` или `temp_{file_unique_id}`) — гарантированно полный. Частичным может быть только `*.part.*`. +2. Каждый Telegram RPC ограничен таймаутом. Глобальный RPC-гейт не может удерживаться дольше таймаута. +3. Временные ошибки Telegram (FloodWait) никогда не превращаются в постоянные HTTP-ответы (404). +4. Event loop не выполняет CPU-работу > ~50 мс за раз. +5. Каждая фоновая задача supervised: её смерть видна в логах на CRITICAL и/или она перезапускается. +6. Healthcheck не зависит от Telegram RPC и от обхода файловой системы. + +## Процесс + +- Ветка `fix/stability` от `main`, коммит на каждую стадию (на `main` не коммитим). +- Порядок стадий = порядок деплоя: после стадий 1 и 2 уже можно выкатываться и наблюдать. +- После каждой стадии: pytest + ручные сценарии из стадии 7 + код-ревью. +- Перед началом: пересоздать/обновить venv под requirements (Kurigram 2.2.22), прогнать 174 существующих теста как baseline. Добавить dev-зависимости: pytest-asyncio, httpx (для TestClient). + +--- + +## Стадия 1 — устранение зависаний (минимальный диф, максимальный эффект) + +### 1.1 Таймауты на все RPC под глобальным гейтом + +Файлы: tg_cache.py, config.py. + +- Добавить в config `tg_rpc_timeout` (env `TG_RPC_TIMEOUT`, default 60). +- `cached_get_chat_history` (tg_cache.py:142-144): итерацию истории собрать в корутину и обернуть в `asyncio.wait_for` **внутри** `async with tg_rpc()`: + ```python + async with tg_rpc(): + async def _collect(): + # Full paginated history; bounded by the outer wait_for + return [m async for m in client.get_chat_history(channel_id, limit=limit)] + messages = await asyncio.wait_for(_collect(), timeout=Config["tg_rpc_timeout"]) + ``` +- `cached_get_chat` (tg_cache.py:216-217): `await asyncio.wait_for(client.get_chat(channel_id), timeout=...)` внутри гейта. + +**Где не наебаться:** +- `wait_for` должен оборачивать **только RPC**, не `_sem.acquire()`. Ожидание в очереди гейта — легитимный backpressure; если холдер ограничен таймаутом, очередь всегда дренируется. Обернёшь acquire — получишь ложные таймауты при штатной очереди из 47 фидов miniflux. +- `async for` нельзя завернуть в wait_for напрямую — только через промежуточную корутину (как в эскизе). +- CancelledError не глотать: `except Exception` в вызывающих местах её и так не ловит (это BaseException в 3.11) — не «улучшать» до `except BaseException`. + +### 1.2 Таймауты и гейт на остальные живые RPC + +- `_reply_enrichment` (rss_generator.py:505): обернуть `client.get_messages` в `async with tg_rpc()` + `wait_for`. +- `PostParser.get_post` (post_parser.py:98): `wait_for(..., 30)`; заодно удалить `print(...)` (post_parser.py:95). +- `/health` (api_server.py:906): `wait_for(client.client.get_me(), 10)`. + +### 1.3 Починка фонового воркера и очереди + +Файл: api_server.py. + +- `background_download_worker` (620-637): `task_done()` вызывать только если элемент был реально получен: + ```python + while True: + item = await download_queue.get() # cancellation propagates cleanly here + channel, post_id, file_unique_id = item + try: + async with BACKGROUND_DOWNLOAD_SEMAPHORE: + await download_media_file(channel, post_id, file_unique_id) + await asyncio.sleep(2) + except errors.FloodWait as e: + logger.warning(f"bg_download_floodwait: sleeping {e.value}s") + await asyncio.sleep(min(int(e.value) + 5, 900)) + except Exception as e: + logger.error(f"Background download error ...: {e}") + finally: + download_queue.task_done() + ``` +- `download_new_files` (605-609): `await download_queue.put(...)` → `download_queue.put_nowait(...)` — иначе `except asyncio.QueueFull` остаётся мёртвым кодом, а заполненная очередь навсегда блокирует весь `cache_media_files` (включая удаление старых файлов). + +**Где не наебаться:** +- Сейчас смерть воркера убивает и уборку кэша (цепочка: воркер умер → очередь заполнилась → `await put()` завис → свипер больше не крутится). После фикса проверить тестом: воркер, у которого download постоянно бросает исключение, продолжает жить и `queue.join()` завершается. +- FloodWait в воркере надо ловить **до** Exception и спать, иначе воркер будет молотить телеграм под флудом. + +### 1.4 Supervision фоновых задач + +- В lifespan повесить `add_done_callback` на `background_task` и `worker_task`: если таск завершился не через CancelledError — лог CRITICAL с exception + перезапуск таска (обёртка `_supervised(factory)` с ограничением частоты рестартов, например не чаще раза в 60 с). + +### Тесты стадии 1 +- Гейт: мок «зависшего» RPC (asyncio.Event, который никогда не сеттится) → первый вызов отваливается по таймауту, второй проходит; пермит не утёк. +- Воркер: download бросает Exception/FloodWait → воркер жив, task_done сбалансирован. +- `_TgRpcGate`: отмена во время ожидания spacing не теряет пермит (уже реализовано — закрепить тестом). + +### DoD стадии 1 +Ни один путь кода не ждёт Telegram без таймаута; воркер не умирает молча; pytest зелёный. + +--- + +## Стадия 2 — статика: большие видео, FloodWait, семафор + +### 2.1 Единый путь скачивания с атомарным rename (главный фикс флаки) + +Файл: api_server.py, telegram_client.py. + +Сейчас большие видео (>100MB) качаются напрямую в финальное имя `temp_{fid}` (379-396): конкурентный запрос видит частичный файл и отдаёт его; при таймауте обрезок не удаляется и целый час отдаётся как «готовый». + +- Выделить хелпер: + ```python + async def _download_atomic(file_id: str, final_path: str, timeout: float) -> str: + # Download to a unique partial path, validate, atomically rename. + part_path = f"{final_path}.part.{uuid.uuid4().hex}" + try: + await client.safe_download_media(file_id, part_path, timeout=timeout) + if not os.path.exists(part_path) or os.path.getsize(part_path) == 0: + raise ZeroSizeFileError(...) + if not os.path.exists(final_path): + os.rename(part_path, final_path) # atomic on POSIX + return final_path + finally: + # Always clean up our partial file (timeout, cancel, race loser) + if os.path.exists(part_path): + try: os.remove(part_path) + except OSError: pass + ``` +- Использовать его и для обычных файлов, и для больших видео (финальное имя больших остаётся `temp_{fid}` — семантика «не кэшировать постоянно, чистить по TTL» сохраняется). +- `safe_download_media` научить принимать `timeout` параметром; для больших видео таймаут от размера: `min(1800, max(120, file_size // (256*1024)))` (≈256 KB/s минимально допустимая скорость), env-конфигурируемо. +- Проверку `if os.path.exists(temp_file_path): return` (382-384) заменить: существующий `temp_{fid}` теперь **гарантированно полный** (появляется только через rename) — отдавать можно смело. + +**Где не наебаться:** +- Единый суффикс `.part.{hex}` вместо старого `.tmp.{hex}`. Обновить regex свипера (api_server.py:536) на новый суффикс, **оставив** и старый паттерн на переходный период (на диске могут лежать старые обрезки). +- Инвариант «финальное имя = полный файл» ломается, если хоть один путь пишет мимо `.part.` — после правки grep-ом убедиться, что `download_media` больше нигде не получает финальный путь. +- В `download_new_files` (598-600) проверка «temp существует → скипаем» остаётся корректной: полный файл есть — качать не надо. Частичные `.part.` под неё не попадают — это правильно. +- Первый запрос большого видео всё ещё отвечает только после полного скачивания (минуты). Это осознанное ограничение; прогрессивный стриминг через `client.stream_media` — отдельная большая фича, в этот план не входит (пометить как future work). + +### 2.2 Дедупликация конкурентных скачиваний (in-flight registry) + +- Модульный `_inflight: dict[tuple[str, int, str], asyncio.Future]`. +- Первый запрос: создаёт future, запускает скачивание **отдельным** `asyncio.create_task` (не в контексте HTTP-запроса!), в finally сеттит result/exception и удаляет ключ. Остальные: `await`ят future. + +**Где не наебаться (классическая ловушка):** +- Если качать прямо в корутине первого HTTP-запроса, отключение его клиента отменит скачивание, future никогда не завершится, и все ожидающие повиснут. Поэтому: скачивание — в detached task; ожидающие делают `await asyncio.wait_for(fut, timeout)`. +- В finally у task обязательно и `set_exception`, и `pop` ключа — иначе после первой ошибки ключ навсегда «занят» отработавшим future. +- `set_exception` на future, который никто не await-ит, даст "exception was never retrieved" warning — допустимо, но можно гасить через `fut.exception()` в done-callback. + +### 2.3 FloodWait → 429 в /media + +- В `get_media` добавить обработчик **до** `except errors.RPCError` (api_server.py:1001): + ```python + except errors.FloodWait as e: + retry_after = min(int(e.value) + random.randint(1, 30), 300) + return Response(status_code=429, content="Telegram flood wait", + headers={"Retry-After": str(retry_after)}) + ``` + +**Где не наебаться:** порядок except-веток решает: FloodWait — подкласс RPCError, поставишь после — не сработает никогда. Закрепить тестом (мок download_media_file бросает FloodWait → ответ 429, не 404). + +### 2.4 Не удалять большое видео «из-под зрителя» + +- При каждой отдаче файла с именем `temp_*` — `await asyncio.to_thread(os.utime, path)` (touch mtime). Свипер (порог 1 час по mtime) перестанет удалять активно просматриваемые файлы. + +### 2.5 Ограничить ожидание HTTP-семафора + +- Вместо `async with HTTP_DOWNLOAD_SEMAPHORE`: явный `acquire` под `wait_for(30)`; таймаут → 503 + `Retry-After: 30`. Release — только если acquire удался (try/finally вокруг критической секции, а не вокруг acquire). + +### Тесты стадии 2 +- Конкурентные запросы большого видео с медленным фейковым download: второй запрос НЕ получает частичный файл (либо ждёт future, либо 503/504), после завершения оба получают полный. +- Таймаут download → `.part.` удалён, финального имени нет. +- FloodWait → 429 c Retry-After. +- Свипер: чистит и `.part.`, и старые `.tmp.`, не трогает свежие. + +### DoD стадии 2 +Ни при каком сценарии клиенту не может быть отдан неполный файл; флуд отдаёт 429; обрезки не живут дольше часа. + +--- + +## Стадия 3 — отдача файлов через FileResponse, чистка HTTP-слоя + +### 3.1 Заменить самодельный стриминг на `FileResponse` + +Файл: api_server.py (`prepare_file_response`, 203-335). + +Starlette 0.45.3 сам обрабатывает Range/If-Range/206/416/multipart, ставит Accept-Ranges/ETag/Last-Modified и читает файл эффективно (без нашего to_thread на каждые 64KB): + +```python +return FileResponse( + file_path, + media_type=media_type, + filename=os.path.basename(file_path), # sets Content-Disposition: inline; filename* + content_disposition_type="inline", + headers={"Cache-Control": "public, max-age=86400, immutable"}, + background=background, +) +``` + +Оставить: пре-чек существования (404), MIME-логику (magic + SQLite-кэш типа). Удалить: ручной парсинг Range, `file_chunk_generator`, ручные заголовки. + +**Где не наебаться:** +- **Сначала тесты, потом свап.** Написать httpx TestClient-тесты на текущее поведение (bytes=0-499, bytes=500-, bytes=-500, start за EOF → 416, мусорный заголовок), затем перейти на FileResponse и осознанно принять расхождения (например, Starlette на кривой заголовок может отдать 200-полный вместо 416 — это допустимо по RFC 7233). +- FileResponse проверяет файл на этапе отправки: пре-чек на 404 оставить обязательно. +- `filename*=UTF-8''...` FileResponse формирует сам — руками Content-Disposition не собирать, иначе задвоится. + +### 3.2 Убрать BaseHTTPMiddleware + +- `RequestLoggingMiddleware` (api_server.py:54-64) удалить. Логирование запросов: либо `--access-log` uvicorn на debug-уровне, либо чистый ASGI-middleware из ~10 строк (без BaseHTTPMiddleware): меньше оверхеда на каждый чанк стриминга и никаких сюрпризов с отменой/фоновыми тасками. + +### 3.3 Расширить дефолтный executor + +- В lifespan: `loop.set_default_executor(ThreadPoolExecutor(max_workers=32, thread_name_prefix="io"))`. Даже после ухода per-chunk чтений остаются SQLite/magic/pickle/os.walk — 5–6 дефолтных потоков в контейнере мало. + +### DoD стадии 3 +Range-тесты зелёные; profiling-запрос большого файла не порождает поток-на-чанк; поведение эндпоинта эквивалентно (кроме осознанных RFC-допущений). + +--- + +## Стадия 4 — гигиена event loop (генерация фидов) + +### 4.1 `raw_message` — лениво + +- `process_message(..., include_raw: bool = False)`: `str(message)` (полная сериализация каждого поста! post_parser.py:523) выполнять только для JSON-выдачи и debug-HTML. В фидах — никогда. + +### 4.2 Убрать side-effect IO из process_message (обезвредить ловушку до 4.3) + +Сейчас `_generate_html_media` → `_save_media_file_ids` (post_parser.py:676, 977-1028) делает `asyncio.get_running_loop()` + `create_task`. **Если перенести рендер в поток (4.3) без этой правки, `get_running_loop()` бросит RuntimeError, ветка упадёт в `except` с логом — и записи media id молча перестанут сохраняться, а фоновый прогрев кэша умрёт.** + +- Рефакторинг: `_save_media_file_ids` не пишет в БД, а складывает `(channel, post_id, file_unique_id, ts)` в `self._pending_media_ids` инстанса PostParser (инстанс создаётся на запрос — потокобезопасно, т.к. рендер идёт в одном потоке). +- После рендера вызывающий код (rss_generator / get_post) один раз делает `await asyncio.to_thread(upsert_media_file_ids_bulk_sync, DB_PATH, entries)` — новая функция в file_io.py с `executemany`. Это заодно даёт батчинг апсертов (часть стадии 5). +- Счётчик `_persist_pending_count` и create_task-механику удалить. + +### 4.3 Рендер фида — в поток + +- `_create_time_based_media_groups`, `_create_messages_groups`, `_trim_messages_groups`, `_render_messages_groups` — внутри нет ни одного await; превратить в обычные sync-функции и выполнять одним `await asyncio.to_thread(_render_pipeline, ...)` из `generate_channel_rss` / `generate_channel_html`. +- `copy.deepcopy(messages)` (rss_generator.py:38) уезжает в поток вместе со всем остальным. + +**Где не наебаться:** +- GIL: перенос CPU-работы в поток не делает её бесплатной — луп перестаёт стоять колом, но замедляется. Поэтому 4.1/4.4 (сокращение работы) обязательны, а не опциональны. +- Внутри потока не должно остаться ни `create_task`, ни `get_running_loop` (это ровно 4.2), ни обращений к asyncio-примитивам. +- deepcopy живых Pyrogram-объектов сегодня работает — код не менять, только переместить; отдельным тестом убедиться, что deepcopy Message из pickle-кэша не падает. Полный отказ от deepcopy (не-мутирующая группировка через map message_id → group_id) — отдельный опциональный рефакторинг, в эту стадию не тащить. + +### 4.4 Сократить sanitize-проходы + +Сейчас на каждое сообщение bleach вызывается 4+ раз (body: post_parser.py:672, media: 752, footer: 833, reactions: 920), плюс финальный проход по всему фиду. Bleach — самая дорогая CPU-часть (есть diag_sanitize_slow > 50 мс на вызов). + +- Правило: **один sanitize на каждую выходную границу**, не на каждый фрагмент: + - RSS/HTML-фиды: финальный проход уже есть (rss_generator.py:443, 643) → внутренние per-message проходы убрать. + - `/html/{channel}/{post_id}`: единственный проход добавить в `_format_html` (или в эндпоинте) — сейчас он полагается на внутренние. + - `/json/...`: html-поля внутри JSON должны оставаться чищеными → один проход по body+footer в `process_message`. + +**Где не наебаться (XSS!):** +- Перед удалением внутренних проходов построить карту «выходная точка → где чистится», и только потом резать. Ни один выходной путь не должен остаться без ровно одного прохода. +- Добавить тесты безопасности: мок-сообщение с `