From 576e16d14d1eece9ba138979422b2ca96fde9f54 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 3 Jun 2026 21:49:52 -0400 Subject: [PATCH] fix(download): release DB connections across the gallery-dl subprocess MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backfill events were STILL stranding empty after the timeout-ladder fix. Worker logs showed the salvage path working ("Download timeout for anduo/patreon after 1170.0s (18 files written)") but then: Retry in 3s: DBAPIError(ConnectionDoesNotExistError: connection was closed in the middle of operation) ...succeeded in 0.149s <- in-flight guard no-op Root cause: DownloadService held the async + sync DB connections checked out across the entire (≤19.5-min backfill) gallery-dl subprocess. The server reaps the idle connection, so phase 3's first query hits a dead socket. That DBAPIError trips download_source's autoretry_for, the retry re-enters _phase1_setup, sees the event still 'running', returns in_flight and no-ops — leaving the event to be stranded empty by the recovery sweep. pool_pre_ping was already on both engines but can't help a *held* connection (it only validates on pool checkout). Fix: - DownloadService.download_source closes the async + sync sessions after phase 1, before the subprocess, so phase 3 re-acquires a live connection (matches the class's "Phase 2 — no DB connection" docstring). - The per-task async engine switches to NullPool so phase 3 always opens a fresh connection rather than a pooled one the server may have reaped. Tests: assert connections are released before gdl.download runs and the event still finalizes; assert the task engine uses NullPool. Also fixes a stale 1800s->1170s comment. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/download_service.py | 17 +++++- backend/app/tasks/_async_session.py | 11 +++- tests/test_download_service.py | 78 ++++++++++++++++++++++++ 3 files changed, 104 insertions(+), 2 deletions(-) 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()