diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 92b499e..f965a56 100644 --- a/backend/app/services/download_service.py +++ b/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 diff --git a/backend/app/tasks/_async_session.py b/backend/app/tasks/_async_session.py index ea7a796..e8c9094 100644 --- a/backend/app/tasks/_async_session.py +++ b/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 diff --git a/tests/test_download_service.py b/tests/test_download_service.py index cc07f05..8d15e60 100644 --- a/tests/test_download_service.py +++ b/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()