From a00a2786e36f67aaeafd249cf5722a0f0dab0b17 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 7 Jun 2026 21:02:39 -0400 Subject: [PATCH] fix(tags): normalize task fails fast on lock + logs progress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit normalize_tags_task ran to the 40-min hard limit with zero logs (operator- flagged 2026-06-07). Cause: a per-group merge repoints series_page (via _repoint_series_pages); during the wedged 0040 migration that held ACCESS EXCLUSIVE on series_page, the merge's UPDATE blocked on that lock. The time-box check is at the top of the group loop, so a statement blocked mid-group never yields back to it — the task sat until the Celery hard kill. No logs because the only log fired per *finished* group. - Set lock_timeout=30s on the normalize session (opt-in server_settings on the async factory). A blocked merge now raises, the per-group handler rolls back + counts an error, and the loop continues — one stuck group can't strand the chunk, and the budget checkpoint stays effective. - Log group count at start + a heartbeat every 25 groups, so a long/slow run is diagnosable instead of silent. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/tag_service.py | 14 ++++++++++++++ backend/app/tasks/_async_session.py | 21 ++++++++++++++++++--- backend/app/tasks/admin.py | 11 ++++++++++- 3 files changed, 42 insertions(+), 4 deletions(-) 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.