bvandeusen
65211a3f2f
The inline INSERT…SELECT backfill in migration 0045 wrapped the table creation and a ~100 GB pass over image_record.tagger_predictions in one transaction: nothing committed until the end, it was unmonitorable, and an earlier MATERIALIZED-CTE form spilled the full 100 GB to temp on NFS. A deploy got stuck on it for ~2h with image_prediction never appearing. Split the concerns: - 0045 now creates ONLY the table + indexes (instant DDL → web boots). - New backend.app.tasks.admin.backfill_image_predictions_task copies the >= store-floor predictions from the JSON into image_prediction, batched by id window and committed per chunk: live progress, resumable (re-enqueues from the last committed id), idempotent (ON CONFLICT DO NOTHING). json_each stays in the DB executor streaming each window — no Python-side 100 GB load, no materialization. - POST /api/admin/maintenance/backfill-predictions + a Maintenance-tab card to trigger the one-time run after upgrading. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
419 lines
18 KiB
Python
419 lines
18 KiB
Python
"""FC-3k: admin destructive Celery tasks.
|
|
|
|
Two long-running ops on the maintenance queue. task_run lifecycle is
|
|
captured automatically by FC-3i signals — these tasks just return
|
|
their summary dict so it lands in task_run.metadata (via Celery's
|
|
result backend) for the dashboard to surface.
|
|
|
|
Soft/hard time limits inherit the FC-3i recovery sweep: a runaway
|
|
task gets killed and flipped to status='timeout' by
|
|
recover_stalled_task_runs.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy.exc import DBAPIError, OperationalError
|
|
|
|
from ..celery_app import celery
|
|
from ..services import cleanup_service
|
|
from ._sync_engine import sync_session_factory as _sync_session_factory
|
|
|
|
log = logging.getLogger(__name__)
|
|
IMAGES_ROOT = Path("/images")
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.delete_artist_cascade_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=1800, time_limit=2400, # 30 min / 40 min
|
|
)
|
|
def delete_artist_cascade_task(self, *, artist_id: int) -> dict:
|
|
"""Wraps cleanup_service.delete_artist_cascade. Returns the
|
|
service's summary dict for FC-3i task_run.metadata capture."""
|
|
SessionLocal = _sync_session_factory()
|
|
with SessionLocal() as session:
|
|
return cleanup_service.delete_artist_cascade(
|
|
session, artist_id=artist_id, images_root=IMAGES_ROOT,
|
|
)
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.bulk_delete_images_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=900, time_limit=1200, # 15 min / 20 min
|
|
)
|
|
def bulk_delete_images_task(self, *, image_ids: list[int]) -> dict:
|
|
"""Wraps cleanup_service.delete_images."""
|
|
SessionLocal = _sync_session_factory()
|
|
with SessionLocal() as session:
|
|
return cleanup_service.delete_images(
|
|
session, image_ids=image_ids, images_root=IMAGES_ROOT,
|
|
)
|
|
|
|
|
|
# Time-box one chunk well under the soft limit so a large archive back-catalog
|
|
# can't run the task into the Celery time limit (or hog the maintenance_long
|
|
# lane). The task re-enqueues itself with the resume cursor until the scan is
|
|
# exhausted — mirrors normalize_tags_task (operator-asked 2026-06-07: reasonable
|
|
# timeout, then re-queue so other work keeps flowing).
|
|
_REEXTRACT_CHUNK_SECONDS = 600
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.reextract_archive_attachments_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=1800, time_limit=2400, # 30 min / 40 min
|
|
)
|
|
def reextract_archive_attachments_task(self, after_id: int = 0) -> dict:
|
|
"""Wraps cleanup_service.reextract_archive_attachments (#713 part 2):
|
|
re-extract PostAttachments that are actually archives but were filed
|
|
opaquely before the magic-byte gate, and link their members to the post.
|
|
|
|
Time-boxed + self-resuming: scans attachments after ``after_id`` and, on a
|
|
chunk cut, re-enqueues from where it stopped so a big backlog finishes across
|
|
chunks instead of dying at the soft limit."""
|
|
SessionLocal = _sync_session_factory()
|
|
with SessionLocal() as session:
|
|
summary = cleanup_service.reextract_archive_attachments(
|
|
session, images_root=IMAGES_ROOT,
|
|
time_budget_seconds=_REEXTRACT_CHUNK_SECONDS, after_id=after_id,
|
|
)
|
|
# More attachments past this chunk's cursor — continue in the next.
|
|
if summary.get("partial") and summary.get("resume_after_id", 0) > after_id:
|
|
log.info(
|
|
"reextract chunk done (%d scanned, %d archives, resume after id %s) "
|
|
"— re-enqueuing to continue",
|
|
summary.get("scanned", 0), summary.get("archives", 0),
|
|
summary["resume_after_id"],
|
|
)
|
|
reextract_archive_attachments_task.delay(summary["resume_after_id"])
|
|
return summary
|
|
|
|
|
|
# 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
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.normalize_tags_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=1800, time_limit=2400, # 30 min / 40 min
|
|
)
|
|
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. 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
|
|
from ._async_session import async_session_factory
|
|
|
|
async def _run() -> dict:
|
|
# 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.
|
|
return await normalize_existing_tags(
|
|
session, dry_run=False,
|
|
time_budget_seconds=_NORMALIZE_CHUNK_SECONDS,
|
|
)
|
|
finally:
|
|
await async_engine.dispose()
|
|
|
|
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
|
|
|
|
|
|
# Time-box one rescan chunk well under the soft limit and re-enqueue from the
|
|
# cursor — scoring every post against its artist's series is O(posts) and grows
|
|
# with the library (FC-6.3). Mirrors normalize_tags_task.
|
|
_SERIES_RESCAN_CHUNK_SECONDS = 600
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.rescan_series_suggestions_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=1800, time_limit=2400, # 30 min / 40 min
|
|
)
|
|
def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict:
|
|
"""Score posts against their artist's series and write pending suggestions
|
|
(FC-6.3). Settings-gated; time-boxed + self-resuming from a post-id cursor.
|
|
Per-task async engine (NullPool) under its own asyncio loop, like normalize."""
|
|
import asyncio
|
|
|
|
from ..models import ImportSettings
|
|
from ..services.series_match_service import SeriesMatchService
|
|
from ._async_session import async_session_factory
|
|
|
|
async def _run() -> dict:
|
|
async_factory, async_engine = async_session_factory()
|
|
try:
|
|
async with async_factory() as session:
|
|
settings = await ImportSettings.load(session)
|
|
if not settings.series_suggest_enabled:
|
|
return {"skipped": "series suggestions disabled"}
|
|
threshold = settings.series_suggest_threshold
|
|
return await SeriesMatchService(session).rescan(
|
|
threshold=threshold,
|
|
time_budget_seconds=_SERIES_RESCAN_CHUNK_SECONDS,
|
|
after_post_id=after_post_id,
|
|
)
|
|
finally:
|
|
await async_engine.dispose()
|
|
|
|
summary = asyncio.run(_run())
|
|
if summary.get("partial") and summary.get("resume_after_id", 0) > after_post_id:
|
|
log.info(
|
|
"rescan_series_suggestions chunk done (%d scanned, %d suggested, "
|
|
"resume after %s) — re-enqueuing",
|
|
summary.get("scanned", 0), summary.get("suggested", 0),
|
|
summary["resume_after_id"],
|
|
)
|
|
rescan_series_suggestions_task.delay(summary["resume_after_id"])
|
|
return summary
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.prune_low_confidence_predictions_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=3600, time_limit=4200, # 60 min / 70 min
|
|
)
|
|
def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict:
|
|
"""One-time #764 backfill: drop tagger_predictions entries below the DB
|
|
store floor (ml_settings.tagger_store_floor) from existing image_record
|
|
rows, and clamp any allowlist min_confidence below the floor up to it.
|
|
|
|
The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire
|
|
near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites
|
|
each row to the new floor. Keyset by id ASC (restart-safe via after_id);
|
|
idempotent — already-pruned rows rewrite to themselves and are skipped.
|
|
Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on
|
|
image_record afterward to return the disk to the OS.
|
|
|
|
The keep predicate (confidence >= floor) mirrors Tagger.infer's store
|
|
gate so backfilled rows match what new imports store. Self-resumes on the
|
|
soft time limit (re-enqueues from the last committed id)."""
|
|
from celery.exceptions import SoftTimeLimitExceeded
|
|
from sqlalchemy import select, update
|
|
|
|
from ..models import ImageRecord, MLSettings, TagAllowlist
|
|
|
|
SessionLocal = _sync_session_factory()
|
|
scanned = 0
|
|
pruned = 0
|
|
clamped = 0
|
|
last_id = after_id
|
|
try:
|
|
with SessionLocal() as session:
|
|
floor = session.execute(
|
|
select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
|
|
).scalar_one()
|
|
# Clamp allowlist thresholds below the new floor once, on the
|
|
# first pass (#764 consumer #4) — a sub-floor min_confidence can't
|
|
# apply more permissively now that nothing below it is stored.
|
|
if after_id == 0:
|
|
clamped = session.execute(
|
|
update(TagAllowlist)
|
|
.where(TagAllowlist.min_confidence < floor)
|
|
.values(min_confidence=floor)
|
|
).rowcount or 0
|
|
session.commit()
|
|
|
|
while True:
|
|
rows = session.execute(
|
|
select(ImageRecord.id, ImageRecord.tagger_predictions)
|
|
.where(ImageRecord.id > last_id)
|
|
.where(ImageRecord.tagger_predictions.is_not(None))
|
|
.order_by(ImageRecord.id.asc())
|
|
.limit(500)
|
|
).all()
|
|
if not rows:
|
|
break
|
|
for image_id, preds in rows:
|
|
scanned += 1
|
|
if not preds:
|
|
continue
|
|
kept = {
|
|
name: p for name, p in preds.items()
|
|
if float(p.get("confidence", 0.0)) >= floor
|
|
}
|
|
if len(kept) != len(preds):
|
|
session.execute(
|
|
update(ImageRecord)
|
|
.where(ImageRecord.id == image_id)
|
|
.values(tagger_predictions=kept)
|
|
)
|
|
pruned += 1
|
|
session.commit()
|
|
last_id = rows[-1].id # advance only after commit, for resume
|
|
except SoftTimeLimitExceeded:
|
|
log.warning(
|
|
"prune_low_confidence_predictions soft-limited at id=%s "
|
|
"(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned,
|
|
)
|
|
prune_low_confidence_predictions_task.delay(last_id)
|
|
return {
|
|
"partial": True, "last_id": last_id,
|
|
"scanned": scanned, "pruned": pruned,
|
|
}
|
|
|
|
log.info(
|
|
"prune_low_confidence_predictions complete: floor=%s scanned=%d "
|
|
"pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped,
|
|
)
|
|
return {
|
|
"floor": floor, "scanned": scanned, "pruned": pruned,
|
|
"allowlist_clamped": clamped, "last_id": last_id,
|
|
}
|
|
|
|
|
|
# Backfill image_prediction from image_record.tagger_predictions (#768).
|
|
# Deliberately NOT done in migration 0045: a single INSERT…SELECT over the
|
|
# ~100 GB TOAST is one transaction — invisible until commit, unmonitorable, and
|
|
# the MATERIALIZED-CTE form spilled the whole 100 GB to temp on NFS. Instead we
|
|
# walk image_record in id WINDOWS, running a bounded INSERT…SELECT over each
|
|
# window and committing per chunk: progress is visible (image_prediction grows
|
|
# live), it's resumable (re-enqueues from the last committed id), and json_each
|
|
# stays in the DB executor streaming each window (no Python-side 100 GB load, no
|
|
# materialization). Idempotent via ON CONFLICT DO NOTHING.
|
|
_BACKFILL_PRED_CHUNK_SECONDS = 600 # re-enqueue boundary, like normalize_tags
|
|
_BACKFILL_PRED_ID_WINDOW = 2000 # image_record ids per committed batch
|
|
|
|
|
|
@celery.task(
|
|
name="backend.app.tasks.admin.backfill_image_predictions_task",
|
|
bind=True,
|
|
autoretry_for=(OperationalError, DBAPIError),
|
|
retry_backoff=15, retry_backoff_max=180, max_retries=1,
|
|
soft_time_limit=1800, time_limit=2400, # 30 min / 40 min
|
|
)
|
|
def backfill_image_predictions_task(self, after_id: int = 0) -> dict:
|
|
"""One-time #768 backfill: copy each image_record's stored tagger
|
|
predictions (the >= store-floor entries) from the tagger_predictions JSON
|
|
into the normalized image_prediction table.
|
|
|
|
Batched by id window + committed per chunk so it's monitorable and
|
|
resumable; idempotent (ON CONFLICT DO NOTHING) so re-running is safe.
|
|
Filters to >= ml_settings.tagger_store_floor (default 0.70) so the table
|
|
stays small even from the full pre-prune JSON tail. Guards json_each against
|
|
non-object rows (scalar/null tagger_predictions → "cannot deconstruct a
|
|
scalar") via an inline CASE. Self-resumes on the soft time limit."""
|
|
import time
|
|
|
|
from celery.exceptions import SoftTimeLimitExceeded
|
|
from sqlalchemy import func, select, text
|
|
|
|
from ..models import ImageRecord, MLSettings
|
|
|
|
_INSERT_WINDOW = text(
|
|
"""
|
|
INSERT INTO image_prediction (image_record_id, raw_name, category, score)
|
|
SELECT ir.id,
|
|
je.key,
|
|
COALESCE(je.value ->> 'category', 'general'),
|
|
(je.value ->> 'confidence')::double precision
|
|
FROM image_record ir,
|
|
json_each(
|
|
CASE WHEN json_typeof(ir.tagger_predictions) = 'object'
|
|
THEN ir.tagger_predictions
|
|
ELSE '{}'::json END
|
|
) je
|
|
WHERE ir.id > :lo AND ir.id <= :hi
|
|
AND je.value ->> 'confidence' IS NOT NULL
|
|
AND (je.value ->> 'confidence')::double precision >= :floor
|
|
ON CONFLICT (image_record_id, raw_name) DO NOTHING
|
|
"""
|
|
)
|
|
|
|
SessionLocal = _sync_session_factory()
|
|
started = time.monotonic()
|
|
last_id = after_id
|
|
inserted = 0
|
|
windows = 0
|
|
with SessionLocal() as session:
|
|
floor = session.execute(
|
|
select(MLSettings.tagger_store_floor).where(MLSettings.id == 1)
|
|
).scalar_one()
|
|
max_id = session.execute(
|
|
select(func.max(ImageRecord.id))
|
|
).scalar() or 0
|
|
|
|
try:
|
|
while last_id < max_id:
|
|
hi = last_id + _BACKFILL_PRED_ID_WINDOW
|
|
res = session.execute(
|
|
_INSERT_WINDOW, {"lo": last_id, "hi": hi, "floor": floor}
|
|
)
|
|
session.commit()
|
|
inserted += res.rowcount or 0
|
|
windows += 1
|
|
last_id = hi # advance only after commit, for resume
|
|
if time.monotonic() - started > _BACKFILL_PRED_CHUNK_SECONDS:
|
|
log.info(
|
|
"backfill_image_predictions chunk done (windows=%d "
|
|
"inserted=%d up to id=%d/%d) — re-enqueuing",
|
|
windows, inserted, min(last_id, max_id), max_id,
|
|
)
|
|
backfill_image_predictions_task.delay(last_id)
|
|
return {
|
|
"partial": True, "last_id": last_id, "max_id": max_id,
|
|
"inserted": inserted, "windows": windows,
|
|
}
|
|
except SoftTimeLimitExceeded:
|
|
log.warning(
|
|
"backfill_image_predictions soft-limited at id=%d "
|
|
"(inserted=%d) — re-enqueuing", last_id, inserted,
|
|
)
|
|
backfill_image_predictions_task.delay(last_id)
|
|
return {
|
|
"partial": True, "last_id": last_id, "max_id": max_id,
|
|
"inserted": inserted, "windows": windows,
|
|
}
|
|
|
|
log.info(
|
|
"backfill_image_predictions complete: floor=%s inserted=%d windows=%d "
|
|
"max_id=%d", floor, inserted, windows, max_id,
|
|
)
|
|
return {
|
|
"floor": floor, "inserted": inserted, "windows": windows,
|
|
"max_id": max_id, "last_id": max_id,
|
|
}
|