From 7d8b9c3d90af5004d13cea4af93bb04775271967 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 23 May 2026 10:44:16 -0400 Subject: [PATCH] fix(tasks): share one sync engine per worker process to stop connection-pool leak Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/tasks/_sync_engine.py | 36 +++++++++++++++++++++++++++++++ backend/app/tasks/import_file.py | 11 ++-------- backend/app/tasks/maintenance.py | 11 ++-------- backend/app/tasks/ml.py | 11 ++-------- backend/app/tasks/scan.py | 10 ++------- backend/app/tasks/thumbnail.py | 11 +--------- 6 files changed, 45 insertions(+), 45 deletions(-) create mode 100644 backend/app/tasks/_sync_engine.py diff --git a/backend/app/tasks/_sync_engine.py b/backend/app/tasks/_sync_engine.py new file mode 100644 index 0000000..508e490 --- /dev/null +++ b/backend/app/tasks/_sync_engine.py @@ -0,0 +1,36 @@ +"""Process-level sync engine for Celery task modules. + +Each task module used to call ``create_engine(...)`` on every invocation, +which leaked engines (and Postgres connections) — high-fire-rate tasks +like ``import_media_file`` would exhaust ``max_connections`` within +minutes during a bulk migration. + +This module owns one engine per process. Celery prefork forks before any +task runs, so each worker process lazily initializes its own engine on +the first task and reuses it for the rest of its life. +""" + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from ..config import get_config + +_ENGINE = None +_SESSIONMAKER = None + + +def sync_session_factory(): + """Return a process-wide ``sessionmaker`` bound to a single engine.""" + global _ENGINE, _SESSIONMAKER + if _SESSIONMAKER is None: + cfg = get_config() + _ENGINE = create_engine( + cfg.database_url_sync, + future=True, + pool_pre_ping=True, + pool_size=5, + max_overflow=5, + pool_recycle=300, + ) + _SESSIONMAKER = sessionmaker(_ENGINE, expire_on_commit=False) + return _SESSIONMAKER diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index d53de74..4eee360 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -5,20 +5,13 @@ updates the ImportTask state machine + ImportBatch counters atomically. from datetime import UTC, datetime from pathlib import Path -from sqlalchemy import create_engine, select, update -from sqlalchemy.orm import sessionmaker +from sqlalchemy import select, update from ..celery_app import celery -from ..config import get_config from ..models import ImportBatch, ImportSettings, ImportTask from ..services.importer import Importer from ..services.thumbnailer import Thumbnailer - - -def _sync_session_factory(): - cfg = get_config() - engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True) - return sessionmaker(engine, expire_on_commit=False) +from ._sync_engine import sync_session_factory as _sync_session_factory IMAGES_ROOT = Path("/images") diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index ced907f..4b23ac0 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -6,13 +6,12 @@ from datetime import UTC, datetime, timedelta from pathlib import Path from PIL import Image -from sqlalchemy import create_engine, delete, select, update -from sqlalchemy.orm import sessionmaker +from sqlalchemy import delete, select, update from ..celery_app import celery -from ..config import get_config from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask from ..utils.phash import compute_phash +from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) @@ -23,12 +22,6 @@ VERIFY_PAGE = 200 FFPROBE_TIMEOUT_SECONDS = 10 -def _sync_session_factory(): - cfg = get_config() - engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True) - return sessionmaker(engine, expire_on_commit=False) - - @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") def recover_interrupted_tasks() -> int: """Find ImportTask rows stuck in 'processing' for >30 min and re-queue them. diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 13d4394..a208376 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -8,23 +8,16 @@ apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions from pathlib import Path -from sqlalchemy import create_engine, select -from sqlalchemy.orm import sessionmaker +from sqlalchemy import select from ..celery_app import celery -from ..config import get_config from ..models import ImageRecord, MLSettings +from ._sync_engine import sync_session_factory as _sync_session_factory IMAGES_ROOT = Path("/images") VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"} -def _sync_session_factory(): - cfg = get_config() - engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True) - return sessionmaker(engine, expire_on_commit=False) - - def _is_video(path: Path) -> bool: return path.suffix.lower() in VIDEO_EXTS diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 02b1838..c4581f7 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -11,14 +11,14 @@ import asyncio from datetime import UTC, datetime from pathlib import Path -from sqlalchemy import create_engine, select +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine -from sqlalchemy.orm import sessionmaker from ..celery_app import celery from ..config import get_config from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask from ..services.scheduler_service import select_due_sources +from ._sync_engine import sync_session_factory as _sync_session_factory def _iter_import_files(import_root: Path): @@ -35,12 +35,6 @@ def _iter_import_files(import_root: Path): yield entry -def _sync_session_factory(): - cfg = get_config() - engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True) - return sessionmaker(engine, expire_on_commit=False) - - @celery.task(name="backend.app.tasks.scan.scan_directory", bind=True) def scan_directory(self, triggered_by: str = "manual", mode: str = "quick") -> int: diff --git a/backend/app/tasks/thumbnail.py b/backend/app/tasks/thumbnail.py index 68fa665..61bed80 100644 --- a/backend/app/tasks/thumbnail.py +++ b/backend/app/tasks/thumbnail.py @@ -7,24 +7,15 @@ so they deserve their own queue lane. from pathlib import Path -from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker - from ..celery_app import celery -from ..config import get_config from ..models import ImageRecord from ..services.importer import is_video from ..services.thumbnailer import Thumbnailer +from ._sync_engine import sync_session_factory as _sync_session_factory IMAGES_ROOT = Path("/images") -def _sync_session_factory(): - cfg = get_config() - engine = create_engine(cfg.database_url_sync, future=True, pool_pre_ping=True) - return sessionmaker(engine, expire_on_commit=False) - - @celery.task(name="backend.app.tasks.thumbnail.generate_thumbnail", bind=True) def generate_thumbnail(self, image_id: int) -> dict: SessionLocal = _sync_session_factory()