diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index f14367a1..2d34e118 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -212,9 +212,14 @@ celery_app.conf.beat_schedule = _build_beat_schedule() @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' with a stale heartbeat (>5 min) 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. + 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. """ from sqlalchemy import text @@ -230,8 +235,6 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: FROM kn_scrape_runs WHERE status = 'running' AND objects_snapshot IS NOT NULL - AND COALESCE(heartbeat_at, started_at) - < NOW() - INTERVAL '5 minutes' ORDER BY started_at ASC LIMIT 20 """ @@ -286,9 +289,10 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: # старых runs сохранена в nspd_scrape_runs — никаких side-effects. # NSPD geo-jobs: bulk-fetcher с собственной resume-логикой через - # nspd_geo_jobs / nspd_geo_targets. Жадный resume: status='running' со - # stale heartbeat (>10мин) → re-enqueue с тем же job_id (task сам прочитает - # pending targets и продолжит). + # 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: @@ -300,8 +304,6 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: SET status = 'queued', error = COALESCE(error, 'auto-resume at worker_ready') WHERE status IN ('running', 'paused') - AND COALESCE(heartbeat_at, started_at, created_at) - < NOW() - INTERVAL '10 minutes' RETURNING job_id """ ) diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index fc686508..54fc3668 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -96,7 +96,7 @@ services: # контейнера задан settings.objective_anton_sqlite_path (default # /data/anton-sqlite/analysis.db). - /opt/gendesign/site-finder:/data/anton-sqlite:ro - command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=5", "--queues=celery,scrape_kn,geo"] + command: ["celery", "-A", "app.workers.celery_app", "worker", "--loglevel=info", "--concurrency=8", "--queues=celery,scrape_kn,geo"] beat: # Lean backend-образ (без Chromium) — beat только триггерит таски в Redis.