Merge pull request 'fix(download): release DB connections across the gallery-dl subprocess' (#60) from dev into main

backend/app/services/download_service.py

@@ -91,10 +91,25 @@ class DownloadService:
             return setup["event_id"]
         ctx = setup
 
+        # Release the phase-1 DB connections before the (up to ~19.5-min in
+        # backfill) gallery-dl subprocess. Held checked-out across that idle
+        # window, the asyncpg/psycopg connections get reaped by the server,
+        # and phase 3's first query then hits a dead socket
+        # (asyncpg ConnectionDoesNotExistError) → download_source autoretry →
+        # _phase1_setup's in-flight guard no-ops the retry → the event
+        # strands empty for the recovery sweep (Anduo #40014, 2026-06-04).
+        # pool_pre_ping can't help a *held* connection — it only validates on
+        # pool checkout. Closing returns them to the pool so phase 3 re-
+        # acquires a live one (the async task engine uses NullPool, the sync
+        # engine pre_ping + pool_recycle=300). This is what makes the
+        # "Phase 2 — no DB connection" contract in the class docstring true.
+        await self.async_session.close()
+        self.sync_session.close()
+
         source_config = SourceConfig.from_dict(ctx["config_overrides"] or {})
         # alembic 0031 / plan #544: derive skip_value + timeout from the
         # source's backfill_runs_remaining counter. When > 0, walk the full
-        # post history (skip: True + 1800s); when 0, exit gallery-dl after
+        # post history (skip: True + 1170s); when 0, exit gallery-dl after
         # 20 contiguous archived items (skip: "exit:20" + the default
         # 870s). Operator sets backfill via POST /api/sources/{id}/backfill.
         backfill_remaining = ctx.get("backfill_runs_remaining", 0) or 0

backend/app/tasks/_async_session.py

@@ -10,6 +10,7 @@ disposes it (``await engine.dispose()``) when its loop ends.
 """
 
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
 
 from ..config import get_config
 
@@ -17,5 +18,13 @@ from ..config import get_config
 def async_session_factory():
     """Return ``(sessionmaker, engine)`` bound to a fresh async engine."""
     cfg = get_config()
-    engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True)
+    # NullPool: this engine lives for ONE task (created + disposed per
+    # asyncio.run loop), so intra-task connection pooling buys nothing and
+    # actively bit us — download_source releases its phase-1 connection
+    # before a multi-minute gallery-dl subprocess, and a *pooled* idle
+    # connection would be reaped by the server and handed back dead to
+    # phase 3 (asyncpg ConnectionDoesNotExistError, Anduo #40014). NullPool
+    # opens a fresh real connection on each checkout, so phase 3 always
+    # reconnects clean; pre_ping is then redundant.
+    engine = create_async_engine(cfg.database_url, future=True, poolclass=NullPool)
     return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine

tests/test_download_service.py

@@ -613,3 +613,81 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
     assert len(thumb_calls) == 2
     assert len(ml_calls) == 2
     assert sorted(thumb_calls) == sorted(ml_calls)
+
+
+@pytest.mark.asyncio
+async def test_releases_db_connections_before_subprocess(
+    db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
+):
+    """Phase 1's DB connections must be released BEFORE the (up to ~19.5-min
+    in backfill) gallery-dl subprocess, so they don't idle-die and strand
+    phase 3 with ConnectionDoesNotExistError (Anduo #40014). Spy on the
+    async session close and assert it happened before gdl.download runs;
+    phase 3 must still finalize the event."""
+    from backend.app.services.download_service import DownloadService
+    from backend.app.services.importer import Importer
+
+    _artist, source = seed_artist_and_source
+
+    closed = {"async": False}
+    orig_close = db.close
+
+    async def spy_close():
+        closed["async"] = True
+        await orig_close()
+
+    monkeypatch.setattr(db, "close", spy_close)
+
+    seen = {}
+
+    async def fake_download(url, **k):
+        seen["closed_before_download"] = closed["async"]
+        return _make_fake_dl_result(success=True, written_paths=[], stdout="")
+
+    fake_gdl = MagicMock()
+    fake_gdl.download = fake_download
+    fake_gdl._compute_run_stats = lambda *a, **k: {
+        "exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
+        "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
+    }
+    fake_gdl._extract_errors_warnings = lambda *a, **k: ""
+    fake_gdl._truncate_log = lambda x, **k: x
+
+    sync_settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    importer = Importer(
+        session=db_sync, images_root=tmp_path, import_root=tmp_path,
+        thumbnailer=Thumbnailer(images_root=tmp_path), settings=sync_settings,
+    )
+    cred_service = CredentialService(
+        db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
+    )
+
+    svc = DownloadService(
+        async_session=db, sync_session=db_sync,
+        gdl=fake_gdl, importer=importer, cred_service=cred_service,
+    )
+    event_id = await svc.download_source(source.id)
+
+    assert seen["closed_before_download"] is True
+    ev = (await db.execute(
+        select(DownloadEvent).where(DownloadEvent.id == event_id)
+    )).scalar_one()
+    assert ev.status == "ok"
+
+
+@pytest.mark.asyncio
+async def test_async_task_engine_uses_nullpool():
+    """The per-task async engine must use NullPool so phase 3 re-acquires a
+    fresh connection instead of a pooled one the server reaped during the
+    long subprocess (Anduo #40014)."""
+    from sqlalchemy.pool import NullPool
+
+    from backend.app.tasks._async_session import async_session_factory
+
+    _factory, engine = async_session_factory()
+    try:
+        assert isinstance(engine.sync_engine.pool, NullPool)
+    finally:
+        await engine.dispose()