diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py index c33121a..3f77d9a 100644 --- a/backend/app/api/admin.py +++ b/backend/app/api/admin.py @@ -348,3 +348,15 @@ async def trigger_reextract_archives(): async_result = reextract_archive_attachments_task.delay() return jsonify({"task_id": async_result.id, "status": "queued"}), 202 + + +@admin_bp.route("/maintenance/prune-predictions", methods=["POST"]) +async def trigger_prune_predictions(): + """Operator-triggered #764 backfill: drop stored tagger predictions below + the current ml_settings.tagger_store_floor and clamp allowlist thresholds + up to it. Shrinks image_record's TOAST (~100 GB of sub-0.70 scores). + Idempotent + self-resuming; runs on the maintenance_long lane.""" + from ..tasks.admin import prune_low_confidence_predictions_task + + async_result = prune_low_confidence_predictions_task.delay() + return jsonify({"task_id": async_result.id, "status": "queued"}), 202 diff --git a/backend/app/tasks/admin.py b/backend/app/tasks/admin.py index c6a1b08..7123a71 100644 --- a/backend/app/tasks/admin.py +++ b/backend/app/tasks/admin.py @@ -207,3 +207,99 @@ def rescan_series_suggestions_task(self, after_post_id: int = 0) -> dict: ) rescan_series_suggestions_task.delay(summary["resume_after_id"]) return summary + + +@celery.task( + name="backend.app.tasks.admin.prune_low_confidence_predictions_task", + bind=True, + autoretry_for=(OperationalError, DBAPIError), + retry_backoff=15, retry_backoff_max=180, max_retries=1, + soft_time_limit=3600, time_limit=4200, # 60 min / 70 min +) +def prune_low_confidence_predictions_task(self, after_id: int = 0) -> dict: + """One-time #764 backfill: drop tagger_predictions entries below the DB + store floor (ml_settings.tagger_store_floor) from existing image_record + rows, and clamp any allowlist min_confidence below the floor up to it. + + The Camie tagger emits ~10k tags; the old 0.05 floor stored the entire + near-zero tail, bloating image_record's TOAST to ~100 GB. This rewrites + each row to the new floor. Keyset by id ASC (restart-safe via after_id); + idempotent — already-pruned rows rewrite to themselves and are skipped. + Rewriting rows generates bloat, so run VACUUM FULL / pg_repack on + image_record afterward to return the disk to the OS. + + The keep predicate (confidence >= floor) mirrors Tagger.infer's store + gate so backfilled rows match what new imports store. Self-resumes on the + soft time limit (re-enqueues from the last committed id).""" + from celery.exceptions import SoftTimeLimitExceeded + from sqlalchemy import select, update + + from ..models import ImageRecord, MLSettings, TagAllowlist + + SessionLocal = _sync_session_factory() + scanned = 0 + pruned = 0 + clamped = 0 + last_id = after_id + try: + with SessionLocal() as session: + floor = session.execute( + select(MLSettings.tagger_store_floor).where(MLSettings.id == 1) + ).scalar_one() + # Clamp allowlist thresholds below the new floor once, on the + # first pass (#764 consumer #4) — a sub-floor min_confidence can't + # apply more permissively now that nothing below it is stored. + if after_id == 0: + clamped = session.execute( + update(TagAllowlist) + .where(TagAllowlist.min_confidence < floor) + .values(min_confidence=floor) + ).rowcount or 0 + session.commit() + + while True: + rows = session.execute( + select(ImageRecord.id, ImageRecord.tagger_predictions) + .where(ImageRecord.id > last_id) + .where(ImageRecord.tagger_predictions.is_not(None)) + .order_by(ImageRecord.id.asc()) + .limit(500) + ).all() + if not rows: + break + for image_id, preds in rows: + scanned += 1 + if not preds: + continue + kept = { + name: p for name, p in preds.items() + if float(p.get("confidence", 0.0)) >= floor + } + if len(kept) != len(preds): + session.execute( + update(ImageRecord) + .where(ImageRecord.id == image_id) + .values(tagger_predictions=kept) + ) + pruned += 1 + session.commit() + last_id = rows[-1].id # advance only after commit, for resume + except SoftTimeLimitExceeded: + log.warning( + "prune_low_confidence_predictions soft-limited at id=%s " + "(scanned=%d pruned=%d) — re-enqueuing", last_id, scanned, pruned, + ) + prune_low_confidence_predictions_task.delay(last_id) + return { + "partial": True, "last_id": last_id, + "scanned": scanned, "pruned": pruned, + } + + log.info( + "prune_low_confidence_predictions complete: floor=%s scanned=%d " + "pruned=%d allowlist_clamped=%d", floor, scanned, pruned, clamped, + ) + return { + "floor": floor, "scanned": scanned, "pruned": pruned, + "allowlist_clamped": clamped, "last_id": last_id, + } diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 76b85c5..69761d1 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -12,6 +12,7 @@ + @@ -31,6 +32,7 @@ import MLBackfillCard from './MLBackfillCard.vue' import CentroidRecomputeCard from './CentroidRecomputeCard.vue' import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue' import MLThresholdSliders from './MLThresholdSliders.vue' +import PrunePredictionsCard from './PrunePredictionsCard.vue' import AllowlistTable from './AllowlistTable.vue' import AliasTable from './AliasTable.vue' import DbMaintenanceCard from './DbMaintenanceCard.vue' diff --git a/frontend/src/components/settings/PrunePredictionsCard.vue b/frontend/src/components/settings/PrunePredictionsCard.vue new file mode 100644 index 0000000..f3af5dc --- /dev/null +++ b/frontend/src/components/settings/PrunePredictionsCard.vue @@ -0,0 +1,58 @@ + + + diff --git a/tests/test_tasks_admin.py b/tests/test_tasks_admin.py index 948dd39..d975259 100644 --- a/tests/test_tasks_admin.py +++ b/tests/test_tasks_admin.py @@ -42,6 +42,68 @@ def test_bulk_delete_images_task_registered(): ) +def test_prune_low_confidence_predictions_task_registered(): + assert ( + "backend.app.tasks.admin.prune_low_confidence_predictions_task" + in celery.tasks + ) + + +@pytest.mark.asyncio +async def test_prune_low_confidence_predictions(db_sync, tmp_path): + # #764: drop stored tagger predictions below the store floor (default + # 0.70) and clamp allowlist thresholds up to it. + from backend.app.models import Tag, TagAllowlist, TagKind + from backend.app.tasks.admin import prune_low_confidence_predictions_task + + f0 = tmp_path / "p0.jpg" + f0.write_bytes(b"x") + img0 = ImageRecord( + path=str(f0), sha256=f"{0:064x}", size_bytes=10, mime="image/jpeg", + origin="imported_filesystem", + tagger_predictions={ + "keep_high": {"category": "general", "confidence": 0.92}, + "keep_edge": {"category": "general", "confidence": 0.70}, + "drop_mid": {"category": "general", "confidence": 0.40}, + "drop_tiny": {"category": "general", "confidence": 0.06}, + }, + ) + db_sync.add(img0) + f1 = tmp_path / "p1.jpg" + f1.write_bytes(b"x") + img1 = ImageRecord( + path=str(f1), sha256=f"{1:064x}", size_bytes=10, mime="image/jpeg", + origin="imported_filesystem", + tagger_predictions={"only": {"category": "general", "confidence": 0.99}}, + ) + db_sync.add(img1) + tag = Tag(name="lowthr-tag", kind=TagKind.general) + db_sync.add(tag) + db_sync.flush() + db_sync.add(TagAllowlist(tag_id=tag.id, min_confidence=0.30)) + db_sync.commit() + img0_id, img1_id, tag_id = img0.id, img1.id, tag.id + + result = prune_low_confidence_predictions_task.delay().get() + assert result["floor"] == pytest.approx(0.70) + assert result["pruned"] == 1 # only img0 had sub-floor entries + assert result["allowlist_clamped"] == 1 + + db_sync.expire_all() + p0 = db_sync.execute( + select(ImageRecord.tagger_predictions).where(ImageRecord.id == img0_id) + ).scalar_one() + assert set(p0) == {"keep_high", "keep_edge"} # >=0.70 kept, <0.70 dropped + p1 = db_sync.execute( + select(ImageRecord.tagger_predictions).where(ImageRecord.id == img1_id) + ).scalar_one() + assert set(p1) == {"only"} # already clean — untouched + clamped = db_sync.execute( + select(TagAllowlist.min_confidence).where(TagAllowlist.tag_id == tag_id) + ).scalar_one() + assert clamped == pytest.approx(0.70) + + # --- delete_artist_cascade_task -------------------------------------