feat(tradein/scheduler): cooperative SIGTERM-drain — in-flight scrape unit commits before exit (phase 2) #2070

Merged

lekss361 merged 3 commits from feat/tradein-scheduler-graceful-drain into main

2026-06-28 17:20:12 +00:00

lekss361 commented

2026-06-28 17:16:38 +00:00

Owner

Зачем

Фаза 2 развязки scraper-деплоя (после #2065 фаза 0+1). На деплое Docker шлёт scraper'у SIGTERM (stop_grace_period: 120s). Раньше scheduler_main делал жёсткий task.cancel() → карточка гибла в полёте. Теперь SIGTERM кооперативный: текущая единица работы доделывается и коммитится, процесс выходит чисто в рамках grace.

Механизм

app/core/shutdown.py (новый, leaf-модуль, без circular import): request_shutdown() / shutdown_requested() поверх module-level asyncio.Event.
scheduler_main.py: SIGTERM/SIGINT → request_shutdown() (вместо cancel). _await_scheduler гонит координатор против asyncio.wait_for(task, 100s); не уложился → hard-cancel fallback (нормальный crash по-прежнему всплывает через task.result()).
scheduler.py scheduler_loop: на shutdown_requested() — не клеймит новое, выходит из tick-loop.
Кооперативные точки: задачи honor'ят флаг в тех же чек-пойнтах, где уже чекают is_cancelled — avito_detail_backfill (между карточками → break → mark_done partial; snapshot = pending-query, само-резюмится), import_rosreestr_dkp (между батчами → update_heartbeat + mark_done partial). Drain = mark_done(partial), не mark_cancelled (это user-cancel семантика, не тронута).

Pre-push fix (нашёл code-review)

_await_scheduler дренил только координатор, а работа бежит в detached create_task детях (26 trigger_* сайтов) → asyncio.run teardown хард-кансельнул бы их (тот же #1182). Фикс: registry _inflight_tasks (helper _spawn_tracked заменил все 26 create_task-сайтов) + _drain_inflight() (один asyncio.wait, бюджет 80с < 100с < 120с), вызывается из scheduler_loop на shutdown. (Выбран точный registry, а не asyncio.all_tasks() — последний включил бы родительский _run → circular wait.)

Честный scope-boundary

~10 других is_cancelled-чек-пойнтов в scrape_pipeline.py (city-sweeps/full-loads) не заведены в кооперативный drain — они коммитят инкрементально (теряют ≤1 карточку) и упираются в safety-net hard-cancel. Это приемлемая граница фазы 2; полное покрытие — follow-up.

Безвредность

Без SIGTERM shutdown_requested() == False → поведение идентично текущему. is_cancelled/reap_zombies семантика не тронута. Дренированный run = done (reaper не пометит зомби).
Идемпотентность: per-card/per-batch commit + pending-query/cursor-resnap → даже hard-cancel не теряет данные (следующий run перечитает).

Тесты

tests/core/test_shutdown.py, tests/test_scheduler_main.py (drain calls request_shutdown не cancel; wait_for-timeout → hard-cancel), tests/test_scheduler.py (load-bearing: detached кооперативный ребёнок дренится, НЕ кансельнут; non-coop → timeout; no-op; loop→drain wiring), tests/tasks/test_avito_detail_backfill.py (shutdown → break + mark_done partial). Локально: ruff-100 clean, pytest целевой 59 passed, regression sweep 254 passed.

Refs #1182. На мерж — после ОК (правка критичного планировщика).

## Зачем Фаза 2 развязки scraper-деплоя (после #2065 фаза 0+1). На деплое Docker шлёт scraper'у SIGTERM (`stop_grace_period: 120s`). Раньше `scheduler_main` делал жёсткий `task.cancel()` → карточка гибла в полёте. Теперь SIGTERM **кооперативный**: текущая единица работы доделывается и коммитится, процесс выходит чисто в рамках grace. ## Механизм - **`app/core/shutdown.py`** (новый, leaf-модуль, без circular import): `request_shutdown()` / `shutdown_requested()` поверх module-level `asyncio.Event`. - **`scheduler_main.py`**: SIGTERM/SIGINT → `request_shutdown()` (вместо cancel). `_await_scheduler` гонит координатор против `asyncio.wait_for(task, 100s)`; не уложился → hard-cancel fallback (нормальный crash по-прежнему всплывает через `task.result()`). - **`scheduler.py` `scheduler_loop`**: на `shutdown_requested()` — не клеймит новое, выходит из tick-loop. - **Кооперативные точки**: задачи honor'ят флаг в тех же чек-пойнтах, где уже чекают `is_cancelled` — `avito_detail_backfill` (между карточками → break → `mark_done` partial; snapshot = pending-query, само-резюмится), `import_rosreestr_dkp` (между батчами → `update_heartbeat` + `mark_done` partial). Drain = `mark_done`(partial), **не** `mark_cancelled` (это user-cancel семантика, не тронута). ## Pre-push fix (нашёл code-review) `_await_scheduler` дренил только координатор, а работа бежит в **detached `create_task`** детях (26 `trigger_*` сайтов) → `asyncio.run` teardown хард-кансельнул бы их (тот же #1182). Фикс: registry `_inflight_tasks` (helper `_spawn_tracked` заменил все 26 `create_task`-сайтов) + `_drain_inflight()` (один `asyncio.wait`, бюджет 80с < 100с < 120с), вызывается из `scheduler_loop` на shutdown. (Выбран точный registry, а не `asyncio.all_tasks()` — последний включил бы родительский `_run` → circular wait.) ## Честный scope-boundary ~10 других `is_cancelled`-чек-пойнтов в `scrape_pipeline.py` (city-sweeps/full-loads) **не** заведены в кооперативный drain — они коммитят инкрементально (теряют ≤1 карточку) и упираются в safety-net hard-cancel. Это приемлемая граница фазы 2; полное покрытие — follow-up. ## Безвредность - Без SIGTERM `shutdown_requested()` == False → поведение идентично текущему. `is_cancelled`/`reap_zombies` семантика не тронута. Дренированный run = `done` (reaper не пометит зомби). - Идемпотентность: per-card/per-batch commit + pending-query/cursor-resnap → даже hard-cancel не теряет данные (следующий run перечитает). ## Тесты `tests/core/test_shutdown.py`, `tests/test_scheduler_main.py` (drain calls request_shutdown не cancel; wait_for-timeout → hard-cancel), `tests/test_scheduler.py` (**load-bearing**: detached кооперативный ребёнок дренится, НЕ кансельнут; non-coop → timeout; no-op; loop→drain wiring), `tests/tasks/test_avito_detail_backfill.py` (shutdown → break + mark_done partial). Локально: ruff-100 clean, `pytest` целевой **59 passed**, regression sweep **254 passed**. Refs #1182. На мерж — после ОК (правка критичного планировщика).

lekss361 added 3 commits 2026-06-28 17:16:39 +00:00

feat(tradein/scraper): cooperative SIGTERM-drain for graceful deploy (#1182 Phase 2) 848e6d3bdd

Деплой (docker recreate tradein-scraper) шлёт SIGTERM с stop_grace_period=120s
(Phase 1). Раньше scheduler_main отвечал hard task.cancel() → бегущий scrape-unit
получал CancelledError посреди браузер-карточки → карточка гибла без COMMIT.

Теперь SIGTERM/SIGINT лишь выставляют кооперативный флаг; tick-loop и длинные
задачи опрашивают его на СВОИХ существующих between-unit checkpoint'ах,
докоммичивают текущий unit и выходят сами.

- app/core/shutdown.py: новый standalone-модуль (asyncio.Event + request_shutdown
  / shutdown_requested / wait_for_shutdown), без app-зависимостей → нет циклов.
- scheduler_main._run: SIGTERM → request_shutdown() вместо task.cancel(); ждём
  добровольного drain'а, safety-net wait_for(100s < docker grace) с fallback на
  hard-cancel для некооперирующей задачи.
- scheduler_loop: проверка флага в начале тика и после каждого dispatch — не
  reap/claim новые run'ы во время shutdown, выходим из loop'а.
- avito_detail_backfill: break на границе карточки рядом с budget-guard; snapshot
  pending-query идемпотентен → следующий run сам резюмит остаток.
- import_rosreestr_dkp: расширен is_cancelled checkpoint — drain делает mark_done
  (partial), не mark_cancelled; user-cancel семантика без изменений.

Без SIGTERM поведение идентично прежнему. reap_zombies не трогает drained-run'ы
(они mark_done, не 'running').

Tests: tests/core/test_shutdown.py, scheduler_main drain+timeout-fallback,
avito_detail_backfill partial-drain. 26 + 29 scheduler passed.

fix(tradein/scraper): surface scheduler_loop crash in _await_scheduler done-branch 6f76b565e5

В ветке `task.done()` без shutdown глушилось реальное исключение задачи. Зовём
task.result() → пробрасываем настоящий сбой scheduler_loop наружу (loud crash,
как делал прежний `await task`), вместо тихого "exited cleanly".

fix(tradein/scraper): drain detached run-children on SIGTERM, not just coordinator (#1182 Phase 2)

CI / changes (pull_request) Successful in 6s

Details

CI / backend-tests (pull_request) Has been skipped

Details

CI / frontend-tests (pull_request) Has been skipped

Details

CI / openapi-codegen-check (pull_request) Has been skipped

Details

566b2f9617

Pre-push review: _await_scheduler ждал только scheduler_loop COORDINATOR, но вся
scrape-работа крутится в detached asyncio.create_task детях (каждый trigger_* делал
`task = create_task(_run())` без join). На SIGTERM coordinator выходил из while True
и завершался → asyncio.run() teardown хард-кансельил ещё бегущих детей mid-await =
ровно #1182 failure mode. Кооперативный checkpoint спасал ребёнка лишь когда его
residual-время случайно перекрывало drain — вероятностно, не гарантированно.

Fix — coordinator теперь дренажит детей перед выходом:
- _spawn_tracked(coro): централизованный detached-spawn, кладёт задачу в module-level
  registry _inflight_tasks (strong-ref = RUF006 keep-alive) + done-callback ретривит
  exception и убирает из set'а. Заменил 26 одинаковых
  `task = create_task(_run()); task.add_done_callback(...)` сайтов.
- _drain_inflight(): один asyncio.wait по живым детям с бюджетом
  _CHILD_DRAIN_TIMEOUT_S=80s (< scheduler_main 100s < docker grace 120s). Кооперативные
  дети (avito_detail_backfill, rosreestr-executor) дочекивают карточку/батч + mark_done
  и резолвятся; некооперативные упираются в timeout и падают на внешний hard-cancel.
- scheduler_loop по выходу из tick-loop (только по SIGTERM) зовёт await _drain_inflight().

NB: raw asyncio.all_tasks()-minus-self здесь НЕЛЬЗЯ — в нашей топологии он захватывает
_run parent-task (блокирован на wait_for(coordinator)) и shutdown_waiter → циклическое
ожидание coordinator↔_run, всегда упирающееся в timeout. Точный registry это исключает.

Tests: tests/test_scheduler.py — detached cooperative child drained-not-cancelled,
non-cooperative child timeout→left for hard-cancel, no-op без детей, scheduler_loop→drain
wiring. Обновил 3 source-inspection теста под новый _spawn_tracked паттерн.