fix(maintenance): time-box + self-resume the archive re-extract task
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 17s
CI / backend-lint-and-test (push) Successful in 25s
CI / integration (push) Failing after 3m3s

reextract_archive_attachments loaded ALL PostAttachments and ran in one pass up
to a 30-min soft limit, then died without re-enqueueing — a large archive
backlog would only ever partially process. And a naive re-run can't advance: an
already-extracted archive is still an archive on disk, so it'd re-extract the
same first batch forever.

Give it a real cursor + time-box + self-resume (mirrors normalize_tags_task,
operator-asked 2026-06-07: reasonable timeout, then re-queue so other work keeps
flowing):
- service scans attachments with id > after_id in ascending order, time-boxes
  the chunk, and reports partial=True + resume_after_id (last scanned id).
- task passes a 600s budget and re-enqueues itself from the cursor until the
  scan is exhausted. Routes on the maintenance_long lane.
- This is independent of the maintenance_long lane isolation (already shipped) —
  that stops long tasks starving the quick maintenance queue; this stops the
  re-extract itself dying on a big backlog.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-07 12:03:13 -04:00
parent 5201fab088
commit a73d9327d8
3 changed files with 125 additions and 6 deletions
+35 -3
View File
@@ -12,6 +12,7 @@ the one-and-done GS/IR migration tooling.)
from __future__ import annotations
import logging
import time
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
@@ -723,7 +724,13 @@ def _reextract_archive_to_post(
sidecar_path.unlink(missing_ok=True)
def reextract_archive_attachments(session: Session, *, images_root: Path) -> dict:
def reextract_archive_attachments(
session: Session,
*,
images_root: Path,
time_budget_seconds: float | None = None,
after_id: int = 0,
) -> dict:
"""Re-process existing PostAttachments that are ACTUALLY archives but were
filed opaquely before #713 part 1 (extension-only is_archive missed mangled /
extension-less Patreon attachment names). For each: extract the members,
@@ -731,6 +738,14 @@ def reextract_archive_attachments(session: Session, *, images_root: Path) -> dic
Idempotent — members dedupe by sha256, the archive dedupes by sha — so it's
safe to run repeatedly. Returns a summary dict for task_run.metadata.
Time-boxed + resumable: scans PostAttachments in ascending id order starting
after ``after_id``. When ``time_budget_seconds`` elapses, stops and reports
``partial=True`` + ``resume_after_id`` (the last scanned id) so the task can
re-enqueue itself and continue — a large archive back-catalog can't run the
task into the Celery time limit or hog the maintenance lane. A bare re-run
(after_id=0) would never advance because an already-extracted archive is
still an archive on disk, so the cursor is what guarantees forward progress.
"""
from ..models import ImportSettings, Post, PostAttachment, Source
from ..tasks.ml import tag_and_embed
@@ -742,7 +757,7 @@ def reextract_archive_attachments(session: Session, *, images_root: Path) -> dic
summary = {
"scanned": 0, "archives": 0, "members_imported": 0,
"posts_touched": 0, "skipped_no_post": 0, "skipped_no_artist": 0,
"errors": 0,
"errors": 0, "partial": False, "resume_after_id": after_id,
}
settings = ImportSettings.load_sync(session)
importer = Importer(
@@ -751,11 +766,15 @@ def reextract_archive_attachments(session: Session, *, images_root: Path) -> dic
)
attachments = session.execute(
select(PostAttachment).order_by(PostAttachment.id)
select(PostAttachment)
.where(PostAttachment.id > after_id)
.order_by(PostAttachment.id)
).scalars().all()
enqueue_ids: list[int] = []
start = time.monotonic()
for att in attachments:
summary["scanned"] += 1
summary["resume_after_id"] = att.id
stored = Path(att.path)
try:
if not stored.is_file() or not is_archive(stored):
@@ -795,6 +814,19 @@ def reextract_archive_attachments(session: Session, *, images_root: Path) -> dic
summary["posts_touched"] += 1
enqueue_ids.extend(ids)
# Time-box the chunk. resume_after_id already points at this attachment,
# so the next run starts strictly after it. Checked after the commit so a
# half-extracted archive never straddles the boundary.
if (
time_budget_seconds is not None
and time.monotonic() - start >= time_budget_seconds
):
summary["partial"] = True
break
else:
# Loop ran to exhaustion — nothing left to resume.
summary["partial"] = False
# Thumbnails + ML for the newly-imported members (best-effort; off the
# critical path — a Redis hiccup must not fail the whole re-extract).
for img_id in enqueue_ids:
+26 -3
View File
@@ -57,6 +57,14 @@ def bulk_delete_images_task(self, *, image_ids: list[int]) -> dict:
)
# 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,
@@ -64,15 +72,30 @@ def bulk_delete_images_task(self, *, image_ids: list[int]) -> dict:
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) -> dict:
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."""
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:
return cleanup_service.reextract_archive_attachments(
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