bvandeusen
576e16d14d
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) <noreply@anthropic.com>
31 lines
1.5 KiB
Python
31 lines
1.5 KiB
Python
"""Per-invocation async session factory for Celery task modules.
|
|
|
|
Async engine connections are bound to the event loop. Each Celery task
|
|
runs its async body under a fresh ``asyncio.run()`` loop, so it needs its
|
|
own engine created (and disposed) within that loop — a process-wide async
|
|
engine would reuse loop-bound connections across tasks and raise "attached
|
|
to a different loop". So unlike the process-wide sync engine in
|
|
``_sync_engine.py``, this returns a fresh engine per call; the caller
|
|
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
|
|
|
|
|
|
def async_session_factory():
|
|
"""Return ``(sessionmaker, engine)`` bound to a fresh async engine."""
|
|
cfg = get_config()
|
|
# 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
|