diff --git a/api_server.py b/api_server.py index 6201bcf..6fbe2ea 100644 --- a/api_server.py +++ b/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)) 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/file_io.py b/file_io.py index 965bc96..fbdfb89 100644 --- a/file_io.py +++ b/file_io.py @@ -69,6 +69,24 @@ def upsert_media_file_id_sync(db_path: str, channel: str, post_id: int, file_uni ) +def upsert_media_file_ids_bulk_sync(db_path: str, entries: List[tuple]) -> None: + """Insert or replace multiple media file ID records in a single transaction. + + entries: iterable of (channel, post_id, file_unique_id, added) tuples. + Uses executemany for batched upserts (one connection, one commit). + """ + if not entries: + return + with _db_connection(db_path) as conn: + conn.executemany( + """INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) + VALUES (?, ?, ?, ?) + ON CONFLICT(channel, post_id, file_unique_id) + DO UPDATE SET added = excluded.added""", + entries, + ) + + def update_media_file_access_sync(db_path: str, channel: str, post_id: int, file_unique_id: str, added: float) -> None: """Update the access timestamp for an existing media file ID record.""" with _db_connection(db_path) as conn: @@ -78,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: diff --git a/post_parser.py b/post_parser.py index 3e75f49..32ca6f1 100644 --- a/post_parser.py +++ b/post_parser.py @@ -21,16 +21,13 @@ from pyrogram.enums import MessageMediaType from bleach.css_sanitizer import CSSSanitizer from bleach import clean as HTMLSanitizer from config import get_settings -from file_io import upsert_media_file_id_sync, DB_PATH +from file_io import upsert_media_file_ids_bulk_sync, DB_PATH from url_signer import generate_media_digest Config = get_settings() logger = logging.getLogger(__name__) -# Module-level counter for in-flight persist tasks (used only in diagnostics) -_persist_pending_count = 0 - #tests #http://127.0.0.1:8000/post/html/DragorWW_space/114 — video @@ -65,6 +62,12 @@ _persist_pending_count = 0 class PostParser: def __init__(self, client): self.client = client + # Media file-id records collected during rendering. _save_media_file_ids + # appends (channel, post_id, file_unique_id, ts) tuples here instead of + # touching asyncio/DB directly (see 4.2). A fresh PostParser is created per + # request and rendering runs in a single thread, so this list is thread-safe. + # The caller flushes it once via upsert_media_file_ids_bulk_sync after render. + self._pending_media_ids: List[tuple] = [] @staticmethod def get_all_possible_flags() -> List[str]: @@ -106,7 +109,14 @@ class PostParser: logger.error(f"post_not_found_or_empty: channel {prepared_channel_id}, post_id {post_id}") return None - processed_message = self.process_message(message) + # Single-post outputs need sanitized body/footer (no whole-feed pass exists here). + # raw_message is only needed for JSON output and for debug HTML. + include_raw = (output_type == 'json') or debug + processed_message = self.process_message(message, include_raw=include_raw, sanitize=True) + + # Flush media file-id records collected during rendering with a single bulk upsert. + await self._flush_pending_media_ids() + if output_type == 'html': return self._format_html(processed_message, debug) elif output_type == 'json': @@ -480,11 +490,15 @@ class PostParser: html_content.append(f'
{data["html"]["body"]}
') html_content.append(f'
{data["html"]["footer"]}
') - # Add raw JSON debug output if debug is enabled + # Add raw JSON debug output if debug is enabled. + # raw_message is the full str(message) serialization and may contain user-controlled + # text with HTML/JS — escape it before dropping it into the 
 (bleach can't run
+        # here because 
 is not in the allowed-tags whitelist and would be stripped).
         if debug:
