diff --git a/tradein-mvp/backend/app/scheduler_main.py b/tradein-mvp/backend/app/scheduler_main.py index 59254465..57e17551 100644 --- a/tradein-mvp/backend/app/scheduler_main.py +++ b/tradein-mvp/backend/app/scheduler_main.py @@ -101,15 +101,23 @@ def _should_run() -> bool: return settings.scheduler_enable -async def _await_scheduler(task: asyncio.Task[None]) -> None: +async def _await_scheduler(task: asyncio.Task[None]) -> bool: """Дождаться завершения scheduler-задачи с кооперативным SIGTERM-drain'ом. + Возвращает True, если задачу пришлось хард-кансельнуть (grace истёк), False — если + она вышла сама. Вызывающий обязан различать эти исходы в логах (#3391): строка + «drained cleanly» после hard-cancel'а — ложь, а именно она печаталась на проде + 07.09 сразу за WARNING'ом о превышении grace. + - Нет shutdown: scheduler_loop бесконечен → задача никогда не завершается, ждём её как есть (процесс просто работает). - shutdown запрошен, задача ещё бежит: bounded `wait_for(_DRAIN_TIMEOUT_S)` — кооперативная задача докоммитит текущий unit на ближайшем checkpoint'е и выйдет сама. Если превысила grace → hard-cancel + suppress CancelledError, чтобы некооперативная задача не подвесила процесс за пределами docker stop_grace_period. + Пометку `interrupted` своим in-flight прогонам kit-scheduler успевает поставить в + обработчике CancelledError (SchedulerContext.mark_inflight_interrupted): между + hard-cancel'ом и SIGKILL'ом остаётся 20 с docker-grace (120s − 100s). """ # Гонка «задача завершилась сама» против «пришёл SIGTERM»: отсчёт safety-net'а # должен стартовать от МОМЕНТА запроса drain'а, а не от старта процесса. @@ -127,7 +135,7 @@ async def _await_scheduler(task: asyncio.Task[None]) -> None: # упал (сохраняем прежнюю loud-crash семантику `await task`), а не глушит его. task.result() logger.info("scheduler_main: scheduler task exited cleanly") - return + return False logger.info( "scheduler_main: SIGTERM-drain — waiting up to %.0fs for in-flight unit to commit", @@ -136,6 +144,7 @@ async def _await_scheduler(task: asyncio.Task[None]) -> None: try: await asyncio.wait_for(task, timeout=_DRAIN_TIMEOUT_S) logger.info("scheduler_main: scheduler drained and exited cleanly") + return False except TimeoutError: logger.warning( "scheduler_main: drain exceeded %.0fs grace — hard-cancelling scheduler task", @@ -144,6 +153,7 @@ async def _await_scheduler(task: asyncio.Task[None]) -> None: task.cancel() with suppress(asyncio.CancelledError): await task + return True async def _run_kit_scheduler() -> None: @@ -224,8 +234,12 @@ async def _run() -> None: # Windows dev: signal handlers через loop не поддерживаются logger.warning("scheduler_main: loop.add_signal_handler not supported (Windows dev)") - await _await_scheduler(task) + hard_cancelled = await _await_scheduler(task) + if hard_cancelled: + # Дрейн НЕ был чистым: задача не вышла сама, её сняли. WARNING об этом уже + # напечатан в _await_scheduler — второй строкой её не «переобъявляем». + return if shutdown_requested(): logger.info("scheduler_main: scheduler drained cleanly (SIGTERM)") else: diff --git a/tradein-mvp/backend/tests/test_3391_drain_marks_inflight_app_tasks.py b/tradein-mvp/backend/tests/test_3391_drain_marks_inflight_app_tasks.py new file mode 100644 index 00000000..12cce176 --- /dev/null +++ b/tradein-mvp/backend/tests/test_3391_drain_marks_inflight_app_tasks.py @@ -0,0 +1,266 @@ +"""SIGTERM-drain снимает с 'running' прогоны app-task'ов, не доживших до финализации (#3391). + +Прод 07.09 02:36 UTC (первый настоящий SIGTERM-drain после #3363): деплой пересоздал +`tradein-scraper`, kit-scheduler напечатал «draining 2 in-flight run task(s)», через 100 с +`scheduler_main` хард-кансельнул задачу — и оба бэкфилла (cian_detail_backfill 6167, +cian_history_backfill 6173) остались в scrape_runs со статусом 'running' до boot-reap'а +следующего контейнера, где стали 'zombie' с `boot_reaped=true`. `interrupted=1` при дрейне +пишут только kit-пайплайны (#3363) и DKP-импорт: у задач, чьё тело живёт в `app`, метку +ставить некому. + +Проверяется НЕ «функция вызвалась», а значение в строке прогона: + 1. grace истёк → строка застрявшей задачи 'done' + counters.interrupted=1, прежние + счётчики (в т.ч. чекпоинт) на месте; + 2. соседняя задача, успевшая финализироваться сама, НЕ перезаписана (без interrupted); + 3. hard-cancel из scheduler_main (CancelledError прямо в `asyncio.wait` дрейна) даёт тот + же результат — это ровно прод-путь 07.09, ветка таймаута его не покрывает; + 4. `scheduler_main` после hard-cancel'а больше не печатает «drained cleanly». + +Фейковый `mark_done` counters ЗАМЕНЯЕТ, а не мержит — так ведёт себя боевая app-копия +(`app.services.scrape_runs`, #3390), которая и инжектируется в прод-контекст. Голый +`{"interrupted": 1}` на этом фейке стёр бы чекпоинт, и (1) покраснел бы. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import types +from contextlib import suppress +from typing import Any +from unittest.mock import MagicMock + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +import pytest +from scraper_kit.orchestration import scheduler as kit_sched +from scraper_kit.orchestration.scheduler import Handler, SchedulerContext, _dispatch + +import app.scheduler_main as sm +from app.core import shutdown as sd + + +class _Table: + """Мини-`scrape_runs`: id → {status, counters}.""" + + def __init__(self) -> None: + self.rows: dict[int, dict[str, Any]] = {} + self._next_id = 6167 + + def claim(self, source: str) -> int: + run_id = self._next_id + self._next_id += 6 + self.rows[run_id] = {"status": "running", "counters": {}, "source": source} + return run_id + + +class _FakeResult: + def __init__(self, *, fetchone: Any = None, scalar: Any = None) -> None: + self._fetchone = fetchone + self._scalar = scalar + + def fetchone(self) -> Any: + return self._fetchone + + def scalar(self) -> Any: + return self._scalar + + +class _FakeDb: + """Session поверх `_Table`: claim-путь `_claim_run` + SELECT статуса дрейна.""" + + def __init__(self, table: _Table, *, running_states: list[bool] | None = None) -> None: + self.table = table + self._running = list(running_states or []) + self.closed = False + + def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult: + sql = str(stmt) + if "SELECT status, counters FROM scrape_runs" in sql: + row = self.table.rows.get(int((params or {})["id"])) + if row is None: + return _FakeResult(fetchone=None) + return _FakeResult( + fetchone=types.SimpleNamespace(status=row["status"], counters=row["counters"]) + ) + if "SELECT 1 FROM scrape_runs" in sql: + return _FakeResult(fetchone=(1,) if self._running.pop(0) else None) + if "pg_try_advisory_xact_lock" in sql: + return _FakeResult(scalar=True) + return _FakeResult() + + def commit(self) -> None: + pass + + def rollback(self) -> None: + pass + + def close(self) -> None: + self.closed = True + + +class _FakeRuns: + """ctx.runs: create_run + mark_done с боевым гейтом `WHERE status = 'running'`.""" + + def __init__(self, table: _Table) -> None: + self.table = table + + def create_run(self, db: Any, *, source: str, params: dict[str, Any]) -> int: + return self.table.claim(source) + + def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None: + row = self.table.rows.get(run_id) + if row is None or row["status"] != "running": + return # боевой UPDATE ... WHERE status = 'running' — no-op + row["status"] = "done" + row["counters"] = dict(counters) # app-копия ЗАМЕНЯЕТ counters (#3390) + + +def _make_sched(source: str) -> dict[str, Any]: + return { + "id": 1, + "source": source, + "enabled": True, + "window_start_hour": 2, + "window_end_hour": 5, + "default_params": {}, + } + + +def _ctx(table: _Table) -> SchedulerContext: + return SchedulerContext( + config=MagicMock(), + matcher=MagicMock(), + enrichment=MagicMock(), + session_factory=lambda: _FakeDb(table), + runs=_FakeRuns(table), + ) + + +async def _dispatch_stuck(ctx: SchedulerContext, table: _Table, started: asyncio.Event) -> int: + """app-task, который на дрейн не реагирует (долгий await вместо checkpoint'а).""" + + async def _job(db: Any, run_id: int, params: dict[str, Any], c: Any) -> None: + started.set() + await asyncio.sleep(10) + + run_id = await _dispatch( + Handler(_job, "cian_detail_backfill"), + _FakeDb(table, running_states=[False, False]), + _make_sched("cian_detail_backfill"), + ctx, + ) + assert run_id is not None + await asyncio.wait_for(started.wait(), timeout=2.0) + # Пульс до дрейна: собранное этим прогоном и его чекпоинт. + table.rows[run_id]["counters"] = {"lots_fetched": 120, "done_buckets": ["1:0:5"]} + return run_id + + +async def _cancel_inflight(ctx: SchedulerContext) -> None: + for task in list(ctx._inflight_tasks): + task.cancel() + with suppress(asyncio.CancelledError): + await task + + +# ── 1-2. grace истёк: застрявший помечен, самофинализировавшийся не тронут ─────── + + +async def test_drain_grace_expiry_marks_stuck_run_interrupted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(kit_sched, "_CHILD_DRAIN_TIMEOUT_S", 0.05) + table = _Table() + ctx = _ctx(table) + + stuck_id = await _dispatch_stuck(ctx, table, asyncio.Event()) + + finished = asyncio.Event() + + async def _quick(db: Any, run_id: int, params: dict[str, Any], c: Any) -> None: + c.runs.mark_done(db, run_id, {"lots_fetched": 5}) + finished.set() + + quick_id = await _dispatch( + Handler(_quick, "cian_history_backfill"), + _FakeDb(table, running_states=[False, False]), + _make_sched("cian_history_backfill"), + ctx, + ) + await asyncio.wait_for(finished.wait(), timeout=2.0) + + await ctx.drain_inflight() + + assert table.rows[stuck_id]["status"] == "done" + assert table.rows[stuck_id]["counters"]["interrupted"] == 1 + # Чекпоинт и собранное пережили пометку — иначе резюм подхватывать нечего. + assert table.rows[stuck_id]["counters"]["done_buckets"] == ["1:0:5"] + assert table.rows[stuck_id]["counters"]["lots_fetched"] == 120 + # Финализировавшийся сам — не перезаписан: ни статуса, ни лишней метки. + assert table.rows[quick_id]["status"] == "done" + assert table.rows[quick_id]["counters"] == {"lots_fetched": 5} + + await _cancel_inflight(ctx) + + +# ── 3. hard-cancel из scheduler_main — прод-путь 07.09 ────────────────────────── + + +async def test_drain_hard_cancel_marks_stuck_run_interrupted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """CancelledError прилетает прямо в `asyncio.wait` дрейна (grace scheduler_main истёк).""" + monkeypatch.setattr(kit_sched, "_CHILD_DRAIN_TIMEOUT_S", 30.0) + table = _Table() + ctx = _ctx(table) + stuck_id = await _dispatch_stuck(ctx, table, asyncio.Event()) + + drain = asyncio.create_task(ctx.drain_inflight()) + await asyncio.sleep(0) # дать дрейну дойти до asyncio.wait + await asyncio.sleep(0) + drain.cancel() + with suppress(asyncio.CancelledError): + await drain + + assert table.rows[stuck_id]["status"] == "done" + assert table.rows[stuck_id]["counters"]["interrupted"] == 1 + + await _cancel_inflight(ctx) + + +# ── 4. scheduler_main: после hard-cancel'а «drained cleanly» не печатается ─────── + + +@pytest.fixture() +def _reset_shutdown() -> Any: + sd.reset_shutdown() + yield + sd.reset_shutdown() + + +async def test_hard_cancel_does_not_log_drained_cleanly( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + _reset_shutdown: Any, +) -> None: + """Прод 07.09: WARNING о hard-cancel'е и INFO «drained cleanly» стояли подряд.""" + caplog.set_level(logging.INFO) + monkeypatch.setattr(sm, "_DRAIN_TIMEOUT_S", 0.05) + started = asyncio.Event() + + async def _stub_loop() -> None: + started.set() + await asyncio.sleep(30) # некооперативная задача: shutdown не смотрит + + monkeypatch.setattr(sm, "_run_kit_scheduler", _stub_loop) + + task = asyncio.create_task(sm._run()) + await asyncio.wait_for(started.wait(), timeout=2.0) + sd.request_shutdown() + await asyncio.wait_for(task, timeout=5.0) + + assert "hard-cancelling scheduler task" in caplog.text + assert "drained cleanly" not in caplog.text + assert "drained and exited cleanly" not in caplog.text diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py index 711ddd0e..8a992ac7 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py @@ -32,7 +32,7 @@ from __future__ import annotations import asyncio import logging import random -from collections.abc import Callable, Coroutine, Mapping +from collections.abc import Callable, Coroutine, Iterable, Mapping from dataclasses import dataclass, field from datetime import UTC, datetime, time, timedelta from typing import TYPE_CHECKING, Any @@ -348,42 +348,117 @@ class SchedulerContext: runs: Any = _kit_runs # #1182 P2: strong-ref'ы detached run-задач для graceful SIGTERM-drain'а. _inflight_tasks: set[asyncio.Task[None]] = field(default_factory=set) + # #3391: task → run_id заклеймленного прогона. Без него дрейн знает, что задача + # не дожила до финализации, но не знает, КАКУЮ строку scrape_runs снимать с + # 'running' (claim-логика run_id никуда не публиковала). + _inflight_run_ids: dict[asyncio.Task[None], int] = field(default_factory=dict) - def spawn_tracked(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]: + def spawn_tracked( + self, coro: Coroutine[Any, Any, None], *, run_id: int | None = None + ) -> asyncio.Task[None]: """create_task + регистрация в _inflight_tasks для graceful-drain'а (#1182 P2). strong-ref в set'е держит задачу до завершения (RUF006) и даёт scheduler_loop'у дождаться её на SIGTERM-drain'е; done-callback ретривит exception и убирает задачу. + + `run_id` (#3391) — строка scrape_runs этой задачи; None у задач без прогона. """ task = asyncio.create_task(coro) self._inflight_tasks.add(task) + if run_id is not None: + self._inflight_run_ids[task] = run_id def _on_done(t: asyncio.Task[None]) -> None: self._inflight_tasks.discard(t) + self._inflight_run_ids.pop(t, None) if not t.cancelled(): t.exception() task.add_done_callback(_on_done) return task + def mark_inflight_interrupted(self, tasks: Iterable[asyncio.Task[None]]) -> int: + """Снять с 'running' прогоны задач, не доживших до собственной финализации (#3391). + + Прод 07.09 (первый настоящий SIGTERM-drain после #3363): два app-task бэкфилла + (cian_detail_backfill 6167, cian_history_backfill 6173) не уложились в grace, + scheduler_main их хард-кансельнул — и строки остались 'running' до boot-reap'а + следующего контейнера, где стали 'zombie' с `boot_reaped=true`. `interrupted=1` + при дрейне пишут только kit-пайплайны (#3363) и DKP-импорт: у задач, чьё тело + живёт в app и о дрейне не знает, метку ставить некому. Ставим её ЗДЕСЬ — в + единственной точке, через которую проходит любая detached run-задача. + + Синхронно (sync-сессия, ни одного await): вызывается в том числе из except + CancelledError, где лишний await мог бы не дожить до конца. + + Статус строки перечитывается перед записью: задача, успевшая финализироваться + сама (done/failed/banned), не перезаписывается. Гонку добивает сам `mark_done` + (`WHERE status = 'running'`) — но тогда счётчики уже прочитаны, и их merge был бы + холостым. counters берём из строки и дописываем `interrupted`, а не отдаём + `{"interrupted": 1}` голым: app-копия `mark_done` counters ЗАМЕНЯЕТ, а не мержит + (#3390) — голый словарь стёр бы всю бухгалтерию прогона, включая чекпоинт. + """ + run_ids = [rid for t in tasks if (rid := self._inflight_run_ids.get(t)) is not None] + if not run_ids: + return 0 + marked = 0 + db = self.session_factory() + try: + for run_id in run_ids: + try: + row = db.execute( + text("SELECT status, counters FROM scrape_runs WHERE id = :id"), + {"id": run_id}, + ).fetchone() + if row is None or row.status != "running": + continue + counters = dict(row.counters or {}) + counters["interrupted"] = 1 + self.runs.mark_done(db, run_id, counters) + marked += 1 + except Exception: + # Один непроходимый прогон не должен утащить остальные: дрейну + # осталось секунды до SIGKILL, а каждая незакрытая строка — это + # ещё один 'zombie' у следующего контейнера. + logger.exception("scheduler: drain — не удалось пометить run_id=%d", run_id) + finally: + db.close() + if marked: + logger.warning( + "scheduler: drain — %d прогон(ов) сняты с 'running' как interrupted: %s", + marked, + run_ids, + ) + return marked + async def drain_inflight(self) -> None: """Дождаться завершения detached run-задач перед teardown'ом (#1182 P2). Кооперативные дети дочекивают текущий unit + mark_done и резолвятся; - некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S и остаются на внешний - hard-cancel fallback. Не busy-spin — один asyncio.wait. + некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S либо во внешний hard-cancel + из scheduler_main — и в обоих случаях их строки помечаются `interrupted` + (#3391), иначе они доживают в 'running' до boot-reap'а. Не busy-spin — один + asyncio.wait. """ pending = [t for t in self._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) + try: + _done, still = await asyncio.wait(pending, timeout=_CHILD_DRAIN_TIMEOUT_S) + except asyncio.CancelledError: + # Hard-cancel из scheduler_main (grace 100s истёк раньше нашего дрейна — + # такт tick-сна + _CHILD_DRAIN_TIMEOUT_S могут превысить его). Пишем метку + # синхронно и пробрасываем отмену дальше. + self.mark_inflight_interrupted([t for t in pending if not t.done()]) + raise if still: logger.warning( "scheduler: %d task(s) did not drain in %.0fs — leaving for hard-cancel", len(still), _CHILD_DRAIN_TIMEOUT_S, ) + self.mark_inflight_interrupted(still) else: logger.info("scheduler: all in-flight run task(s) drained cleanly") @@ -897,7 +972,7 @@ async def _dispatch( finally: run_db.close() - ctx.spawn_tracked(_run()) + ctx.spawn_tracked(_run(), run_id=run_id) logger.info("scheduler: triggered %s run_id=%d", handler.log_name, run_id) return run_id