fix(worker): periodic zombie cleanup via beat instead of worker_ready
worker_ready signal handler was NOT firing in our setup (verified via DB breadcrumb after 3 deploys — zero rows of stage='worker_ready' in nspd_geo_log). Root cause of unreliability unknown — possibly Celery internals, possibly compose recreate timing. Either way, after every redeploy users had to manually cancel/resume jobs to keep them moving. Replace signal-based resume with periodic beat task: - cleanup_zombies runs every minute (* * * * *) - Finds nspd_geo_jobs in status running/paused with heartbeat >2 min stale - Sets status='queued' + apply_async with queue=geo - Idempotent — if no zombies, no-op worker_ready handler kept (with FK-fix breadcrumb on NULL job_id) for diagnostic purposes — if signal ever does fire, we'll have evidence.
This commit is contained in:
parent
17feaa408e
commit
caa467fb7e
2 changed files with 63 additions and 1 deletions
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue