feat(fc-cleanup): scan_library_for_rule Celery task + maintenance-queue registration — Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

2026-05-26 01:21:37 -04:00
parent 4f2ceaaf31
commit 6ed2021ad6
3 changed files with 267 additions and 0 deletions
@@ -33,6 +33,7 @@ def make_celery() -> Celery:
            "backend.app.tasks.download",
            "backend.app.tasks.backup",
            "backend.app.tasks.admin",
+            "backend.app.tasks.library_audit",
        ],
    )
    app.conf.update(
@@ -47,6 +48,7 @@ def make_celery() -> Celery:
            "backend.app.tasks.migration.*": {"queue": "maintenance"},
            "backend.app.tasks.backup.*": {"queue": "maintenance"},
            "backend.app.tasks.admin.*": {"queue": "maintenance"},
+            "backend.app.tasks.library_audit.*": {"queue": "maintenance"},
        },
        # Heavy ML tasks need fair dispatch — see ImageRepo's precedent.
        task_acks_late=True,
@@ -0,0 +1,167 @@
+"""scan_library_for_rule Celery task — iterates image_record in keyset-
+paginated batches, evaluates the audit rule per image, populates
+LibraryAuditRun.matched_ids. Runs on the maintenance queue with a 2h soft
+time limit (plenty of margin for 100k+ image libraries at ~100ms PIL
+decode + histogram per image).
+
+State machine:
+  start: status='running'
+  end success: status='ready'
+  end error: status='error', error=traceback
+  oversize: status='error', error='matched too many images; tighten threshold'
+  external cancel: scan sees status='cancelled' between batches, exits.
+"""
+
+import logging
+import traceback
+from datetime import UTC, datetime
+
+from celery.exceptions import SoftTimeLimitExceeded
+from PIL import Image
+from sqlalchemy import select, update
+from sqlalchemy.exc import DBAPIError, OperationalError
+
+from ..celery_app import celery
+from ..models import ImageRecord, LibraryAuditRun
+from ..services.audits import single_color, transparency
+from ._sync_engine import sync_session_factory as _sync_session_factory
+
+log = logging.getLogger(__name__)
+
+_BATCH = 500
+_PROGRESS_TICK = 100
+_MAX_MATCHED = 50_000
+
+_RULES = {
+    "transparency": transparency.evaluate,
+    "single_color": single_color.evaluate,
+}
+
+
+@celery.task(
+    name="backend.app.tasks.library_audit.scan_library_for_rule",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=7200,
+    time_limit=7500,
+)
+def scan_library_for_rule(self, audit_id: int) -> dict:
+    """See module docstring. Returns a small summary dict for eager-mode
+    test assertions (real workers ignore the return value)."""
+    SessionLocal = _sync_session_factory()
+    try:
+        with SessionLocal() as session:
+            audit = session.get(LibraryAuditRun, audit_id)
+            if audit is None:
+                return {"audit_id": audit_id, "status": "missing"}
+            evaluate = _RULES.get(audit.rule)
+            if evaluate is None:
+                _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
+            while True:
+                # Cancellation check between batches.
+                current_status = session.execute(
+                    select(LibraryAuditRun.status)
+                    .where(LibraryAuditRun.id == audit_id)
+                ).scalar_one()
+                if current_status == "cancelled":
+                    return {"audit_id": audit_id, "status": "cancelled"}
+                rows = session.execute(
+                    select(ImageRecord.id, ImageRecord.path)
+                    .where(ImageRecord.id > last_id)
+                    .where(ImageRecord.mime.like("image/%"))
+                    .order_by(ImageRecord.id.asc())
+                    .limit(_BATCH)
+                ).all()
+                if not rows:
+                    break
+                for image_id, image_path in rows:
+                    last_id = image_id
+                    scanned += 1
+                    try:
+                        with Image.open(image_path) as im:
+                            try:
+                                if evaluate(im, **params):
+                                    matched.append(image_id)
+                            except Exception as exc:  # noqa: BLE001
+                                log.warning(
+                                    "audit %s: rule evaluate failed on %s: %s",
+                                    audit_id, image_path, exc,
+                                )
+                    except FileNotFoundError:
+                        log.warning(
+                            "audit %s: image_record %s file missing at %s; skipping",
+                            audit_id, image_id, image_path,
+                        )
+                    except OSError as exc:
+                        log.warning(
+                            "audit %s: PIL load failed for %s: %s",
+                            audit_id, image_path, exc,
+                        )
+                    if len(matched) > _MAX_MATCHED:
+                        _mark_error(
+                            session, audit_id,
+                            f"matched > {_MAX_MATCHED} images; "
+                            "tighten threshold and re-run",
+                        )
+                        return {"audit_id": audit_id, "status": "error"}
+                    if scanned % _PROGRESS_TICK == 0:
+                        session.execute(
+                            update(LibraryAuditRun)
+                            .where(LibraryAuditRun.id == audit_id)
+                            .values(scanned_count=scanned)
+                        )
+                        session.commit()
+            # Final state.
+            session.execute(
+                update(LibraryAuditRun)
+                .where(LibraryAuditRun.id == audit_id)
+                .values(
+                    scanned_count=scanned,
+                    matched_count=len(matched),
+                    matched_ids=matched,
+                    status="ready",
+                    finished_at=datetime.now(UTC),
+                )
+            )
+            session.commit()
+            return {
+                "audit_id": audit_id,
+                "status": "ready",
+                "scanned": scanned,
+                "matched": len(matched),
+            }
+    except SoftTimeLimitExceeded:
+        with SessionLocal() as session:
+            _mark_error(session, audit_id, "soft_time_limit exceeded (>7200s)")
+        raise
+    except (OperationalError, DBAPIError):
+        # Retryable per the decorator; leave row in 'running' and let
+        # autoretry try again. Recovery sweep catches if all retries fail.
+        raise
+    except Exception:  # noqa: BLE001
+        tb = traceback.format_exc()
+        with SessionLocal() as session:
+            _mark_error(session, audit_id, tb)
+        raise
+
+
+def _mark_error(session, audit_id: int, error_msg: str) -> None:
+    session.execute(
+        update(LibraryAuditRun)
+        .where(LibraryAuditRun.id == audit_id)
+        .values(
+            status="error",
+            error=error_msg,
+            finished_at=datetime.now(UTC),
+        )
+    )
+    session.commit()
@@ -0,0 +1,98 @@
+"""Tests for scan_library_for_rule Celery task.
+
+Eager mode is used so the task runs synchronously in-test and we can
+assert state via column selects (post-DML ORM access banned per
+reference-async-coredml-test-assertions)."""
+
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from backend.app import celery_app
+from backend.app.models import ImageRecord, LibraryAuditRun
+
+import backend.app.tasks.library_audit  # noqa: F401 — registration
+
+pytestmark = pytest.mark.integration
+
+
+def _mk_image(db_sync, tmp_path, *, mode, color, name):
+    path = tmp_path / name
+    Image.new(mode, (10, 10), color).save(path)
+    short_sha = f"fake-{name}"
+    rec = ImageRecord(
+        path=str(path),
+        sha256=short_sha + "0" * (64 - len(short_sha)),
+        size_bytes=path.stat().st_size,
+        mime="image/png",
+        width=10, height=10,
+        origin="imported_filesystem",
+        integrity_status="ok",
+    )
+    db_sync.add(rec)
+    db_sync.flush()
+    return rec, path
+
+
+def test_scan_library_for_rule_populates_matched_ids_for_transparency(
+    db_sync, tmp_path, monkeypatch,
+):
+    transparent_rec, _ = _mk_image(
+        db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="trans.png",
+    )
+    opaque_rec, _ = _mk_image(
+        db_sync, tmp_path, mode="RGBA", color=(200, 0, 0, 255), name="opaque.png",
+    )
+    audit = LibraryAuditRun(
+        rule="transparency", params={"threshold": 0.5},
+        status="running", matched_ids=[],
+    )
+    db_sync.add(audit)
+    db_sync.commit()
+    audit_id = audit.id
+
+    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", True)
+    from backend.app.tasks.library_audit import scan_library_for_rule
+    scan_library_for_rule.run(audit_id)
+    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", False)
+
+    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 transparent_rec.id in matched
+    assert opaque_rec.id not in matched
+    assert status == "ready"
+
+
+def test_scan_library_for_rule_skips_missing_files_gracefully(
+    db_sync, tmp_path, monkeypatch,
+):
+    rec, path = _mk_image(
+        db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="ghost.png",
+    )
+    path.unlink()  # delete the file but leave the DB row
+    audit = LibraryAuditRun(
+        rule="transparency", params={"threshold": 0.5},
+        status="running", matched_ids=[],
+    )
+    db_sync.add(audit)
+    db_sync.commit()
+    audit_id = audit.id
+
+    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", True)
+    from backend.app.tasks.library_audit import scan_library_for_rule
+    scan_library_for_rule.run(audit_id)
+    monkeypatch.setattr(celery_app.celery.conf, "task_always_eager", False)
+
+    status = db_sync.execute(
+        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
+    ).scalar_one()
+    matched = db_sync.execute(
+        select(LibraryAuditRun.matched_ids).where(LibraryAuditRun.id == audit_id)
+    ).scalar_one()
+    # Missing file is skipped (warning logged), audit completes successfully.
+    assert status == "ready"
+    assert rec.id not in matched