diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index c798bfb..4616812 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -743,6 +743,10 @@ async def normalize_existing_tags( "sample": sample, } start = time.monotonic() + log.info( + "normalize_existing_tags: %d group(s) need changes (budget=%ss)", + len(touched), time_budget_seconds, + ) for done, (key, members) in enumerate(touched): # Time-box: stop cleanly before the Celery limit kills us mid-group and # strands the run as a timeout. The caller re-enqueues to finish the @@ -754,6 +758,16 @@ async def normalize_existing_tags( summary["partial"] = True summary["remaining"] = len(touched) - done break + # Heartbeat so a long run is diagnosable instead of silent — the timeout + # operator-flagged 2026-06-07 produced zero logs because the only log was + # per finished group and it was stuck mid-group on a lock. + if done and done % 25 == 0: + log.info( + "normalize_existing_tags: %d/%d groups (%d merged, %d errors, " + "%.0fs elapsed)", + done, len(touched), summary["merged"], summary["errors"], + time.monotonic() - start, + ) canonical = key[2] names_by_id = dict(members) # Survivor: prefer a member already named canonically (no rename, no diff --git a/backend/app/tasks/_async_session.py b/backend/app/tasks/_async_session.py index e8c9094..0252dc3 100644 --- a/backend/app/tasks/_async_session.py +++ b/backend/app/tasks/_async_session.py @@ -15,8 +15,15 @@ from sqlalchemy.pool import NullPool from ..config import get_config -def async_session_factory(): - """Return ``(sessionmaker, engine)`` bound to a fresh async engine.""" +def async_session_factory(*, server_settings: dict | None = None): + """Return ``(sessionmaker, engine)`` bound to a fresh async engine. + + ``server_settings`` (optional) are applied by asyncpg as per-connection GUCs + on connect. Because NullPool opens a fresh real connection per checkout, every + transaction in the task inherits them — used to set ``lock_timeout`` so a + statement blocked on a lock fails fast instead of hanging to the Celery hard + limit (operator-flagged normalize_tags timeout 2026-06-07). + """ cfg = get_config() # NullPool: this engine lives for ONE task (created + disposed per # asyncio.run loop), so intra-task connection pooling buys nothing and @@ -26,5 +33,13 @@ def async_session_factory(): # 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) + connect_args = {} + if server_settings: + connect_args["server_settings"] = { + k: str(v) for k, v in server_settings.items() + } + engine = create_async_engine( + cfg.database_url, future=True, poolclass=NullPool, + connect_args=connect_args, + ) return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine diff --git a/backend/app/tasks/admin.py b/backend/app/tasks/admin.py index 8096707..c6a1b08 100644 --- a/backend/app/tasks/admin.py +++ b/backend/app/tasks/admin.py @@ -126,7 +126,16 @@ def normalize_tags_task(self) -> dict: from ._async_session import async_session_factory async def _run() -> dict: - async_factory, async_engine = async_session_factory() + # lock_timeout=30s: a per-group merge repoints FKs across image_tag and + # series_page; if a statement blocks on a lock (e.g. behind a schema + # migration holding ACCESS EXCLUSIVE on series_page — the exact wedge that + # made this task run to the 40-min hard limit with no progress, + # operator-flagged 2026-06-07), it now fails fast. The per-group handler + # catches it (rollback + error++) and the loop continues, so one blocked + # group can't strand the whole chunk. + async_factory, async_engine = async_session_factory( + server_settings={"lock_timeout": "30s"} + ) try: async with async_factory() as session: # normalize_existing_tags commits per group internally.