From d9d502a60d52f6cb36f0d8fa4b008b56b59ba7db Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 23:57:06 -0400 Subject: [PATCH] fix(tags): time-box + self-resume the tag standardization (stop the 40-min timeout) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit normalize_tags_task timed out at the 40-min hard limit on a large back-catalog (the first run recases the whole booru vocabulary) — operator-flagged, and it monopolized the concurrency-1 maintenance queue while doing so. normalize_existing_tags now takes time_budget_seconds: the live run stops cleanly at the budget and reports {partial, remaining}. The task runs 600s chunks and re-enqueues itself until nothing remains (idempotent — commits per group, so the next chunk skips already-canonical groups). Short chunks let the recovery sweep and other maintenance tasks interleave instead of being blocked for 40 minutes. Frontend: the Standardize button is now fire-and-forget ("Queued — runs in the background; re-run Preview to confirm") instead of poll-until-done, which would have falsely reported "complete" after the first chunk. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/tag_service.py | 35 +++++++++++++++++-- backend/app/tasks/admin.py | 30 +++++++++++++--- .../settings/TagMaintenanceCard.vue | 23 +++++------- tests/test_tag_normalize.py | 23 ++++++++++++ 4 files changed, 90 insertions(+), 21 deletions(-) diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index bfdef07..255311e 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -1,6 +1,7 @@ """Tag CRUD + autocomplete + image-tag association.""" import logging +import time from collections.abc import Sequence from dataclasses import dataclass @@ -637,7 +638,10 @@ def _best_connected(tag_ids: list[int], counts: dict[int, int]) -> int: async def normalize_existing_tags( - session: AsyncSession, *, dry_run: bool = False + session: AsyncSession, + *, + dry_run: bool = False, + time_budget_seconds: float | None = None, ) -> dict: """Convert the back-catalog to the #701 canonical tag form. @@ -647,6 +651,14 @@ async def normalize_existing_tags( survivor to the canonical form. Idempotent: a group that is already a lone canonical tag is a no-op, so re-running is safe. + A first run over a fresh back-catalog can touch tens of thousands of tags + (the whole booru-derived vocabulary needs recasing) and won't finish inside + one Celery time limit — it timed out at 40 min (operator-flagged 2026-06-07). + `time_budget_seconds` time-boxes the live run: it stops cleanly at the budget + and reports `partial`/`remaining` so the caller can re-enqueue and continue. + Because it commits per group and is idempotent, the next run just picks up + the groups still needing change. + dry_run=True returns a projection (counts + a sample of the changes) with no mutations. Live runs commit per group and isolate failures per group so one bad group can't strand the rest. @@ -656,7 +668,7 @@ async def normalize_existing_tags( "total_changes": T, "sample": [{"to", "from": [...], "kind", "merge"}]} Returns (live): {"groups_processed", "merged", "renamed", "aliases_created", "errors", - "sample": [...]} + "total_changes", "remaining", "partial", "sample": [...]} """ rows = ( await session.execute( @@ -715,9 +727,23 @@ async def normalize_existing_tags( "renamed": 0, "aliases_created": 0, "errors": 0, + "total_changes": len(touched), + "remaining": len(touched), + "partial": False, "sample": sample, } - for key, members in touched: + start = time.monotonic() + 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 + # rest (idempotent — already-canonical groups are skipped next pass). + if ( + time_budget_seconds is not None + and time.monotonic() - start >= time_budget_seconds + ): + summary["partial"] = True + summary["remaining"] = len(touched) - done + break canonical = key[2] names_by_id = dict(members) # Survivor: prefer a member already named canonically (no rename, no @@ -752,4 +778,7 @@ async def normalize_existing_tags( log.warning( "tag normalize failed for group %r: %s", canonical, exc ) + else: + # Loop finished without hitting the time budget — nothing left to do. + summary["remaining"] = 0 return summary diff --git a/backend/app/tasks/admin.py b/backend/app/tasks/admin.py index f115387..f9dafd4 100644 --- a/backend/app/tasks/admin.py +++ b/backend/app/tasks/admin.py @@ -82,11 +82,21 @@ def reextract_archive_attachments_task(self) -> dict: retry_backoff=15, retry_backoff_max=180, max_retries=1, soft_time_limit=1800, time_limit=2400, # 30 min / 40 min ) +# Time-box one chunk well under the soft limit so a large back-catalog (the +# first run recases the whole booru vocabulary) can't run the task into the +# Celery time limit — it timed out at 40 min, operator-flagged 2026-06-07. The +# task re-enqueues itself until nothing remains (idempotent — already-canonical +# groups are skipped). 600s keeps each chunk short enough that the recovery +# sweep and other maintenance tasks interleave on the concurrency-1 queue. +_NORMALIZE_CHUNK_SECONDS = 600 + + def normalize_tags_task(self) -> dict: """Wraps tag_service.normalize_existing_tags (#714): Title-Case the back-catalog and merge case/whitespace-variant duplicate tags via the - tested async merge path. Runs under its own asyncio loop + per-task async - engine (NullPool, disposed when the loop ends), mirroring download_source.""" + tested async merge path. Time-boxed + self-resuming so a huge first run + finishes across chunks instead of timing out. Runs under its own asyncio + loop + per-task async engine (NullPool), mirroring download_source.""" import asyncio from ..services.tag_service import normalize_existing_tags @@ -97,8 +107,20 @@ def normalize_tags_task(self) -> dict: try: async with async_factory() as session: # normalize_existing_tags commits per group internally. - return await normalize_existing_tags(session, dry_run=False) + return await normalize_existing_tags( + session, dry_run=False, + time_budget_seconds=_NORMALIZE_CHUNK_SECONDS, + ) finally: await async_engine.dispose() - return asyncio.run(_run()) + summary = asyncio.run(_run()) + # More groups to canonicalize than fit this chunk — continue in the next. + if summary.get("partial") and summary.get("remaining", 0) > 0: + log.info( + "normalize_tags_task chunk done (%d processed, %d remaining) — " + "re-enqueuing to continue", + summary.get("groups_processed", 0), summary["remaining"], + ) + normalize_tags_task.delay() + return summary diff --git a/frontend/src/components/settings/TagMaintenanceCard.vue b/frontend/src/components/settings/TagMaintenanceCard.vue index 2232074..b1d996b 100644 --- a/frontend/src/components/settings/TagMaintenanceCard.vue +++ b/frontend/src/components/settings/TagMaintenanceCard.vue @@ -185,13 +185,9 @@ :loading="normCommitting" @click="onNormCommit" >Standardize {{ normPreview.total_changes }} tag group(s) - - - + + Queued ✓ — runs in the background (you can leave this page). It + processes in chunks, so re-run “Preview” later to confirm it's all done. @@ -289,13 +285,12 @@ async function onNormCommit() { normCommitting.value = true normResult.value = null try { - // Long op (FK repoints): enqueue the maintenance task, then tail the - // activity dashboard until its row reaches a terminal status. (The - // per-run summary dict isn't exposed by /activity/runs, so we surface - // the terminal status — the dry-run preview is the detailed view.) - const { task_id: taskId } = await store.normalizeTags({ dryRun: false }) - const row = await store.pollTaskUntilDone(taskId) - normResult.value = row?.status || 'ok' + // Fire-and-forget: the task is time-boxed and self-resuming across chunks + // (a large back-catalog can't finish in one run), so we DON'T poll-until- + // done — that would falsely report "complete" after the first chunk. Just + // confirm it's queued; the operator can re-run Preview later to verify. + await store.normalizeTags({ dryRun: false }) + normResult.value = 'queued' normPreview.value = { total_changes: 0, tags_to_rename: 0, collisions: 0, tags_to_merge: 0, sample: [] } } finally { normCommitting.value = false diff --git a/tests/test_tag_normalize.py b/tests/test_tag_normalize.py index b9c8472..2cd247a 100644 --- a/tests/test_tag_normalize.py +++ b/tests/test_tag_normalize.py @@ -123,6 +123,29 @@ async def test_live_merges_case_variants_and_repoints_images(db): assert assoc == 3 +@pytest.mark.asyncio +async def test_time_budget_stops_partial_and_reports_remaining(db): + """A zero budget stops before the first group so a huge back-catalog can't + run the task into the Celery time limit; it reports partial/remaining so the + caller re-enqueues to continue (operator-flagged 40-min timeout 2026-06-07).""" + await _raw_tag(db, "alpha", TagKind.general) + await _raw_tag(db, "beta", TagKind.general) + + summary = await normalize_existing_tags(db, dry_run=False, time_budget_seconds=0) + + assert summary["partial"] is True + assert summary["groups_processed"] == 0 + assert summary["remaining"] == summary["total_changes"] >= 2 + # Nothing recased yet — the budget cut before any work. + assert await _count_named(db, "alpha", TagKind.general) == 1 + + # A full (unbudgeted) run finishes and reports nothing remaining. + done = await normalize_existing_tags(db, dry_run=False) + assert done["partial"] is False + assert done["remaining"] == 0 + assert await _count_named(db, "Alpha", TagKind.general) == 1 + + @pytest.mark.asyncio async def test_idempotent_second_run_is_noop(db): await _raw_tag(db, "kafka", TagKind.character)