fix(audit): chunk + self-resume library scans (stop the 2h queue-hog timeouts)

scan_library_for_rule ran one 2-hour pass that timed out on large libraries and
held the concurrency-1 maintenance queue the whole time, starving vacuum/backup/
normalize (operator-flagged — it was the dominant entry in the 24h failures).

It now runs ~10-min chunks and re-enqueues itself until the library is
exhausted, matching the operator's preferred pattern (reasonable timeout → retry
queued → other things process between). New columns (alembic 0039):
resume_after_id persists the keyset cursor so a chunk continues where the last
left off; last_progress_at lets the recovery sweep tell a progressing multi-
chunk audit from a dead one (it now measures staleness from last_progress_at,
not started_at). Matches accumulate across chunks. soft/hard limits dropped
2h→15/16.7 min so the in-chunk budget fires first; a soft-limit backstop
re-enqueues to resume instead of erroring the whole run.

Tests: time-box → re-enqueue (status stays running); resume carries prior
matches and appends new ones. Existing full-scan tests unchanged (small sets
finish in one chunk).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

alembic/versions/0039_library_audit_resume.py

@@ -0,0 +1,40 @@
+"""library_audit_run: resume cursor + progress timestamp for chunked scans
+
+Revision ID: 0039
+Revises: 0038
+Create Date: 2026-06-07
+
+scan_library_for_rule used to run one 2h pass that timed out on large libraries
+and monopolized the concurrency-1 maintenance queue (operator-flagged). It now
+runs short time-boxed chunks that re-enqueue: `resume_after_id` persists the
+keyset cursor so the next chunk continues where it left off, and
+`last_progress_at` lets the recovery sweep tell a progressing multi-chunk audit
+from a genuinely stuck one.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0039"
+down_revision: Union[str, None] = "0038"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "library_audit_run",
+        sa.Column(
+            "resume_after_id", sa.Integer, nullable=False, server_default="0"
+        ),
+    )
+    op.add_column(
+        "library_audit_run",
+        sa.Column("last_progress_at", sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("library_audit_run", "last_progress_at")
+    op.drop_column("library_audit_run", "resume_after_id")

backend/app/models/library_audit_run.py

@@ -35,3 +35,10 @@ class LibraryAuditRun(Base):
     matched_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     matched_ids: Mapped[list[int]] = mapped_column(JSONB, nullable=False, default=list)
     error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # Chunked-scan state (alembic 0039): keyset cursor the next chunk resumes
+    # from, and the last time a chunk made progress (so the recovery sweep can
+    # tell a progressing multi-chunk audit from a stuck one).
+    resume_after_id: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    last_progress_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True,
+    )

backend/app/tasks/library_audit.py

@@ -13,6 +13,7 @@ State machine:
 """
 
 import logging
+import time
 import traceback
 from datetime import UTC, datetime
 
@@ -31,6 +32,12 @@ log = logging.getLogger(__name__)
 _BATCH = 500
 _PROGRESS_TICK = 100
 _MAX_MATCHED = 50_000
