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

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

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)