From fdd5b4e0e23a82e5ad1754dc6cd65e577664e9f5 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Mon, 11 May 2026 16:40:21 +0300 Subject: [PATCH] fix(worker): dispose SQLA engine in worker_process_init + log resume scan 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. --- backend/app/workers/celery_app.py | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/backend/app/workers/celery_app.py b/backend/app/workers/celery_app.py index 2d34e118..0dfcd9da 100644 --- a/backend/app/workers/celery_app.py +++ b/backend/app/workers/celery_app.py @@ -8,7 +8,7 @@ import logging from celery import Celery from celery.schedules import crontab -from celery.signals import worker_ready +from celery.signals import worker_process_init, worker_ready from app.core.config import settings @@ -209,6 +209,29 @@ def _build_beat_schedule() -> dict: 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 @@ -221,6 +244,7 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: >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 @@ -333,3 +357,4 @@ def _resume_zombie_runs(sender=None, **_kwargs) -> None: 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))