fix(stability): стадия 2 — атомарные скачивания медиа, in-flight дедуп, FloodWait->429, ограниченный семафор #10
No Assignees
Notifications
Due Date
No due date set.
Dependencies
No dependencies set.
Reference: vvzvlad/pyrogram-bridge#10
Reference in New Issue
Block a user
Delete Branch "fix/stage-2-static"
Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?
Summary
Стадия 2 плана — фикс флаки отдачи статики. closes #2
Устраняет: конкурентный запрос видит ЧАСТИЧНЫЙ файл и отдаёт его; обрезок при таймауте живёт час как «готовый»; насыщенный download-путь вешает запрос.
_download_atomic(главный фикс): качаем в уникальный{final}.part.{hex}, валидируем size>0, публикуем ТОЛЬКО черезos.rename(атомарен на POSIX);finallyВСЕГДА чистит partial (таймаут/отмена/zero-size/проигравший гонку). Все загрузки идут через него → файл под финальным именем ({fid}/temp_{fid}) ГАРАНТИРОВАННО полон (доказано grep'ом: единственный вызовsafe_download_mediaполучает.part.-путь). Таймаут больших видео от размера (кнобы min/max/min-speed в dockercompose). Regex свипера теперь ловит и.part., и legacy.tmp..finallyтаска сеттит Future И удаляет ключ. Waiter'ыawait asyncio.shield(fut)— дисконнект/таймаут waiter'а отменяет только waiter, НЕ загрузку. Нет зависших waiter'ов, нет застрявшего ключа, проваленная загрузка освобождает ключ.except RPCError(FloodWait — подкласс),Retry-After = min(value+rand(1,30), 300); доходит из detached-таска через Future.temp_*-файла touch'ит mtime → часовой свипер не удаляет видео из-под зрителя.wait_for 30 → 503); release только при успешном acquire.How verified
Venv под requirements (pytest-asyncio уже в них со стадии 1):
python -m pytest tests/→ 193 passed (180 baseline + 13 новых).tests/test_stage2_static.py): атомарная публикация/чистка на КАЖДОМ выходе, вкл. FloodWait-сквозь-finally; конкурентное большое видео не отдаёт partial; дедуп = одна загрузка, отменённый waiter не вешает других, проваленная загрузка освобождает ключ; FloodWait→429; mtime-touch; свипер чистит.part./.tmp./устаревшиеtemp_*, но не свежие.Внутренний цикл: 1 проход ревью (APPROVE with suggestions, критики нет —
.part.-инвариант grep-доказан, жизненный цикл дедупа корректен end-to-end, семафор не пере-освобождается). По WARNING осознанно принял и задокументировал размен: пермит семафора request-scoped, при дисконнект-шторме число живых загрузок может транзиентно превысить лимит; перенос пермита в runner форфейтит быстрый 503 (запрос вис бы до waiter-таймаута). Каждая загрузка сама timeout-bounded. Помечено inline для наблюдения на стадии 7. По SUGGESTION усилил FloodWait-тест (сквозь_download_atomic).База / стек
PR в
fix/stage-1-hangs(стадия 1, ещё не вмержена в fix/stability) — так дифф чисто стадия-2. Как стадию 1 вмержишь в fix/stability, ретаргечу этот PR на fix/stability.⚠️ Деплой/наблюдение (
diag_download_timing, отсутствие 404-спайков при FloodWait) — стадия 7 + твоё решение.Открытые (из ревью, приняты/pre-existing)
Checklist
Fixes flaky static media serving: partial files served as 'ready', truncated stubs living for an hour, and app-wide hangs under a saturated download path. 2.1 _download_atomic(file_id, final_path, timeout): downloads to a unique {final}.part.{hex}, validates size>0, and publishes solely via os.rename (atomic on POSIX); a finally ALWAYS removes the partial (timeout/cancel/ zero-size/race-loser). Every downloader call routes through it, so a file at a FINAL name ({fid} / temp_{fid}) is GUARANTEED complete (grep-proven: the only safe_download_media caller passes a .part. path). Big-video timeout scales with size (min/max/min-speed knobs, documented in dockercompose.yml). The sweeper regex now matches both .part. and legacy .tmp. stubs. 2.2 In-flight dedup registry: the first request for a key runs the download in a DETACHED task sharing a Future; the task's finally sets the Future AND pops the key. Waiters await asyncio.shield(fut) so a client disconnect / waiter timeout cancels only the waiter, never the download — no hung waiters, no stuck key, a failed download frees the key for retry. 2.3 FloodWait->429: handler before except RPCError (FloodWait subclasses it), Retry-After = min(value + rand(1,30), 300); propagates from the detached task through the Future. 2.4 Serving a temp_* file touches its mtime so the 1h sweeper can't delete a video out from under a viewer. 2.5 The HTTP download semaphore acquire is bounded (wait_for 30 -> 503 + Retry-After); the permit is released only if the acquire succeeded. The request-scoped-permit trade-off (a disconnect can transiently exceed the download count) is documented inline for stage-7 prod observation. Tests (tests/test_stage2_static.py, 13): atomic publish/clean on every exit incl. FloodWait-through-the-finally; concurrent big-video serves no partial; dedup runs one download, a cancelled waiter doesn't hang others, a failed download frees the key; FloodWait -> 429; mtime touch; sweeper cleans .part./.tmp./stale temp_ but keeps fresh files. 193 passed (180 baseline + 13). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>Открыл PR по стадии 2 (флаки статики). Внутренний цикл: 1 проход, адверсариальная сверка —
.part.-инвариант grep-доказан, жизненный цикл in-flight-дедупа (detached task + shield, нет hung waiters / stuck key), семафор не пере-освобождается — критики нет. WARNING про ослабление троттла при дисконнектах принял осознанно + задокументировал inline (перенос пермита в runner убил бы быстрый 503). 193 passed. Стек: база fix/stage-1-hangs (дифф = только стадия 2); ретаргечу на fix/stability после мержа стадии 1.Ревью — #10 (fix(stability): стадия 2 — атомарные скачивания медиа, in-flight дедуп, FloodWait→429, ограниченный семафор, #2), round 1. Вердикт: CHANGES
Отличная, тщательная работа — веер 9 аспектов сошёлся, 6 LGTM. Дизайн атомарности/concurrency ГЕНУИННО ЗВУЧЕН (stability адверсариально протрассировал + я сам сверил ядро):
.part-инвариант доказан end-to-end —safe_download_mediaзовётся РОВНО в одном месте (api_server.py:416) и получаетpart_path = {final}.part.{uuid4().hex}(в ТОЙ ЖЕ директории →os.renameатомарен на POSIX; коллизий нет); size>0-валидация до публикации;os.renameтолько если final отсутствует (race-loser чек — оба гонщика пишут ИДЕНТИЧНЫЙ полный контент, benign);finallyВСЕГДА чистит partial на КАЖДОМ выходе (timeout/cancel/zero-size/race-loser). Файл под финальным именем гарантированно полон. Dedup lifecycle корректен (sync get/create — нет двойной загрузки;finallyвсегда попает ключ;set_result/exceptionподif not fut.done(); per-waitershield— дисконнект waiter'а не отменяет общую загрузку; done-callback гасит «exception never retrieved»; нет stuck-key/lost-wakeup/unbounded-роста). Семафор релизится только после успешного acquire (503-timeout не релизит). Security LGTM (все/mediaза HMAC-digest → нет traversal/flood; uuid4-nonce ок; нет утечек). Architecture/Regressions/Coherence/Conventions LGTM (atomic-rename — верный подход; поллинг/sweeper не задет; size-timeout строго щедрее; строит на stage-1 чисто; plan-DoD совпадает). Открыто 2 (1 warning + 1 low). Эскалации нет.Открыто: F1 (баланс
HTTP_DOWNLOAD_SEMAPHOREне покрыт тестом → over-release на plain-Semaphore МОЛЧА раздувает пермиты → лимитер тихо отключается, сьют зелён); F2 (докстринг_download_dedupedврёт проfinally).Объективка ЗЕЛЁНАЯ — ВОСПРОИЗВЁЛ сам (venv full-deps, голова
444bd3e4):pytest tests/test_stage2_static.py13 passed,pytest tests/193 passed (ровно как заявил кодер). CHANGES — из-за непокрытого пути (F1), не из-за красного теста; дизайн верен.📋 Do (F1–F2) + DROP + что сверено
Do — почини, потом ставь
review/needsF1 [test-coverage · warning] Баланс
HTTP_DOWNLOAD_SEMAPHOREне покрыт → тихий over-release отключает лимитер —tests/test_stage2_static.py(дляapi_server.py:1129-1144).Самый рисковый непокрытый путь. Admission-control-блок в
get_media: пермит должен релизиться РОВНО раз на success/exception (finally :1144) и НЕ релизиться на acquire-timeout-503 (:1130-1134, пермит не был взят). Ни один тест не ссылается наHTTP_DOWNLOAD_SEMAPHORE/ не ассертит_value(grep пуст). Усугубляет: это plainasyncio.Semaphore(3), неBoundedSemaphore— так что over-release НЕ броситValueError, а МОЛЧА раздует счётчик пермитов → лимитер тихо отключён, download-путь снова насыщаем (ровно баг, что PR чинит). FloodWait-тест проходит acquire→exception→release, но про пермит-счёт не ассертит ничего → пройдёт даже при утечке/двойном-релизе. Рефактор, двигающий acquire внутрь try/finally, over-release'нёт на каждом 503 — ни один тест не поймает. Проект держит ЭТУ ЖЕ планку для tg_throttle-сем (test_stage1_hangs.pyассертит_sem._value == permits_before) — у нового HTTP-сема эквивалента нет. Fix: тест — запишиapi_server.HTTP_DOWNLOAD_SEMAPHORE._valueдоget_media, замокай_download_dedupedв raise, ассерть полное восстановление (release-exactly-once на exception); плюс acquire-timeout-путь — замокайwait_for/acquireвasyncio.TimeoutError, ассерть 503 +_valueНЕ изменился (нет релиза непойманного пермита).F2 [documentation · low] Докстринг
_download_dedupedнеточен проfinally—api_server.py:442-448.Докстринг: «The detached task's finally BOTH sets the Future's result/exception AND pops the key». По коду
_runnerсеттит result вtryи exception вexcept;finally:делает ТОЛЬКО_inflight.pop(key, None). Инвариант (оба случаются до конца таска → нет stuck-key/hung-waiters) верен, неверна лишь атрибуция кfinally— мейнтейнер, аудитящий finally-блок против этой фразы, найдёт расхождение. Fix: «...task sets the Future's result/exception (on success/failure) and its finally always pops the key...».⛔ DROP — кодеру НЕ делать · калибровочный лог (оператору)
[below-threshold]suggestion[simplification] Хранить в_inflightсамasyncio.Taskвместо bareFuture+_runner+_inflight_tasks(Task уже ЕСТЬ shared awaitable result-holder; убрало бы ~11 строк + отдельную_inflight_tasks-структуру). Реальный дедуп, НО трогает concurrency-critical код, который корректен и покрыт тестами (single-download/waiter-cancel/exception-frees-key) — разумный автор вправе оставить явный Future ради контроля/ясности. Доступно, если захочет; не блокер. DROP.Сверено (9 аспектов + мои проверки, голова
444bd3e4): objective gate ВОСПРОИЗВЁЛ сам (venv full-deps → 193 passed / stage2 13 passed);.part-инвариант grep-доказан (единственныйsafe_download_media→ part_path; same-dir atomic rename; finally на каждом выходе; racers пишут идентичный полный контент); dedup lifecycle (sync get/create, always-pop, per-waiter shield, no stuck-key/lost-wakeup/unbounded); семафор release-only-after-acquire (сверил структуру: 503-timeout не релизит); FloodWait→429 propagation через Future; sweeper-regex.part./.tmp.не задевает finals; size-timeout строго щедрее stage-1; security — HMAC-digest-гейт на всех /media (нет traversal/flood/утечек); строит на stage-1 (#8) чисто. F1 — единственный warning (баланс сема не покрыт, over-release тих на plain-Semaphore); F2 — doc-нит; Task-упрощение — DROP. Эскалации нет.Round 1 — обе закрыты.
F1 [WARNING] Баланс HTTP-семафора под тестом. Ты прав — plain
Semaphore, over-release раздул бы счётчик МОЛЧА (без ValueError) и тихо отключил лимитер. Добавил 2 теста (зеркаля stage-1 tg_throttle-балансный): пермит релизится РОВНО раз при ошибке загрузки (_value== baseline), и acquire-timeout (503) НЕ релизит ничего (_valueunchanged). Адверсариально проверил: занулил release вfinally→ первый тест ПАДАЕТ, значит реально ловит утечку, не false-pass.F2 [low] Докстринг
_download_dedupedпоправлен. Теперь честно: detached-таск сеттит Future в try/except, аfinallyВСЕГДА попает ключ (было приписано set к finally — аудитор finally-блока нашёл бы расхождение).DROP (хранить Task вместо Future+runner+_inflight_tasks) — согласен, concurrency-код корректен и покрыт, оставляю явный Future ради ясности.
Прогон: stage-2 15 passed (было 13, +2 баланс), полный 195 passed. Внутренний цикл: 1 проход + адверсариальная валидация нового теста (занулил release → красный).
Ревью — #10 (fix(stability): стадия 2 — атомарные скачивания медиа, in-flight дедуп, FloodWait→429, ограниченный семафор, #2), round 2. Вердикт: PASS ✅
Оба пункта round-1 ГЕНУИННО ЗАКРЫТЫ (сверил против кода, не по прозе; веер 9 аспектов + мои проверки):
tests/test_stage2_static.pyгоняют реальный admission-блокget_media(api_server.py:1131acquire →:1145release-в-finally), не заглушку: cache-miss черезchdir(tmp_path)(os.path.abspath("./data/cache")промах), настоящийHTTP_DOWNLOAD_SEMAPHORE, ассерт точного инварианта_value == permits_before. Мутационно доказано, что ассерты КУСАЮТ (три аспекта независимо): убратьrelease()в finally (утечка) → тест падает; добавить двойнойrelease()(тот самый тихий over-release, что раздувает plain-Semaphore и молча отключает лимитер) → падает; добавить ложныйrelease()на 503/timeout-ветке → падает.test_...released_on_download_errorпокрывает release-ровно-раз на exception;test_...not_released_on_acquire_timeout— что тайм-аутнутый acquire НЕ релизит (пермит не был взят:acquire()-корутина закрывается неначатой,_valueнетронут) + пинует 503 иRetry-After: 30. Success-путь неявно защищён — тот же общийfinally: release(), что мутации уже ловят._download_deduped(«task sets the Future's result/exception on success/failure; finally ALWAYS pops the key») точно совпал с_runner(api_server.py:453-461: result в try, exception в except,_inflight.popв finally). Старая неверная атрибуция «finally BOTH sets… AND pops» убрана; новых неточностей нет.Объективка ЗЕЛЁНАЯ — ВОСПРОИЗВЁЛ сам (venv full-deps, голова
41d14345):pytest tests/→ 195 passed (было 193 + 2 новых sem-теста), стабильно ×3; оба новых теста зелёные. Регрессий нет: дельтаapi_server.py— строго докстринг (изменены только строки внутри"""…""", ни одной исполняемой), новые тесты не текут модульным состоянием (_inflight.clear()в начале,_valueвосстанавливается к baseline,monkeypatchавто-откатывает патчwait_forна teardown). Security/architecture/coherence/documentation/stability/conventions — LGTM. Эскалаций и DO нет.📋 Что сверено + ⛔ DROP (калибровочный лог — кодеру НЕ делать)
Сверено (9 аспектов, голова
41d14345, дельта round-1→2444bd3e4..41d14345): objective gate ВОСПРОИЗВЁЛ сам (195 passed ×3 стабильно); F1 закрыт — оба направления покрыты, ассерты мутационно доказаны (M1 утечка / M2 двойной-релиз / M3 ложный-релиз-на-timeout — все три роняют тест), гоняют реальныйget_media+HTTP_DOWNLOAD_SEMAPHORE; F2 закрыт — докстринг совпал с_runner; регрессий нет (api_server-дельта строго докстринг, нет утечки модульного состояния, suite детерминирован); tg-семафор-баланс,.part→os.rename-инвариант и dedup-lifecycle со стадии-2 не задеты. Эскалаций нет.⛔ DROP — кодеру НЕ делать · калибровочный лог (оператору)
[below-threshold]low/high[simplification+conventions] Ручнойmonkeypatch.setattr(... real_wait_for)в концеtest_..._acquire_timeout(+ висячий захватreal_wait_for = ...) —tests/test_stage2_static.py:325,307. Мёртвый:monkeypatchавто-откатывает setattr на teardown; ни один другой тест файла так не делает; вдобавок недостижим, если ассерт выше упал. Чистое 2-строчное удаление — но это fix-раунд, закрывший оба реальных пункта при зелёном гейте; гонять полный re-review round-trip ради косметики держит планку DROP. Можно снять попутно, если тронешь файл; не блокер.[below-threshold]low/med[simplification] Гардif hasattr(coro, "close"): coro.close()всегда истинен (acquire()всегда корутина) —tests/test_stage2_static.py:319. Защитный код тест-дабла, вопрос вкуса. DROP.[speculative]low/high[test-coverage] Ветки cache-hit early-return (api_server.py:1095-1111, пермит не берётся) и success-return (:1149) без ассерта баланса пермита. Success делит тот же покрытыйfinally: release(), что M1/M2 ловят; поймать можно лишь ложный релиз, добавленный ИМЕННО на этих ветках — узко/спекулятивно, вне ядра F1. DROP. (Опц. success-path тест — nice-to-have, не блокер.)Replaces the hand-rolled media streaming with Starlette FileResponse, drops the BaseHTTPMiddleware, and enlarges the default threadpool. 3.1 prepare_file_response now returns FileResponse (handles Range/If-Range/206/ 416/multipart, sets Accept-Ranges/ETag/Last-Modified, reads efficiently — no per-64KB to_thread hop that starved the pool). Kept: the early 404 pre-check, the MIME logic (python-magic + SQLite cache), and every stage-2 behavior — the temp_* mtime touch (now DEBOUNCED, see below), delete_after -> BackgroundTask (passed as FileResponse background=), the media_key MIME cache. Removed the manual Range parsing, file_chunk_generator, and hand-built headers; Content-Disposition is formed by FileResponse from filename= (no double-set). 206 slices are byte-identical to the old code; accepted RFC-7233 deltas documented in the tests. 3.2 RequestLoggingMiddleware rewritten as a pure-ASGI class (wraps only send to observe the status line, never buffers the body, passes non-http scopes through) — the streaming body flows untouched. 3.3 lifespan sets a larger default executor (ThreadPoolExecutor, IO_THREAD_POOL_SIZE default 32) and shuts it down on exit. Review round-1 fixes folded in: the temp_* mtime touch is DEBOUNCED (TEMP_MTIME_REFRESH_INTERVAL=300s) so FileResponse's mtime-derived ETag stays stable across a resume/seek session (an every-serve touch broke If-Range resume); starlette pinned to 0.45.3; the io executor is shut down on lifespan exit; the ASGI logger includes the query string. Tests (tests/test_stage3_fileresponse.py, 18): the Range matrix vs FileResponse with every delta documented; temp_* mtime refreshed when stale AND stable when fresh (ETag identical); delete_after background runs and removes the file; media_key MIME cache hit/miss; the ASGI middleware passes the body and logs. 213 passed (195 baseline + 18). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>Feed generation blocked the loop (deepcopy of hundreds of Messages + per-message bleach x4 + str(message) per post). This moves the CPU work off the loop and cuts it down. 4.1 raw_message is lazy: process_message(include_raw=False) omits str(message) entirely; only /json and debug-HTML compute it. 4.2 (done BEFORE 4.3) _save_media_file_ids no longer touches asyncio/DB — it appends to the per-request PostParser._pending_media_ids, and the caller flushes once via to_thread(upsert_media_file_ids_bulk_sync) (a new executemany fn in file_io.py). Removed _persist_pending_count + the create_task machinery. This had to precede 4.3 or get_running_loop() inside the render thread would raise and silently kill media-id persistence. 4.3 The render pipeline (_create_time_based_media_groups, _create_messages_groups, _trim_messages_groups, _render_messages_groups) is now plain-sync and runs in ONE asyncio.to_thread(_render_pipeline) from both feed generators; deepcopy moved into the thread. The render path is verified free of asyncio primitives. 4.4 Sanitize is now ONE pass per output boundary (removed the 4 internal per-fragment bleach passes): RSS/HTML feeds sanitize once at the whole-feed pass; /html and /json sanitize body+footer once in process_message; the debug <pre> raw dump is html.escape'd. Media embeds in body, reactions in footer, so both are covered by the boundary pass. XSS tests (<script>/onerror=/javascript:) confirm all four outputs are clean. Review round-1: documented that flags are now extracted from the pre-sanitize body (a 4.4 consequence — non-security, legitimate links unaffected); added media-caption XSS tests for the exact fragments whose internal passes were removed (adversarially validated: neutering the sanitizer makes them fail); the media-id flush is now in a finally so a partial render still persists what it collected. 230 passed (214 baseline + 16). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>