fix(tags): time-box + self-resume the tag standardization (stop the 40-min timeout)
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -185,13 +185,9 @@
|
||||
:loading="normCommitting"
|
||||
@click="onNormCommit"
|
||||
>Standardize {{ normPreview.total_changes }} tag group(s)</v-btn>
|
||||
<span
|
||||
v-if="normResult"
|
||||
class="ml-3 text-caption"
|
||||
:class="normResult === 'ok' ? 'text-success' : 'text-warning'"
|
||||
>
|
||||
<template v-if="normResult === 'ok'">Standardization complete ✓</template>
|
||||
<template v-else>Finished with status: {{ normResult }}</template>
|
||||
<span v-if="normResult === 'queued'" class="ml-3 text-caption text-success">
|
||||
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.
|
||||
</span>
|
||||
</div>
|
||||
</v-card-text>
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user