From dd3c126e4a8ba037c7e241e14d24d4529b2e86dc Mon Sep 17 00:00:00 2001 From: lekss361 Date: Fri, 15 May 2026 00:04:54 +0300 Subject: [PATCH] refactor(workers): split celery_app.py god-object into 3 modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per audit batch #127 P3 god objects (issue #134). Split layout: - celery_app.py: 270 → 35 lines (config only) - beat_schedule.py: 207 lines NEW (DB-driven + fallback + hardcoded) - lifecycle.py: 168 lines NEW (worker_process_init + worker_ready handlers) No business logic changes — move-only structural refactor. Public API (celery_app.conf.beat_schedule) unchanged. Verify: - beat keys: kn-region-66, objective_sync, refresh-ekb-districts-medians, nspd-geo-zombie-cleanup, poi-sync-weekly, noise-sync-weekly - ruff + ruff-format clean - AST parse OK Vault: Module_Beat_Schedule + Module_Worker_Lifecycle NEW; Module_Celery_App updated. Closes #134 Refs: #127 --- backend/app/workers/beat_schedule.py | 241 +++++++++++++++ backend/app/workers/celery_app.py | 430 +-------------------------- backend/app/workers/lifecycle.py | 216 ++++++++++++++ 3 files changed, 465 insertions(+), 422 deletions(-) create mode 100644 backend/app/workers/beat_schedule.py create mode 100644 backend/app/workers/lifecycle.py diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py new file mode 100644 index 00000000..e7ce5146 --- /dev/null +++ b/backend/app/workers/beat_schedule.py @@ -0,0 +1,241 @@ +"""Beat schedule builder for Celery. + +Priority order: +1. DB-driven — читает таблицу job_settings при старте beat-контейнера. +2. Fallback — env-based расписание (SCRAPE_KN_CRON, OBJECTIVE_SYNC_CRON), + используется если БД недоступна или таблица пустая. +3. Hardcoded entries — добавляются поверх DB-расписания (не в job_settings): + - refresh-ekb-districts-medians (аналитика, ежемесячно 5-го в 04:00) + - nspd-geo-zombie-cleanup (каждую минуту, замена нестабильного worker_ready) + - poi-sync-weekly (OSM POI, понедельник 03:00) + - noise-sync-weekly (OSM шум, понедельник 03:30) + +Публичный API: build_beat_schedule() → dict. +""" + +import logging + +from celery.schedules import crontab + +from app.core.config import settings + +logger = logging.getLogger(__name__) + +# Словарь job_type → celery task name (полное dotted name). +# Используется при построении beat_schedule из job_settings. +TASK_MAP: dict[str, str] = { + "scrape_kn": "tasks.scrape_kn.scrape_kn_region", + "objective_sync": "tasks.scrape_objective.sync_all_groups", + # nspd_geo и objective_etl не имеют cron — запускаются вручную. + # При необходимости можно добавить сюда. +} + + +def _parse_cron(spec: str) -> crontab: + """Parse 'M H DoM Mon DoW' crontab string into Celery crontab. Empty fields default to '*'.""" + parts = spec.strip().split() + if len(parts) != 5: + raise ValueError(f"crontab spec must have 5 fields, got: {spec!r}") + minute, hour, dom, month, dow = parts + return crontab( + minute=minute, + hour=hour, + day_of_month=dom, + month_of_year=month, + day_of_week=dow, + ) + + +def _default_regions() -> list[int]: + return [int(x.strip()) for x in settings.scrape_kn_default_regions.split(",") if x.strip()] + + +def _build_beat_schedule_from_db() -> dict: + """Строит beat_schedule из job_settings БД. + + Возвращает пустой словарь при любой ошибке — caller добавит fallback. + """ + try: + from app.core.db import SessionLocal + from app.services.job_settings import get_all_safe + + rows = get_all_safe(SessionLocal) + except Exception as e: + logger.warning("build_beat_schedule: не удалось прочитать job_settings: %s", e) + return {} + + schedule: dict = {} + for row in rows: + jt = row["job_type"] + if not row.get("enabled"): + continue + cron_str = row.get("cron_schedule") + if not cron_str: + # NULL cron = только ручной запуск, не добавляем в beat + continue + if jt not in TASK_MAP: + logger.warning( + "build_beat_schedule: job_type '%s' не имеет TASK_MAP-записи — пропускаем", jt + ) + continue + try: + cron = _parse_cron(cron_str) + except Exception as e: + logger.warning( + "build_beat_schedule: невалидный cron '%s' для '%s': %s", cron_str, jt, e + ) + continue + + queue = row.get("queue_name", "celery") or "celery" + + if jt == "scrape_kn": + # scrape_kn запускается отдельной задачей на каждый регион + extra = row.get("extra_config") or {} + regions: list[int] = extra.get("default_regions") or _default_regions() + for rc in regions: + schedule[f"kn-region-{rc}"] = { + "task": TASK_MAP[jt], + "schedule": cron, + "args": [rc, None], + "options": {"queue": queue}, + } + logger.info( + "build_beat_schedule: scrape_kn cron=%r regions=%s queue=%s", + cron_str, + regions, + queue, + ) + else: + kwargs: dict = {} + if jt == "objective_sync": + kwargs = {"triggered_by": "beat"} + schedule[jt] = { + "task": TASK_MAP[jt], + "schedule": cron, + "kwargs": kwargs, + "options": {"queue": queue}, + } + logger.info("build_beat_schedule: %s cron=%r queue=%s", jt, cron_str, queue) + + return schedule + + +def _build_beat_schedule_fallback() -> dict: + """Fallback-расписание из env vars — используется если БД недоступна на старте.""" + logger.warning("beat_schedule: строим fallback из env vars") + schedule: dict = {} + + # scrape_kn + try: + for rc in _default_regions(): + schedule[f"kn-region-{rc}"] = { + "task": "tasks.scrape_kn.scrape_kn_region", + "schedule": _parse_cron(settings.scrape_kn_cron), + "args": [rc, None], + } + except Exception as e: + logger.warning("beat_schedule fallback: scrape_kn failed: %s", e) + + # objective_sync + try: + from app.core.db import SessionLocal + from app.services.objective_sync_config import get_cron_schedule_safe + + cron_str = get_cron_schedule_safe(SessionLocal) + except Exception: + cron_str = settings.objective_sync_cron + + try: + schedule["objective-sync"] = { + "task": "tasks.scrape_objective.sync_all_groups", + "schedule": _parse_cron(cron_str), + "kwargs": {"triggered_by": "beat"}, + } + except Exception as e: + logger.warning("beat_schedule fallback: objective_sync failed: %s", e) + + # refresh ekb-districts medians — ежемесячно 5-го числа в 04:00 МСК + try: + schedule["refresh-ekb-districts-medians"] = { + "task": "tasks.refresh_analytics.refresh_ekb_districts_medians", + "schedule": _parse_cron("0 4 5 * *"), + "kwargs": {"window_months": 24, "min_deals": 50}, + } + except Exception as e: + logger.warning("beat_schedule fallback: refresh_analytics failed: %s", e) + + return schedule + + +def build_beat_schedule() -> dict: + """Строит beat_schedule: сначала из DB, при неудаче — fallback на env. + + Всегда добавляет refresh-ekb-districts-medians (нет в job_settings). + """ + schedule: dict = {} + try: + schedule = _build_beat_schedule_from_db() + except Exception as e: + logger.error("build_beat_schedule: неожиданная ошибка: %s — fallback", e) + + if not schedule: + # БД пустая или недоступна — полный fallback + return _build_beat_schedule_fallback() + + # Всегда добавляем refresh_analytics (этот job нет в job_settings, + # он технический и не требует конфигурации через UI). + if "refresh-ekb-districts-medians" not in schedule: + try: + schedule["refresh-ekb-districts-medians"] = { + "task": "tasks.refresh_analytics.refresh_ekb_districts_medians", + "schedule": _parse_cron("0 4 5 * *"), + "kwargs": {"window_months": 24, "min_deals": 50}, + "options": {"queue": "celery"}, + } + except Exception as e: + logger.warning("beat_schedule: refresh_analytics failed: %s", e) + + # Zombie cleanup: каждую минуту проверяет nspd_geo_jobs с heartbeat > 2 мин + # и re-enqueue'ит их. Замена worker_ready signal handler'а — он не fires + # стабильно в нашем setup, beat-task надёжнее. + schedule["nspd-geo-zombie-cleanup"] = { + "task": "tasks.nspd_geo.cleanup_zombies", + "schedule": _parse_cron("* * * * *"), + "options": {"queue": "celery"}, + } + + # OSM POI sync — еженедельно в понедельник в 03:00 МСК + schedule["poi-sync-weekly"] = { + "task": "tasks.poi_sync.sync_osm_poi_ekb", + "schedule": _parse_cron("0 3 * * mon"), + "options": {"queue": "celery"}, + } + + # OSM noise sources sync — еженедельно в понедельник в 03:30 МСК + # (через 30 мин после poi-sync, чтобы не нагружать Overpass одновременно) + schedule["noise-sync-weekly"] = { + "task": "tasks.noise_sync.sync_osm_noise_sources_ekb", + "schedule": _parse_cron("30 3 * * mon"), + "options": {"queue": "celery"}, + } + + # NSPD quarter dump refresh — DISABLED 2026-05-14 per Bug_NSPD_WMS_NotBulk + # post-mortem (vault: fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md). + # + # Action item #1: disable beat schedule до Sprint 2 fix (grid sampling + # rewrite). harvest_quarter в текущей реализации пишет почти пустые dumps + # из-за single-pixel WMS GetFeatureInfo bug. Запуск Mon 04:00 МСК потратит + # rate-limit budget и заполнит nspd_quarter_dumps мусором. + # + # Task code остаётся в tasks/nspd_sync.py — re-enable после Sprint 2 grid + # sampling rewrite (см. Bug_NSPD_WMS_NotBulk_2026_May14 → Sprint 2 fix-strategy). + # До тех пор harvest_quarter можно вызывать вручную через admin endpoint. + # + # schedule["nspd-harvest-stale-quarters"] = { + # "task": "tasks.nspd_sync.harvest_stale_quarters", + # "schedule": _parse_cron("0 4 * * mon"), + # "kwargs": {"region_code": 66, "max_age_days": 90, "batch_size": 50}, + # "options": {"queue": "celery"}, + # } + + return schedule diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 7adb9afa..ecaa911a 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -1,48 +1,17 @@ -"""Celery app + beat schedule. +"""Celery app — single source of truth для Celery configuration. -Beat schedule строится из таблицы job_settings при старте beat-контейнера. -При недоступности БД — fallback на env-based расписание (старое поведение). +Beat schedule build → app/workers/beat_schedule.py. +Worker lifecycle hooks (process_init, worker_ready) → app/workers/lifecycle.py. """ import logging from celery import Celery -from celery.schedules import crontab -from celery.signals import worker_process_init, worker_ready from app.core.config import settings logger = logging.getLogger(__name__) -# Словарь job_type → celery task name (полное dotted name). -# Используется при построении beat_schedule из job_settings. -TASK_MAP: dict[str, str] = { - "scrape_kn": "tasks.scrape_kn.scrape_kn_region", - "objective_sync": "tasks.scrape_objective.sync_all_groups", - # nspd_geo и objective_etl не имеют cron — запускаются вручную. - # При необходимости можно добавить сюда. -} - - -def _parse_cron(spec: str) -> crontab: - """Parse 'M H DoM Mon DoW' crontab string into Celery crontab. Empty fields default to '*'.""" - parts = spec.strip().split() - if len(parts) != 5: - raise ValueError(f"crontab spec must have 5 fields, got: {spec!r}") - minute, hour, dom, month, dow = parts - return crontab( - minute=minute, - hour=hour, - day_of_month=dom, - month_of_year=month, - day_of_week=dow, - ) - - -def _default_regions() -> list[int]: - return [int(x.strip()) for x in settings.scrape_kn_default_regions.split(",") if x.strip()] - - celery_app = Celery( "gendesign", broker=settings.redis_url, @@ -61,393 +30,10 @@ celery_app = Celery( ) celery_app.conf.timezone = "Europe/Moscow" +# Apply beat schedule (DB → fallback → hardcoded entries) +from app.workers.beat_schedule import build_beat_schedule # noqa: E402 -def _build_beat_schedule_from_db() -> dict: - """Строит beat_schedule из job_settings БД. +celery_app.conf.beat_schedule = build_beat_schedule() - Возвращает пустой словарь при любой ошибке — caller добавит fallback. - """ - try: - from app.core.db import SessionLocal - from app.services.job_settings import get_all_safe - - rows = get_all_safe(SessionLocal) - except Exception as e: - logger.warning("build_beat_schedule: не удалось прочитать job_settings: %s", e) - return {} - - schedule: dict = {} - for row in rows: - jt = row["job_type"] - if not row.get("enabled"): - continue - cron_str = row.get("cron_schedule") - if not cron_str: - # NULL cron = только ручной запуск, не добавляем в beat - continue - if jt not in TASK_MAP: - logger.warning( - "build_beat_schedule: job_type '%s' не имеет TASK_MAP-записи — пропускаем", jt - ) - continue - try: - cron = _parse_cron(cron_str) - except Exception as e: - logger.warning( - "build_beat_schedule: невалидный cron '%s' для '%s': %s", cron_str, jt, e - ) - continue - - queue = row.get("queue_name", "celery") or "celery" - - if jt == "scrape_kn": - # scrape_kn запускается отдельной задачей на каждый регион - extra = row.get("extra_config") or {} - regions: list[int] = extra.get("default_regions") or _default_regions() - for rc in regions: - schedule[f"kn-region-{rc}"] = { - "task": TASK_MAP[jt], - "schedule": cron, - "args": [rc, None], - "options": {"queue": queue}, - } - logger.info( - "build_beat_schedule: scrape_kn cron=%r regions=%s queue=%s", - cron_str, - regions, - queue, - ) - else: - kwargs: dict = {} - if jt == "objective_sync": - kwargs = {"triggered_by": "beat"} - schedule[jt] = { - "task": TASK_MAP[jt], - "schedule": cron, - "kwargs": kwargs, - "options": {"queue": queue}, - } - logger.info("build_beat_schedule: %s cron=%r queue=%s", jt, cron_str, queue) - - return schedule - - -def _build_beat_schedule_fallback() -> dict: - """Fallback-расписание из env vars — используется если БД недоступна на старте.""" - logger.warning("beat_schedule: строим fallback из env vars") - schedule: dict = {} - - # scrape_kn - try: - for rc in _default_regions(): - schedule[f"kn-region-{rc}"] = { - "task": "tasks.scrape_kn.scrape_kn_region", - "schedule": _parse_cron(settings.scrape_kn_cron), - "args": [rc, None], - } - except Exception as e: - logger.warning("beat_schedule fallback: scrape_kn failed: %s", e) - - # objective_sync - try: - from app.core.db import SessionLocal - from app.services.objective_sync_config import get_cron_schedule_safe - - cron_str = get_cron_schedule_safe(SessionLocal) - except Exception: - cron_str = settings.objective_sync_cron - - try: - schedule["objective-sync"] = { - "task": "tasks.scrape_objective.sync_all_groups", - "schedule": _parse_cron(cron_str), - "kwargs": {"triggered_by": "beat"}, - } - except Exception as e: - logger.warning("beat_schedule fallback: objective_sync failed: %s", e) - - # refresh ekb-districts medians — ежемесячно 5-го числа в 04:00 МСК - try: - schedule["refresh-ekb-districts-medians"] = { - "task": "tasks.refresh_analytics.refresh_ekb_districts_medians", - "schedule": _parse_cron("0 4 5 * *"), - "kwargs": {"window_months": 24, "min_deals": 50}, - } - except Exception as e: - logger.warning("beat_schedule fallback: refresh_analytics failed: %s", e) - - return schedule - - -def _build_beat_schedule() -> dict: - """Строит beat_schedule: сначала из DB, при неудаче — fallback на env. - - Всегда добавляет refresh-ekb-districts-medians (нет в job_settings). - """ - schedule: dict = {} - try: - schedule = _build_beat_schedule_from_db() - except Exception as e: - logger.error("_build_beat_schedule: неожиданная ошибка: %s — fallback", e) - - if not schedule: - # БД пустая или недоступна — полный fallback - return _build_beat_schedule_fallback() - - # Всегда добавляем refresh_analytics (этот job нет в job_settings, - # он технический и не требует конфигурации через UI). - if "refresh-ekb-districts-medians" not in schedule: - try: - schedule["refresh-ekb-districts-medians"] = { - "task": "tasks.refresh_analytics.refresh_ekb_districts_medians", - "schedule": _parse_cron("0 4 5 * *"), - "kwargs": {"window_months": 24, "min_deals": 50}, - "options": {"queue": "celery"}, - } - except Exception as e: - logger.warning("beat_schedule: refresh_analytics failed: %s", e) - - # Zombie cleanup: каждую минуту проверяет nspd_geo_jobs с heartbeat > 2 мин - # и re-enqueue'ит их. Замена worker_ready signal handler'а — он не fires - # стабильно в нашем setup, beat-task надёжнее. - schedule["nspd-geo-zombie-cleanup"] = { - "task": "tasks.nspd_geo.cleanup_zombies", - "schedule": _parse_cron("* * * * *"), - "options": {"queue": "celery"}, - } - - # OSM POI sync — еженедельно в понедельник в 03:00 МСК - schedule["poi-sync-weekly"] = { - "task": "tasks.poi_sync.sync_osm_poi_ekb", - "schedule": _parse_cron("0 3 * * mon"), - "options": {"queue": "celery"}, - } - - # OSM noise sources sync — еженедельно в понедельник в 03:30 МСК - # (через 30 мин после poi-sync, чтобы не нагружать Overpass одновременно) - schedule["noise-sync-weekly"] = { - "task": "tasks.noise_sync.sync_osm_noise_sources_ekb", - "schedule": _parse_cron("30 3 * * mon"), - "options": {"queue": "celery"}, - } - - # NSPD quarter dump refresh — DISABLED 2026-05-14 per Bug_NSPD_WMS_NotBulk - # post-mortem (vault: fixes/Bug_NSPD_WMS_NotBulk_2026_May14.md). - # - # Action item #1: disable beat schedule до Sprint 2 fix (grid sampling - # rewrite). harvest_quarter в текущей реализации пишет почти пустые dumps - # из-за single-pixel WMS GetFeatureInfo bug. Запуск Mon 04:00 МСК потратит - # rate-limit budget и заполнит nspd_quarter_dumps мусором. - # - # Task code остаётся в tasks/nspd_sync.py — re-enable после Sprint 2 grid - # sampling rewrite (см. Bug_NSPD_WMS_NotBulk_2026_May14 → Sprint 2 fix-strategy). - # До тех пор harvest_quarter можно вызывать вручную через admin endpoint. - # - # schedule["nspd-harvest-stale-quarters"] = { - # "task": "tasks.nspd_sync.harvest_stale_quarters", - # "schedule": _parse_cron("0 4 * * mon"), - # "kwargs": {"region_code": 66, "max_age_days": 90, "batch_size": 50}, - # "options": {"queue": "celery"}, - # } - - return schedule - - -celery_app.conf.beat_schedule = _build_beat_schedule() - - -@worker_process_init.connect -def _reset_db_connections(sender=None, **_kwargs) -> None: - """Dispose SQLAlchemy engine in each prefork child process. - - Background: Celery prefork pool fork()'s child processes from the master - that already imported SQLAlchemy + psycopg. The child inherits open TCP - sockets to PostgreSQL, but psycopg connections cannot be safely shared - across processes after fork — first use raises: - psycopg.ProgrammingError: can't change 'autocommit' now: - connection in transaction status INTRANS - - Fix: dispose the engine in each child. SQLAlchemy will lazily open fresh - connections per-child. Standard Celery + SQLA recipe. - """ - try: - from app.core.db import engine - - engine.dispose() - logger.info("worker_process_init: SQLAlchemy engine disposed for fork") - except Exception as e: - logger.warning("worker_process_init: engine dispose failed: %s", e) - - -@worker_ready.connect -def _resume_zombie_runs(sender=None, **_kwargs) -> None: - """When a worker finishes booting (after redeploy/restart), find any sweep - that was 'running' or 'paused' and re-enqueue a resume task. Each resume - creates a fresh run_id linked via resumed_from_run_id; the original row is - marked 'zombie' so the audit trail is preserved. - - No time threshold: by definition, on worker_ready ANY 'running' row is a - zombie because there is no active worker. Previously we required heartbeat - >5 min, which skipped jobs interrupted seconds before redeploy — they - stayed in 'running' status forever and required manual cancel/resume. - """ - logger.info("worker_ready: resume scan starting") - from sqlalchemy import text - - # Persistent breadcrumb: write to nspd_geo_log so we can confirm via DB - # query whether worker_ready actually fired (independent of container logs). - # NULL job_id to avoid FK violation (FK is to nspd_geo_jobs.job_id which - # has no 0 row). - try: - from app.core.db import SessionLocal - - _db = SessionLocal() - try: - _db.execute( - text( - "INSERT INTO nspd_geo_log (job_id, level, stage, message) " - "VALUES (NULL, 'info', 'worker_ready', 'worker_ready signal fired')" - ) - ) - _db.commit() - except Exception: - _db.rollback() - raise - finally: - _db.close() - except Exception as _e: - logger.warning("worker_ready: breadcrumb insert failed: %s", _e) - - from app.core.db import SessionLocal - - # ── kn_scrape_runs resume ── - db = SessionLocal() - ids: list[int] = [] - try: - rows = ( - db.execute( - text( - """ - SELECT run_id - FROM kn_scrape_runs - WHERE status = 'running' - AND objects_snapshot IS NOT NULL - ORDER BY started_at ASC - LIMIT 20 - """ - ) - ) - .mappings() - .all() - ) - if rows: - ids = [int(r["run_id"]) for r in rows] - # Помечаем найденные как 'zombie' одним апдейтом — resume создаст новые - # run_id со ссылкой resumed_from_run_id. - db.execute( - text( - """ - UPDATE kn_scrape_runs - SET status = 'zombie', - finished_at = NOW(), - error = COALESCE(error, - 'auto-zombie at worker_ready, resume scheduled') - WHERE run_id = ANY(:ids) - """ - ), - {"ids": ids}, - ) - db.commit() - else: - logger.info("worker_ready: нет stale kn runs для resume") - except Exception as e: - logger.exception("worker_ready kn resume scan failed: %s", e) - try: - db.rollback() - except Exception: - pass - finally: - db.close() - - # Enqueue resume tasks. Late import to avoid circular at module load. - from app.workers.tasks.scrape_kn import resume_kn_run - - for rid in ids: - try: - resume_kn_run.apply_async(args=[rid]) - logger.info("worker_ready: resume_kn_run enqueued for run=%s", rid) - except Exception as e: - logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e) - - # NSPD-runs (старый Playwright-scraper) resume УДАЛЁН 2026-05-11. Скрапер - # снят с эксплуатации в пользу bulk geo-fetcher (nspd_geo). История - # старых runs сохранена в nspd_scrape_runs — никаких side-effects. - - # NSPD geo-jobs: bulk-fetcher с собственной resume-логикой через - # nspd_geo_jobs / nspd_geo_targets. Resume любых 'running' / 'paused' jobs - # — на worker_ready по определению нет активных воркеров, всё running == - # zombie. Раньше требовали heartbeat >10мин, что пропускало jobs убитых - # за минуту до редеплоя и оставляло их вечно висеть. - db = SessionLocal() - geo_resume_jobs: list[int] = [] - try: - rows = ( - db.execute( - text( - """ - UPDATE nspd_geo_jobs - SET status = 'queued', - error = COALESCE(error, 'auto-resume at worker_ready') - WHERE status IN ('running', 'paused') - RETURNING job_id - """ - ) - ) - .mappings() - .all() - ) - db.commit() - geo_resume_jobs = [int(r["job_id"]) for r in rows] - for jid in geo_resume_jobs: - logger.info("worker_ready: NSPD geo job=%s — resume scheduled", jid) - except Exception as e: - logger.warning("worker_ready nspd_geo resume scan failed: %s", e) - try: - db.rollback() - except Exception: - pass - finally: - db.close() - - if geo_resume_jobs: - from app.workers.tasks.nspd_geo import process_nspd_geo_job - - for jid in geo_resume_jobs: - try: - process_nspd_geo_job.apply_async(args=[jid], queue="geo") - logger.info("worker_ready: process_nspd_geo_job enqueued job=%s", jid) - except Exception as e: - logger.warning("worker_ready: failed to enqueue geo resume job=%s: %s", jid, e) - logger.info("worker_ready: resume scan finished (geo_jobs=%d)", len(geo_resume_jobs)) - - # Sanity check: nspd_quarter_dumps table must exist (migration 88). - # Logs critical error but does NOT crash the worker — table may be absent - # in dev/staging before migration is applied. - try: - from app.core.db import SessionLocal as _SessionLocal - - _db = _SessionLocal() - try: - _db.execute(text("SELECT 1 FROM nspd_quarter_dumps LIMIT 0")) - logger.info("worker_ready: nspd_quarter_dumps table OK") - except Exception as _te: - logger.critical( - "worker_ready: nspd_quarter_dumps table missing or inaccessible — " - "apply migration 88_nspd_quarter_dumps.sql before using harvest_quarter. " - "Error: %s", - _te, - ) - finally: - _db.close() - except Exception as _e: - logger.warning("worker_ready: nspd_quarter_dumps sanity check failed: %s", _e) +# Register lifecycle hooks (import for side-effect — signal decorator registration) +from app.workers import lifecycle # noqa: E402, F401 diff --git a/backend/app/workers/lifecycle.py b/backend/app/workers/lifecycle.py new file mode 100644 index 00000000..ea12b219 --- /dev/null +++ b/backend/app/workers/lifecycle.py @@ -0,0 +1,216 @@ +"""Worker lifecycle hooks — SQLAlchemy engine dispose on fork, zombie-resume on ready. + +Handlers регистрируются через Celery signals декораторами при импорте модуля. +Импортируй этот модуль как side-effect из celery_app.py: + + from app.workers import lifecycle # noqa: F401 + +worker_process_init — dispose SQLAlchemy engine в каждом prefork child-процессе, + чтобы избежать shared TCP-сокетов к PostgreSQL после fork(). + +worker_ready — при рестарте воркера находит все 'running'/'paused' записи + kn_scrape_runs и nspd_geo_jobs и re-enqueue'ит их как zombie-resume. +""" + +import logging + +from celery.signals import worker_process_init, worker_ready + +logger = logging.getLogger(__name__) + + +@worker_process_init.connect +def _reset_db_connections(sender=None, **_kwargs) -> None: + """Dispose SQLAlchemy engine in each prefork child process. + + Background: Celery prefork pool fork()'s child processes from the master + that already imported SQLAlchemy + psycopg. The child inherits open TCP + sockets to PostgreSQL, but psycopg connections cannot be safely shared + across processes after fork — first use raises: + psycopg.ProgrammingError: can't change 'autocommit' now: + connection in transaction status INTRANS + + Fix: dispose the engine in each child. SQLAlchemy will lazily open fresh + connections per-child. Standard Celery + SQLA recipe. + """ + try: + from app.core.db import engine + + engine.dispose() + logger.info("worker_process_init: SQLAlchemy engine disposed for fork") + except Exception as e: + logger.warning("worker_process_init: engine dispose failed: %s", e) + + +@worker_ready.connect +def _resume_zombie_runs(sender=None, **_kwargs) -> None: + """When a worker finishes booting (after redeploy/restart), find any sweep + that was 'running' or 'paused' and re-enqueue a resume task. Each resume + creates a fresh run_id linked via resumed_from_run_id; the original row is + marked 'zombie' so the audit trail is preserved. + + No time threshold: by definition, on worker_ready ANY 'running' row is a + zombie because there is no active worker. Previously we required heartbeat + >5 min, which skipped jobs interrupted seconds before redeploy — they + stayed in 'running' status forever and required manual cancel/resume. + """ + logger.info("worker_ready: resume scan starting") + from sqlalchemy import text + + # Persistent breadcrumb: write to nspd_geo_log so we can confirm via DB + # query whether worker_ready actually fired (independent of container logs). + # NULL job_id to avoid FK violation (FK is to nspd_geo_jobs.job_id which + # has no 0 row). + try: + from app.core.db import SessionLocal + + _db = SessionLocal() + try: + _db.execute( + text( + "INSERT INTO nspd_geo_log (job_id, level, stage, message) " + "VALUES (NULL, 'info', 'worker_ready', 'worker_ready signal fired')" + ) + ) + _db.commit() + except Exception: + _db.rollback() + raise + finally: + _db.close() + except Exception as _e: + logger.warning("worker_ready: breadcrumb insert failed: %s", _e) + + from app.core.db import SessionLocal + + # ── kn_scrape_runs resume ── + db = SessionLocal() + ids: list[int] = [] + try: + rows = ( + db.execute( + text( + """ + SELECT run_id + FROM kn_scrape_runs + WHERE status = 'running' + AND objects_snapshot IS NOT NULL + ORDER BY started_at ASC + LIMIT 20 + """ + ) + ) + .mappings() + .all() + ) + if rows: + ids = [int(r["run_id"]) for r in rows] + # Помечаем найденные как 'zombie' одним апдейтом — resume создаст новые + # run_id со ссылкой resumed_from_run_id. + db.execute( + text( + """ + UPDATE kn_scrape_runs + SET status = 'zombie', + finished_at = NOW(), + error = COALESCE(error, + 'auto-zombie at worker_ready, resume scheduled') + WHERE run_id = ANY(:ids) + """ + ), + {"ids": ids}, + ) + db.commit() + else: + logger.info("worker_ready: нет stale kn runs для resume") + except Exception as e: + logger.exception("worker_ready kn resume scan failed: %s", e) + try: + db.rollback() + except Exception: + pass + finally: + db.close() + + # Enqueue resume tasks. Late import to avoid circular at module load. + from app.workers.tasks.scrape_kn import resume_kn_run + + for rid in ids: + try: + resume_kn_run.apply_async(args=[rid]) + logger.info("worker_ready: resume_kn_run enqueued for run=%s", rid) + except Exception as e: + logger.warning("worker_ready: failed to enqueue resume for run=%s: %s", rid, e) + + # NSPD-runs (старый Playwright-scraper) resume УДАЛЁН 2026-05-11. Скрапер + # снят с эксплуатации в пользу bulk geo-fetcher (nspd_geo). История + # старых runs сохранена в nspd_scrape_runs — никаких side-effects. + + # NSPD geo-jobs: bulk-fetcher с собственной resume-логикой через + # nspd_geo_jobs / nspd_geo_targets. Resume любых 'running' / 'paused' jobs + # — на worker_ready по определению нет активных воркеров, всё running == + # zombie. Раньше требовали heartbeat >10мин, что пропускало jobs убитых + # за минуту до редеплоя и оставляло их вечно висеть. + db = SessionLocal() + geo_resume_jobs: list[int] = [] + try: + rows = ( + db.execute( + text( + """ + UPDATE nspd_geo_jobs + SET status = 'queued', + error = COALESCE(error, 'auto-resume at worker_ready') + WHERE status IN ('running', 'paused') + RETURNING job_id + """ + ) + ) + .mappings() + .all() + ) + db.commit() + geo_resume_jobs = [int(r["job_id"]) for r in rows] + for jid in geo_resume_jobs: + logger.info("worker_ready: NSPD geo job=%s — resume scheduled", jid) + except Exception as e: + logger.warning("worker_ready nspd_geo resume scan failed: %s", e) + try: + db.rollback() + except Exception: + pass + finally: + db.close() + + if geo_resume_jobs: + from app.workers.tasks.nspd_geo import process_nspd_geo_job + + for jid in geo_resume_jobs: + try: + process_nspd_geo_job.apply_async(args=[jid], queue="geo") + logger.info("worker_ready: process_nspd_geo_job enqueued job=%s", jid) + except Exception as e: + logger.warning("worker_ready: failed to enqueue geo resume job=%s: %s", jid, e) + logger.info("worker_ready: resume scan finished (geo_jobs=%d)", len(geo_resume_jobs)) + + # Sanity check: nspd_quarter_dumps table must exist (migration 88). + # Logs critical error but does NOT crash the worker — table may be absent + # in dev/staging before migration is applied. + try: + from app.core.db import SessionLocal as _SessionLocal + + _db = _SessionLocal() + try: + _db.execute(text("SELECT 1 FROM nspd_quarter_dumps LIMIT 0")) + logger.info("worker_ready: nspd_quarter_dumps table OK") + except Exception as _te: + logger.critical( + "worker_ready: nspd_quarter_dumps table missing or inaccessible — " + "apply migration 88_nspd_quarter_dumps.sql before using harvest_quarter. " + "Error: %s", + _te, + ) + finally: + _db.close() + except Exception as _e: + logger.warning("worker_ready: nspd_quarter_dumps sanity check failed: %s", _e)