From 6ed2021ad6d145302d583dfca17445219b9180cf Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 26 May 2026 01:21:37 -0400 Subject: [PATCH] =?UTF-8?q?feat(fc-cleanup):=20scan=5Flibrary=5Ffor=5Frule?= =?UTF-8?q?=20Celery=20task=20+=20maintenance-queue=20registration=20?= =?UTF-8?q?=E2=80=94=20Co-Authored-By:=20Claude=20Opus=204.7=20(1M=20conte?= =?UTF-8?q?xt)=20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/celery_app.py | 2 + backend/app/tasks/library_audit.py | 167 +++++++++++++++++++++++++++++ tests/test_tasks_library_audit.py | 98 +++++++++++++++++ 3 files changed, 267 insertions(+) create mode 100644 backend/app/tasks/library_audit.py create mode 100644 tests/test_tasks_library_audit.py diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 6ba84cb..3068ca2 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -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, diff --git a/backend/app/tasks/library_audit.py b/backend/app/tasks/library_audit.py new file mode 100644 index 0000000..dd66a23 --- /dev/null +++ b/backend/app/tasks/library_audit.py @@ -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() diff --git a/tests/test_tasks_library_audit.py b/tests/test_tasks_library_audit.py new file mode 100644 index 0000000..3599dfc --- /dev/null +++ b/tests/test_tasks_library_audit.py @@ -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