diff --git a/tradein-mvp/backend/app/core/shutdown.py b/tradein-mvp/backend/app/core/shutdown.py new file mode 100644 index 00000000..294134b1 --- /dev/null +++ b/tradein-mvp/backend/app/core/shutdown.py @@ -0,0 +1,48 @@ +"""Кооперативный shutdown-флаг для scraper-планировщика (#1182 Phase 2). + +Зачем: деплой (docker recreate `tradein-scraper`) шлёт SIGTERM с +`stop_grace_period=120s`. Раньше `scheduler_main` отвечал hard `task.cancel()` → +бегущий scrape-unit получал CancelledError посреди браузер-карточки → карточка +гибла без COMMIT. Теперь SIGTERM лишь ВЫСТАВЛЯЕТ этот флаг; tick-loop планировщика +и длинные задачи опрашивают `shutdown_requested()` на СВОИХ существующих +between-unit checkpoint'ах, докоммичивают текущий unit и выходят сами. + +Модуль намеренно standalone (зависит только от `asyncio`): планировщик импортирует +задачи, а задачи должны читать флаг — общий нижний слой без app-зависимостей +исключает циклический импорт. + +Без SIGTERM `shutdown_requested()` всегда False → поведение идентично прежнему. +""" + +from __future__ import annotations + +import asyncio + +# Module-level Event: создаётся на import'е. В Python 3.10+ asyncio.Event не +# привязывается к loop'у в __init__ (loop берётся лениво в .wait()), поэтому +# set()/is_set() безопасны даже до запуска event loop'а. +_shutdown_event = asyncio.Event() + + +def request_shutdown() -> None: + """Запросить кооперативный drain. Идемпотентно — повторные вызовы безвредны.""" + _shutdown_event.set() + + +def shutdown_requested() -> bool: + """True после первого request_shutdown() (идёт кооперативный drain).""" + return _shutdown_event.is_set() + + +async def wait_for_shutdown() -> None: + """Заблокироваться до request_shutdown(). Нужно safety-net drain'у в scheduler_main.""" + await _shutdown_event.wait() + + +def reset_shutdown() -> None: + """Сбросить флаг — ТОЛЬКО для тестов. Пересоздаёт Event (а не .clear()): asyncio.Event + кеширует event loop при первом .wait(), а pytest-asyncio даёт каждому тесту НОВЫЙ loop; + свежий объект избегает 'bound to a different event loop'. В проде не вызывается + (один asyncio.run → один loop), поэтому переприсвоение глобала здесь безопасно.""" + global _shutdown_event + _shutdown_event = asyncio.Event() diff --git a/tradein-mvp/backend/app/scheduler_main.py b/tradein-mvp/backend/app/scheduler_main.py index 807a7a5c..51f356cc 100644 --- a/tradein-mvp/backend/app/scheduler_main.py +++ b/tradein-mvp/backend/app/scheduler_main.py @@ -20,6 +20,7 @@ import sys from contextlib import suppress from app.core.config import settings +from app.core.shutdown import request_shutdown, shutdown_requested, wait_for_shutdown logging.basicConfig( level=logging.INFO, @@ -28,6 +29,12 @@ logging.basicConfig( logger = logging.getLogger(__name__) +# Safety net: после запроса drain'а ждём добровольного выхода задачи не дольше +# этого порога, который ОБЯЗАН быть < docker stop_grace_period (120s). Если +# некооперативная задача не вышла за это время — hard-cancel, чтобы процесс не +# подвис до SIGKILL за пределами grace. Module-level → monkeypatch'абельно в тестах. +_DRAIN_TIMEOUT_S = 100.0 + # Мониторинг ошибок — GlitchTip (Sentry-совместимый, #396). # Только integrations без Starlette/FastAPI (у нас нет ASGI-приложения здесь). # SqlalchemyIntegration + HttpxIntegration + LoggingIntegration покрывают scraper-слой. @@ -60,29 +67,82 @@ def _should_run() -> bool: return settings.scheduler_enable +async def _await_scheduler(task: asyncio.Task[None]) -> None: + """Дождаться завершения scheduler-задачи с кооперативным SIGTERM-drain'ом. + + - Нет shutdown: scheduler_loop бесконечен → задача никогда не завершается, ждём её + как есть (процесс просто работает). + - shutdown запрошен, задача ещё бежит: bounded `wait_for(_DRAIN_TIMEOUT_S)` — + кооперативная задача докоммитит текущий unit на ближайшем checkpoint'е и выйдет + сама. Если превысила grace → hard-cancel + suppress CancelledError, чтобы + некооперативная задача не подвесила процесс за пределами docker stop_grace_period. + """ + # Гонка «задача завершилась сама» против «пришёл SIGTERM»: отсчёт safety-net'а + # должен стартовать от МОМЕНТА запроса drain'а, а не от старта процесса. + shutdown_waiter = asyncio.create_task(wait_for_shutdown()) + try: + await asyncio.wait({task, shutdown_waiter}, return_when=asyncio.FIRST_COMPLETED) + finally: + shutdown_waiter.cancel() + with suppress(asyncio.CancelledError): + await shutdown_waiter + + if task.done(): + # Задача вышла сама (drained на checkpoint'е; без сигнала сюда не попадаем). + # .result() пробрасывает реальное исключение, если scheduler_loop неожиданно + # упал (сохраняем прежнюю loud-crash семантику `await task`), а не глушит его. + task.result() + logger.info("scheduler_main: scheduler task exited cleanly") + return + + logger.info( + "scheduler_main: SIGTERM-drain — waiting up to %.0fs for in-flight unit to commit", + _DRAIN_TIMEOUT_S, + ) + try: + await asyncio.wait_for(task, timeout=_DRAIN_TIMEOUT_S) + logger.info("scheduler_main: scheduler drained and exited cleanly") + except TimeoutError: + logger.warning( + "scheduler_main: drain exceeded %.0fs grace — hard-cancelling scheduler task", + _DRAIN_TIMEOUT_S, + ) + task.cancel() + with suppress(asyncio.CancelledError): + await task + + async def _run() -> None: - """Async entrypoint: запустить scheduler_loop() с чистой отменой по SIGTERM/SIGINT.""" + """Async entrypoint: scheduler_loop() с кооперативным SIGTERM/SIGINT-drain'ом. + + SIGTERM/SIGINT → request_shutdown() (НЕ task.cancel()): бегущий scrape-unit + докоммитит текущую карточку и выйдет сам на ближайшем between-unit checkpoint'е + (scheduler_loop / avito_detail_backfill / import_rosreestr_dkp). Safety-net в + _await_scheduler гарантирует выход в пределах docker grace. + """ from app.services.scheduler import scheduler_loop task = asyncio.create_task(scheduler_loop()) loop = asyncio.get_running_loop() - def _cancel(_signum: int, _frame: object = None) -> None: - logger.info("scheduler_main: signal received — cancelling scheduler task") - task.cancel() + def _on_signal(signum: int) -> None: + logger.info("scheduler_main: signal %d received — requesting cooperative drain", signum) + request_shutdown() try: - loop.add_signal_handler(signal.SIGTERM, lambda: _cancel(signal.SIGTERM)) - loop.add_signal_handler(signal.SIGINT, lambda: _cancel(signal.SIGINT)) + loop.add_signal_handler(signal.SIGTERM, lambda: _on_signal(signal.SIGTERM)) + loop.add_signal_handler(signal.SIGINT, lambda: _on_signal(signal.SIGINT)) except NotImplementedError: # Windows dev: signal handlers через loop не поддерживаются logger.warning("scheduler_main: loop.add_signal_handler not supported (Windows dev)") - with suppress(asyncio.CancelledError): - await task + await _await_scheduler(task) - logger.info("scheduler_main: scheduler cancelled cleanly") + if shutdown_requested(): + logger.info("scheduler_main: scheduler drained cleanly (SIGTERM)") + else: + logger.info("scheduler_main: scheduler task exited") if __name__ == "__main__": diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 281369f5..92bb33ce 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -132,6 +132,7 @@ from __future__ import annotations import asyncio import logging import random +from collections.abc import Coroutine from datetime import UTC, datetime, time, timedelta from typing import Any @@ -140,6 +141,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import SessionLocal +from app.core.shutdown import shutdown_requested from app.services import scrape_runs as runs_mod from app.services.scrape_pipeline import ( run_avito_city_sweep, @@ -157,6 +159,63 @@ logger = logging.getLogger(__name__) SCHEDULER_TICK_SEC = 60 ZOMBIE_THRESHOLD_HOURS = 6 +# #1182 Phase 2: graceful SIGTERM-drain. Scrape-работа крутится в detached +# asyncio.create_task (см. _spawn_tracked) — coordinator (scheduler_loop) их +# НЕ join'ит. Без явного дренажа asyncio.run() teardown хард-кансельнул бы их +# mid-await (тот самый #1182 failure mode). Регистр strong-ref'ов даёт coordinator'у +# дождаться детей перед выходом из tick-loop'а. +_inflight_tasks: set[asyncio.Task[None]] = set() + +# Бюджет дренажа детей < scheduler_main._DRAIN_TIMEOUT_S (100s) < docker grace (120s): +# coordinator успевает дождаться кооперативных детей (avito_detail_backfill, +# rosreestr-executor дочекивают карточку/батч + mark_done и резолвятся) ВНУТРИ +# внешнего wait_for, а некооперативные упираются в этот timeout и падают на внешний +# hard-cancel fallback. +_CHILD_DRAIN_TIMEOUT_S = 80.0 + + +def _spawn_tracked(coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]: + """create_task + регистрация задачи в _inflight_tasks для graceful-drain'а (#1182 P2). + + Заменяет прежний `task = create_task(_run()); task.add_done_callback(...)`: + strong-ref в set'е держит задачу до завершения (RUF006) и даёт scheduler_loop'у + дождаться её на SIGTERM-drain'е; done-callback ретривит exception (чтобы не было + 'Task exception was never retrieved') и убирает задачу из set'а. + """ + task = asyncio.create_task(coro) + _inflight_tasks.add(task) + + def _on_done(t: asyncio.Task[None]) -> None: + _inflight_tasks.discard(t) + if not t.cancelled(): + t.exception() + + task.add_done_callback(_on_done) + return task + + +async def _drain_inflight() -> None: + """Дождаться завершения detached run-задач перед teardown'ом процесса (#1182 P2). + + Зовётся из scheduler_loop при выходе из tick-loop'а по SIGTERM-drain'у. Кооперативные + дети дочекивают текущий unit (карточку/батч), делают mark_done и резолвятся; + некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S и остаются на внешний hard-cancel + fallback (scheduler_main 100s + docker 120s). Не busy-spin — один asyncio.wait. + """ + pending = [t for t in _inflight_tasks if not t.done()] + if not pending: + return + logger.info("scheduler: draining %d in-flight run task(s) on shutdown", len(pending)) + _done, still = await asyncio.wait(pending, timeout=_CHILD_DRAIN_TIMEOUT_S) + if still: + logger.warning( + "scheduler: %d task(s) did not drain in %.0fs — leaving for hard-cancel", + len(still), + _CHILD_DRAIN_TIMEOUT_S, + ) + else: + logger.info("scheduler: all in-flight run task(s) drained cleanly") + def compute_next_run_at( window_start_hour: int, @@ -368,9 +427,7 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any] finally: run_db.close() - task = asyncio.create_task(_run()) - # Keep reference to avoid GC before task completes (RUF006) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered city_sweep run_id=%d", run_id) return run_id @@ -406,9 +463,7 @@ async def trigger_avito_newbuilding_sweep_run( finally: run_db.close() - task = asyncio.create_task(_run()) - # Keep reference to avoid GC before task completes (RUF006) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered newbuilding_sweep run_id=%d", run_id) return run_id @@ -446,9 +501,7 @@ async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any finally: run_db.close() - task = asyncio.create_task(_run()) - # Keep reference to avoid GC before task completes (RUF006) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered yandex_city_sweep run_id=%d", run_id) return run_id @@ -506,8 +559,7 @@ async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) - finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered cian_history_backfill run_id=%d", run_id) return run_id @@ -542,8 +594,7 @@ async def trigger_cian_city_sweep_run(db: Session, schedule_row: dict[str, Any]) finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered cian_city_sweep run_id=%d", run_id) return run_id @@ -578,8 +629,7 @@ async def trigger_cian_full_load_run(db: Session, schedule_row: dict[str, Any]) finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered cian_full_load run_id=%d", run_id) return run_id @@ -624,8 +674,7 @@ async def trigger_avito_full_load_run(db: Session, schedule_row: dict[str, Any]) finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered avito_full_load run_id=%d", run_id) return run_id @@ -668,8 +717,7 @@ async def trigger_avito_full_load_exhaustive_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered avito_full_load_exhaustive run_id=%d", run_id) return run_id @@ -705,8 +753,7 @@ async def trigger_domclick_city_sweep_run(db: Session, schedule_row: dict[str, A finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered domclick_city_sweep run_id=%d", run_id) return run_id @@ -809,8 +856,7 @@ async def trigger_yandex_address_backfill_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered yandex_address_backfill run_id=%d", run_id) return run_id @@ -848,8 +894,7 @@ async def trigger_geocode_missing_listings_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered geocode_missing_listings run_id=%d", run_id) return run_id @@ -873,8 +918,7 @@ async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) - finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered rosreestr_dkp_import run_id=%d", run_id) return run_id @@ -908,8 +952,7 @@ async def trigger_listing_source_snapshot_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered listing_source_snapshot run_id=%d", run_id) return run_id @@ -953,8 +996,7 @@ async def trigger_refresh_search_matview_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered refresh_search_matview run_id=%d", run_id) return run_id @@ -988,8 +1030,7 @@ async def trigger_deactivate_stale_avito_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered deactivate_stale_avito run_id=%d", run_id) return run_id @@ -1041,8 +1082,7 @@ async def trigger_deactivate_stale_run(db: Session, schedule_row: dict[str, Any] finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered deactivate_stale source=%s run_id=%d", listing_source, run_id) return run_id @@ -1077,8 +1117,7 @@ async def trigger_sber_index_pull_run(db: Session, schedule_row: dict[str, Any]) finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered sber_index_pull run_id=%d", run_id) return run_id @@ -1115,8 +1154,7 @@ async def trigger_rosreestr_quarter_poll_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered rosreestr_quarter_poll run_id=%d", run_id) return run_id @@ -1158,8 +1196,7 @@ async def trigger_newbuilding_enrich_run(db: Session, schedule_row: dict[str, An finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered newbuilding_enrich run_id=%d", run_id) return run_id @@ -1207,8 +1244,7 @@ async def trigger_yandex_newbuilding_sweep_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered yandex_newbuilding_sweep run_id=%d", run_id) return run_id @@ -1242,8 +1278,7 @@ async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered asking_to_sold_ratio_refresh run_id=%d", run_id) return run_id @@ -1319,9 +1354,28 @@ def import_rosreestr_dkp( try: while True: - if runs_mod.is_cancelled(db, run_id): - logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id) - runs_mod.mark_cancelled(db, run_id) + cancelled = runs_mod.is_cancelled(db, run_id) + if cancelled or shutdown_requested(): + if cancelled: + # User-cancel: семантика без изменений — mark_cancelled. + logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id) + runs_mod.mark_cancelled(db, run_id) + return + # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper). + # Это НЕ user-cancel → mark_done (partial), не mark_cancelled. Курсор + # уже зафиксирован update_heartbeat'ом каждый батч; следующий run + # пере-сканирует с id=0 (INSERT ... ON CONFLICT DO NOTHING идемпотентен) + # → уже вставленные строки пропускаются, остаток до-импортируется. + # mark_done выводит run из 'running' → reap_zombies его не тронет. + logger.info( + "rosreestr_dkp_import run_id=%d: SIGTERM-drain — committing partial " + "(last_id=%d, batches=%d) and exiting", + run_id, + last_id, + total_batches, + ) + runs_mod.update_heartbeat(db, run_id, counters) + runs_mod.mark_done(db, run_id, counters) return # Cursor-based pagination via foreign table gendesign_rosreestr_deals. @@ -1511,8 +1565,7 @@ async def trigger_avito_detail_backfill_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered avito_detail_backfill run_id=%d", run_id) return run_id @@ -1548,8 +1601,7 @@ async def trigger_yandex_detail_backfill_run( finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered yandex_detail_backfill run_id=%d", run_id) return run_id @@ -1590,8 +1642,7 @@ async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, A finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered cadastral_geo_match run_id=%d", run_id) return run_id @@ -1663,8 +1714,7 @@ async def trigger_house_imv_backfill_run(db: Session, schedule_row: dict[str, An finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered house_imv_backfill run_id=%d", run_id) return run_id @@ -1710,8 +1760,7 @@ async def trigger_house_dedup_merge_run(db: Session, schedule_row: dict[str, Any finally: run_db.close() - task = asyncio.create_task(_run()) - task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + _spawn_tracked(_run()) logger.info("scheduler: triggered house_dedup_merge run_id=%d", run_id) return run_id @@ -1742,6 +1791,11 @@ async def scheduler_loop() -> None: # Initial sleep 30s чтобы дать FastAPI startup завершиться await asyncio.sleep(30) while True: + # #1182 Phase 2: кооперативный SIGTERM-drain. Не reap'аем и не claim'аем + # новые run'ы во время shutdown — даём текущему dispatch'у докатиться и выходим. + if shutdown_requested(): + logger.info("scheduler: SIGTERM-drain — stop claiming new runs, exiting tick loop") + break try: db = SessionLocal() try: @@ -1803,8 +1857,27 @@ async def scheduler_loop() -> None: await trigger_house_dedup_merge_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) + + # #1182 Phase 2: после каждого dispatch'а проверяем drain — текущий + # run уже отпущен в свою asyncio-задачу (сам докоммитит/выйдет по + # своему checkpoint'у), а новые в этом тике не запускаем. + if shutdown_requested(): + logger.info( + "scheduler: SIGTERM-drain — stop dispatch after source=%s", source + ) + break finally: db.close() except Exception: logger.exception("scheduler: tick failed") + # Промежуточная проверка перед 60s-сном: выходим сразу, не ждём целый тик. + if shutdown_requested(): + logger.info("scheduler: SIGTERM-drain — exiting tick loop after dispatch") + break await asyncio.sleep(SCHEDULER_TICK_SEC) + + # Tick-loop вышел только по SIGTERM-drain'у (иначе while True бесконечен): дожидаемся + # detached run-задач (_spawn_tracked) — пусть докоммитят текущий unit и сделают + # mark_done, а не дадим asyncio.run() teardown'у хард-кансельнуть их mid-await (#1182). + await _drain_inflight() + logger.info("scheduler: tick loop exited (in-flight drain complete)") diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py index 15e79b6e..490b78bd 100644 --- a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py +++ b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py @@ -26,6 +26,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from app.core.config import settings +from app.core.shutdown import shutdown_requested from app.services import scrape_runs as runs_mod from app.services.scrape_pipeline import _CHROME_HEADERS, _avito_proxies from app.services.scrapers.avito import AvitoScraper @@ -212,6 +213,20 @@ async def run_avito_detail_backfill( ) break + # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper). + # Чисто выходим на границе карточки — каждая карточка уже закоммичена, + # mark_done с текущими счётчиками вызовется после loop'а. Snapshot — + # pending-query (detail_enriched_at IS NULL), per-item идемпотентна → + # следующий run сам до-резюмит остаток, resume_cursor не нужен. + if shutdown_requested(): + logger.info( + "avito_detail_backfill: run_id=%d SIGTERM-drain — stopping at #%d/%d", + run_id, + idx, + len(snapshot), + ) + break + # Delay before each request except the first if do_sleep: await asyncio.sleep(request_delay_sec) diff --git a/tradein-mvp/backend/tests/core/__init__.py b/tradein-mvp/backend/tests/core/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tradein-mvp/backend/tests/core/test_shutdown.py b/tradein-mvp/backend/tests/core/test_shutdown.py new file mode 100644 index 00000000..1f6570f5 --- /dev/null +++ b/tradein-mvp/backend/tests/core/test_shutdown.py @@ -0,0 +1,62 @@ +"""Тесты для app/core/shutdown.py — кооперативный SIGTERM-флаг (#1182 Phase 2).""" + +from __future__ import annotations + +import asyncio + +import pytest + +from app.core import shutdown as sd + + +@pytest.fixture(autouse=True) +def _reset() -> None: + """Event — module-global: сбрасываем до и после каждого теста для изоляции.""" + sd.reset_shutdown() + yield + sd.reset_shutdown() + + +def test_default_not_requested() -> None: + """По умолчанию shutdown не запрошен.""" + assert sd.shutdown_requested() is False + + +def test_request_sets_flag() -> None: + """request_shutdown() выставляет флаг.""" + sd.request_shutdown() + assert sd.shutdown_requested() is True + + +def test_request_is_idempotent() -> None: + """Повторный request_shutdown() безвреден — флаг остаётся True.""" + sd.request_shutdown() + sd.request_shutdown() + assert sd.shutdown_requested() is True + + +def test_reset_clears_flag() -> None: + """reset_shutdown() (test-only) снимает флаг.""" + sd.request_shutdown() + assert sd.shutdown_requested() is True + sd.reset_shutdown() + assert sd.shutdown_requested() is False + + +@pytest.mark.asyncio +async def test_wait_for_shutdown_blocks_until_requested() -> None: + """wait_for_shutdown() висит, пока не вызван request_shutdown().""" + waiter = asyncio.create_task(sd.wait_for_shutdown()) + await asyncio.sleep(0.02) + assert not waiter.done(), "не должен завершиться до request_shutdown()" + + sd.request_shutdown() + await asyncio.wait_for(waiter, timeout=1.0) + assert waiter.done() + + +@pytest.mark.asyncio +async def test_wait_for_shutdown_returns_immediately_if_already_set() -> None: + """Если флаг уже выставлен, wait_for_shutdown() возвращается сразу.""" + sd.request_shutdown() + await asyncio.wait_for(sd.wait_for_shutdown(), timeout=1.0) diff --git a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py index 5593c6cd..94bf1a4f 100644 --- a/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py @@ -12,11 +12,21 @@ sys.modules.setdefault("weasyprint", _wp_mock) import pytest # noqa: E402 +from app.core import shutdown as _sd # noqa: E402 from app.tasks.avito_detail_backfill import ( # noqa: E402 AvitoDetailBackfillResult, run_avito_detail_backfill, ) + +@pytest.fixture(autouse=True) +def _reset_shutdown() -> None: + """shutdown — module-global Event: чистим вокруг каждого теста (изоляция #1182).""" + _sd.reset_shutdown() + yield + _sd.reset_shutdown() + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -42,6 +52,7 @@ _SLEEP = "app.tasks.avito_detail_backfill.asyncio.sleep" _SESSION = "app.tasks.avito_detail_backfill.AsyncSession" _SCRAPER = "app.tasks.avito_detail_backfill.AvitoScraper" _SETTINGS = "app.tasks.avito_detail_backfill.settings" +_SHUTDOWN = "app.tasks.avito_detail_backfill.shutdown_requested" # --------------------------------------------------------------------------- # Tests @@ -139,6 +150,48 @@ async def test_backfill_blocked_abort_after_max_consecutive() -> None: assert mock_scraper.return_value._rotate_ip.call_count == 5 +@pytest.mark.asyncio +async def test_backfill_sigterm_drain_breaks_and_marks_done_partial() -> None: + """#1182 Phase 2: shutdown_requested() True → loop выходит на границе карточки, + mark_done вызывается с ЧАСТИЧНЫМИ счётчиками (не mark_failed, не mark_cancelled). + + side_effect [False, True]: 1-я карточка обрабатывается (attempted=1, enriched=1), + перед 2-й приходит SIGTERM-drain → break. Snapshot — pending-query, остаток + до-резюмит следующий run (resume_cursor не нужен). + """ + snapshot = _make_snapshot(3) + db = _mock_db(snapshot) + runs = MagicMock() + mock_enrichment = MagicMock() + mock_fetch = AsyncMock(return_value=mock_enrichment) + mock_save = MagicMock(return_value=True) + fake_settings = MagicMock(scraper_fetch_mode="cffi") + with ( + patch(_SETTINGS, fake_settings), + patch(_SESSION, return_value=AsyncMock()), + patch(_SCRAPER), + patch(_RUNS, runs), + patch(_FETCH, mock_fetch), + patch(_SAVE, mock_save), + patch(_SLEEP, new_callable=AsyncMock), + patch(_SHUTDOWN, side_effect=[False, True]), + ): + result = await run_avito_detail_backfill( + db, run_id=13, params={"batch_size": 10, "budget_sec": 3600} + ) + + # Обработана только 1-я карточка, на 2-й — drain-break. + assert result.attempted == 1 + assert result.enriched == 1 + assert mock_fetch.call_count == 1 + runs.mark_done.assert_called_once() + runs.mark_failed.assert_not_called() + runs.mark_cancelled.assert_not_called() + # mark_done получил ЧАСТИЧНЫЕ счётчики (attempted=1, а не весь snapshot=3). + done_counters = runs.mark_done.call_args.args[2] + assert done_counters["attempted"] == 1 + + @pytest.mark.asyncio async def test_backfill_budget_guard_stops_loop() -> None: """Budget expired before first listing -> fetch_detail not called.""" diff --git a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py b/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py index 6f8f1bcc..44852cce 100644 --- a/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py @@ -4,6 +4,7 @@ No real network, no real Postgres. The Cian fetch is mocked; the DB session is a lightweight in-memory fake that models just enough of the three target tables to assert (a) rows land and (b) an idempotent re-run does not duplicate them. """ + from __future__ import annotations import os @@ -636,7 +637,9 @@ def test_trigger_claims_run_and_delegates_to_task() -> None: trig_src = inspect.getsource(scheduler.trigger_newbuilding_enrich_run) assert "_claim_run(db, schedule_row)" in trig_src assert "run_newbuilding_enrich" in trig_src - assert "asyncio.create_task(" in trig_src + # #1182 P2: detached spawn централизован в _spawn_tracked (внутри — asyncio.create_task + # + registry strong-ref для graceful SIGTERM-drain). + assert "_spawn_tracked(_run())" in trig_src @pytest.mark.asyncio diff --git a/tradein-mvp/backend/tests/test_house_dedup_merge.py b/tradein-mvp/backend/tests/test_house_dedup_merge.py index 24455db6..89a2e1f6 100644 --- a/tradein-mvp/backend/tests/test_house_dedup_merge.py +++ b/tradein-mvp/backend/tests/test_house_dedup_merge.py @@ -396,9 +396,10 @@ def test_trigger_claims_run_and_runs_in_executor() -> None: # sync DB-only task → run_in_executor (mirror cadastral_geo_match, not a bare await). assert "run_in_executor" in src assert "run_house_dedup_merge" in src - # fresh session inside the spawned task + RUF006 keep-alive callback. + # fresh session inside the spawned task + #1182 P2 detached spawn via _spawn_tracked + # (registry strong-ref = RUF006 keep-alive + graceful SIGTERM-drain). assert "run_db = SessionLocal()" in src - assert "task.add_done_callback(" in src + assert "_spawn_tracked(_run())" in src assert "run_db.close()" in src @@ -535,9 +536,7 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: ).all() assert [r.id for r in survivors] == [900001] # loser's listing re-pointed to keeper. - repointed = db.execute( - _t("SELECT house_id_fk FROM listings WHERE id = 910002") - ).scalar() + repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910002")).scalar() assert repointed == 900001 # price_dynamics deduped — exactly one row for the keeper at that 6-col key (no dup), # and it is the KEEPER's own row (price_per_sqm=100000), not the loser's (999999): @@ -580,10 +579,7 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None: _t("DELETE FROM house_sources WHERE ext_id IN ('SRC-KEEP','SRC-LOSE','EXT-KEEP')") ) db.execute( - _t( - "DELETE FROM house_address_aliases " - "WHERE normalized_address = 'тестдом 1772, 1'" - ) + _t("DELETE FROM house_address_aliases WHERE normalized_address = 'тестдом 1772, 1'") ) db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)")) db.commit() diff --git a/tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py b/tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py index 4606ac41..8bd52594 100644 --- a/tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py +++ b/tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py @@ -54,9 +54,10 @@ def test_trigger_mirrors_backfill_sibling_pattern() -> None: assert "backfill_house_imv" in _TRIGGER_SRC # async service → bare await (NOT run_in_executor, unlike the sync DB-only triggers). assert "run_in_executor" not in _TRIGGER_SRC - # RUF006: keep a reference + done-callback so the task is not GC'd mid-flight. - assert "task = asyncio.create_task(_run())" in _TRIGGER_SRC - assert "task.add_done_callback(" in _TRIGGER_SRC + # #1182 P2: detached spawn через _spawn_tracked — strong-ref в registry (RUF006 + + # graceful SIGTERM-drain), done-callback ретривит exception. Заменил прежний + # `task = asyncio.create_task(_run()); task.add_done_callback(...)`. + assert "_spawn_tracked(_run())" in _TRIGGER_SRC assert "run_db.close()" in _TRIGGER_SRC diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py index 5e70b4e6..6046f815 100644 --- a/tradein-mvp/backend/tests/test_scheduler.py +++ b/tradein-mvp/backend/tests/test_scheduler.py @@ -2,7 +2,7 @@ import asyncio import os -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") @@ -10,9 +10,22 @@ from datetime import UTC, datetime import pytest +import app.services.scheduler as sched +from app.core import shutdown as sd from app.services.scheduler import ZOMBIE_THRESHOLD_HOURS, compute_next_run_at +@pytest.fixture(autouse=True) +def _reset_drain_state() -> None: + """#1182 P2: shutdown-флаг и registry детей — module-global; чистим вокруг каждого + теста (изоляция от detached-задач, что триггеры спавнят в _inflight_tasks).""" + sd.reset_shutdown() + sched._inflight_tasks.clear() + yield + sched._inflight_tasks.clear() + sd.reset_shutdown() + + def test_compute_next_run_in_normal_window() -> None: now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) next_at = compute_next_run_at(2, 5, now=now) @@ -730,3 +743,83 @@ async def test_scheduler_dispatch_routes_avito_full_load_exhaustive( await _sched.trigger_avito_full_load_exhaustive_run(db, sch) assert "avito_full_load_exhaustive" in triggered + + +# --------------------------------------------------------------------------- +# #1182 Phase 2 — graceful SIGTERM-drain детей (detached run-задач) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_drain_inflight_awaits_detached_cooperative_child() -> None: + """Гвоздь Phase 2: _drain_inflight ДОЖИДАЕТСЯ detached-ребёнка (как scrape-run, + отвязанный через _spawn_tracked) до его clean-exit, а НЕ хард-кансельит его. + + Регрессия gap'а: coordinator раньше ждал только себя → asyncio.run teardown + хард-кансельил живых детей mid-await (#1182 failure mode). + """ + clean_exit = asyncio.Event() # == mark_done-эквивалент ребёнка + + async def _child() -> None: + # Кооперативный ребёнок: крутится, пока не запрошен shutdown, потом выходит чисто. + while not sd.shutdown_requested(): + await asyncio.sleep(0.01) + clean_exit.set() + + child = sched._spawn_tracked(_child()) # detached, попадает в _inflight_tasks + await asyncio.sleep(0.02) + assert not child.done(), "ребёнок ещё бежит (shutdown не запрошен)" + + sd.request_shutdown() + await asyncio.wait_for(sched._drain_inflight(), timeout=2.0) + + assert clean_exit.is_set(), "drain дождался clean-exit ребёнка" + assert child.done() + assert not child.cancelled(), "ребёнок вышел сам, НЕ хард-cancel" + + +@pytest.mark.asyncio +async def test_drain_inflight_times_out_on_noncooperative_child( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Некооперативный ребёнок (игнорит shutdown) упирается в _CHILD_DRAIN_TIMEOUT_S и + остаётся бежать — отдаётся внешнему hard-cancel fallback (scheduler_main/docker).""" + monkeypatch.setattr(sched, "_CHILD_DRAIN_TIMEOUT_S", 0.05) + + async def _stuck() -> None: + await asyncio.Event().wait() # никогда не проверяет shutdown_requested() + + child = sched._spawn_tracked(_stuck()) + sd.request_shutdown() + + await asyncio.wait_for(sched._drain_inflight(), timeout=2.0) + + assert not child.done(), "drain НЕ повесил процесс — оставил ребёнка под hard-cancel" + + # cleanup + child.cancel() + with pytest.raises(asyncio.CancelledError): + await child + + +@pytest.mark.asyncio +async def test_drain_inflight_noop_when_no_children() -> None: + """Без in-flight детей _drain_inflight() — мгновенный no-op (без сна/ожидания).""" + assert not sched._inflight_tasks + await asyncio.wait_for(sched._drain_inflight(), timeout=1.0) + + +@pytest.mark.asyncio +async def test_scheduler_loop_calls_drain_on_shutdown(monkeypatch: pytest.MonkeyPatch) -> None: + """scheduler_loop при выходе по SIGTERM-drain'у вызывает _drain_inflight (проводка + coordinator → дренаж детей). Initial sleep(30) и tick-sleep занулены, чтобы тест + был детерминированным и быстрым.""" + drain_spy = AsyncMock() + monkeypatch.setattr(sched, "_drain_inflight", drain_spy) + monkeypatch.setattr(sched.asyncio, "sleep", AsyncMock()) # без реальных 30s/60s + + sd.request_shutdown() # первый же top-of-tick check → break + + await asyncio.wait_for(sched.scheduler_loop(), timeout=2.0) + + drain_spy.assert_awaited_once() diff --git a/tradein-mvp/backend/tests/test_scheduler_main.py b/tradein-mvp/backend/tests/test_scheduler_main.py index 439926f3..a8c49700 100644 --- a/tradein-mvp/backend/tests/test_scheduler_main.py +++ b/tradein-mvp/backend/tests/test_scheduler_main.py @@ -3,6 +3,8 @@ (a) модуль импортируется без реального DB/сети. (b) _should_run() возвращает False при SCHEDULER_ENABLE=false. (c) _run() отменяется чисто без unhandled exceptions. +(d) #1182 Phase 2: SIGTERM → request_shutdown() (не task.cancel); задача + кооперативно drain'ится; safety-net wait_for-timeout добивает зависшую. """ from __future__ import annotations @@ -15,6 +17,15 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost: import pytest import app.scheduler_main as sm +from app.core import shutdown as sd + + +@pytest.fixture(autouse=True) +def _reset_shutdown() -> None: + """shutdown — module-global Event: сбрасываем вокруг каждого теста для изоляции.""" + sd.reset_shutdown() + yield + sd.reset_shutdown() # (a) Импорт модуля @@ -70,3 +81,74 @@ async def test_run_cancels_cleanly(monkeypatch: pytest.MonkeyPatch) -> None: # Тест прошёл если не было unhandled exception assert task.done() + + +# (d) #1182 Phase 2 — кооперативный SIGTERM-drain +@pytest.mark.asyncio +async def test_await_scheduler_drains_on_shutdown() -> None: + """_await_scheduler: кооперативная задача видит флаг и выходит сама — без cancel.""" + drained = False + + async def _coop_loop() -> None: + nonlocal drained + while not sd.shutdown_requested(): + await asyncio.sleep(0.01) + drained = True # докоммитил unit и вышел + + task = asyncio.create_task(_coop_loop()) + runner = asyncio.create_task(sm._await_scheduler(task)) + + await asyncio.sleep(0.03) + assert not runner.done(), "без shutdown _await_scheduler не завершается" + + sd.request_shutdown() + await asyncio.wait_for(runner, timeout=2.0) + + assert drained is True + assert task.done() + assert not task.cancelled(), "кооперативный выход — НЕ hard-cancel" + + +@pytest.mark.asyncio +async def test_await_scheduler_timeout_hard_cancels(monkeypatch: pytest.MonkeyPatch) -> None: + """_await_scheduler: некооперативная задача (игнорит флаг) добивается cancel'ом + по истечении safety-net grace (_DRAIN_TIMEOUT_S).""" + monkeypatch.setattr(sm, "_DRAIN_TIMEOUT_S", 0.05) + + async def _stuck_loop() -> None: + # Игнорирует shutdown_requested() — висит вечно. + await asyncio.Event().wait() + + task = asyncio.create_task(_stuck_loop()) + sd.request_shutdown() # drain уже запрошен + + await asyncio.wait_for(sm._await_scheduler(task), timeout=2.0) + + assert task.done() + assert task.cancelled(), "превысила grace → hard-cancel" + + +@pytest.mark.asyncio +async def test_run_uses_request_shutdown_not_cancel(monkeypatch: pytest.MonkeyPatch) -> None: + """_run(): когда выставлен shutdown, scheduler_loop сам drain'ится и _run выходит, + БЕЗ обращения к task.cancel() (кооперативный путь, не hard-cancel).""" + started = asyncio.Event() + + async def _coop_loop() -> None: + started.set() + while not sd.shutdown_requested(): + await asyncio.sleep(0.01) + + import app.services.scheduler as _sched_mod + + monkeypatch.setattr(_sched_mod, "scheduler_loop", _coop_loop) + + run_task = asyncio.create_task(sm._run()) + await asyncio.wait_for(started.wait(), timeout=2.0) + + # Имитируем то, что делает SIGTERM-хендлер (на Windows add_signal_handler нет). + sm.request_shutdown() + + await asyncio.wait_for(run_task, timeout=2.0) + assert run_task.done() + assert run_task.exception() is None