+# One chunk's wall-clock budget. Was a single 2h pass that timed out on large
+# libraries and held the concurrency-1 maintenance queue the whole time
+# (operator-flagged 2026-06-07). Now: scan ~10 min, persist the keyset cursor +
+# matches, re-enqueue to continue — so backups/vacuum/normalize chunks can
+# interleave. soft/hard limits sit just above so the budget fires first.
+_CHUNK_SECONDS = 600
 
 _RULES = {
     "transparency": transparency.evaluate,
@@ -46,13 +53,16 @@ _RULES = {
     retry_backoff_max=60,
     retry_jitter=True,
     max_retries=3,
-    soft_time_limit=7200,
-    time_limit=7500,
+    soft_time_limit=900,
+    time_limit=1000,
 )
 def scan_library_for_rule(self, audit_id: int) -> dict:
-    """See module docstring. Returns a small summary dict for eager-mode
+    """See module docstring. Time-boxed + self-resuming: one call scans a
+    ~10-min chunk, persists the resume cursor + matches, and re-enqueues itself
+    until the library is exhausted. Returns a small summary dict for eager-mode
     test assertions (real workers ignore the return value)."""
     SessionLocal = _sync_session_factory()
+    start = time.monotonic()
     try:
         with SessionLocal() as session:
             audit = session.get(LibraryAuditRun, audit_id)
@@ -63,9 +73,10 @@ def scan_library_for_rule(self, audit_id: int) -> dict:
                 _mark_error(session, audit_id, f"unknown rule {audit.rule!r}")
                 return {"audit_id": audit_id, "status": "error"}
             params = dict(audit.params or {})
-            matched: list[int] = []
-            scanned = 0
-            last_id = 0
+            # Resume from the previous chunk's persisted state.
+            matched: list[int] = list(audit.matched_ids or [])
+            scanned = audit.scanned_count or 0
+            last_id = audit.resume_after_id or 0
             while True:
                 # Cancellation check between batches.
                 current_status = session.execute(
@@ -74,6 +85,15 @@ def scan_library_for_rule(self, audit_id: int) -> dict:
                 ).scalar_one()
                 if current_status == "cancelled":
                     return {"audit_id": audit_id, "status": "cancelled"}
+                # Time-box: persist the cursor + matches and re-enqueue so the
+                # queue is freed between chunks. The next call resumes here.
+                if time.monotonic() - start >= _CHUNK_SECONDS:
+                    _persist_chunk(session, audit_id, scanned, matched, last_id)
+                    scan_library_for_rule.delay(audit_id)
+                    return {
+                        "audit_id": audit_id, "status": "running",
+                        "partial": True, "scanned": scanned,
+                    }
                 rows = session.execute(
                     select(ImageRecord.id, ImageRecord.path)
                     .where(ImageRecord.id > last_id)
@@ -114,10 +134,16 @@ def scan_library_for_rule(self, audit_id: int) -> dict:
                         )
                         return {"audit_id": audit_id, "status": "error"}
                     if scanned % _PROGRESS_TICK == 0:
+                        # Cheap heartbeat: scanned_count + last_progress_at so the
+                        # recovery sweep sees the multi-chunk audit is alive. The
+                        # cursor + matches are persisted at chunk boundaries.
                         session.execute(
                             update(LibraryAuditRun)
                             .where(LibraryAuditRun.id == audit_id)
-                            .values(scanned_count=scanned)
+                            .values(
+                                scanned_count=scanned,
+                                last_progress_at=datetime.now(UTC),
+                            )
                         )
                         session.commit()
             # Final state.
@@ -128,8 +154,10 @@ def scan_library_for_rule(self, audit_id: int) -> dict:
                     scanned_count=scanned,
                     matched_count=len(matched),
                     matched_ids=matched,
+                    resume_after_id=last_id,
                     status="ready",
                     finished_at=datetime.now(UTC),
+                    last_progress_at=datetime.now(UTC),
                 )
             )
             session.commit()
@@ -140,9 +168,14 @@ def scan_library_for_rule(self, audit_id: int) -> dict:
                 "matched": len(matched),
             }
     except SoftTimeLimitExceeded:
-        with SessionLocal() as session:
-            _mark_error(session, audit_id, "soft_time_limit exceeded (>7200s)")
-        raise
+        # Backstop (the in-chunk budget should fire first): the audit stays
+        # 'running' with its last committed cursor; re-enqueue to continue from
+        # there rather than marking the whole run an error.
+        log.warning(
+            "audit %s: soft time limit hit — re-enqueuing to resume", audit_id,
+        )
+        scan_library_for_rule.delay(audit_id)
+        return {"audit_id": audit_id, "status": "running", "partial": True}
     except (OperationalError, DBAPIError):
         # Retryable per the decorator; leave row in 'running' and let
         # autoretry try again. Recovery sweep catches if all retries fail.
@@ -154,6 +187,23 @@ def scan_library_for_rule(self, audit_id: int) -> dict:
         raise
 
 
+def _persist_chunk(session, audit_id, scanned, matched, last_id) -> None:
+    """Persist a chunk boundary: scanned count, matches so far, and the keyset
+    cursor the next chunk resumes from. Keeps status='running'."""
+    session.execute(
+        update(LibraryAuditRun)
+        .where(LibraryAuditRun.id == audit_id)
+        .values(
+            scanned_count=scanned,
+            matched_count=len(matched),
+            matched_ids=list(matched),
+            resume_after_id=last_id,
+            last_progress_at=datetime.now(UTC),
+        )
+    )
+    session.commit()
+
+
 def _mark_error(session, audit_id: int, error_msg: str) -> None:
     session.execute(
         update(LibraryAuditRun)

backend/app/tasks/maintenance.py

@@ -647,19 +647,30 @@ def recover_stalled_library_audit_runs() -> int:
     guard in start_audit_run — a SIGKILL'd run would block all future
     audits until manual DB surgery. (The guard is now age-aware, but
     this sweep is what makes that work in practice.)
+
+    Measures staleness from last_progress_at (alembic 0039), NOT started_at:
+    a chunked scan stays 'running' across many re-enqueued chunks and can
+    legitimately run for hours on a big library — only flag one that hasn't
+    made progress in the threshold window (a dead chunk that never re-enqueued).
+    Falls back to started_at for pre-0039 / never-ticked rows.
     """
     SessionLocal = _sync_session_factory()
     now = datetime.now(UTC)
     cutoff = now - timedelta(minutes=LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES)
     msg = (
-        f"stranded by recovery sweep (no terminal status after "
+        f"stranded by recovery sweep (no progress for "
         f"{LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES} min)"
     )
     with SessionLocal() as session:
         result = session.execute(
             update(LibraryAuditRun)
             .where(LibraryAuditRun.status == "running")
-            .where(LibraryAuditRun.started_at < cutoff)
+            .where(
+                func.coalesce(
+                    LibraryAuditRun.last_progress_at,
+                    LibraryAuditRun.started_at,
+                ) < cutoff
+            )
             .values(status="error", finished_at=now, error=msg)
         )
         session.commit()

tests/test_tasks_library_audit.py

@@ -69,6 +69,60 @@ def test_scan_library_for_rule_populates_matched_ids_for_transparency(
     assert status == "ready"
 
 
+def test_scan_time_boxes_and_reenqueues(db_sync, tmp_path, monkeypatch):
+    """A zero chunk budget stops before scanning and re-enqueues to continue,
+    leaving the audit 'running' — so a huge library can't run the task into the
+    Celery limit or hog the maintenance queue (operator-flagged 2026-06-07)."""
+    _mk_image(db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="a.png")
+    audit = LibraryAuditRun(
+        rule="transparency", params={"threshold": 0.5},
+        status="running", matched_ids=[],
+    )
+    db_sync.add(audit)
+    db_sync.commit()
+    audit_id = audit.id
+
+    from backend.app.tasks import library_audit as la
+    delays = []
+    monkeypatch.setattr(la.scan_library_for_rule, "delay", lambda aid: delays.append(aid))
+    monkeypatch.setattr(la, "_CHUNK_SECONDS", 0)  # time-box on the first iteration
+    la.scan_library_for_rule.run(audit_id)
+
+    status = db_sync.execute(
+        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
+    ).scalar_one()
+    assert status == "running"      # not finished — handed off to the next chunk
+    assert delays == [audit_id]     # re-enqueued itself
+
+
+def test_scan_resumes_and_accumulates_matched(db_sync, tmp_path, monkeypatch):
+    """A later chunk keeps the prior chunks' matches and appends its own."""
+    trans, _ = _mk_image(
+        db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="t.png",
+    )
+    audit = LibraryAuditRun(
+        rule="transparency", params={"threshold": 0.5},
+        status="running", matched_ids=[424242], resume_after_id=0,
+    )
+    db_sync.add(audit)
+    db_sync.commit()
+    audit_id = audit.id
+
+    from backend.app.tasks import library_audit as la
+    monkeypatch.setattr(la.scan_library_for_rule, "delay", lambda aid: None)
+    la.scan_library_for_rule.run(audit_id)
+
+    matched = db_sync.execute(
+        select(LibraryAuditRun.matched_ids).where(LibraryAuditRun.id == audit_id)
+    ).scalar_one()
+    status = db_sync.execute(
+        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
+    ).scalar_one()
+    assert status == "ready"
+    assert 424242 in matched     # carried over from a prior chunk
+    assert trans.id in matched   # found this chunk
+
+
 def test_scan_library_for_rule_skips_missing_files_gracefully(
     db_sync, tmp_path, monkeypatch,
 ):