From a73d9327d8c23b181f29b1d72ad371ace0b6f867 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 7 Jun 2026 12:03:13 -0400 Subject: [PATCH] fix(maintenance): time-box + self-resume the archive re-extract task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/services/cleanup_service.py | 38 +++++++++++++-- backend/app/tasks/admin.py | 29 +++++++++-- tests/test_reextract_archives.py | 64 +++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 6 deletions(-) diff --git a/backend/app/services/cleanup_service.py b/backend/app/services/cleanup_service.py index 605c215..05c55c8 100644 --- a/backend/app/services/cleanup_service.py +++ b/backend/app/services/cleanup_service.py @@ -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: diff --git a/backend/app/tasks/admin.py b/backend/app/tasks/admin.py index 9a65f1e..4942c26 100644 --- a/backend/app/tasks/admin.py +++ b/backend/app/tasks/admin.py @@ -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 diff --git a/tests/test_reextract_archives.py b/tests/test_reextract_archives.py index b16a1c9..9c6eb47 100644 --- a/tests/test_reextract_archives.py +++ b/tests/test_reextract_archives.py @@ -89,3 +89,67 @@ def test_reextract_links_archive_members_to_post(db_sync, tmp_path, monkeypatch) ) assert again["members_imported"] == 0 assert db_sync.execute(select(ImageRecord)).scalars().all() == images + + +def test_reextract_timebox_resumes_from_cursor(db_sync, tmp_path, monkeypatch): + """A 0-second budget cuts the chunk after the first attachment and reports a + resume cursor; the next run starts strictly after it and finishes the rest.""" + from backend.app.tasks import ml as ml_mod + from backend.app.tasks import thumbnail as thumb_mod + + monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda *a, **k: None) + monkeypatch.setattr(ml_mod.tag_and_embed, "delay", lambda *a, **k: None) + + images_root = tmp_path / "images" + images_root.mkdir() + + artist = Artist(name="Bob", slug="bob") + db_sync.add(artist) + db_sync.flush() + source = Source( + artist_id=artist.id, platform="patreon", + url="https://patreon.com/bob", enabled=True, config_overrides={}, + ) + db_sync.add(source) + db_sync.flush() + post = Post( + source_id=source.id, artist_id=artist.id, external_post_id="42", + post_url="https://www.patreon.com/posts/42", + ) + db_sync.add(post) + db_sync.flush() + + store_dir = images_root / "attachments" / "two" + store_dir.mkdir(parents=True) + att_ids = [] + for n, color in enumerate(("red", "blue")): + arc = store_dir / f"{n}_archive_{color}" + with zipfile.ZipFile(arc, "w") as zf: + zf.writestr(f"{color}.jpg", _jpeg(color)) + sha = hashlib.sha256(arc.read_bytes()).hexdigest() + att = PostAttachment( + post_id=post.id, artist_id=artist.id, sha256=sha, path=str(arc), + original_filename=arc.name, ext="", size_bytes=arc.stat().st_size, + ) + db_sync.add(att) + db_sync.flush() + att_ids.append(att.id) + db_sync.commit() + + # Budget 0 → break right after the first attachment commits. + first = cleanup_service.reextract_archive_attachments( + db_sync, images_root=images_root, time_budget_seconds=0.0, + ) + assert first["partial"] is True + assert first["scanned"] == 1 + assert first["members_imported"] == 1 + assert first["resume_after_id"] == att_ids[0] + + # Resume strictly after the cursor — picks up the second, then runs dry. + second = cleanup_service.reextract_archive_attachments( + db_sync, images_root=images_root, after_id=att_ids[0], + ) + assert second["partial"] is False + assert second["scanned"] == 1 + assert second["members_imported"] == 1 + assert len(db_sync.execute(select(ImageRecord)).scalars().all()) == 2