+            raw_escaped = html.escape(str(data.get("raw_message", "")))
             html_content.append(f'
{data["raw_message"]}
')
+                                f'white-space: pre-wrap;">{raw_escaped}
')
         html_data = '\n'.join(html_content)
         return html_data
 
@@ -500,11 +514,39 @@ class PostParser:
         
         return return_html
 
-    def process_message(self, message: Message) -> Dict[Any, Any]:
-        # Compute html body once — avoids triple _generate_html_body calls
+    def process_message(self, message: Message, include_raw: bool = False, sanitize: bool = True) -> Dict[Any, Any]:
+        """Build the processed representation of a message.
+
+        Args:
+            message: the Pyrogram message.
+            include_raw: when True, compute the (expensive) full ``str(message)``
+                serialization into ``result['raw_message']``. Only JSON output and
+                debug HTML need it; feed generation must pass False to avoid
+                serializing every post.
+            sanitize: when True, run the html body and footer through a single
+                bleach pass each. Single-post HTML and JSON need this (there is no
+                whole-feed pass on those paths). Feed generation passes False and
+                relies on the final sanitize in rss_generator (per-post for RSS,
+                whole-feed for HTML), so no
+                fragment is sanitized more than once.
+        """
+        # Compute html body once — avoids triple _generate_html_body calls.
+        # The internal per-fragment sanitize passes were removed (4.4); sanitize
+        # exactly once per output boundary here when requested.
         html_body = self._generate_html_body(message)
+        # NOTE (stage-4 4.4 consequence): flags are now extracted from the PRE-sanitize
+        # body (the per-fragment sanitize that used to run inside _generate_html_body was
+        # removed). Legitimate links are unaffected — bleach keeps whitelisted
+        #  — so link/foreign_channel/mention flags are identical for
+        # normal content. They can differ ONLY for URL-like text that bleach would strip
+        # (e.g. a URL inside a disallowed attribute); flags are non-security (used for
+        # exclude_flags filtering / display), so this edge divergence is accepted rather
+        # than re-adding a per-message sanitize pass that 4.4 deliberately eliminated.
         flags = self._extract_flags(message, html_body=html_body)
         footer = self.generate_html_footer(message, flags_list=flags)
+        if sanitize:
+            html_body = self._sanitize_html(html_body)
+            footer = self._sanitize_html(footer)
         result = {
             'channel': self.get_channel_username(message),
             'message_id': message.id,
@@ -523,8 +565,10 @@ class PostParser:
             'media_group_id': message.media_group_id,
             'service': getattr(message, "service", None),
             'show_caption_above_media': getattr(message, 'show_caption_above_media', False),
-            'raw_message': str(message)
         }
+        if include_raw:
+            # Full serialization of the message — expensive; only for JSON/debug.
+            result['raw_message'] = str(message)
         
         # Add webpage data if available
         if webpage := getattr(message, "web_page", None):
@@ -671,8 +715,10 @@ class PostParser:
         if message.forward_origin: content_body.append(f"--- Forwarded post end ---") # Forward info end
         content_body.append(f'
')
 
+        # NOTE: sanitize is NOT applied here. Sanitization happens exactly once per
+        # output boundary (process_message for single-post/JSON; in rss_generator the
+        # per-post pass for RSS and the whole-feed pass for HTML). See the map (4.4).
         html_body = '\n'.join(content_body)
-        html_body = self._sanitize_html(html_body)
         return html_body
 
     def _generate_html_media(self, message: Message) -> str:
@@ -751,8 +797,9 @@ class PostParser:
                     content_media.append(webpage_html)
                     content_media.append('')
 
+        # Not sanitized here — this fragment is embedded in the body and sanitized
+        # once at the output boundary (see the 4.4 sanitize coverage map).
         html_media = '\n'.join(content_media)
-        html_media = self._sanitize_html(html_media)
         return html_media
 
 
@@ -832,8 +879,9 @@ class PostParser:
             flags_html = self._format_flags(current_flags)
             content_footer.append('
' + flags_html)
             
+        # Not sanitized here — sanitized once at the output boundary (4.4 coverage map):
+        # process_message for single-post/JSON; per-post (RSS) / whole-feed (HTML) for feeds.
         html_footer = '\n'.join(content_footer)
-        html_footer = self._sanitize_html(html_footer)
         return html_footer
 
 
@@ -919,8 +967,10 @@ class PostParser:
                 if links:
                     parts.append(' | '.join(links))
 
+            # Raw fragment — embedded in the footer, sanitized once at the output
+            # boundary (see the 4.4 sanitize coverage map). Not sanitized here.
             result_html = '
'.join(parts) if parts else None
-            return self._sanitize_html(result_html) if result_html else None
+            return result_html if result_html else None
             
         except Exception as e:
             logger.error(f"reactions_views_links_error: message_id {message.id}, error {str(e)}")
@@ -969,15 +1019,31 @@ class PostParser:
             logger.error(f"file_id_extraction_error: media_type {message.media}, error {str(e)}")
             return None
 
-    async def _persist_media_file_id_async(self, channel: str, post_id: int, file_unique_id: str, added: float) -> None:
-        """Persist a single media file ID record to SQLite (fire-and-forget)."""
+    async def _flush_pending_media_ids(self) -> None:
+        """Persist media file-id records collected during rendering with ONE bulk upsert.
+
+        Called by the caller (get_post / rss_generator) after rendering completes.
+        Runs the blocking SQLite write in a thread. No-op when nothing was collected.
+        """
+        entries = self._pending_media_ids
+        if not entries:
+            return
         try:
-            await asyncio.to_thread(upsert_media_file_id_sync, DB_PATH, channel, post_id, file_unique_id, added)
-            logger.debug(f"persist_media_file_id: upserted {channel}/{post_id}/{file_unique_id}")
+            await asyncio.to_thread(upsert_media_file_ids_bulk_sync, DB_PATH, entries)
+            logger.debug(f"persist_media_file_ids_bulk: upserted {len(entries)} records")
         except Exception as e:
-            logger.error(f"file_id_save_error: error upserting {channel}/{post_id}/{file_unique_id}, error {str(e)}")
+            logger.error(f"file_id_bulk_save_error: error bulk-upserting {len(entries)} records, error {str(e)}")
+        finally:
+            # Clear regardless of outcome so a retry does not double-persist a stale batch.
+            self._pending_media_ids = []
 
     def _save_media_file_ids(self, message: Message) -> None:
+        """Collect a media file-id record for later bulk persistence.
+
+        IMPORTANT (4.2): this runs inside the render thread, so it must NOT touch
+        asyncio or the DB. It only appends to self._pending_media_ids; the caller
+        flushes the batch via _flush_pending_media_ids() after rendering.
+        """
         try:
             channel_username = self.get_channel_username(message)
             if not channel_username:
@@ -1003,29 +1069,8 @@ class PostParser:
 
                 if file_unique_id:
                     added_ts = datetime.now().timestamp()
-                    try:
-                        loop = asyncio.get_running_loop()
-                        if loop.is_running():
-                            global _persist_pending_count
-                            task = loop.create_task(
-                                self._persist_media_file_id_async(channel_username, message.id, file_unique_id, added_ts)
-                            )
-                            _persist_pending_count += 1
-
-                            def _on_task_done(t):
-                                global _persist_pending_count
-                                _persist_pending_count -= 1
-
-                            task.add_done_callback(_on_task_done)
-
-                            if _persist_pending_count > 5:
-                                logger.warning(f"persist_task_queue: {_persist_pending_count} pending _persist_media_file_id_async tasks (msg {message.id})")
-                            logger.debug(f"persist_task_created: queued for {channel_username}/{message.id}/{file_unique_id}, total pending: {_persist_pending_count}")
-                        else:
-                            # Fallback sync path (should not occur during normal FastAPI runtime)
-                            upsert_media_file_id_sync(DB_PATH, channel_username, message.id, file_unique_id, added_ts)
-                    except Exception as e:
-                        logger.error(f"file_id_save_error: error saving {channel_username}/{message.id}/{file_unique_id}, error {str(e)}")
+                    # Thread-safe: just append; the caller persists the batch.
+                    self._pending_media_ids.append((channel_username, message.id, file_unique_id, added_ts))
 
         except Exception as e:
             logger.error(f"file_id_collection_error: message_id {message.id}, error {str(e)}")
diff --git a/rss_generator.py b/rss_generator.py
index a26ca11..0c6f7ea 100644
--- a/rss_generator.py
+++ b/rss_generator.py
@@ -14,6 +14,7 @@ import logging
 import asyncio
 import re
 import time
+from html import escape as html_escape
 from datetime import datetime, timezone
 from types import SimpleNamespace
 from typing import Optional
@@ -30,9 +31,12 @@ Config = get_settings()
 
 logger = logging.getLogger(__name__)
 
-async def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]:
+def _create_time_based_media_groups(messages: list[Message], merge_seconds: int = 5) -> list[Message]:
     """
     Create media groups based on time difference between messages
+
+    Plain synchronous function (contains no await): runs inside the render thread
+    via _render_pipeline. Must not touch asyncio.
     """
     # Deep-copy the input list to avoid mutating cached Message objects (the cache may
     # reuse the same objects across calls with different merge_seconds values)
@@ -92,9 +96,11 @@ async def _create_time_based_media_groups(messages: list[Message], merge_seconds
 
     return messages_sorted
 
-async def _create_messages_groups(messages: list[Message]) -> list[list[Message]]:
+def _create_messages_groups(messages: list[Message]) -> list[list[Message]]:
     """
     Process messages into formatted posts, handling media groups
+
+    Plain synchronous function (contains no await): runs inside the render thread.
     """
     processing_groups: list[list[Message]] = []
     media_groups: dict[str | int, list[Message]] = {}
@@ -136,9 +142,11 @@ async def _create_messages_groups(messages: list[Message]) -> list[list[Message]
     
     return processing_groups
 
-async def _trim_messages_groups(messages_groups: list[list[Message]], limit: int):
+def _trim_messages_groups(messages_groups: list[list[Message]], limit: int):
     """
     Trim messages groups to limit
+
+    Plain synchronous function (contains no await): runs inside the render thread.
     """
     if len(messages_groups) > limit: # Trim groups if they exceed the specified limit
         messages_groups = messages_groups[:limit]
@@ -193,12 +201,13 @@ def processed_message_to_tg_message(processed_message: dict) -> Message:
     return tg_message_mock # type: ignore
 
 
-async def _render_messages_groups(messages_groups: list[list[Message]], 
-                                    post_parser: PostParser, 
-                                    exclude_flags: str | None = None, 
+def _render_messages_groups(messages_groups: list[list[Message]],
+                                    post_parser: PostParser,
+                                    exclude_flags: str | None = None,
                                     exclude_text: str | None = None):
     """
     Render message groups into HTML format
+    Plain synchronous function (contains no await): runs inside the render thread.
     Args:
         messages_groups: List of message groups (each group is a list of messages)
         post_parser: PostParser instance
@@ -213,7 +222,9 @@ async def _render_messages_groups(messages_groups: list[list[Message]],
         try:
             if len(group) == 1: # Single message - simple case
                 one_message = group[0]
-                message_data = post_parser.process_message(one_message)
+                # Feed path: raw_message not needed and sanitize deferred to the final
+                # whole-feed pass, so each fragment is sanitized exactly once.
+                message_data = post_parser.process_message(one_message, include_raw=False, sanitize=False)
                 html_parts = [
                     f'
{message_data["html"]["body"]}
',
                     f'
{message_data["html"]["footer"]}
'
@@ -228,7 +239,7 @@ async def _render_messages_groups(messages_groups: list[list[Message]],
                     'flags': message_data['flags']
                 })
             else: # Multiple messages in group - merge text and html body
-                processed_messages = [post_parser.process_message(msg) for msg in group]
+                processed_messages = [post_parser.process_message(msg, include_raw=False, sanitize=False) for msg in group]
 
                 # Determine main message for header/footer/title
                 main_message = next(  
@@ -309,7 +320,31 @@ async def _render_messages_groups(messages_groups: list[list[Message]],
     rendered_posts.sort(key=lambda x: x['date'] if x['date'] is not None else 0.0, reverse=True)
     return rendered_posts
 
-async def generate_channel_rss(channel: str | int, 
+
+def _render_pipeline(messages: list[Message],
+                     post_parser: PostParser,
+                     limit: int,
+                     exclude_flags: str | None,
+                     exclude_text: str | None,
+                     merge_seconds: int,
+                     time_based_merge: bool):
+    """
+    Full synchronous feed render pipeline (grouping + trimming + rendering).
+
+    Runs entirely in a worker thread via a single asyncio.to_thread call. It contains
+    NO await, NO asyncio primitives, NO create_task/get_running_loop — all the CPU-heavy
+    work (deepcopy, grouping, bleach-free rendering) happens here off the event loop.
+    Media file-id records are accumulated on post_parser._pending_media_ids and flushed
+    by the caller after this returns.
+    """
+    if time_based_merge:
+        messages = _create_time_based_media_groups(messages, merge_seconds)
+    message_groups = _create_messages_groups(messages)
+    message_groups = _trim_messages_groups(message_groups, limit)
+    return _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text)
+
+
+async def generate_channel_rss(channel: str | int,
                                 client: Client, 
                                 limit: int = 20, 
                                 exclude_flags: str | None = None,
@@ -389,14 +424,23 @@ async def generate_channel_rss(channel: str | int,
         messages_elapsed = time.time() - messages_start_time
         logger.debug(f"rss_messages_retrieval_timing: channel {channel}, {len(messages)} messages retrieved in {messages_elapsed:.3f} seconds")
         
-        # Process messages into groups and render them
+        # Process messages into groups and render them.
+        # The whole grouping/trimming/rendering pipeline is CPU-heavy (deepcopy +
+        # per-message rendering) and contains no await, so run it in ONE worker
+        # thread to keep the event loop responsive.
         processing_start_time = time.time()
-        if Config['time_based_merge']:
-            messages = await _create_time_based_media_groups(messages, merge_seconds)
-        message_groups = await _create_messages_groups(messages)
-        message_groups = await _trim_messages_groups(message_groups, limit)
-        final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text)
-        
+        try:
+            final_posts = await asyncio.to_thread(
+                _render_pipeline, messages, post_parser, limit,
+                exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'],
+            )
+        finally:
+            # Persist media file-ids collected during rendering with a single bulk
+            # upsert — in a finally so a partial render still records what it collected
+            # (the flush is best-effort and swallows its own errors, so it cannot mask a
+            # render exception).
+            await post_parser._flush_pending_media_ids()
+
         processing_elapsed = time.time() - processing_start_time
         logger.debug(f"rss_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds")
         
@@ -443,8 +487,13 @@ async def generate_channel_rss(channel: str | int,
                     )
                 sanitized_html = await asyncio.to_thread(_sanitize_sync, post['html'])
             except Exception as e:
+                # FAIL CLOSED: since 4.4 removed the per-fragment passes, post['html'] is
+                # raw channel-controlled HTML, and this is its ONLY sanitize. If bleach
+                # itself throws (e.g. RecursionError on pathological nested HTML), do NOT
+                # emit the raw payload (that would be stored XSS) — html.escape it so the
+                # content survives as inert text.
                 logger.error(f"rss_html_sanitization_error: channel {channel}, message_id {post['message_id']}, error {str(e)}")
-                sanitized_html = post['html']
+                sanitized_html = html_escape(post['html'])
             fe.content(content=sanitized_html, type='CDATA')
             
             if post['date'] is not None:
@@ -601,16 +650,19 @@ async def generate_channel_html(channel: str | int,
         enrichment_elapsed = time.time() - enrichment_start_time
         logger.debug(f"html_reply_enrichment_timing: channel {channel}, replies enriched in {enrichment_elapsed:.3f} seconds")
 
-        # Process messages into groups and render them
+        # Process messages into groups and render them in ONE worker thread
+        # (CPU-heavy, no await) to keep the event loop responsive.
         processing_start_time = time.time()
-        if Config['time_based_merge']:
-            messages = await _create_time_based_media_groups(messages, merge_seconds)
+        try:
+            final_posts = await asyncio.to_thread(
+                _render_pipeline, messages, post_parser, limit,
+                exclude_flags, exclude_text, merge_seconds, Config['time_based_merge'],
+            )
+        finally:
+            # Persist media file-ids collected during rendering with a single bulk
+            # upsert — in a finally so a partial render still records what it collected.
+            await post_parser._flush_pending_media_ids()
 
-        # Process messages into groups and render them
-        message_groups = await _create_messages_groups(messages)
-        message_groups = await _trim_messages_groups(message_groups, limit)
-        final_posts = await _render_messages_groups(message_groups, post_parser, exclude_flags, exclude_text)
-        
         processing_elapsed = time.time() - processing_start_time
         logger.debug(f"html_messages_processing_timing: channel {channel}, {len(final_posts)} posts processed in {processing_elapsed:.3f} seconds")
 
@@ -646,7 +698,11 @@ async def generate_channel_html(channel: str | int,
                 )
             html = await asyncio.to_thread(_sanitize_sync, html)
         except Exception as e:
+            # FAIL CLOSED (see the RSS path): `html` is now the raw concatenated feed
+            # (4.4 removed the per-fragment passes). If bleach throws, html.escape the
+            # whole feed rather than returning raw channel HTML/JS to the client.
             logger.error(f"html_final_sanitization_error: channel {channel}, error {str(e)}")
+            html = html_escape(html)
         
         html_gen_elapsed = time.time() - html_gen_start_time
         logger.debug(f"html_generation_timing: channel {channel}, HTML generated in {html_gen_elapsed:.3f} seconds")
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_stage4_eventloop.py b/tests/test_stage4_eventloop.py
new file mode 100644
index 0000000..c0648ec
--- /dev/null
+++ b/tests/test_stage4_eventloop.py
@@ -0,0 +1,471 @@
+# 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 4 (event-loop hygiene for feed generation) tests.
+
+Covers:
+- 4.1 raw_message laziness: feeds do NOT compute str(message); JSON/debug HTML do.
+- 4.2 side-effect IO removed from process_message: _save_media_file_ids only appends to
+      self._pending_media_ids; the caller flushes once via upsert_media_file_ids_bulk_sync.
+      Also: the render path contains NO create_task / get_running_loop / to_thread.
+- 4.3 render pipeline moved into a single thread: the four render functions are now plain
+      sync functions and actually execute off the main thread; deepcopy of a pickled Message
+      does not crash; a 100-message feed generates correctly.
+- 4.4 sanitize coverage (XSS): a click"
+
+
+class _Str(str):
+    """Minimal stand-in for Pyrogram's Str: .html returns the raw string unchanged,
+    so a malicious payload reaches the pre-sanitization html body just like real text
+    carrying entities would."""
+    @property
+    def html(self):
+        return str(self)
+
+
+def make_message(mid, text=None, media=None, photo_uid=None, username="testchan",
+                 date=None):
+    m = SimpleNamespace()
+    m.id = mid
+    m.date = date or datetime(2024, 1, 1, 12, 0, mid % 60, tzinfo=timezone.utc)
+    m.text = _Str(text) if text is not None else None
+    m.caption = None
+    m.media = media
+    m.web_page = None
+    m.poll = None
+    m.service = None
+    m.forward_origin = None
+    m.reply_to_message = None
+    m.reply_to_message_id = None
+    m.sender_chat = None
+    m.from_user = None
+    m.reactions = None
+    m.views = 100
+    m.media_group_id = None
+    m.show_caption_above_media = False
+    m.chat = SimpleNamespace(id=-1001234567890, username=username)
+    # media sub-objects default to None
+    for attr in ("photo", "video", "document", "audio", "voice",
+                 "video_note", "animation", "sticker"):
+        setattr(m, attr, None)
+    if media == MessageMediaType.PHOTO and photo_uid:
+        m.photo = SimpleNamespace(file_unique_id=photo_uid)
+    return m
+
+
+def _co_names(func):
+    return set(func.__code__.co_names)
+
+
+# ---------------------------------------------------------------------------
+# 4.3 — render functions are plain sync and run off the main thread
+# ---------------------------------------------------------------------------
+
+def test_render_functions_are_sync():
+    for fn in (_create_time_based_media_groups, _create_messages_groups,
+               _trim_messages_groups, _render_messages_groups, _render_pipeline):
+        assert not asyncio.iscoroutinefunction(fn), f"{fn.__name__} must be a plain sync function"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_runs_in_worker_thread(monkeypatch):
+    main_ident = threading.get_ident()
+    seen = {}
+
+    real_render = rss_module._render_messages_groups
+
+    def spy(*args, **kwargs):
+        seen["ident"] = threading.get_ident()
+        return real_render(*args, **kwargs)
+
+    monkeypatch.setattr(rss_module, "_render_messages_groups", spy)
+
+    async def fake_get_chat(client, channel):
+        return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
+
+    async def fake_get_history(client, channel, limit=20):
+        return [make_message(i, text=f"post {i}") for i in range(5)]
+
+    monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
+    monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
+
+    await generate_channel_rss("testchan", client=SimpleNamespace(), limit=5)
+    assert "ident" in seen
+    assert seen["ident"] != main_ident, "render pipeline must run in a worker thread, not the loop thread"
+
+
+def test_deepcopy_of_pickled_message_does_not_crash():
+    from pyrogram.types import Message, Chat
+    from pyrogram.enums import ChatType
+    m = Message(id=7, date=datetime.now(timezone.utc), text="hello",
+                chat=Chat(id=-1001, type=ChatType.CHANNEL, username="testchan"))
+    roundtripped = pickle.loads(pickle.dumps(m))   # mimics the pickle cache
+    clone = copy.deepcopy(roundtripped)            # what _create_time_based_media_groups does
+    assert clone.id == 7
+    assert clone.chat.username == "testchan"
+
+
+# ---------------------------------------------------------------------------
+# 4.2 — no asyncio in the render path; bulk upsert after render
+# ---------------------------------------------------------------------------
+
+def test_render_path_has_no_asyncio_side_effects():
+    banned = {"create_task", "get_running_loop", "to_thread", "ensure_future"}
+    funcs = [
+        _render_pipeline, _create_time_based_media_groups, _create_messages_groups,
+        _trim_messages_groups, _render_messages_groups,
+        PostParser.process_message, PostParser._generate_html_body,
+        PostParser._generate_html_media, PostParser.generate_html_footer,
+        PostParser._reactions_views_links, PostParser._save_media_file_ids,
+        PostParser._sanitize_html,
+    ]
+    for fn in funcs:
+        offenders = _co_names(fn) & banned
+        assert not offenders, f"{fn.__qualname__} references forbidden asyncio names: {offenders}"
+
+
+@pytest.mark.asyncio
+async def test_media_ids_persisted_via_bulk_upsert(monkeypatch):
+    calls = []
+
+    def fake_bulk(db_path, entries):
+        calls.append(list(entries))
+
+    monkeypatch.setattr(pp_module, "upsert_media_file_ids_bulk_sync", fake_bulk)
+
+    async def fake_get_chat(client, channel):
+        return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
+
+    async def fake_get_history(client, channel, limit=20):
+        return [
+            make_message(1, text="just text"),
+            make_message(2, media=MessageMediaType.PHOTO, photo_uid="uid_abc"),
+        ]
+
+    monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
+    monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
+
+    await generate_channel_rss("testchan", client=SimpleNamespace(), limit=10)
+
+    assert len(calls) == 1, "bulk upsert must be called exactly once after render"
+    entries = calls[0]
+    assert len(entries) == 1, "only the photo message contributes a media id"
+    channel, post_id, fid, _ts = entries[0]
+    assert (channel, post_id, fid) == ("testchan", 2, "uid_abc")
+
+
+@pytest.mark.asyncio
+async def test_save_media_file_ids_only_appends(monkeypatch):
+    # Even with a running loop, _save_media_file_ids must not create tasks — just append.
+    parser = PostParser(SimpleNamespace())
+    monkeypatch.setattr(pp_module, "upsert_media_file_ids_bulk_sync",
+                        lambda *a, **k: (_ for _ in ()).throw(AssertionError("must not be called directly")))
+    msg = make_message(3, media=MessageMediaType.PHOTO, photo_uid="uid_x")
+    parser._save_media_file_ids(msg)
+    assert parser._pending_media_ids == [("testchan", 3, "uid_x", parser._pending_media_ids[0][3])]
+
+
+# ---------------------------------------------------------------------------
+# 4.1 — raw_message laziness
+# ---------------------------------------------------------------------------
+
+def test_raw_message_lazy_for_feed():
+    parser = PostParser(SimpleNamespace())
+    result = parser.process_message(make_message(10, text="hi"), include_raw=False, sanitize=False)
+    assert "raw_message" not in result
+
+
+def test_raw_message_present_for_json_and_debug():
+    parser = PostParser(SimpleNamespace())
+    result = parser.process_message(make_message(11, text="hi"), include_raw=True)
+    assert "raw_message" in result
+    assert isinstance(result["raw_message"], str)
+
+
+@pytest.mark.asyncio
+async def test_100_message_feed_generates(monkeypatch):
+    async def fake_get_chat(client, channel):
+        return SimpleNamespace(title="Test", username="testchan", id=-1001234567890)
+
+    async def fake_get_history(client, channel, limit=20):
+        return [make_message(i, text=f"post number {i}") for i in range(1, 101)]
+
+    monkeypatch.setattr("tg_cache.cached_get_chat", fake_get_chat, raising=False)
+    monkeypatch.setattr("tg_cache.cached_get_chat_history", fake_get_history, raising=False)
+
+    rss = await generate_channel_rss("testchan", client=SimpleNamespace(), limit=100)
+    assert rss.count("") == 100
+    assert "post number 50" in rss
+
+
+# ---------------------------------------------------------------------------
+# 4.4 — XSS: payload stripped in all four outputs
+# ---------------------------------------------------------------------------
+
+def _assert_clean(html_str, where):
+    assert "