diff --git a/api_server.py b/api_server.py index 6b497af..6fbe2ea 100644 --- a/api_server.py +++ b/api_server.py @@ -33,7 +33,7 @@ from pyrogram import errors from pyrogram.types import Message from pyrogram.enums import MessageMediaType from fastapi import FastAPI, HTTPException, Response, Request -from fastapi.responses import HTMLResponse, FileResponse +from fastapi.responses import HTMLResponse, FileResponse, JSONResponse from telegram_client import TelegramClient from config import get_settings, setup_logging from rss_generator import generate_channel_rss, generate_channel_html @@ -1065,6 +1065,46 @@ async def get_raw_post_json(channel: str, post_id: int, request: Request, token: raise HTTPException(status_code=500, detail=error_message) from e +@app.get("/ping") +async def ping() -> JSONResponse: + """Lightweight liveness probe for the container healthcheck. + + Reflects process/event-loop liveness (always answers in microseconds) and TG liveness + from the watchdog's last-probe data. It MUST NOT issue any Telegram RPC (no get_me, + no safe_get_messages), touch SQLite, or walk the filesystem — that is the whole point: + it stays instant and truthful even while a real TG RPC is hung. It only reads the + already-recorded watchdog timestamp and the is_connected bool. + """ + age = client.watchdog_last_ok_age() # seconds since last OK probe, None if never + # is_connected is None before client.start() and a bool afterwards; coerce so the JSON + # "connected" field is always a bool (never null) and the pre-start window reports false. + connected = bool(client.client.is_connected) + threshold = Config["tg_ping_unhealthy_after"] + # age is None right after boot: the watchdog hasn't run its first probe yet. Treat that + # as healthy (gate on connected only) so a freshly-started container is not killed before + # its first probe cycle — otherwise start_period would have to cover a full watchdog interval. + # + # The staleness branch (age >= threshold => degraded) is only meaningful while the watchdog + # is running to refresh age. With the watchdog DISABLED (TG_WATCHDOG_ENABLED=false) nothing + # refreshes age — yet a disconnect-flap restart can still stamp it once (see _restart_client, + # which runs before the watchdog-enabled gate), after which age only grows. Letting that + # stale age drive /ping to 503 would spuriously fail the container healthcheck on a live + # connection and trigger an autoheal restart. So gate staleness on the watchdog being on; + # with it off, /ping is a pure connectivity check (no zombie-session detection — that + # TG-liveness signal only exists while the watchdog runs). + healthy = connected and ( + not Config["tg_watchdog_enabled"] or age is None or age < threshold + ) + return JSONResponse( + { + "status": "ok" if healthy else "degraded", + "connected": connected, + "last_probe_age_s": round(age, 1) if age is not None else None, + "threshold_s": threshold, + }, + status_code=200 if healthy else 503, + ) + @app.get("/health") @app.get("/health/{token}") async def health_check(request: Request, token: str | None = None) -> Response: diff --git a/config.py b/config.py index a8fc395..3cc8763 100644 --- a/config.py +++ b/config.py @@ -123,6 +123,17 @@ def get_settings() -> dict[str, Any]: "tg_watchdog_heartbeat_every": _parse_int_env("TG_WATCHDOG_HEARTBEAT_EVERY", 30), "tg_disconnect_flap_limit": _parse_int_env("TG_DISCONNECT_FLAP_LIMIT", 3), "tg_disconnect_flap_window": _parse_int_env("TG_DISCONNECT_FLAP_WINDOW", 120), + # /ping reports TG as unhealthy once the last successful watchdog probe is older than + # this many seconds. Default is derived from the watchdog cadence: it is roughly how + # long the watchdog itself would take to give up and trigger a restart — + # interval * (failures + 1) + timeout. With the defaults (60,3,10) that is 250s, so a + # transient slow probe never flaps /ping, but a genuinely stuck session (no successful + # probe for ~4 min) surfaces as 503 before/around the time the watchdog restarts. + "tg_ping_unhealthy_after": _parse_int_env( + "TG_PING_UNHEALTHY_AFTER", + _parse_int_env("TG_WATCHDOG_INTERVAL", 60) * (_parse_int_env("TG_WATCHDOG_FAILURES", 3) + 1) + + _parse_int_env("TG_WATCHDOG_TIMEOUT", 10), + ), # Media download timeout scales with file size (large videos): the per-download # timeout is clamped to [min, max] seconds, with an effective floor of # `media_download_min_speed` bytes/s (timeout ≈ file_size / min_speed). diff --git a/dockercompose.yml b/dockercompose.yml index c22711a..4fa8bda 100644 --- a/dockercompose.yml +++ b/dockercompose.yml @@ -53,10 +53,15 @@ services: com.centurylinklabs.watchtower.enable: "true" autoheal: true healthcheck: - test: ["CMD", "curl", "-s", "-f", "-o", "/dev/null", "http://127.0.0.1:80/rss/vvzvlad_lytdybr?limit=1"] - interval: 30m + # Lightweight process/loop liveness probe. /ping never touches Telegram or the + # filesystem, so it answers instantly even while a TG RPC is hung — unlike the old + # /rss?limit=1 check, which could exceed the 5s timeout on a cold cache and get the + # container restarted mid-download (corrupted temp files). TG liveness is now judged + # by the in-process watchdog, which /ping reports via its last-probe age. + test: ["CMD", "curl", "-sf", "http://127.0.0.1:80/ping"] + interval: 5m timeout: 5s - retries: 2 + retries: 3 start_period: 30s start_interval: 5s diff --git a/docs/stability-verification.md b/docs/stability-verification.md new file mode 100644 index 0000000..699cf6d --- /dev/null +++ b/docs/stability-verification.md @@ -0,0 +1,148 @@ +# Стадия 7 — сквозная верификация плана стабилизации + +Дата: 2026-07-05. Ветка: `fix/stage-7-verification` (на базе `fix/stage-6-healthcheck`, +кумулятивно содержит стадии 1–6). Эта стадия **ничего не меняет в поведении** прод-кода — +только автоматизирует проверки и фиксирует, что должен наблюдать оператор после деплоя. + +## Итог автотестов + +Каноничный прогон (изоляция `config` через `sys.modules`): `python -m pytest tests/`. + +``` +260 passed +``` + +Раскладка по стадиям (все зелёные): + +| Стадия | Файл тестов | Кол-во | Что покрывает | +|--------|-------------|--------|----------------| +| 1 — анти-зависания | `tests/test_stage1_hangs.py` | 6 | таймаут RPC-гейта без утечки пермита; воркер переживает Exception/FloodWait, `task_done` сбалансирован; отмена в spacing-ожидании не теряет пермит | +| 2 — статика/большие видео | `tests/test_stage2_static.py` | 15 | атомарный `_download_atomic` (publish-on-rename, чистка `.part.` при таймауте/zero-size/гонке); дедуп конкурентных скачиваний; FloodWait→429; touch mtime у `temp_*`; свипер чистит `.part.`+legacy `.tmp.`; баланс HTTP-семафора | +| 3 — FileResponse | `tests/test_stage3_fileresponse.py` | 19 | матрица Range (0-499/500-/-500/за EOF→416/мусор→400/мульти-range→206 multipart); сохранены mtime-touch, delete_after-BackgroundTask, MIME-кэш; чистый ASGI-логгер | +| 4 — гигиена event loop | `tests/test_stage4_eventloop.py` | 19 | ленивый `raw_message`; вынос side-effect IO из `process_message` (bulk upsert media id); рендер-пайплайн ушёл в поток без create_task/get_running_loop; XSS вычищен во всех 4 выводах ровно одним проходом | +| 5 — батчинг SQLite | `tests/test_stage5_sqlite.py` | 8 | кэш-хит пишет в аккумулятор, а не в SQLite; flush→DB; snapshot-then-clear не теряет поздние апдейты; re-queue при сбое; str(channel)-ключи | +| 6 — healthcheck | `tests/test_stage6_healthcheck.py` | 11 | `/ping` 200/503 по connected+age; ноль TG RPC; watchdog отключён→чистая проверка коннекта; `watchdog_last_ok_age()` | +| 7 — интеграция | `tests/test_stage7_integration.py` | 8 | сквозные сценарии (см. ниже) | +| — регрессия парсера | `tests/test_postparser_*.py` | 174 | заголовки/флаги/автор (существовавшие до плана) | + +## Новые интеграционные тесты стадии 7 (DoD → сквозное доказательство) + +`tests/test_stage7_integration.py` драйвит **реальные точки входа** (`get_media`, `ping`, +flush), чтобы поймать регрессии, которые проявляются только при взаимодействии стадий: + +1. **Range на уровне маршрута `/media`** — `test_media_route_range_prefix_0_99`, + `_range_suffix_last_100`, `_range_unsatisfiable_416`. Прогоняют `get_media` (кэш-хит) → + `prepare_file_response` → `FileResponse` через реальный ASGI: `bytes=0-99`→206 с точным + Content-Range/длиной и байтами, `bytes=-100`→206 (суффикс), `bytes=999999999-`→416 (`*/size`). + Ловит: если кэш-хит перестанет доходить до FileResponse (ре-буферизация тела, ручные + заголовки) или сломается связка digest-гейта — Range перестанет работать. (Стадия 3 + пинит это на `prepare_file_response` напрямую; здесь — сшивка стадий 2+3.) +2. **`/ping` быстр и без RPC при висящей операции** — `test_ping_prompt_and_rpc_free_while_slow_op_pending` + и `_reports_degraded_promptly_while_slow_op_pending`. Пока фейковая медленная корутина + (модель зависшего hot-path) висит на `Event`, `ping()` возвращает 200/503 корректно и + **до** завершения медленной операции, при нуле вызовов `get_me`/`safe_get_messages`. + Ловит: рекаплинг `/ping` к TG RPC или к любому awaitable, который может застопориться. +3. **Дедуп + очистка при disconnect через реальный `get_media`** — + `test_get_media_concurrent_shares_one_download_and_drains_registry` (2 конкурентных + запроса → одна скачка, `_inflight` пуст) и `_request_cancel_does_not_stick_registry_or_hang_sibling` + (отмена запроса-«клиента» не оставляет застрявший ключ и не вешает соседа). Ловит: + возврат скачивания в корутину запроса (отмена убила бы download) или потерю `finally`-pop. +4. **str(channel)-ключ access-time end-to-end** — `test_media_cache_hit_flush_updates_str_keyed_db_row`. + Кэш-хит `/media` записывает str-ключ, flush обновляет ту же строку в реальной SQLite + (hit→flush→DB). Сшивает две половины: ключ, который пишет hot-path, и WHERE, по которому + апдейтит flush. Ловит любое расхождение ключа аккумулятора и WHERE bulk-UPDATE (напр. + перестановку колонок в SQL — проверено мутацией: тест краснеет, `added` остаётся stale). + (int/str-хазард самого канала живёт в `download_media_file` и закрыт стадией 5 — + `test_cache_hit_keys_channel_as_str`; здесь покрыта связка get_media+flush.) + +## Соответствие DoD стадий → доказательство + +- **DoD 1** «ни один путь не ждёт TG без таймаута; воркер не умирает молча» → + `test_stage1_hangs.py` (гейт-таймаут, живучесть воркера). Живая проверка supervision + под нагрузкой — **наблюдение оператора** (лог `supervised_task_crashed/…_exited`). +- **DoD 2** «клиенту никогда не отдаётся неполный файл; флуд→429; обрезки не живут >часа» → + `test_stage2_static.py` целиком + интеграционный дедуп-тест стадии 7. +- **DoD 3** «Range-тесты зелёные; поведение эндпоинта эквивалентно (± RFC-допущения)» → + `test_stage3_fileresponse.py` + Range на уровне `/media` (стадия 7). «Нет потока-на-чанк» + — **наблюдение оператора** (нагрузочный запрос большого файла, число io-потоков). +- **DoD 4** «генерация 100-сообщ. фида не блокирует луп (параллельный /ping <100 мс); XSS + зелёный; media id сохраняются» → `test_stage4_eventloop.py` (рендер в потоке, XSS, + bulk upsert) + `/ping`-decoupling стадии 7. Живой замер «<100 мс на проде» — + **наблюдение оператора**. +- **DoD 5** «на кэш-хит /media ноль обращений к SQLite; фоновая запись раз в минуту» → + `test_stage5_sqlite.py` + str-ключ end-to-end стадии 7. +- **DoD 6** «во время зависшего TG RPC /ping отвечает мгновенно (503); контейнер не + рестартится от медленного фида» → `test_stage6_healthcheck.py` + `/ping`-under-slow-op + стадии 7. Отсутствие autoheal-рестартов на холодном кэше — **наблюдение оператора**. + +## Ручные curl-сценарии для оператора (пост-деплой, локально в контейнере) + +Подставить реальный `{channel}/{post_id}/{fid}/{digest}` (валидная подпись обязательна). + +```bash +# 1. Range-тройка (ожидания: 206 / 206 / 416) +curl -s -D- -o /dev/null -H "Range: bytes=0-99" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}" +curl -s -D- -o /dev/null -H "Range: bytes=-100" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}" +curl -s -D- -o /dev/null -H "Range: bytes=999999999-" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}" + +# 2. Параллельные запросы одного БОЛЬШОГО видео (>100 MB), пока не в кэше: +# оба должны получить ПОЛНЫЙ файл (одинаковый размер), без частичной отдачи. +URL="http://127.0.0.1:80/media/{ch}/{pid}/{bigfid}/{digest}" +curl -s -o /tmp/a "$URL" & curl -s -o /tmp/b "$URL" & wait +ls -l /tmp/a /tmp/b # размеры совпадают и равны полному файлу; на диске нет *.part.* + +# 3. Обрыв на середине стрима — нет утечки тасков/фд: +# считать fd до/после и убедиться, что не растут монотонно. +lsof -p $(pgrep -f api_server) | wc -l # baseline +timeout 2 curl -s -o /dev/null "$URL"; sleep 5 # оборвать скачку на середине (несколько раз) +lsof -p $(pgrep -f api_server) | wc -l # не выросло относительно baseline + +# 4. Генерация большого фида + параллельный /ping (<100 мс во время генерации): +curl -s -o /dev/null "http://127.0.0.1:80/rss/{big_channel}" & +for i in $(seq 1 20); do curl -s -o /dev/null -w "%{time_total}\n" "http://127.0.0.1:80/ping"; done +wait # все замеры /ping должны быть заметно < 0.100 s +``` + +> Сценарии 2 и 3 (реальная скачка большого видео + реальный подсчёт fd через `lsof`) в +> headless-тестах **намеренно не подделаны** — им нужен живой сервер, реальные загрузки и +> реальные файловые дескрипторы. Дедуп-инвариант и disconnect-очистка доказаны на уровне +> реестра/`get_media` (тесты стадии 7 №3), но «нет утечки fd на проде» проверяет оператор. + +## Diag-логи для наблюдения после деплоя + +Ожидаемая динамика (сравнить с до-деплойным baseline): + +- `diag_semaphore_wait` — ожидание HTTP-семафора должно **упасть** (реже/меньше секунд). +- `diag_download_timing` — время скачивания стабилизируется; нет длинных «зависаний». +- `diag_sanitize_slow` — почти **исчезнуть** (один sanitize на выходную границу, стадия 4). +- `watchdog: heartbeat` — **продолжаются** штатно (живость TG-сессии). +- На момент FloodWait — **нет всплеска 404** на `/media`; вместо этого `media_flood_wait` + и ответы **429** с `Retry-After` (стадия 2.3). +- Признаки supervision: любые `supervised_task_crashed` / `supervised_task_exited` на + CRITICAL — сигнал разобраться, но задача при этом перезапускается (не тихая смерть). + +## План отката (rollback) + +Единица отката — **стадия целиком**, не отдельный коммит. Каждая стадия живёт на своей +ветке/PR (`fix/stage-1-hangs` … `fix/stage-7-verification`) и, как правило, состоит из +**двух коммитов**: feature-коммит + фикс review-раунда (последний нередко чинит реальный +баг — напр. fail-closed XSS в стадии 4, watchdog-gate в стадии 6). Поэтому `git revert` +одного коммита осиротит фикс review-раунда и даст несогласованный откат. Откатывать нужно +на гранулярности стадии; порядок стадий = порядок деплоя. + +- **Как откатывать стадию** — зависит от того, как ветки влиты в `fix/stability`/`main`: + - если стадия влита **squash-merge** (один коммит на стадию) — `git revert ` + откатывает её целиком, однозначно; + - если стадия влита **merge-коммитом** — `git revert -m 1 ` откатывает весь + вклад ветки (оба коммита) разом; + - если история линейная (rebase/fast-forward) — реверт **всех** коммитов стадии + (`git revert ..` включительно), а не одного. +- Стадии 1 и 2 — низкорисковые, деплоятся первыми; откатываются независимо от остальных. +- Стадия 3 (FileResponse) независима — реверт возвращает прежний ручной стриминг. +- Стадия 4 требует, чтобы 4.2 откатывалась не позже 4.3 (иначе рендер-в-потоке остался бы + без безопасного пути сохранения media id) — откатывать стадию 4 целиком. +- Стадии 5 и 6 независимы, откатываются по отдельности. +- Стадия 7 — только тесты и этот документ: реверт ничего не меняет в проде. + +Прод-деплой, живое наблюдение diag-логов и обновление прод-compose (healthcheck→`/ping`, +вне репозитория) — **зона ответственности оператора** (пункты 3–4 плана стадии 7). diff --git a/telegram_client.py b/telegram_client.py index da9793e..6d4f547 100644 --- a/telegram_client.py +++ b/telegram_client.py @@ -259,6 +259,17 @@ class TelegramClient: # Emergency termination os._exit(1) + def watchdog_last_ok_age(self) -> float | None: + """Seconds since the last successful watchdog probe, or None if none succeeded yet. + + Reads only the already-recorded monotonic timestamp set by the watchdog loop; it + never issues a Telegram RPC, so it is safe to call from the hot /ping path even + while a real RPC is hung. + """ + if self._wd_last_ok_monotonic is None: + return None + return time.monotonic() - self._wd_last_ok_monotonic + async def safe_get_messages(self, channel_id, post_id, max_retries=2): """Wrapper with retry logic for auth errors""" for attempt in range(max_retries): diff --git a/tests/mock_config.py b/tests/mock_config.py index d6bacfe..cfed846 100644 --- a/tests/mock_config.py +++ b/tests/mock_config.py @@ -33,6 +33,7 @@ def get_settings(): "tg_watchdog_heartbeat_every": 30, "tg_disconnect_flap_limit": 3, "tg_disconnect_flap_window": 120, + "tg_ping_unhealthy_after": 250, "media_download_timeout_min": 120, "media_download_timeout_max": 1800, "media_download_min_speed": 256 * 1024, diff --git a/tests/test_stage6_healthcheck.py b/tests/test_stage6_healthcheck.py new file mode 100644 index 0000000..2a1d5ff --- /dev/null +++ b/tests/test_stage6_healthcheck.py @@ -0,0 +1,188 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring +# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +""" +Stage 6 (lightweight /ping healthcheck) regression tests. + +Covers: +- /ping returns 200 "ok" when connected and the last probe is recent (age < threshold). +- /ping returns 503 "degraded" when connected but the last probe is stale (age > threshold). +- /ping returns 503 "degraded" when disconnected, regardless of probe age. +- /ping returns 200 "ok" on a fresh boot (age is None) while connected — a freshly-started + container must NOT be killed before the watchdog's first probe. +- ANTI-REGRESSION (the critical invariant): /ping issues ZERO Telegram RPC. A spy on the + fake client's get_me / safe_get_messages proves neither is ever called. +- TelegramClient.watchdog_last_ok_age(): None when never probed; a positive float afterwards. +""" +import os +import sys +import time + +import pytest + +# Add project root to sys.path and mock the config module (same pattern as the other tests). +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings']) + +from fastapi.testclient import TestClient + +import api_server +from telegram_client import TelegramClient + + +# --------------------------------------------------------------------------- # +# Fakes +# --------------------------------------------------------------------------- # +class _FakeKurigramClient: + """Stands in for TelegramClient.client — exposes is_connected and RPC spies.""" + def __init__(self, is_connected=True): + self.is_connected = is_connected + self.get_me_calls = 0 + + async def get_me(self): + # If /ping ever touches this, the whole point of the endpoint is defeated. + self.get_me_calls += 1 + raise AssertionError("/ping must never call get_me()") + + +class _FakeTelegramClient: + """Stands in for api_server.client with a controllable probe age + RPC spies.""" + def __init__(self, age, is_connected=True): + self._age = age + self.client = _FakeKurigramClient(is_connected=is_connected) + self.safe_get_messages_calls = 0 + + def watchdog_last_ok_age(self): + return self._age + + async def safe_get_messages(self, *args, **kwargs): + self.safe_get_messages_calls += 1 + raise AssertionError("/ping must never call safe_get_messages()") + + +@pytest.fixture +def patch_client(monkeypatch): + """Return a factory that installs a fake api_server.client and yields a TestClient.""" + def _install(age, is_connected=True): + fake = _FakeTelegramClient(age=age, is_connected=is_connected) + monkeypatch.setattr(api_server, "client", fake) + return fake, TestClient(api_server.app) + return _install + + +THRESHOLD = api_server.Config["tg_ping_unhealthy_after"] # 250 in mock_config + + +# --------------------------------------------------------------------------- # +# /ping endpoint behavior +# --------------------------------------------------------------------------- # +def test_ping_healthy_connected_recent(patch_client): + fake, tc = patch_client(age=THRESHOLD - 10, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "ok" + assert body["connected"] is True + assert body["last_probe_age_s"] == round(THRESHOLD - 10, 1) + assert body["threshold_s"] == THRESHOLD + assert fake.client.get_me_calls == 0 + + +def test_ping_degraded_stale_probe(patch_client): + fake, tc = patch_client(age=THRESHOLD + 100, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 503 + body = r.json() + assert body["status"] == "degraded" + assert body["connected"] is True + assert fake.client.get_me_calls == 0 + + +def test_ping_watchdog_disabled_stale_age_still_healthy(patch_client, monkeypatch): + # With the watchdog OFF, nothing refreshes age (a disconnect-flap restart can stamp it + # once, then it only grows). A stale age must NOT drive /ping to 503 on a live connection + # — that would spuriously fail the healthcheck and trigger an autoheal restart. So with the + # watchdog disabled, /ping is a pure connectivity check: connected + stale age => healthy. + monkeypatch.setitem(api_server.Config, "tg_watchdog_enabled", False) + fake, tc = patch_client(age=THRESHOLD + 100, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "ok" + assert body["connected"] is True + assert fake.client.get_me_calls == 0 + + +def test_ping_degraded_disconnected(patch_client): + # Even with a fresh probe age, a disconnected client is unhealthy. + fake, tc = patch_client(age=1.0, is_connected=False) + r = tc.get("/ping") + assert r.status_code == 503 + body = r.json() + assert body["status"] == "degraded" + assert body["connected"] is False + + +def test_ping_degraded_disconnected_even_when_age_none(patch_client): + fake, tc = patch_client(age=None, is_connected=False) + r = tc.get("/ping") + assert r.status_code == 503 + assert r.json()["status"] == "degraded" + + +def test_ping_fresh_boot_age_none_is_healthy(patch_client): + # Right after boot the watchdog hasn't probed yet (age None); connected => healthy. + fake, tc = patch_client(age=None, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "ok" + assert body["last_probe_age_s"] is None + assert body["threshold_s"] == THRESHOLD + + +def test_ping_pre_start_connected_none_is_degraded_bool(patch_client): + # Before client.start(), Kurigram's is_connected is None. /ping must not 500: it coerces + # to a bool, so "connected" is false (never null) and the endpoint reports 503 degraded. + fake, tc = patch_client(age=None, is_connected=None) + r = tc.get("/ping") + assert r.status_code == 503 + body = r.json() + assert body["status"] == "degraded" + assert body["connected"] is False # bool, not null + assert fake.client.get_me_calls == 0 + + +def test_ping_issues_no_tg_rpc(patch_client): + """The critical invariant: /ping never issues any Telegram RPC in any branch.""" + for age, connected in [(1.0, True), (THRESHOLD + 500, True), (None, True), (1.0, False)]: + fake, tc = patch_client(age=age, is_connected=connected) + tc.get("/ping") + assert fake.client.get_me_calls == 0, f"get_me called (age={age}, connected={connected})" + assert fake.safe_get_messages_calls == 0, f"safe_get_messages called (age={age}, connected={connected})" + + +def test_ping_route_needs_no_token(patch_client): + # /ping is unauthenticated by design (no token path variant); it just works. + fake, tc = patch_client(age=1.0, is_connected=True) + assert tc.get("/ping").status_code == 200 + + +# --------------------------------------------------------------------------- # +# TelegramClient.watchdog_last_ok_age accessor +# --------------------------------------------------------------------------- # +def test_watchdog_last_ok_age_none_when_never_probed(): + tgc = TelegramClient() + assert tgc._wd_last_ok_monotonic is None + assert tgc.watchdog_last_ok_age() is None + + +def test_watchdog_last_ok_age_positive_after_probe(): + tgc = TelegramClient() + tgc._wd_last_ok_monotonic = time.monotonic() - 5 + age = tgc.watchdog_last_ok_age() + assert age is not None + assert age >= 5.0 + # Sanity: a plausible upper bound so we didn't accidentally read the wrong field. + assert age < 60.0 diff --git a/tests/test_stage7_integration.py b/tests/test_stage7_integration.py new file mode 100644 index 0000000..a950e23 --- /dev/null +++ b/tests/test_stage7_integration.py @@ -0,0 +1,349 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring +# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +""" +Stage 7 — cross-stage END-TO-END integration tests. + +Unlike the per-stage suites (which pin one seam in isolation), these drive the real +public entry points (`get_media`, `ping`, the access-time flush) so a regression that +only shows up when the stages interact goes red. Every scenario here maps to one of the +plan's "Стадия 7 — сквозные ручные сценарии"; the ones that genuinely need a running +server + real downloads + lsof (fd-leak counting) are NOT faked here — they stay in +docs/stability-verification.md for the operator's prod observation. + +Automated here: +- Range semantics at the /media ROUTE level (get_media -> prepare_file_response -> + FileResponse driven through real ASGI): bytes=0-99 -> 206, bytes=-100 -> 206, + bytes=999999999- -> 416. (Stage 3 pins these on prepare_file_response directly; this + adds the end-to-end assertion that get_media's cache-hit branch reaches FileResponse + with a live Range still honored — i.e. stages 2+3 wired together.) +- /ping (stage 6) stays fast + correct while a deliberately-slow op is in flight, issuing + ZERO Telegram RPC — proving the healthcheck is decoupled from the hot/blocked paths. +- In-flight dedup + disconnect cleanup (stages 1/2) through the REAL get_media entry + point: concurrent requests share one download and the _inflight registry drains; a + cancelled request (client disconnect) leaves neither a stuck key nor a hung sibling. +- str(channel) access-time key consistency (stage 5) end-to-end: a /media cache hit for + an int-ish channel records the str-keyed timestamp, and the flush UPDATE matches the + str-keyed DB row (hit -> flush -> DB), the exact affinity gotcha the plan warns about. +""" +import os +import sys +import time +import sqlite3 +import asyncio +from types import SimpleNamespace + +import pytest + +# Add project root to sys.path and mock the config module (same pattern as the other tests). +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings']) + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +import api_server +from pyrogram import errors +from file_io import init_db_sync + + +# 2048 deterministic bytes so Range slices are byte-checkable. +BODY = bytes(range(256)) * 8 +SIZE = len(BODY) + + +def _media_app(): + """A bare app that mounts the REAL get_media (and ping) with NO lifespan, so + client.start() never runs — a pure cache hit never touches Telegram, and FileResponse + computes Range/206/416 at send time, which only happens when driven through ASGI.""" + app = FastAPI() + app.add_api_route("/media/{channel}/{post_id}/{file_unique_id}/{digest}", api_server.get_media, methods=["GET"]) + app.add_api_route("/media/{channel}/{post_id}/{file_unique_id}", api_server.get_media, methods=["GET"]) + return app + + +def _seed_cache(tmp_path, channel, post_id, fid, body=BODY): + cache_dir = tmp_path / "data" / "cache" / str(channel) / str(post_id) + cache_dir.mkdir(parents=True, exist_ok=True) + (cache_dir / fid).write_bytes(body) + return cache_dir / fid + + +# --------------------------------------------------------------------------- # +# Range semantics at the /media ROUTE level (stages 2 + 3 wired together). +# Plan scenario: `curl -H "Range: bytes=0-99" / "bytes=-100" / "bytes=999999999-"`. +# Regression caught: any change that makes get_media's cache-hit branch stop reaching +# FileResponse (e.g. re-buffering the body, hand-rolling headers, dropping the media_key +# path) or that breaks the digest gate wiring — the Range would stop being honored. +# --------------------------------------------------------------------------- # +def test_media_route_range_prefix_0_99(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + _seed_cache(tmp_path, "chan", 3, "fidR") + monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True) + # Keep the MIME path DB-free; the FileResponse/Range machinery is what we exercise. + monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None) + monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None) + + c = TestClient(_media_app()) + r = c.get("/media/chan/3/fidR/anydigest", headers={"Range": "bytes=0-99"}) + assert r.status_code == 206 + assert r.headers["content-range"] == f"bytes 0-99/{SIZE}" + assert r.headers["content-length"] == "100" + assert r.content == BODY[:100] + assert r.headers["accept-ranges"] == "bytes" + + +def test_media_route_range_suffix_last_100(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + _seed_cache(tmp_path, "chan", 3, "fidS") + monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True) + monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None) + monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None) + + c = TestClient(_media_app()) + r = c.get("/media/chan/3/fidS/anydigest", headers={"Range": "bytes=-100"}) + assert r.status_code == 206 + assert r.headers["content-range"] == f"bytes {SIZE - 100}-{SIZE - 1}/{SIZE}" + assert r.headers["content-length"] == "100" + assert r.content == BODY[-100:] + + +def test_media_route_range_unsatisfiable_416(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + _seed_cache(tmp_path, "chan", 3, "fidU") + monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True) + monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None) + monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None) + + c = TestClient(_media_app()) + r = c.get("/media/chan/3/fidU/anydigest", headers={"Range": "bytes=999999999-"}) + assert r.status_code == 416 + # Starlette's 416 Content-Range is `*/size` (documented stage-3 RFC-7233 delta). + assert r.headers["content-range"] == f"*/{SIZE}" + + +# --------------------------------------------------------------------------- # +# /ping (stage 6) stays fast + correct while a slow op is in flight, zero TG RPC. +# Plan scenario: "Генерация фида на 100+ сообщений + параллельный /ping -> ping < 100 мс". +# We model the concurrent slow op as a coroutine parked on an Event that is NEVER set +# during the ping, and assert ping resolves while it is still pending AND touches no RPC. +# Regression caught: re-coupling /ping to a TG RPC (get_me / safe_get_messages) or to any +# awaitable that a blocked hot path could stall — the ping would no longer return promptly +# or the spy counts would go non-zero. +# --------------------------------------------------------------------------- # +class _FakeKurigram: + def __init__(self, is_connected=True): + self.is_connected = is_connected + self.get_me_calls = 0 + + async def get_me(self): + self.get_me_calls += 1 + raise AssertionError("/ping must never call get_me()") + + +class _FakeTelegramClient: + def __init__(self, age, is_connected=True): + self._age = age + self.client = _FakeKurigram(is_connected=is_connected) + self.safe_get_messages_calls = 0 + + def watchdog_last_ok_age(self): + return self._age + + async def safe_get_messages(self, *a, **k): + self.safe_get_messages_calls += 1 + raise AssertionError("/ping must never call safe_get_messages()") + + +async def test_ping_prompt_and_rpc_free_while_slow_op_pending(monkeypatch): + threshold = api_server.Config["tg_ping_unhealthy_after"] + fake = _FakeTelegramClient(age=threshold - 10, is_connected=True) + monkeypatch.setattr(api_server, "client", fake) + + # A concurrent slow operation (a stand-in for a hung feed/RPC hot path) parked on an + # Event we deliberately never set for the duration of the ping. + gate = asyncio.Event() + + async def slow_op(): + await gate.wait() + + slow = asyncio.create_task(slow_op()) + await asyncio.sleep(0) # let slow_op start and park on the gate + + # The real proof of decoupling: ping() returns under a tight deadline while the slow op + # is parked, AND issues zero TG RPC. wait_for reds if ping ever blocks; the spies (which + # raise if touched) red if ping recouples to any RPC. `not slow.done()` is only a sanity + # check that ping did not somehow drive the parked op — the gate keeps it pending anyway. + resp = await asyncio.wait_for(api_server.ping(), timeout=1.0) + + assert resp.status_code == 200 # correct health while connected + fresh + assert not slow.done() # sanity: slow op still parked, ping did not await it + assert fake.client.get_me_calls == 0 # decoupled: zero TG RPC + assert fake.safe_get_messages_calls == 0 + + gate.set() + await slow + + +async def test_ping_reports_degraded_promptly_while_slow_op_pending(monkeypatch): + threshold = api_server.Config["tg_ping_unhealthy_after"] + fake = _FakeTelegramClient(age=threshold + 100, is_connected=True) # stale probe + monkeypatch.setattr(api_server, "client", fake) + + gate = asyncio.Event() + + async def slow_op(): + await gate.wait() + + slow = asyncio.create_task(slow_op()) + await asyncio.sleep(0) + + resp = await asyncio.wait_for(api_server.ping(), timeout=1.0) + + assert resp.status_code == 503 # stale watchdog probe -> degraded, still instant + assert not slow.done() # sanity: slow op still parked, ping did not await it + assert fake.client.get_me_calls == 0 + assert fake.safe_get_messages_calls == 0 + + gate.set() + await slow + + +# --------------------------------------------------------------------------- # +# In-flight dedup + disconnect cleanup (stages 1/2) through the REAL get_media entry. +# Plan scenario: "Параллельные запросы одного большого видео -> нет частичной отдачи" and +# "Отключение клиента на середине стрима -> нет утечки тасков/фд". +# The pure-unit stage-2 tests pin _download_deduped directly; these drive get_media so the +# HTTP semaphore + dedup registry + serve path are proven wired together. +# Regression caught: moving the download back into the request coroutine (so a disconnect +# cancels it), or dropping the finally-pop, would leave a stuck _inflight key / hung sibling. +# --------------------------------------------------------------------------- # +async def test_get_media_concurrent_shares_one_download_and_drains_registry(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + api_server._inflight.clear() + monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True) + # Serve step is not under test here; keep it to a sentinel so we assert on dedup + registry. + sentinel = object() + + async def fake_prepare(file_path, request=None, delete_after=False, media_key=None): + return sentinel + + monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare) + + calls = [] + + async def slow_dl(channel, post_id, fid): + calls.append(fid) + await asyncio.sleep(0.05) # real overlap window for the two requests + return (f"/final/{fid}", False) + + monkeypatch.setattr(api_server, "download_media_file", slow_dl) + + req = SimpleNamespace(headers={}) + t1 = asyncio.create_task(api_server.get_media("chan", 9, "vfid", request=req, digest="x")) + t2 = asyncio.create_task(api_server.get_media("chan", 9, "vfid", request=req, digest="x")) + r1, r2 = await asyncio.gather(t1, t2) + + assert r1 is sentinel and r2 is sentinel + assert calls == ["vfid"] # exactly ONE real download, shared by both requests + assert not api_server._inflight # registry drained — no forever-busy key + + +async def test_get_media_request_cancel_does_not_stick_registry_or_hang_sibling(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + api_server._inflight.clear() + monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True) + sentinel = object() + + async def fake_prepare(file_path, request=None, delete_after=False, media_key=None): + return sentinel + + monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare) + + gate = asyncio.Event() + calls = [] + + async def held_dl(channel, post_id, fid): + calls.append(fid) + await gate.wait() # hold the shared download open until we release it + return (f"/final/{fid}", False) + + monkeypatch.setattr(api_server, "download_media_file", held_dl) + + req = SimpleNamespace(headers={}) + t1 = asyncio.create_task(api_server.get_media("chan", 9, "cfid", request=req, digest="x")) + await asyncio.sleep(0.02) # t1 registers the future + starts the detached download + t2 = asyncio.create_task(api_server.get_media("chan", 9, "cfid", request=req, digest="x")) + await asyncio.sleep(0.02) + + # First client disconnects: its request coroutine is cancelled mid-wait. + t1.cancel() + with pytest.raises(asyncio.CancelledError): + await t1 + + # The detached download is unaffected; releasing it resolves the surviving request. + gate.set() + r2 = await asyncio.wait_for(t2, timeout=2.0) + assert r2 is sentinel + assert calls == ["cfid"] # download ran exactly once (not restarted) + assert not api_server._inflight # registry drained despite the disconnect + + +# --------------------------------------------------------------------------- # +# str(channel) access-time key consistency (stage 5) END-TO-END: hit -> flush -> DB. +# Plan gotcha #9: "Ключи SQLite: channel всегда str(...)"; if the key the hot path RECORDS +# and the key the flush UPDATEs by ever diverge, the timestamp silently stops updating and +# the file eventually falls out of the cache. Stage 5 pins hit and flush separately; this +# stitches them: the /media cache-hit records a (str-channel) key, and the flush's bulk +# UPDATE must land on that exact DB row. (get_media's route channel is already a str, so the +# str() there is a defensive no-op; the live int/str hazard is in download_media_file, pinned +# by stage-5's test_cache_hit_keys_channel_as_str — this test covers the get_media+flush leg.) +# Regression caught: any accumulator-key vs UPDATE-WHERE inconsistency (e.g. a transposed +# key-column order in the bulk SQL, verified to turn this test red) leaves `added` stale. +# --------------------------------------------------------------------------- # +async def test_media_cache_hit_flush_updates_str_keyed_db_row(monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + api_server._access_updates = {} + + channel, post_id, fid = "12345", 7, "fidINT" # int-ish channel, passed as the route str + _seed_cache(tmp_path, channel, post_id, fid) + + db = str(tmp_path / "access.db") + init_db_sync(db) + # Seed the row keyed by the STRING channel with an old access time. + conn = sqlite3.connect(db) + conn.execute( + "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)", + (channel, post_id, fid, 1.0), + ) + conn.commit() + conn.close() + monkeypatch.setattr(api_server, "DB_PATH", db) + + monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True) + sentinel = object() + + async def fake_prepare(file_path, request=None, media_key=None): + return sentinel + + monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare) + + before = time.time() + resp = await api_server.get_media(channel, post_id, fid, request=SimpleNamespace(headers={}), digest="x") + assert resp is sentinel + + # Hot path recorded the str-keyed access time (never the raw int form). + assert (channel, post_id, fid) in api_server._access_updates + + # Flush the accumulator; the bulk UPDATE must match the str-keyed row and refresh `added`. + await api_server._flush_access_updates() + assert api_server._access_updates == {} + + conn = sqlite3.connect(db) + added = conn.execute( + "SELECT added FROM media_file_ids WHERE channel = ? AND post_id = ? AND file_unique_id = ?", + (channel, post_id, fid), + ).fetchone()[0] + conn.close() + assert added >= before # refreshed from the stale 1.0 to ~now via the str-keyed WHERE