Merge pull request 'fix(stability): стадия 5 — батчинг записи access-time в SQLite' (#15) from fix/stage-5-sqlite into fix/stage-4-eventloop

api_server.py

@@ -33,14 +33,15 @@ 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
 from post_parser import PostParser
 from url_signer import verify_media_digest, generate_media_digest
 from file_io import (DB_PATH, init_db_sync, get_all_media_file_ids_sync,
-                     update_media_file_access_sync, remove_media_file_ids_sync,
+                     update_media_file_access_sync, update_media_file_access_bulk_sync,
+                     remove_media_file_ids_sync,
                      get_mime_type_sync, set_mime_type_sync)
 
 # Global python-magic instance for MIME type detection
@@ -139,6 +140,59 @@ async def _supervised(factory, name: str, min_restart_interval: float = 60.0):
             await asyncio.sleep(min_restart_interval - elapsed)
 
 
+# Access-time write accumulator. A /media cache hit used to touch SQLite on the hot path
+# (a threadpool hop + connect + UPDATE per request), which starves the threadpool under
+# active RSS polling. Instead a cache hit just records the timestamp here — a dict write
+# on the single-threaded event loop is cheap and atomic — and a periodic background task
+# flushes the whole batch to SQLite in one executemany. Keys use str(channel) to stay
+# consistent with the string form written at insert time and to not lean on SQLite's
+# implicit column-affinity coercion (the channel column is TEXT, so a bound int would be
+# coerced and still match — but we key the accumulator by the same type we store, rather
+# than depend on that).
+ACCESS_FLUSH_INTERVAL = 60  # seconds between access-time flushes
+_access_updates: dict[tuple[str, int, str], float] = {}
+
+
+async def _flush_access_updates() -> None:
+    """Flush the accumulated access timestamps to SQLite in one bulk UPDATE.
+
+    Snapshot-then-clear atomically on the loop: capture the current dict reference and
+    replace the module global with a fresh empty dict in ONE synchronous step (before any
+    await), so cache-hit writes arriving DURING the flush land in the new dict and are not
+    lost. The bulk UPDATE runs off-loop via asyncio.to_thread. An empty batch is a no-op.
+    """
+    global _access_updates
+    if not _access_updates:
+        return
+    pending = _access_updates
+    _access_updates = {}
+    entries = [(channel, post_id, file_unique_id, added)
+               for (channel, post_id, file_unique_id), added in pending.items()]
+    try:
+        await asyncio.to_thread(update_media_file_access_bulk_sync, DB_PATH, entries)
+    except Exception:
+        # Bulk write failed: re-queue this batch so the access-times are not lost (a lost
+        # timestamp would eventually evict a still-used file from the 20-day cache). Use
+        # setdefault so any FRESHER write accumulated during the flush is never overwritten
+        # by our stale snapshot. Runs on the loop with no await before the mutation, so this
+        # is race-free. Re-raise so the flush loop logs it.
+        for key, added in pending.items():
+            _access_updates.setdefault(key, added)
+        raise
+
+
+async def _access_flush_loop() -> None:
+    """Periodically flush the access-time accumulator (runs under _supervised)."""
+    while True:
+        await asyncio.sleep(ACCESS_FLUSH_INTERVAL)
+        try:
+            await _flush_access_updates()
+        except Exception as e:
+            # Log and keep looping: a transient SQLite error must not drop the batch's
+            # successors. (_supervised still restarts us if this ever raises out.)
+            logger.error(f"access_flush_error: {e}")
+
+
 @asynccontextmanager
 async def lifespan(_: FastAPI):
     setup_logging(Config["log_level"])
@@ -161,9 +215,11 @@ async def lifespan(_: FastAPI):
     # CRITICAL and restarted, so a crash can no longer silently stop cache sweeping or downloads.
     background_task = asyncio.create_task(_supervised(cache_media_files, "cache_media_files"))
     worker_task = asyncio.create_task(_supervised(background_download_worker, "background_download_worker"))
+    access_flush_task = asyncio.create_task(_supervised(_access_flush_loop, "access_flush_loop"))
     yield
     background_task.cancel() # Cleanup
     worker_task.cancel()
+    access_flush_task.cancel()
     try:
         await background_task
     except asyncio.CancelledError:
@@ -172,6 +228,17 @@ async def lifespan(_: FastAPI):
         await worker_task
     except asyncio.CancelledError:
         pass
+    try:
+        await access_flush_task
+    except asyncio.CancelledError:
+        pass
+    # Final flush AFTER the loop task is cancelled (no race with a loop-driven flush) and
+    # BEFORE the threadpool is shut down (to_thread still has its executor), so the last
+    # <=ACCESS_FLUSH_INTERVAL seconds of access-times are persisted on shutdown.
+    try:
+        await _flush_access_updates()
+    except Exception as e:
+        logger.error(f"access_flush_shutdown_error: {e}")
     await client.stop()
     # Shut the io threadpool down so its threads don't linger past a reload/restart.
     io_executor.shutdown(wait=False)
@@ -538,16 +605,11 @@ async def download_media_file(channel: Union[str, int], post_id: int, file_uniqu
                 logger.error(f"cleanup_error: Failed to remove zero-size cached file {cache_path}: {e}")
             # Do not raise error here, proceed to download below
         else:
-            # File exists and is not zero size, update access timestamp and return
+            # File exists and is not zero size, record access timestamp and return.
+            # Record into the accumulator instead of touching SQLite on the hot path; the
+            # background flush persists it. Key channel as str(channel) — see _access_updates.
             logger.info(f"Found cached media file: {cache_path}")
-            try:
-                await asyncio.to_thread(
-                    update_media_file_access_sync,
-                    DB_PATH, str(channel), post_id, file_unique_id,
-                    datetime.now().timestamp()
-                )
-            except Exception as e:
-                logger.error(f"Failed to update timestamp for {channel}/{post_id}/{file_unique_id}: {str(e)}")
+            _access_updates[(str(channel), post_id, file_unique_id)] = datetime.now().timestamp()
             return cache_path, False
 
     file_id = await find_file_id_in_message(message, file_unique_id)
@@ -1003,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:
@@ -1074,17 +1176,9 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re
             if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0:
                 # File is already in cache — skip semaphore and serve directly
                 logger.info(f"pre_semaphore_cache_hit: {channel}/{post_id}/{file_unique_id}")
-                # Fire-and-forget timestamp update with error handling to avoid silent failures
-                async def _update_access(_ch, _pid, _fid):
-                    try:
-                        await asyncio.to_thread(
-                            update_media_file_access_sync,
-                            DB_PATH, str(_ch), _pid, _fid,
-                            datetime.now().timestamp()
-                        )
-                    except Exception as _e:
-                        logger.warning(f"Failed to update access time for {_ch}/{_pid}/{_fid}: {_e}")
-                asyncio.create_task(_update_access(channel, post_id, file_unique_id))
+                # Record the access time into the accumulator instead of firing a per-hit
+                # SQLite write. Key channel as str(channel) — see _access_updates.
+                _access_updates[(str(channel), post_id, file_unique_id)] = datetime.now().timestamp()
                 return await prepare_file_response(cache_path, request=request,
                                                    media_key=(str(channel), post_id, file_unique_id))
 

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).

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
 

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 <squash-commit>`
+    откатывает её целиком, однозначно;
+  - если стадия влита **merge-коммитом** — `git revert -m 1 <merge-commit>` откатывает весь
+    вклад ветки (оба коммита) разом;
+  - если история линейная (rebase/fast-forward) — реверт **всех** коммитов стадии
+    (`git revert <feature>..<review-fix>` включительно), а не одного.
+- Стадии 1 и 2 — низкорисковые, деплоятся первыми; откатываются независимо от остальных.
+- Стадия 3 (FileResponse) независима — реверт возвращает прежний ручной стриминг.
+- Стадия 4 требует, чтобы 4.2 откатывалась не позже 4.3 (иначе рендер-в-потоке остался бы
+  без безопасного пути сохранения media id) — откатывать стадию 4 целиком.
+- Стадии 5 и 6 независимы, откатываются по отдельности.
+- Стадия 7 — только тесты и этот документ: реверт ничего не меняет в проде.
+
+Прод-деплой, живое наблюдение diag-логов и обновление прод-compose (healthcheck→`/ping`,
+вне репозитория) — **зона ответственности оператора** (пункты 3–4 плана стадии 7).

file_io.py

@@ -96,6 +96,27 @@ def update_media_file_access_sync(db_path: str, channel: str, post_id: int, file
         )
 
 
+def update_media_file_access_bulk_sync(db_path: str, entries: List[tuple]) -> None:
+    """Update the access timestamp for multiple existing media file ID records.
+
+    entries: iterable of (channel, post_id, file_unique_id, added) tuples.
+    Uses executemany (one connection, one commit) so a batch of cache-hit access
+    updates costs a single SQLite transaction instead of one connect+UPDATE per hit.
+    Rows that do not exist are simply not matched by the WHERE clause (no-op), mirroring
+    the single-row update_media_file_access_sync. An empty batch is a no-op.
+    """
+    if not entries:
+        return
+    with _db_connection(db_path) as conn:
+        conn.executemany(
+            "UPDATE media_file_ids SET added = ? WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+            # Reorder each (channel, post_id, file_unique_id, added) tuple to match the
+            # UPDATE's placeholder order (added first, then the WHERE key columns).
+            [(added, channel, post_id, file_unique_id)
+             for (channel, post_id, file_unique_id, added) in entries],
+        )
+
+
 def get_all_media_file_ids_sync(db_path: str) -> List[dict]:
     """Return all rows from media_file_ids as a list of dicts."""
     with _db_connection(db_path) as conn:

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):

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,

tests/test_stage5_sqlite.py

@@ -0,0 +1,263 @@
+# 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 5 (SQLite access-time batching) tests.
+
+Covers:
+- 5.1 accumulator: a /media cache hit records into api_server._access_updates and does NOT
+      call update_media_file_access_sync (no synchronous SQLite write on the hot path).
+- 5.1 flush: seeding _access_updates + running the flush once writes the accumulated
+      timestamps to a real temp DB (hit -> flush -> value updated in DB), and clears the dict.
+- 5.1 bulk fn update_media_file_access_bulk_sync: empty no-op, multi-row, and updating an
+      EXISTING row's `added` (WHERE matches on the str channel key).
+- 5.1 snapshot-then-clear: an update arriving DURING the flush lands in the fresh dict and
+      is not lost.
+- gotcha: str(channel) key discipline — an int-ish channel on the hot path keys the
+      accumulator (and thus the UPDATE) by the string form.
+"""
+import os
+import sys
+import sqlite3
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings'])
+
+import api_server
+from file_io import init_db_sync, update_media_file_access_bulk_sync
+
+
+def _fake_plain_message():
+    """A non-poll, non-video message so download_media_file takes the normal cache flow."""
+    return SimpleNamespace(media=None, video=None, empty=False)
+
+
+@pytest.fixture(autouse=True)
+def _clean_accumulator():
+    """Each test starts with an empty accumulator and restores it afterwards."""
+    api_server._access_updates = {}
+    yield
+    api_server._access_updates = {}
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 hot path: cache hit records into the accumulator, no synchronous SQLite.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_cache_hit_records_accumulator_no_sqlite(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    channel, post_id, fid = "testchan", 5, "fidHIT"
+    cache_dir = tmp_path / "data" / "cache" / channel / str(post_id)
+    cache_dir.mkdir(parents=True)
+    (cache_dir / fid).write_bytes(b"cached-bytes")
+
+    async def fake_get(_cid, _pid):
+        return _fake_plain_message()
+    monkeypatch.setattr(api_server.client, "safe_get_messages", fake_get)
+
+    # Spy: the single-row synchronous updater must NOT be called on the hot path.
+    called = []
+    monkeypatch.setattr(api_server, "update_media_file_access_sync",
+                        lambda *a, **k: called.append(a))
+
+    path, delete_after = await api_server.download_media_file(channel, post_id, fid)
+
+    assert path == str(cache_dir / fid)
+    assert delete_after is False
+    assert called == []  # no synchronous SQLite access-write happened
+    assert (channel, post_id, fid) in api_server._access_updates
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 hot path (DoD guard): get_media's pre-semaphore cache-hit — the hottest
+# changed site, the one this PR exists for — records into the accumulator and
+# does NO synchronous SQLite access-write. Mirrors the download_media_file spy
+# so a regression re-introducing a per-hit write into THIS branch goes red.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_get_media_pre_semaphore_cache_hit_no_sqlite(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    channel, post_id, fid = "gmchan", 11, "fidGM"
+    cache_dir = tmp_path / "data" / "cache" / channel / str(post_id)
+    cache_dir.mkdir(parents=True)
+    (cache_dir / fid).write_bytes(b"cached-bytes")
+
+    # Bypass the HMAC digest gate and the FileResponse machinery — the test targets
+    # only the access-time write on the pre-semaphore cache-hit branch.
+    monkeypatch.setattr(api_server, "verify_media_digest", lambda *a, **k: True)
+    sentinel = object()
+    async def fake_prepare(cache_path, request=None, media_key=None):
+        return sentinel
+    monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
+
+    # Spy: the single-row synchronous updater must NOT be called on the hot path.
+    called = []
+    monkeypatch.setattr(api_server, "update_media_file_access_sync",
+                        lambda *a, **k: called.append(a))
+
+    resp = await api_server.get_media(channel, post_id, fid, request=object(), digest="x")
+
+    assert resp is sentinel
+    assert called == []  # no synchronous SQLite access-write on the hottest path
+    assert (channel, post_id, fid) in api_server._access_updates
+
+
+# --------------------------------------------------------------------------- #
+# gotcha: str(channel) key discipline on the hot path (int-ish channel).
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_cache_hit_keys_channel_as_str(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    channel_int, post_id, fid = 12345, 7, "fidINT"
+    cache_dir = tmp_path / "data" / "cache" / str(channel_int) / str(post_id)
+    cache_dir.mkdir(parents=True)
+    (cache_dir / fid).write_bytes(b"x")
+
+    async def fake_get(_cid, _pid):
+        return _fake_plain_message()
+    monkeypatch.setattr(api_server.client, "safe_get_messages", fake_get)
+
+    await api_server.download_media_file(channel_int, post_id, fid)
+
+    assert ("12345", post_id, fid) in api_server._access_updates      # string form
+    assert (channel_int, post_id, fid) not in api_server._access_updates  # never the raw int
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 flush: hit -> flush -> the accumulated timestamp is written to the DB.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_flush_writes_accumulated_timestamps(tmp_path, monkeypatch):
+    db = str(tmp_path / "flush.db")
+    init_db_sync(db)
+    # Seed two existing rows with an old timestamp.
+    conn = sqlite3.connect(db)
+    conn.executemany(
+        "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)",
+        [("chA", 1, "fA", 1.0), ("chB", 2, "fB", 2.0)],
+    )
+    conn.commit()
+    conn.close()
+
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+    api_server._access_updates = {
+        ("chA", 1, "fA"): 111.0,
+        ("chB", 2, "fB"): 222.0,
+    }
+
+    await api_server._flush_access_updates()
+
+    # Dict cleared after flush.
+    assert api_server._access_updates == {}
+
+    conn = sqlite3.connect(db)
+    rows = dict(((c, p, f), a) for c, p, f, a in
+                conn.execute("SELECT channel, post_id, file_unique_id, added FROM media_file_ids"))
+    conn.close()
+    assert rows[("chA", 1, "fA")] == 111.0
+    assert rows[("chB", 2, "fB")] == 222.0
+
+
+@pytest.mark.asyncio
+async def test_flush_empty_is_noop(monkeypatch):
+    api_server._access_updates = {}
+    # Must not raise and must not touch the threadpool/DB.
+    monkeypatch.setattr(api_server, "update_media_file_access_bulk_sync",
+                        lambda *a, **k: pytest.fail("bulk sync should not run for an empty batch"))
+    await api_server._flush_access_updates()
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 snapshot-then-clear: an update arriving DURING the flush is not lost.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_snapshot_then_clear_does_not_lose_late_update(tmp_path, monkeypatch):
+    db = str(tmp_path / "race.db")
+    init_db_sync(db)
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+
+    late_key = ("chLate", 9, "fLate")
+
+    def fake_bulk(_db, entries):
+        # Simulates a cache hit landing WHILE the flush's to_thread runs: because the flush
+        # already replaced the module dict with a fresh one, this write goes into the NEW dict.
+        api_server._access_updates[late_key] = 999.0
+
+    monkeypatch.setattr(api_server, "update_media_file_access_bulk_sync", fake_bulk)
+
+    api_server._access_updates = {("chA", 1, "fA"): 111.0}
+    await api_server._flush_access_updates()
+
+    # The snapshot (chA) was flushed and cleared; the late update survives in the new dict.
+    assert api_server._access_updates == {late_key: 999.0}
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 re-queue on failure: a failed bulk write is not lost; setdefault keeps a
+# fresher write that arrived during the flush.
+# --------------------------------------------------------------------------- #
+@pytest.mark.asyncio
+async def test_failed_flush_requeues_batch_without_clobbering_fresh(tmp_path, monkeypatch):
+    db = str(tmp_path / "fail.db")
+    init_db_sync(db)
+    monkeypatch.setattr(api_server, "DB_PATH", db)
+
+    stale_key = ("chStale", 1, "fS")   # only in the failing snapshot
+    fresh_key = ("chFresh", 2, "fF")   # re-written FRESHER during the flush
+
+    def fake_bulk(_db, _entries):
+        # A newer cache hit for fresh_key lands in the fresh dict WHILE the bulk write runs,
+        # then the write fails. The re-queue must restore stale_key but must NOT overwrite
+        # the newer fresh_key value (setdefault, not assignment).
+        api_server._access_updates[fresh_key] = 999.0
+        raise sqlite3.OperationalError("database is locked")
+
+    monkeypatch.setattr(api_server, "update_media_file_access_bulk_sync", fake_bulk)
+
+    api_server._access_updates = {stale_key: 1.0, fresh_key: 2.0}
+    with pytest.raises(sqlite3.OperationalError):
+        await api_server._flush_access_updates()
+
+    # stale_key restored with its snapshot value; fresh_key keeps the NEWER value (not 2.0).
+    assert api_server._access_updates == {stale_key: 1.0, fresh_key: 999.0}
+
+
+# --------------------------------------------------------------------------- #
+# 5.1 bulk fn: empty no-op, multi-row, and existing-row update via str channel key.
+# --------------------------------------------------------------------------- #
+def test_bulk_access_update_real_sql(tmp_path):
+    db = str(tmp_path / "bulk.db")
+    init_db_sync(db)
+
+    # Empty batch is a no-op (no crash).
+    update_media_file_access_bulk_sync(db, [])
+
+    # Seed existing rows.
+    conn = sqlite3.connect(db)
+    conn.executemany(
+        "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)",
+        [("chA", 1, "fA", 1.0), ("chB", 2, "fB", 2.0)],
+    )
+    conn.commit()
+    conn.close()
+
+    # Multi-row update of EXISTING rows: WHERE matches on the str channel key.
+    update_media_file_access_bulk_sync(db, [
+        ("chA", 1, "fA", 500.0),
+        ("chB", 2, "fB", 600.0),
+    ])
+
+    conn = sqlite3.connect(db)
+    rows = dict(((c, p, f), a) for c, p, f, a in
+                conn.execute("SELECT channel, post_id, file_unique_id, added FROM media_file_ids"))
+    # A row keyed by an int channel does NOT match the string "chA" WHERE (documents the gotcha).
+    n = conn.execute("SELECT COUNT(*) FROM media_file_ids WHERE channel = 1").fetchone()[0]
+    conn.close()
+    assert rows[("chA", 1, "fA")] == 500.0
+    assert rows[("chB", 2, "fB")] == 600.0
+    assert n == 0

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

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