fix(tasks): share one sync engine per worker process to stop connection-pool leak

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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

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 import select, update
-from sqlalchemy.orm import sessionmaker
 
 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
-
+from ._sync_engine import sync_session_factory as _sync_session_factory
-
-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)
 
 
 IMAGES_ROOT = Path("/images")

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 import delete, select, update
-from sqlalchemy.orm import sessionmaker
 
 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.

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 import select
-from sqlalchemy.orm import sessionmaker
 
 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
 

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:

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()