diff --git a/alembic/versions/0039_library_audit_resume.py b/alembic/versions/0039_library_audit_resume.py new file mode 100644 index 0000000..6cfb8f9 --- /dev/null +++ b/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") diff --git a/backend/app/models/library_audit_run.py b/backend/app/models/library_audit_run.py index ca5aa82..a2d4bc2 100644 --- a/backend/app/models/library_audit_run.py +++ b/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, + ) diff --git a/backend/app/tasks/library_audit.py b/backend/app/tasks/library_audit.py index dd66a23..6f2b26c 100644 --- a/backend/app/tasks/library_audit.py +++ b/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) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 2be4a26..807d0e6 100644 --- a/backend/app/tasks/maintenance.py +++ b/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() diff --git a/tests/test_tasks_library_audit.py b/tests/test_tasks_library_audit.py index b10c35b..f26d528 100644 --- a/tests/test_tasks_library_audit.py +++ b/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, ):