fdd5b4e0e2
Two related fixes: 1. worker_process_init handler disposes the SQLAlchemy engine in each prefork child. Without this, child processes inherit open psycopg sockets from the parent. First use in a child raises ProgrammingError: can't change 'autocommit' now: connection in transaction status INTRANS. This was killing 1 of every 5 parallel geo jobs on cold start (job 13 in latest bulk run). 2. Add logger.info at start/end of worker_ready resume handler so we can see in worker logs whether it actually fired and how many jobs it resumed.
360 lines
13 KiB
Python
360 lines
13 KiB
Python
"""Celery app + beat schedule.
|
||
|
||
Beat schedule строится из таблицы job_settings при старте beat-контейнера.
|
||
При недоступности БД — fallback на env-based расписание (старое поведение).
|
||
"""
|
||
|
||
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,
|
||
backend=settings.redis_url,
|
||
include=[
|
||
"app.workers.tasks.scrape_kn",
|
||
"app.workers.tasks.refresh_analytics",
|
||
"app.workers.tasks.scrape_objective",
|
||
"app.workers.tasks.objective_etl",
|
||
"app.workers.tasks.nspd_geo",
|
||
],
|
||
)
|
||
celery_app.conf.timezone = "Europe/Moscow"
|
||
|
||
|
||
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)
|
||
|
||
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
|
||
|
||
from app.core.db import SessionLocal
|
||
|
||
db = SessionLocal()
|
||
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 not rows:
|
||
logger.info("worker_ready: нет stale runs для resume")
|
||
return
|
||
|
||
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()
|
||
except Exception as e:
|
||
logger.exception("worker_ready resume scan failed: %s", e)
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
return
|
||
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))
|