diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 188078b1..fe1dbc6f 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -203,6 +203,15 @@ def _build_beat_schedule() -> dict: 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"}, + } + return schedule @@ -249,6 +258,8 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: # 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 @@ -257,7 +268,7 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: _db.execute( text( "INSERT INTO nspd_geo_log (job_id, level, stage, message) " - "VALUES (0, 'info', 'worker_ready', 'worker_ready signal fired')" + "VALUES (NULL, 'info', 'worker_ready', 'worker_ready signal fired')" ) ) _db.commit() diff --git a/backend/app/workers/tasks/nspd_geo.py b/backend/app/workers/tasks/nspd_geo.py index aa62a394..c3000b23 100644 --- a/backend/app/workers/tasks/nspd_geo.py +++ b/backend/app/workers/tasks/nspd_geo.py @@ -496,3 +496,54 @@ def process_nspd_geo_job(self: Any, job_id: int) -> dict[str, Any]: raise finally: db.close() + + +@celery_app.task(name="tasks.nspd_geo.cleanup_zombies") +def cleanup_zombies() -> dict[str, Any]: + """Periodic zombie cleanup — runs every minute via beat schedule. + + Catches nspd_geo_jobs in status='running' / 'paused' with stale heartbeat + (>2 min) and re-enqueues them. Replaces the unreliable worker_ready signal + handler — beat is independent of worker lifecycle and fires every minute. + + Idempotent: if no zombies, does nothing. If a job is genuinely active, its + heartbeat will be fresh and the WHERE clause won't match. + """ + db = SessionLocal() + resumed: list[int] = [] + try: + rows = ( + db.execute( + text( + """ + UPDATE nspd_geo_jobs + SET status = 'queued', + error = COALESCE(error, 'auto-resume by beat cleanup') + WHERE status IN ('running', 'paused') + AND heartbeat_at < NOW() - INTERVAL '2 minutes' + RETURNING job_id + """ + ) + ) + .mappings() + .all() + ) + db.commit() + resumed = [int(r["job_id"]) for r in rows] + for jid in resumed: + try: + process_nspd_geo_job.apply_async(args=[jid], queue="geo") + logger.info("cleanup_zombies: re-enqueued job=%s", jid) + except Exception as e: + logger.warning("cleanup_zombies: enqueue job=%s failed: %s", jid, e) + if resumed: + logger.info("cleanup_zombies: resumed %d zombie job(s): %s", len(resumed), resumed) + except Exception as e: + logger.exception("cleanup_zombies failed: %s", e) + try: + db.rollback() + except Exception: + pass + finally: + db.close() + return {"resumed_count": len(resumed), "resumed_job_ids": resumed}