diff --git a/api_server.py b/api_server.py index ea8ce36..6fbe2ea 100644 --- a/api_server.py +++ b/api_server.py @@ -15,7 +15,6 @@ import re import uuid import mimetypes from typing import List, Union, Any -from urllib.parse import quote import json from datetime import datetime @@ -23,9 +22,9 @@ import time from contextlib import asynccontextmanager import random import asyncio +from concurrent.futures import ThreadPoolExecutor import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) -from starlette.middleware.base import BaseHTTPMiddleware from starlette.background import BackgroundTask import sys @@ -34,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, StreamingResponse +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 @@ -51,14 +51,35 @@ magic_mime = magic.Magic(mime=True) class ZeroSizeFileError(Exception): """Custom exception for zero-size files found or downloaded.""" -class RequestLoggingMiddleware(BaseHTTPMiddleware): - async def dispatch(self, request: Request, call_next): - # Log only method and URL at debug level to avoid flooding logs on active RSS polling - logger.debug(f"Request: {request.method} {request.url}") +class RequestLoggingMiddleware: + """Pure-ASGI request logger (no BaseHTTPMiddleware). + + BaseHTTPMiddleware runs the downstream app in a separate anyio task and pumps the + response through an in-memory stream, which adds per-request overhead and interacts + badly with streaming bodies, background tasks and client cancellation. This plain + ASGI middleware only wraps `send` to observe the response status line, so it never + buffers the body — the FileResponse stream flows straight through untouched. + """ + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope["type"] != "http": + await self.app(scope, receive, send) + return + # Log only method and path (with query, matching the old request.url logging) at + # debug level to avoid flooding logs on active RSS polling. + _qs = scope.get("query_string") or b"" + _path = scope["path"] + (f"?{_qs.decode('latin-1')}" if _qs else "") + logger.debug(f"Request: {scope['method']} {_path}") + + async def send_wrapper(message): + if message["type"] == "http.response.start": + logger.debug(f"Response status: {message['status']}") + await send(message) + try: - response = await call_next(request) - logger.debug(f"Response status: {response.status_code}") - return response + await self.app(scope, receive, send_wrapper) except Exception as e: logger.error(f"Request processing error: {str(e)}") raise @@ -71,12 +92,118 @@ Config = get_settings() HTTP_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) # semaphore for live HTTP media requests BACKGROUND_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(2) # semaphore for background cache worker download_queue = asyncio.Queue(maxsize=100) +# How stale a temp_* file's mtime must be before a serve refreshes it (keeps the 1h +# sweeper from deleting an actively-viewed video). Well below 1h so the file stays alive, +# but large enough that the mtime — and thus FileResponse's ETag — is stable within any +# such window; a view running longer than one interval costs at most one safe 200 +# If-Range restart per interval (a full re-fetch, never corruption). +TEMP_MTIME_REFRESH_INTERVAL = 300 # seconds + +# In-flight download dedup registry: maps (channel, post_id, file_unique_id) to the +# shared Future of an ongoing download. The FIRST request for a key runs the download in +# a DETACHED task and shares its Future; concurrent requests await that Future instead of +# starting their own download. Detaching the download means a client disconnecting (which +# cancels its request coroutine) can never cancel the download nor leave the Future +# forever-pending — so waiters can't hang. See _download_deduped. +_inflight: dict[tuple[str, int, str], asyncio.Future] = {} +# Strong refs to detached download tasks so they are not garbage-collected mid-flight. +_inflight_tasks: set[asyncio.Task] = set() + +async def _supervised(factory, name: str, min_restart_interval: float = 60.0): + """Run factory() forever, restarting it if it dies with a non-cancellation error. + + A background loop that returns or raises (anything except CancelledError) is logged + at CRITICAL and restarted, but successive (re)starts are spaced at least + min_restart_interval seconds apart so a hard-failing task can't spin the event loop. + CancelledError (shutdown) is propagated to the child and stops supervision. + """ + while True: + start = time.monotonic() + task = asyncio.create_task(factory(), name=name) + try: + await task + # A supervised background loop is not expected to return on its own. + logger.critical(f"supervised_task_exited: {name} returned unexpectedly; restarting") + except asyncio.CancelledError: + # Shutdown or external cancel: propagate to the child, then stop supervising. + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + raise + except Exception as e: + logger.critical(f"supervised_task_crashed: {name} died with {e!r}; restarting", exc_info=True) + # Rate-limit restarts: keep successive starts at least min_restart_interval apart. + elapsed = time.monotonic() - start + if elapsed < min_restart_interval: + 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"]) - + # Enlarge the default threadpool: SQLite/python-magic/pickle/os.walk all run via + # asyncio.to_thread, and the interpreter default (min(32, cpu+4) = 5-6 on a 1-2 CPU + # container) is too small under load. Configurable via IO_THREAD_POOL_SIZE. + loop = asyncio.get_running_loop() + io_executor = ThreadPoolExecutor(max_workers=Config["io_thread_pool_size"], thread_name_prefix="io") + loop.set_default_executor(io_executor) + base_cache_dir = os.path.abspath("./data/cache") os.makedirs(base_cache_dir, exist_ok=True) # Create cache directory @@ -84,11 +211,15 @@ async def lifespan(_: FastAPI): await asyncio.to_thread(init_db_sync, DB_PATH) await client.start() - background_task = asyncio.create_task(cache_media_files()) # Start background task - worker_task = asyncio.create_task(background_download_worker()) # Start download worker + # Supervise the background tasks: if either dies (not via cancellation) it is logged + # 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: @@ -97,7 +228,20 @@ 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) app = FastAPI(title="Pyrogram Bridge", lifespan=lifespan) app.add_middleware(RequestLoggingMiddleware) @@ -201,9 +345,43 @@ async def delayed_delete_file(file_path: str, delay: int = 300) -> None: async def prepare_file_response(file_path: str, request: Request, delete_after: bool = False, - media_key: tuple[str, int, str] | None = None) -> StreamingResponse: - """Prepare a streaming file response with HTTP Range request support.""" - if not os.path.exists(file_path): + media_key: tuple[str, int, str] | None = None) -> Response: + """Serve a cached media file via Starlette's FileResponse. + + FileResponse handles Range/If-Range/206/416/multipart and sets + Accept-Ranges/ETag/Last-Modified itself, and reads the file efficiently (no per-64KB + to_thread hop that starved the threadpool). We keep: the early 404 pre-check, the MIME + logic (python-magic + SQLite type cache), the stage-2 temp_* mtime touch, and the + delete_after BackgroundTask. + """ + # `request` is unused now that FileResponse parses the Range header itself, but the + # signature is kept for call-site compatibility (and future needs). + + # Keep an actively-viewed large-video temp file alive: refresh its mtime so the 1h + # sweeper (which deletes temp_* by mtime) won't remove it out from under a viewer. + # DEBOUNCED: FileResponse derives ETag/Last-Modified from mtime, so touching on EVERY + # serve would change the validators per request. We only refresh when the mtime is + # already older than TEMP_MTIME_REFRESH_INTERVAL — far below the 1h sweeper window, so + # the file stays alive, and the ETag is stable within any such window (a view running + # longer than one interval costs at most one safe 200 If-Range restart per interval). + if os.path.basename(file_path).startswith("temp_"): + try: + age = time.time() - await asyncio.to_thread(os.path.getmtime, file_path) + if age > TEMP_MTIME_REFRESH_INTERVAL: + await asyncio.to_thread(os.utime, file_path, None) + except OSError as e: + logger.debug(f"Failed to refresh mtime for {file_path}: {e}") + + # Take ONE authoritative stat and hand it to FileResponse as stat_result. This both + # (a) preserves the 404 semantics: FileResponse with stat_result=None re-stats at + # send-time and raises a RuntimeError (-> 500, escaping this handler's try/except) if + # the file was swept between here and the send; and (b) makes the ETag/Last-Modified + # reflect exactly the mtime observed after the optional touch above. The remaining + # narrow window (deleted between this stat and FileResponse's own open) truncates the + # body — that pre-existed the FileResponse migration and is not handled here. + try: + stat_result = await asyncio.to_thread(os.stat, file_path) + except FileNotFoundError: raise HTTPException(status_code=404, detail="File not found") media_type: str | None = None @@ -230,110 +408,120 @@ async def prepare_file_response(file_path: str, request: Request, delete_after: logger.debug(f"Determined media type for {os.path.basename(file_path)}: {media_type}") - try: - total_size = os.path.getsize(file_path) - except FileNotFoundError: - raise HTTPException(status_code=404, detail="File not found") - range_header = request.headers.get("range") - - if range_header: - # Parse Range header according to RFC 7233 - try: - range_value = range_header.strip() - if not range_value.startswith("bytes="): - raise ValueError("Only bytes ranges are supported") - range_spec = range_value[len("bytes="):] - if range_spec.startswith("-"): - # Suffix range: bytes=-N (last N bytes) - suffix_length = int(range_spec[1:]) - start = max(0, total_size - suffix_length) - end = total_size - 1 - elif range_spec.endswith("-"): - # Open-ended range: bytes=START- - start = int(range_spec[:-1]) - end = total_size - 1 - else: - # Full range: bytes=START-END - start_str, end_str = range_spec.split("-", 1) - start = int(start_str) - end = int(end_str) - except (ValueError, IndexError) as e: - logger.debug(f"Invalid Range header '{range_header}': {e}") - return Response( - status_code=416, - headers={"Content-Range": f"bytes */{total_size}"} - ) - - # Clamp end to file size - 1 (RFC 7233 allows end >= total_size) - end = min(end, total_size - 1) - - # If start is beyond file size, return 416 - if start >= total_size: - return Response( - status_code=416, - headers={"Content-Range": f"bytes */{total_size}"} - ) - - content_length = end - start + 1 - status_code = 206 - headers = { - "Content-Disposition": ( - f"inline; filename=\"{os.path.basename(file_path)}\"; " - f"filename*=UTF-8''{quote(os.path.basename(file_path))}" - ), - # Files are addressed by file_unique_id which is immutable in Telegram, - # so it is safe to cache them aggressively on the client side. - "Cache-Control": "public, max-age=86400, immutable", - "Accept-Ranges": "bytes", - "Content-Range": f"bytes {start}-{end}/{total_size}", - "Content-Length": str(content_length), - } - else: - # No Range header — serve full file with status 200 - start = 0 - end = total_size - 1 - status_code = 200 - headers = { - "Content-Disposition": ( - f"inline; filename=\"{os.path.basename(file_path)}\"; " - f"filename*=UTF-8''{quote(os.path.basename(file_path))}" - ), - # Files are addressed by file_unique_id which is immutable in Telegram, - # so it is safe to cache them aggressively on the client side. - "Cache-Control": "public, max-age=86400, immutable", - "Accept-Ranges": "bytes", - "Content-Length": str(total_size), - } - - chunk_size = 64 * 1024 # 64 KB chunks - - async def file_chunk_generator(): - """Async generator that reads file in chunks; each chunk is a separate open/seek/read/close.""" - bytes_remaining = end - start + 1 - offset = start - while bytes_remaining > 0: - to_read = min(chunk_size, bytes_remaining) - def read_at(path, off, size): - # Open, seek, read, and close within a single thread call to avoid fd leaks - with open(path, "rb") as f: - f.seek(off) - return f.read(size) - data = await asyncio.to_thread(read_at, file_path, offset, to_read) - if not data: - break - bytes_remaining -= len(data) - offset += len(data) - yield data - + # Delete the temporary file once the response has been fully sent (stage-2 delete_after). + # FileResponse runs this BackgroundTask after streaming the body. background = BackgroundTask(delayed_delete_file, file_path) if delete_after else None - return StreamingResponse( - content=file_chunk_generator(), - status_code=status_code, + + # FileResponse handles Range/If-Range/206/416/multipart and sets + # Accept-Ranges/ETag/Last-Modified itself (from the stat_result we pass). Do NOT + # hand-build Content-Disposition: FileResponse forms it from filename= — + # `inline; filename="x"` for an ASCII name, adding `filename*=UTF-8''x` only for a + # non-ASCII name. It uses setdefault, so a manual header would OVERRIDE it, not double + # it; letting FileResponse own it keeps the RFC 5987 encoding correct. + # + # Files are addressed by file_unique_id which is immutable in Telegram, so it is safe + # to cache them aggressively on the client side. + return FileResponse( + file_path, media_type=media_type, - headers=headers, + filename=os.path.basename(file_path), + content_disposition_type="inline", + stat_result=stat_result, + headers={"Cache-Control": "public, max-age=86400, immutable"}, background=background, ) +def _media_download_timeout(file_size: int) -> float: + """Scale the download timeout with file size. + + Effective floor of `media_download_min_speed` bytes/s (timeout ≈ size / min_speed), + clamped to [media_download_timeout_min, media_download_timeout_max] seconds. + A non-positive/unknown size falls back to the minimum timeout. + """ + min_t = Config["media_download_timeout_min"] + max_t = Config["media_download_timeout_max"] + min_speed = Config["media_download_min_speed"] + if file_size <= 0 or min_speed <= 0: + return float(min_t) + return float(min(max_t, max(min_t, file_size // min_speed))) + + +async def _download_atomic(file_id: str, final_path: str, timeout: float) -> str: + """Download a media file through a unique partial path and atomically publish it. + + Invariant enforced here: a file that exists at a FINAL name (`{file_unique_id}` or + `temp_{file_unique_id}`) is ALWAYS complete. The downloader only ever writes to a + unique `{final_path}.part.{hex}` path; the file appears at its final name solely via + os.rename (atomic on POSIX). The finally block ALWAYS removes our partial (on timeout, + cancel, zero-size, or losing a rename race), so no stub is ever served or left behind. + + Raises ZeroSizeFileError if the download produced a missing/zero-size file. + """ + part_path = f"{final_path}.part.{uuid.uuid4().hex}" + try: + await client.safe_download_media(file_id, part_path, timeout=timeout) + if not os.path.exists(part_path) or os.path.getsize(part_path) == 0: + raise ZeroSizeFileError( + f"Downloaded file for {final_path} is zero size or missing after download attempt." + ) + # Publish atomically, but only if nobody else already produced the final file + # (a concurrent request that won the race). rename is atomic on POSIX. + if not os.path.exists(final_path): + os.rename(part_path, final_path) + return final_path + finally: + # Always clean up our partial: timeout, cancellation, zero-size, or race loser + # (rename skipped because final_path already existed). + if os.path.exists(part_path): + try: + os.remove(part_path) + except OSError as e: + logger.warning(f"cleanup_error: Failed to remove partial file {part_path}: {e}") + + +async def _download_deduped(channel: Union[str, int], post_id: int, file_unique_id: str) -> tuple[Union[str, None], bool]: + """Deduplicate concurrent downloads of the same media by (channel, post_id, fid). + + The first request for a key runs download_media_file in a DETACHED task and shares its + Future; concurrent requests await the same Future (bounded by wait_for). The detached + task sets the Future's result/exception (on success/failure) and its finally ALWAYS + pops the key — so both happen before the task ends: a completed Future never leaves the + key stuck, and a client disconnect (cancelling only the awaiting request coroutine) can + neither cancel the download nor hang other waiters. + """ + key = (str(channel), post_id, file_unique_id) + fut = _inflight.get(key) + if fut is None: + loop = asyncio.get_running_loop() + fut = loop.create_future() + _inflight[key] = fut + + async def _runner(): + try: + result = await download_media_file(channel, post_id, file_unique_id) + if not fut.done(): + fut.set_result(result) + except BaseException as e: # noqa: BLE001 — must forward ANY failure to waiters + if not fut.done(): + fut.set_exception(e) + finally: + _inflight.pop(key, None) + + task = asyncio.create_task(_runner()) + _inflight_tasks.add(task) + task.add_done_callback(_inflight_tasks.discard) + # Retrieve the exception in a done-callback to silence "exception was never + # retrieved" if every waiter disconnects before awaiting the Future. + fut.add_done_callback(lambda f: f.cancelled() or f.exception()) + + # Waiter timeout: a generous safety net above the maximum per-download timeout (the + # download is itself internally bounded), so a live download always completes first. + waiter_timeout = float(Config["media_download_timeout_max"]) + 120.0 + # shield so a timed-out / cancelled waiter does NOT cancel the shared Future that the + # detached task owns and other waiters depend on. + return await asyncio.wait_for(asyncio.shield(fut), timeout=waiter_timeout) + + async def download_media_file(channel: Union[str, int], post_id: int, file_unique_id: str) -> tuple[Union[str, None], bool]: """ Download media file from Telegram and save to cache @@ -377,18 +565,27 @@ async def download_media_file(channel: Union[str, int], post_id: int, file_uniqu logger.error(f"Failed to get video file size for message {post_id}: {str(e)}") if is_large_video: - # For large video, download without permanent caching; use a temporary file + # For large video, download without permanent caching; use a temporary file. temp_file_path = os.path.join(post_dir, f"temp_{file_unique_id}") - if os.path.exists(temp_file_path): + # A temp_{fid} file now only ever appears via atomic rename (see _download_atomic), + # so its presence GUARANTEES a complete file — serve it directly. + if os.path.exists(temp_file_path) and os.path.getsize(temp_file_path) > 0: logger.info(f"Temporary file {temp_file_path} already exists, serving cached large video") return temp_file_path, False file_id = await find_file_id_in_message(message, file_unique_id) if not file_id: logger.error(f"Media with file_unique_id {file_unique_id} not found in message {post_id}") raise HTTPException(status_code=404, detail="File not found in message") - logger.info(f"Downloading large video file {file_unique_id} to temporary path {temp_file_path}") + # Scale the timeout with size (≈256 KB/s floor) so slow big-video downloads aren't + # cut short at the fixed 120s used for regular files. try: - file_path = await client.safe_download_media(file_id, temp_file_path) + file_size = int(message.video.file_size or 0) + except Exception: + file_size = 0 + timeout = _media_download_timeout(file_size) + logger.info(f"Downloading large video file {file_unique_id} to temporary path {temp_file_path} (timeout={timeout:.0f}s)") + try: + file_path = await _download_atomic(file_id, temp_file_path, timeout) except asyncio.TimeoutError: logger.error(f"Timeout downloading large video {file_unique_id}") raise HTTPException(status_code=504, detail="Download timeout") @@ -408,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) @@ -437,46 +629,17 @@ async def download_media_file(channel: Union[str, int], post_id: int, file_uniqu raise HTTPException(status_code=404, detail="File not found in message") - # Use a unique temp path to avoid race conditions when concurrent requests download the same file - temp_path = cache_path + f".tmp.{uuid.uuid4().hex}" + # Download through a unique `.part.` file and atomically publish to the final cache + # path. _download_atomic owns the zero-size check, the race-loser cleanup, and the + # rename-only-if-absent logic, keeping the "final name = complete file" invariant. try: - file_path = await client.safe_download_media(file_id, temp_path) + file_path = await _download_atomic(file_id, cache_path, timeout=float(Config["media_download_timeout_min"])) except asyncio.TimeoutError: logger.error(f"Timeout downloading media {file_unique_id}") - # Clean up the temp file if it was created - if os.path.exists(temp_path): - try: - os.remove(temp_path) - except OSError: - pass raise HTTPException(status_code=504, detail="Download timeout") - # Check if the downloaded temp file exists and has a size greater than zero - if not file_path or not os.path.exists(temp_path) or os.path.getsize(temp_path) == 0: - logger.error(f"download_failed_zero_size: Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing.") - # Attempt to clean up the invalid temp file - if os.path.exists(temp_path): - try: - os.remove(temp_path) - logger.info(f"Removed zero-size temp file: {temp_path}") - except OSError as e: - logger.error(f"cleanup_error: Failed to remove zero-size temp file {temp_path}: {e}") - # Raise specific error to indicate download failure - raise ZeroSizeFileError(f"Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing after download attempt.") - - # Atomic rename: if another concurrent request already wrote the final file, discard our temp copy - if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0: - logger.info(f"Concurrent download already completed for {file_unique_id}, discarding temp file") - try: - os.remove(temp_path) - except OSError as e: - logger.warning(f"cleanup_error: Failed to remove temp file {temp_path}: {e}") - return cache_path, False - - # Atomically move temp file to final cache path (atomic on POSIX) - os.rename(temp_path, cache_path) logger.info(f"Downloaded media file {file_unique_id} to {cache_path}") - return cache_path, False + return file_path, False def remove_old_cached_files_sync(media_files: list, cache_dir: str) -> tuple[list, int]: @@ -532,8 +695,9 @@ def remove_old_cached_files_sync(media_files: list, cache_dir: str) -> tuple[lis for root, _, files in os.walk(cache_dir): for file in files: is_large_video_temp = file.startswith("temp_") - # Match exactly the format generated in download_media_file: {name}.tmp.{32 hex chars} - is_race_temp = bool(re.match(r'^.+\.tmp\.[0-9a-f]{32}$', file)) + # Match the partial-download suffixes: the new `.part.{32 hex}` and the legacy + # `.tmp.{32 hex}` (old stubs may still be on disk after the rename to .part.). + is_race_temp = bool(re.match(r'^.+\.(part|tmp)\.[0-9a-f]{32}$', file)) if not (is_large_video_temp or is_race_temp): continue file_path = os.path.join(root, file) @@ -602,7 +766,10 @@ async def download_new_files(media_files: list, cache_dir: str) -> None: cache_path = os.path.join(post_dir, file_unique_id) if not os.path.exists(cache_path): try: - await download_queue.put((channel, post_id, file_unique_id)) + # put_nowait so a full queue raises QueueFull instead of blocking + # cache_media_files (and thus the sweeper) forever. `await put()` + # never raises QueueFull, which made the except below dead code. + download_queue.put_nowait((channel, post_id, file_unique_id)) files_queued += 1 logger.debug(f"Queued for background download: {channel}/{post_id}/{file_unique_id}") except asyncio.QueueFull: @@ -620,19 +787,22 @@ async def download_new_files(media_files: list, cache_dir: str) -> None: async def background_download_worker(): """Worker that processes downloads from queue""" while True: + # get() is OUTSIDE the try so task_done() in finally always balances exactly one + # successful get(). Cancellation propagates cleanly here (nothing to unbalance). + item = await download_queue.get() + channel, post_id, file_unique_id = item + logger.info(f"Background download: {channel}/{post_id}/{file_unique_id}") try: - channel, post_id, file_unique_id = await download_queue.get() - logger.info(f"Background download: {channel}/{post_id}/{file_unique_id}") - - try: - async with BACKGROUND_DOWNLOAD_SEMAPHORE: # limit concurrent background downloads - await download_media_file(channel, post_id, file_unique_id) - await asyncio.sleep(2) - except Exception as e: - logger.error(f"Background download error for {channel}/{post_id}/{file_unique_id}: {e}") - + async with BACKGROUND_DOWNLOAD_SEMAPHORE: # limit concurrent background downloads + await download_media_file(channel, post_id, file_unique_id) + await asyncio.sleep(2) + except errors.FloodWait as e: + # Must be caught BEFORE the generic Exception (FloodWait subclasses RPCError), + # otherwise the worker would hammer Telegram while under a flood wait. + logger.warning(f"bg_download_floodwait: {channel}/{post_id}/{file_unique_id} sleeping {e.value}s") + await asyncio.sleep(min(int(e.value) + 5, 900)) except Exception as e: - logger.error(f"Background download worker error: {e}") + logger.error(f"Background download error for {channel}/{post_id}/{file_unique_id}: {e}") finally: download_queue.task_done() @@ -878,7 +1048,12 @@ async def get_raw_post_json(channel: str, post_id: int, request: Request, token: if isinstance(channel, str) and channel.startswith('-100'): channel_id = int(channel) - message = await client.client.get_messages(channel_id, post_id) + # Bound the RPC (stage-1 DoD: every Telegram RPC has a timeout). This + # endpoint is not under the tg_rpc gate, so a hang here only blocks this + # one request, but leaving it unbounded still violates the invariant. + message = await asyncio.wait_for( + client.client.get_messages(channel_id, post_id), timeout=30 + ) if not message: raise HTTPException(status_code=404, detail="Post not found") @@ -890,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: @@ -903,8 +1118,9 @@ async def health_check(request: Request, token: str | None = None) -> Response: logger.info(f"Local request, skipping token check for health check.") try: - me = await client.client.get_me() - + # Bound the Telegram RPC so a hung get_me cannot hang the healthcheck. + me = await asyncio.wait_for(client.client.get_me(), timeout=10) + # Offload heavy filesystem scanning to threadpool cache_stats = await asyncio.to_thread(calculate_cache_stats) @@ -934,7 +1150,7 @@ async def health_check(request: Request, token: str | None = None) -> Response: @app.get("/media/{channel}/{post_id}/{file_unique_id}/{digest}", response_model=None) @app.get("/media/{channel}/{post_id}/{file_unique_id}", response_model=None) -async def get_media(channel: str, post_id: int, file_unique_id: str, request: Request, digest: str | None = None) -> StreamingResponse|Response: +async def get_media(channel: str, post_id: int, file_unique_id: str, request: Request, digest: str | None = None) -> Response: try: url = f"{channel}/{post_id}/{file_unique_id}" if not verify_media_digest(url, digest): @@ -960,29 +1176,46 @@ 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)) _sem_wait_start = _time.monotonic() - async with HTTP_DOWNLOAD_SEMAPHORE: # limit concurrent live HTTP downloads + # Bound the wait for a live-download permit: a saturated semaphore must not + # hang the request indefinitely. wait_for wraps ONLY the acquire; the permit + # is released in the finally below only if the acquire actually succeeded (a + # timed-out acquire never holds a permit, so it must not release one). + # + # CONSCIOUS TRADE-OFF (stage-2 review): the permit is held by the REQUEST, not + # the detached download task. So this bounds concurrent *requests* (admission + # control with a fast 503 on saturation), not strictly concurrent *downloads*: + # if a client disconnects, its request coroutine releases the permit while the + # detached download keeps running, so a burst of disconnects across DIFFERENT + # files can transiently exceed the limit of live Telegram downloads. This is + # deliberate — the download surviving a disconnect is the whole point of the + # in-flight registry, and moving the permit into the runner would forfeit the + # fast 503 (a saturated request would instead hang on the future for up to the + # waiter timeout). Each download is itself timeout-bounded, so the overrun is + # transient. If prod observation (stage 7) shows real download-count overrun + # under disconnect churn, strengthen with a separate download-side limiter. + try: + await asyncio.wait_for(HTTP_DOWNLOAD_SEMAPHORE.acquire(), timeout=30) + except asyncio.TimeoutError: + logger.warning(f"http_semaphore_timeout: {channel}/{post_id}/{file_unique_id} waited >30s for HTTP_DOWNLOAD_SEMAPHORE") + return Response(status_code=503, content="Server busy, please retry", + headers={"Retry-After": "30"}) + try: _sem_wait = _time.monotonic() - _sem_wait_start if _sem_wait > 0.5: logger.warning(f"diag_semaphore_wait: {channel}/{post_id}/{file_unique_id} waited {_sem_wait:.3f}s for HTTP_DOWNLOAD_SEMAPHORE") _dl_start = _time.monotonic() - file_path, delete_after = await download_media_file(channel_id, post_id, file_unique_id) + file_path, delete_after = await _download_deduped(channel_id, post_id, file_unique_id) _dl_elapsed = _time.monotonic() - _dl_start logger.info(f"diag_download_timing: {channel}/{post_id}/{file_unique_id} download_media_file took {_dl_elapsed:.3f}s (semaphore_wait={_sem_wait:.3f}s)") + finally: + HTTP_DOWNLOAD_SEMAPHORE.release() if not file_path: raise HTTPException(status_code=404, detail="File not found") if file_path: @@ -998,6 +1231,13 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re except HTTPException: raise + except errors.FloodWait as e: + # MUST precede `except errors.RPCError` (FloodWait subclasses RPCError): a temporary + # Telegram throttle must return a retryable 429, never a permanent 404. + retry_after = min(int(e.value) + random.randint(1, 30), 300) + logger.warning(f"media_flood_wait: {channel}/{post_id}/{file_unique_id} retry after {retry_after}s") + return Response(status_code=429, content="Telegram flood wait", + headers={"Retry-After": str(retry_after)}) except errors.RPCError as e: logger.error(f"Media request RPC error for {channel}/{post_id}/{file_unique_id}: {type(e).__name__} - {str(e)}") raise HTTPException(status_code=404, detail="File not found in Telegram") from e diff --git a/config.py b/config.py index 8ef91c6..3cc8763 100644 --- a/config.py +++ b/config.py @@ -112,6 +112,9 @@ def get_settings() -> dict[str, Any]: "show_bridge_link": os.getenv("SHOW_BRIDGE_LINK", "False").strip() in ["True", "true"], "show_post_flags": os.getenv("SHOW_POST_FLAGS", "False").strip() in ["True", "true"], "proxy": proxy, + # Hard cap (seconds) on any single live Telegram RPC held under the global RPC gate, + # so a hung MTProto call can never pin the gate (and the whole app) indefinitely. + "tg_rpc_timeout": _parse_int_env("TG_RPC_TIMEOUT", 60), "tg_watchdog_enabled": os.getenv("TG_WATCHDOG_ENABLED", "true").strip().lower() not in ["false", "0", "no", "off", "disable", "disabled"], "tg_watchdog_interval": _parse_int_env("TG_WATCHDOG_INTERVAL", 60), "tg_watchdog_timeout": _parse_int_env("TG_WATCHDOG_TIMEOUT", 10), @@ -120,4 +123,25 @@ 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). + "media_download_timeout_min": _parse_int_env("MEDIA_DOWNLOAD_TIMEOUT_MIN", 120), + "media_download_timeout_max": _parse_int_env("MEDIA_DOWNLOAD_TIMEOUT_MAX", 1800), + "media_download_min_speed": _parse_int_env("MEDIA_DOWNLOAD_MIN_SPEED", 256 * 1024), + # Size of the asyncio default ThreadPoolExecutor. SQLite/python-magic/pickle/os.walk + # all run via asyncio.to_thread; the interpreter default (min(32, cpu+4)) is only 5-6 + # on a 1-2 CPU container, which starves those under load. 32 gives ample headroom. + "io_thread_pool_size": _parse_int_env("IO_THREAD_POOL_SIZE", 32), } diff --git a/dockercompose.yml b/dockercompose.yml index a9156fa..4fa8bda 100644 --- a/dockercompose.yml +++ b/dockercompose.yml @@ -24,6 +24,11 @@ services: # TG_CHAT_CACHE_TTL_HOURS: 12 # TTL for cached channel info (title/username/id); removes GetFullChannel from the poll hot path (default: 12) # TG_RPC_CONCURRENCY: 1 # Max concurrent live Telegram RPC calls — global throttle (default: 1) # TG_RPC_MIN_INTERVAL_MS: 500 # Minimum gap between live Telegram RPC starts, ms (default: 500) + # TG_RPC_TIMEOUT: 60 # Max seconds a single live Telegram RPC may run before timing out (default: 60) + # MEDIA_DOWNLOAD_TIMEOUT_MIN: 120 # Min per-download timeout, seconds — also the timeout for regular (non-large) files (default: 120) + # MEDIA_DOWNLOAD_TIMEOUT_MAX: 1800 # Max per-download timeout, seconds — cap for the largest videos (default: 1800) + # MEDIA_DOWNLOAD_MIN_SPEED: 262144 # Assumed floor download speed, bytes/s — large-video timeout ≈ file_size / this, clamped to [MIN,MAX] (default: 262144 = 256 KB/s) + # IO_THREAD_POOL_SIZE: 32 # Size of the asyncio default threadpool for blocking I/O (SQLite/python-magic/pickle/os.walk); raise on a busy 1-2 CPU box (default: 32) PYROGRAM_BRIDGE_URL: https://pgbridge.example.com API_PORT: 80 TOKEN: ХХХ @@ -48,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/plans/2026-07-05-stability-fix-plan.md b/plans/2026-07-05-stability-fix-plan.md new file mode 100644 index 0000000..75d4540 --- /dev/null +++ b/plans/2026-07-05-stability-fix-plan.md @@ -0,0 +1,347 @@ +# План стабилизации pyrogram-bridge: статика + зависания + +Дата: 2026-07-05. База: аудит api_server.py / telegram_client.py / tg_throttle.py / tg_cache.py / file_io.py / rss_generator.py / post_parser.py. + +Проверенные факты, на которых строится план: +- `FloodWait` — подкласс `RPCError` (проверено на установленном Kurigram). +- Starlette 0.45.3 поддерживает Range в `FileResponse` из коробки (проверено по исходникам в venv). +- Дефолтный executor `asyncio.to_thread` = min(32, cpu+4) потоков; в контейнере на 1–2 CPU это 5–6. +- В venv стоит Kurigram 2.2.4, в requirements закреплён 2.2.22 — локальное окружение не соответствует прод-образу. + +## Целевые инварианты (что должно стать верным после ремонта) + +1. Любой файл в кэше с финальным именем (`{file_unique_id}` или `temp_{file_unique_id}`) — гарантированно полный. Частичным может быть только `*.part.*`. +2. Каждый Telegram RPC ограничен таймаутом. Глобальный RPC-гейт не может удерживаться дольше таймаута. +3. Временные ошибки Telegram (FloodWait) никогда не превращаются в постоянные HTTP-ответы (404). +4. Event loop не выполняет CPU-работу > ~50 мс за раз. +5. Каждая фоновая задача supervised: её смерть видна в логах на CRITICAL и/или она перезапускается. +6. Healthcheck не зависит от Telegram RPC и от обхода файловой системы. + +## Процесс + +- Ветка `fix/stability` от `main`, коммит на каждую стадию (на `main` не коммитим). +- Порядок стадий = порядок деплоя: после стадий 1 и 2 уже можно выкатываться и наблюдать. +- После каждой стадии: pytest + ручные сценарии из стадии 7 + код-ревью. +- Перед началом: пересоздать/обновить venv под requirements (Kurigram 2.2.22), прогнать 174 существующих теста как baseline. Добавить dev-зависимости: pytest-asyncio, httpx (для TestClient). + +--- + +## Стадия 1 — устранение зависаний (минимальный диф, максимальный эффект) + +### 1.1 Таймауты на все RPC под глобальным гейтом + +Файлы: tg_cache.py, config.py. + +- Добавить в config `tg_rpc_timeout` (env `TG_RPC_TIMEOUT`, default 60). +- `cached_get_chat_history` (tg_cache.py:142-144): итерацию истории собрать в корутину и обернуть в `asyncio.wait_for` **внутри** `async with tg_rpc()`: + ```python + async with tg_rpc(): + async def _collect(): + # Full paginated history; bounded by the outer wait_for + return [m async for m in client.get_chat_history(channel_id, limit=limit)] + messages = await asyncio.wait_for(_collect(), timeout=Config["tg_rpc_timeout"]) + ``` +- `cached_get_chat` (tg_cache.py:216-217): `await asyncio.wait_for(client.get_chat(channel_id), timeout=...)` внутри гейта. + +**Где не наебаться:** +- `wait_for` должен оборачивать **только RPC**, не `_sem.acquire()`. Ожидание в очереди гейта — легитимный backpressure; если холдер ограничен таймаутом, очередь всегда дренируется. Обернёшь acquire — получишь ложные таймауты при штатной очереди из 47 фидов miniflux. +- `async for` нельзя завернуть в wait_for напрямую — только через промежуточную корутину (как в эскизе). +- CancelledError не глотать: `except Exception` в вызывающих местах её и так не ловит (это BaseException в 3.11) — не «улучшать» до `except BaseException`. + +### 1.2 Таймауты и гейт на остальные живые RPC + +- `_reply_enrichment` (rss_generator.py:505): обернуть `client.get_messages` в `async with tg_rpc()` + `wait_for`. +- `PostParser.get_post` (post_parser.py:98): `wait_for(..., 30)`; заодно удалить `print(...)` (post_parser.py:95). +- `/health` (api_server.py:906): `wait_for(client.client.get_me(), 10)`. + +### 1.3 Починка фонового воркера и очереди + +Файл: api_server.py. + +- `background_download_worker` (620-637): `task_done()` вызывать только если элемент был реально получен: + ```python + while True: + item = await download_queue.get() # cancellation propagates cleanly here + channel, post_id, file_unique_id = item + try: + async with BACKGROUND_DOWNLOAD_SEMAPHORE: + await download_media_file(channel, post_id, file_unique_id) + await asyncio.sleep(2) + except errors.FloodWait as e: + logger.warning(f"bg_download_floodwait: sleeping {e.value}s") + await asyncio.sleep(min(int(e.value) + 5, 900)) + except Exception as e: + logger.error(f"Background download error ...: {e}") + finally: + download_queue.task_done() + ``` +- `download_new_files` (605-609): `await download_queue.put(...)` → `download_queue.put_nowait(...)` — иначе `except asyncio.QueueFull` остаётся мёртвым кодом, а заполненная очередь навсегда блокирует весь `cache_media_files` (включая удаление старых файлов). + +**Где не наебаться:** +- Сейчас смерть воркера убивает и уборку кэша (цепочка: воркер умер → очередь заполнилась → `await put()` завис → свипер больше не крутится). После фикса проверить тестом: воркер, у которого download постоянно бросает исключение, продолжает жить и `queue.join()` завершается. +- FloodWait в воркере надо ловить **до** Exception и спать, иначе воркер будет молотить телеграм под флудом. + +### 1.4 Supervision фоновых задач + +- В lifespan повесить `add_done_callback` на `background_task` и `worker_task`: если таск завершился не через CancelledError — лог CRITICAL с exception + перезапуск таска (обёртка `_supervised(factory)` с ограничением частоты рестартов, например не чаще раза в 60 с). + +### Тесты стадии 1 +- Гейт: мок «зависшего» RPC (asyncio.Event, который никогда не сеттится) → первый вызов отваливается по таймауту, второй проходит; пермит не утёк. +- Воркер: download бросает Exception/FloodWait → воркер жив, task_done сбалансирован. +- `_TgRpcGate`: отмена во время ожидания spacing не теряет пермит (уже реализовано — закрепить тестом). + +### DoD стадии 1 +Ни один путь кода не ждёт Telegram без таймаута; воркер не умирает молча; pytest зелёный. + +--- + +## Стадия 2 — статика: большие видео, FloodWait, семафор + +### 2.1 Единый путь скачивания с атомарным rename (главный фикс флаки) + +Файл: api_server.py, telegram_client.py. + +Сейчас большие видео (>100MB) качаются напрямую в финальное имя `temp_{fid}` (379-396): конкурентный запрос видит частичный файл и отдаёт его; при таймауте обрезок не удаляется и целый час отдаётся как «готовый». + +- Выделить хелпер: + ```python + async def _download_atomic(file_id: str, final_path: str, timeout: float) -> str: + # Download to a unique partial path, validate, atomically rename. + part_path = f"{final_path}.part.{uuid.uuid4().hex}" + try: + await client.safe_download_media(file_id, part_path, timeout=timeout) + if not os.path.exists(part_path) or os.path.getsize(part_path) == 0: + raise ZeroSizeFileError(...) + if not os.path.exists(final_path): + os.rename(part_path, final_path) # atomic on POSIX + return final_path + finally: + # Always clean up our partial file (timeout, cancel, race loser) + if os.path.exists(part_path): + try: os.remove(part_path) + except OSError: pass + ``` +- Использовать его и для обычных файлов, и для больших видео (финальное имя больших остаётся `temp_{fid}` — семантика «не кэшировать постоянно, чистить по TTL» сохраняется). +- `safe_download_media` научить принимать `timeout` параметром; для больших видео таймаут от размера: `min(1800, max(120, file_size // (256*1024)))` (≈256 KB/s минимально допустимая скорость), env-конфигурируемо. +- Проверку `if os.path.exists(temp_file_path): return` (382-384) заменить: существующий `temp_{fid}` теперь **гарантированно полный** (появляется только через rename) — отдавать можно смело. + +**Где не наебаться:** +- Единый суффикс `.part.{hex}` вместо старого `.tmp.{hex}`. Обновить regex свипера (api_server.py:536) на новый суффикс, **оставив** и старый паттерн на переходный период (на диске могут лежать старые обрезки). +- Инвариант «финальное имя = полный файл» ломается, если хоть один путь пишет мимо `.part.` — после правки grep-ом убедиться, что `download_media` больше нигде не получает финальный путь. +- В `download_new_files` (598-600) проверка «temp существует → скипаем» остаётся корректной: полный файл есть — качать не надо. Частичные `.part.` под неё не попадают — это правильно. +- Первый запрос большого видео всё ещё отвечает только после полного скачивания (минуты). Это осознанное ограничение; прогрессивный стриминг через `client.stream_media` — отдельная большая фича, в этот план не входит (пометить как future work). + +### 2.2 Дедупликация конкурентных скачиваний (in-flight registry) + +- Модульный `_inflight: dict[tuple[str, int, str], asyncio.Future]`. +- Первый запрос: создаёт future, запускает скачивание **отдельным** `asyncio.create_task` (не в контексте HTTP-запроса!), в finally сеттит result/exception и удаляет ключ. Остальные: `await`ят future. + +**Где не наебаться (классическая ловушка):** +- Если качать прямо в корутине первого HTTP-запроса, отключение его клиента отменит скачивание, future никогда не завершится, и все ожидающие повиснут. Поэтому: скачивание — в detached task; ожидающие делают `await asyncio.wait_for(fut, timeout)`. +- В finally у task обязательно и `set_exception`, и `pop` ключа — иначе после первой ошибки ключ навсегда «занят» отработавшим future. +- `set_exception` на future, который никто не await-ит, даст "exception was never retrieved" warning — допустимо, но можно гасить через `fut.exception()` в done-callback. + +### 2.3 FloodWait → 429 в /media + +- В `get_media` добавить обработчик **до** `except errors.RPCError` (api_server.py:1001): + ```python + except errors.FloodWait as e: + retry_after = min(int(e.value) + random.randint(1, 30), 300) + return Response(status_code=429, content="Telegram flood wait", + headers={"Retry-After": str(retry_after)}) + ``` + +**Где не наебаться:** порядок except-веток решает: FloodWait — подкласс RPCError, поставишь после — не сработает никогда. Закрепить тестом (мок download_media_file бросает FloodWait → ответ 429, не 404). + +### 2.4 Не удалять большое видео «из-под зрителя» + +- При каждой отдаче файла с именем `temp_*` — `await asyncio.to_thread(os.utime, path)` (touch mtime). Свипер (порог 1 час по mtime) перестанет удалять активно просматриваемые файлы. + +### 2.5 Ограничить ожидание HTTP-семафора + +- Вместо `async with HTTP_DOWNLOAD_SEMAPHORE`: явный `acquire` под `wait_for(30)`; таймаут → 503 + `Retry-After: 30`. Release — только если acquire удался (try/finally вокруг критической секции, а не вокруг acquire). + +### Тесты стадии 2 +- Конкурентные запросы большого видео с медленным фейковым download: второй запрос НЕ получает частичный файл (либо ждёт future, либо 503/504), после завершения оба получают полный. +- Таймаут download → `.part.` удалён, финального имени нет. +- FloodWait → 429 c Retry-After. +- Свипер: чистит и `.part.`, и старые `.tmp.`, не трогает свежие. + +### DoD стадии 2 +Ни при каком сценарии клиенту не может быть отдан неполный файл; флуд отдаёт 429; обрезки не живут дольше часа. + +--- + +## Стадия 3 — отдача файлов через FileResponse, чистка HTTP-слоя + +### 3.1 Заменить самодельный стриминг на `FileResponse` + +Файл: api_server.py (`prepare_file_response`, 203-335). + +Starlette 0.45.3 сам обрабатывает Range/If-Range/206/416/multipart, ставит Accept-Ranges/ETag/Last-Modified и читает файл эффективно (без нашего to_thread на каждые 64KB): + +```python +return FileResponse( + file_path, + media_type=media_type, + filename=os.path.basename(file_path), # sets Content-Disposition: inline; filename* + content_disposition_type="inline", + headers={"Cache-Control": "public, max-age=86400, immutable"}, + background=background, +) +``` + +Оставить: пре-чек существования (404), MIME-логику (magic + SQLite-кэш типа). Удалить: ручной парсинг Range, `file_chunk_generator`, ручные заголовки. + +**Где не наебаться:** +- **Сначала тесты, потом свап.** Написать httpx TestClient-тесты на текущее поведение (bytes=0-499, bytes=500-, bytes=-500, start за EOF → 416, мусорный заголовок), затем перейти на FileResponse и осознанно принять расхождения (например, Starlette на кривой заголовок может отдать 200-полный вместо 416 — это допустимо по RFC 7233). +- FileResponse проверяет файл на этапе отправки: пре-чек на 404 оставить обязательно. +- `filename*=UTF-8''...` FileResponse формирует сам — руками Content-Disposition не собирать, иначе задвоится. + +### 3.2 Убрать BaseHTTPMiddleware + +- `RequestLoggingMiddleware` (api_server.py:54-64) удалить. Логирование запросов: либо `--access-log` uvicorn на debug-уровне, либо чистый ASGI-middleware из ~10 строк (без BaseHTTPMiddleware): меньше оверхеда на каждый чанк стриминга и никаких сюрпризов с отменой/фоновыми тасками. + +### 3.3 Расширить дефолтный executor + +- В lifespan: `loop.set_default_executor(ThreadPoolExecutor(max_workers=32, thread_name_prefix="io"))`. Даже после ухода per-chunk чтений остаются SQLite/magic/pickle/os.walk — 5–6 дефолтных потоков в контейнере мало. + +### DoD стадии 3 +Range-тесты зелёные; profiling-запрос большого файла не порождает поток-на-чанк; поведение эндпоинта эквивалентно (кроме осознанных RFC-допущений). + +--- + +## Стадия 4 — гигиена event loop (генерация фидов) + +### 4.1 `raw_message` — лениво + +- `process_message(..., include_raw: bool = False)`: `str(message)` (полная сериализация каждого поста! post_parser.py:523) выполнять только для JSON-выдачи и debug-HTML. В фидах — никогда. + +### 4.2 Убрать side-effect IO из process_message (обезвредить ловушку до 4.3) + +Сейчас `_generate_html_media` → `_save_media_file_ids` (post_parser.py:676, 977-1028) делает `asyncio.get_running_loop()` + `create_task`. **Если перенести рендер в поток (4.3) без этой правки, `get_running_loop()` бросит RuntimeError, ветка упадёт в `except` с логом — и записи media id молча перестанут сохраняться, а фоновый прогрев кэша умрёт.** + +- Рефакторинг: `_save_media_file_ids` не пишет в БД, а складывает `(channel, post_id, file_unique_id, ts)` в `self._pending_media_ids` инстанса PostParser (инстанс создаётся на запрос — потокобезопасно, т.к. рендер идёт в одном потоке). +- После рендера вызывающий код (rss_generator / get_post) один раз делает `await asyncio.to_thread(upsert_media_file_ids_bulk_sync, DB_PATH, entries)` — новая функция в file_io.py с `executemany`. Это заодно даёт батчинг апсертов (часть стадии 5). +- Счётчик `_persist_pending_count` и create_task-механику удалить. + +### 4.3 Рендер фида — в поток + +- `_create_time_based_media_groups`, `_create_messages_groups`, `_trim_messages_groups`, `_render_messages_groups` — внутри нет ни одного await; превратить в обычные sync-функции и выполнять одним `await asyncio.to_thread(_render_pipeline, ...)` из `generate_channel_rss` / `generate_channel_html`. +- `copy.deepcopy(messages)` (rss_generator.py:38) уезжает в поток вместе со всем остальным. + +**Где не наебаться:** +- GIL: перенос CPU-работы в поток не делает её бесплатной — луп перестаёт стоять колом, но замедляется. Поэтому 4.1/4.4 (сокращение работы) обязательны, а не опциональны. +- Внутри потока не должно остаться ни `create_task`, ни `get_running_loop` (это ровно 4.2), ни обращений к asyncio-примитивам. +- deepcopy живых Pyrogram-объектов сегодня работает — код не менять, только переместить; отдельным тестом убедиться, что deepcopy Message из pickle-кэша не падает. Полный отказ от deepcopy (не-мутирующая группировка через map message_id → group_id) — отдельный опциональный рефакторинг, в эту стадию не тащить. + +### 4.4 Сократить sanitize-проходы + +Сейчас на каждое сообщение bleach вызывается 4+ раз (body: post_parser.py:672, media: 752, footer: 833, reactions: 920), плюс финальный проход по всему фиду. Bleach — самая дорогая CPU-часть (есть diag_sanitize_slow > 50 мс на вызов). + +- Правило: **один sanitize на каждую выходную границу**, не на каждый фрагмент: + - RSS/HTML-фиды: финальный проход уже есть (rss_generator.py:443, 643) → внутренние per-message проходы убрать. + - `/html/{channel}/{post_id}`: единственный проход добавить в `_format_html` (или в эндпоинте) — сейчас он полагается на внутренние. + - `/json/...`: html-поля внутри JSON должны оставаться чищеными → один проход по body+footer в `process_message`. + +**Где не наебаться (XSS!):** +- Перед удалением внутренних проходов построить карту «выходная точка → где чистится», и только потом резать. Ни один выходной путь не должен остаться без ровно одного прохода. +- Добавить тесты безопасности: мок-сообщение с `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 "