diff --git a/alembic/versions/0044_ml_settings_tagger_store_floor.py b/alembic/versions/0044_ml_settings_tagger_store_floor.py new file mode 100644 index 0000000..e019e36 --- /dev/null +++ b/alembic/versions/0044_ml_settings_tagger_store_floor.py @@ -0,0 +1,37 @@ +"""ml_settings.tagger_store_floor + +The ingest confidence floor below which tagger predictions are not stored, +promoted from the TAGGER_STORE_FLOOR env var to a DB-backed, UI-tunable +setting. Default 0.70 (was an env default of 0.05): the suggestion path +already filters at 0.70 and the centroid/learned path covers low-confidence +preferred tags, so the sub-0.70 tail was redundant weight — it had grown +image_record's TOAST to ~100 GB. See plan-task #764. + +Revision ID: 0044 +Revises: 0043 +Create Date: 2026-06-10 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0044" +down_revision: Union[str, None] = "0043" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "ml_settings", + sa.Column( + "tagger_store_floor", sa.Float(), + nullable=False, server_default="0.7", + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "tagger_store_floor") 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/api/ml_admin.py b/backend/app/api/ml_admin.py index 574e6f5..ef3d2fa 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -13,6 +13,7 @@ _EDITABLE = ( "suggestion_threshold_general", "centroid_similarity_threshold", "min_reference_images", + "tagger_store_floor", ) @@ -30,6 +31,7 @@ async def get_settings(): "suggestion_threshold_general": s.suggestion_threshold_general, "centroid_similarity_threshold": s.centroid_similarity_threshold, "min_reference_images": s.min_reference_images, + "tagger_store_floor": s.tagger_store_floor, "tagger_model_version": s.tagger_model_version, "embedder_model_version": s.embedder_model_version, } @@ -47,13 +49,45 @@ async def patch_settings(): s = ( await session.execute(select(MLSettings).where(MLSettings.id == 1)) ).scalar_one() + + # Merge the patch over current values, then validate the result as a + # whole — the store-floor invariant couples three fields, so they + # can't be checked one at a time. + proposed = {f: getattr(s, f) for f in _EDITABLE} for field in _EDITABLE: if field in body: - setattr(s, field, body[field]) + proposed[field] = body[field] + + err = _validate(proposed) + if err is not None: + return jsonify({"error": err}), 400 + + for field in _EDITABLE: + setattr(s, field, proposed[field]) await session.commit() return await get_settings() +def _validate(p: dict) -> str | None: + """Returns an error string if the proposed settings are invalid, else None. + + Invariant (plan-task #764): the per-category suggestion thresholds can't + drop below tagger_store_floor — nothing below the floor is stored, so a + lower threshold would silently surface nothing in that gap. The UI clamps + the sliders to the floor; this is the server-side backstop. + """ + floor = p["tagger_store_floor"] + if not (0.0 <= floor <= 1.0): + return "tagger_store_floor must be between 0 and 1" + for cat in ("character", "general"): + if p[f"suggestion_threshold_{cat}"] < floor: + return ( + f"suggestion_threshold_{cat} cannot be below tagger_store_floor " + f"({floor}) — predictions below the floor are not stored" + ) + return None + + @ml_admin_bp.route("/backfill", methods=["POST"]) async def trigger_backfill(): from ..tasks.ml import backfill diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py index 0371415..0df6e5b 100644 --- a/backend/app/api/system_activity.py +++ b/backend/app/api/system_activity.py @@ -147,6 +147,7 @@ async def list_runs(): """Paginated task_run history. Query params: queue= filter to one queue status= filter to one status (running/ok/error/timeout/retry) + task= case-insensitive substring match on task_name limit= default 50, max 200 before_id= cursor for keyset pagination @@ -161,6 +162,7 @@ async def list_runs(): queue = request.args.get("queue") status = request.args.get("status") + task = request.args.get("task") before_id_raw = request.args.get("before_id") before_id = int(before_id_raw) if before_id_raw else None @@ -170,6 +172,11 @@ async def list_runs(): stmt = stmt.where(TaskRun.queue == queue) if status: stmt = stmt.where(TaskRun.status == status) + if task: + # Task names contain literal underscores (download_source, + # vacuum_analyze) — escape LIKE wildcards so a search for + # "vacuum_analyze" doesn't treat "_" as a single-char match. + stmt = stmt.where(TaskRun.task_name.ilike(f"%{_escape_like(task)}%", escape="\\")) if before_id is not None: stmt = stmt.where(TaskRun.id < before_id) stmt = stmt.limit(limit + 1) @@ -225,6 +232,12 @@ async def list_failures(): }) +def _escape_like(value: str) -> str: + """Escape SQL LIKE/ILIKE metacharacters so user search text is matched + literally. Pairs with `escape="\\"` on the .ilike() call.""" + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + def _row_to_dict(r: TaskRun) -> dict: return { "id": r.id, diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 5c3e858..e28686d 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -28,6 +28,15 @@ class MLSettings(Base): centroid_similarity_threshold: Mapped[float] = mapped_column( Float, nullable=False, default=0.55 ) + # Ingest floor: tagger predictions below this confidence are not stored + # (tagger.Tagger.infer). Default 0.70 — the suggestion path already + # filters at 0.70 and the centroid/learned path covers low-confidence + # preferred tags, so the sub-0.70 tail is redundant weight (it had + # bloated image_record's TOAST to ~100 GB; plan-task #764). Operator- + # tunable via Settings → ML; must stay ≤ the suggestion thresholds. + tagger_store_floor: Mapped[float] = mapped_column( + Float, nullable=False, default=0.70 + ) min_reference_images: Mapped[int] = mapped_column( Integer, nullable=False, default=5 ) diff --git a/backend/app/services/backup_service.py b/backend/app/services/backup_service.py index caa7cfd..ac59b93 100644 --- a/backend/app/services/backup_service.py +++ b/backend/app/services/backup_service.py @@ -24,11 +24,17 @@ from pathlib import Path _BACKUPS_DIRNAME = "_backups" -# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The -# Celery soft limit signals the Python process; subprocess.Popen in a -# blocking syscall ignores that signal. These bound the worst case. -_DB_SUBPROCESS_TIMEOUT_S = 12 * 60 # 12 min (Celery soft is 10 min) -_IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60 # 7 hr (Celery soft is 6 hr) +# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery +# soft limit signals the Python process; subprocess.Popen in a blocking syscall +# ignores that signal, so these bound the worst case directly. Each sits just +# UNDER its task's Celery soft_time_limit so the bounded-kill (_run_bounded) is +# the primary guard and fires cleanly before Celery's soft/hard limits — which +# matters because an NFS D-state hang defeats even Celery's SIGKILL (the failure +# that wedged the maintenance lane for hours, #739). +# backup_db_task: soft=1800s / hard=2100s → 1700s +# backup_images_task: soft=21600s / hard=23400s → 21000s +_DB_SUBPROCESS_TIMEOUT_S = 1700 # ~28 min, under the 30-min DB soft limit +_IMAGES_SUBPROCESS_TIMEOUT_S = 21000 # ~5.8 hr, under the 6-hr images soft limit # Grace after SIGKILL to reap the child. If it can't be reaped in this window # (an uninterruptible NFS D-state — the failure mode that wedged the # concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we @@ -115,18 +121,21 @@ def backup_db( to persist into BackupRun. Raises on subprocess failure.""" ts = _now_ts() out_dir = _backups_dir(images_root) - sql_path = out_dir / f"fc_db_{ts}.sql" + # Custom format (-Fc): compressed (much smaller on NFS) and restored with + # pg_restore. The .dump extension marks it as non-SQL. The BackupRun field + # is still named sql_path — it's just "the db artifact path". + sql_path = out_dir / f"fc_db_{ts}.dump" # Dump to LOCAL disk first, then move the finished file to the (NFS) backups # dir. pg_dump's long phase is then a DB-socket wait + local writes — both # killable — instead of an NFS write that can hang uninterruptibly. Only the # final move touches NFS, and it's a bounded single-file step. - fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".sql") + fd, tmp_name = tempfile.mkstemp(prefix="fc_db_", suffix=".dump") os.close(fd) tmp_path = Path(tmp_name) try: _run_bounded( [ - "pg_dump", "--no-owner", "--no-acl", + "pg_dump", "--no-owner", "--no-acl", "-Fc", "-f", str(tmp_path), _libpq_url(db_url), ], _DB_SUBPROCESS_TIMEOUT_S, @@ -184,8 +193,8 @@ def backup_images( def restore_db(*, db_url: str, sql_path: Path) -> None: - """Wipe public schema, then load from .sql. Raises on subprocess - failure; partial-restore state is the caller's concern.""" + """Wipe public schema, then load from the custom-format dump. Raises on + subprocess failure; partial-restore state is the caller's concern.""" libpq = _libpq_url(db_url) subprocess.run( [ @@ -194,8 +203,9 @@ def restore_db(*, db_url: str, sql_path: Path) -> None: ], capture_output=True, check=True, timeout=120, ) + # Custom-format (-Fc) dumps are restored with pg_restore, not psql. subprocess.run( - ["psql", libpq, "-f", str(sql_path)], + ["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)], capture_output=True, check=True, timeout=_DB_SUBPROCESS_TIMEOUT_S, ) diff --git a/backend/app/services/ml/allowlist.py b/backend/app/services/ml/allowlist.py index dc8aeab..08a6ac7 100644 --- a/backend/app/services/ml/allowlist.py +++ b/backend/app/services/ml/allowlist.py @@ -9,7 +9,7 @@ from sqlalchemy import delete, select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession -from ...models import Tag, TagAllowlist, TagSuggestionRejection +from ...models import MLSettings, Tag, TagAllowlist, TagSuggestionRejection from ...models.tag import image_tag from .aliases import AliasService @@ -91,12 +91,25 @@ class AllowlistService: ) await self.dismiss(image_id, tag_id) + async def _store_floor(self) -> float: + return ( + await self.session.execute( + select(MLSettings.tagger_store_floor).where(MLSettings.id == 1) + ) + ).scalar_one() + async def update_threshold( self, tag_id: int, min_confidence: float ) -> None: row = await self.session.get(TagAllowlist, tag_id) if row is not None: - row.min_confidence = min_confidence + # An allowlist tag can't auto-apply more permissively than the + # ingest store floor — predictions below tagger_store_floor aren't + # stored, so a lower min_confidence would behave identically to the + # floor. Clamp so the stored threshold matches actual behavior + # (#764). + floor = await self._store_floor() + row.min_confidence = max(min_confidence, floor) async def remove(self, tag_id: int) -> None: await self.session.execute( diff --git a/backend/app/services/ml/tagger.py b/backend/app/services/ml/tagger.py index ddf17b9..6348595 100644 --- a/backend/app/services/ml/tagger.py +++ b/backend/app/services/ml/tagger.py @@ -33,8 +33,13 @@ _MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie" _MODEL_FILE = f"{MODEL_NAME}.onnx" _METADATA_FILE = f"{MODEL_NAME}-metadata.json" -# Below this confidence, predictions aren't stored (keeps the JSON compact). -STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05")) +# Ingest floor below which predictions aren't stored (keeps the JSON compact). +# DEFAULT/fallback only — the live value is DB-backed +# (ml_settings.tagger_store_floor) and passed into infer() per call by the ml +# task. 0.70: the suggestion path already filters there and the centroid path +# covers lower-confidence preferred tags, so the sub-0.70 tail is redundant +# (it had bloated image_record's TOAST to ~100 GB; plan-task #764). +DEFAULT_STORE_FLOOR = 0.70 # The categories FC-2b surfaces in the UI. Others (meta/rating/year) are # still stored but the suggestion service filters them out. @@ -145,10 +150,13 @@ class Tagger: arr = arr.transpose(2, 0, 1) # HWC -> CHW return arr[np.newaxis, :, :, :] # NCHW - def infer(self, image_path: Path) -> dict[str, TagPrediction]: + def infer( + self, image_path: Path, *, store_floor: float = DEFAULT_STORE_FLOOR, + ) -> dict[str, TagPrediction]: """Run Camie v2 on one image. Returns {name: TagPrediction} with - confidence >= STORE_FLOOR (across all categories — the suggestion - service does category filtering later). + confidence >= store_floor (across all categories — the suggestion + service does category filtering later). store_floor is the DB-backed + ml_settings.tagger_store_floor, passed in by the ml task. v2 emits multiple outputs; we use the refined predictions (output[1] per onnx_inference.py). Sigmoid is applied to raw @@ -167,7 +175,7 @@ class Tagger: cats = self._tag_categories for idx, score in enumerate(probs): conf = float(score) - if conf < STORE_FLOOR: + if conf < store_floor: continue if idx >= len(names): # Output longer than metadata declared — shouldn't happen but 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/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 7e81786..f53216d 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -120,6 +120,15 @@ IMPORT_BATCH_KEEP_DAYS = 30 # files); time_limit=2100. QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "ml": 25, + # download_source legitimately walks 5-25 min (Patreon/gallery-dl + # deep creators); its hard time_limit is DOWNLOAD_HARD_TIME_LIMIT + # (1500s = 25m). The 5-min default flagged healthy in-flight walks as + # phantom 'RecoverySweep' failures (System Activity showed errors the + # Subscriptions view correctly didn't — the download finished ok and + # reset the source). 30 clears the 25-min limit with buffer and lines + # up with DOWNLOAD_STALL_THRESHOLD_MINUTES (30) so a genuine hard kill + # is swept by the task-run AND event sweeps together. Audit 2026-06-10. + "download": 30, # Audit 2026-06-02 — maintenance/scan queues run tasks that # legitimately exceed the 5-min default (verify_integrity at 70m # hard, scan_directory at 70m hard, apply_allowlist_tags / diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 7703447..ff09258 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -127,7 +127,10 @@ def tag_and_embed(self, image_id: int) -> dict: phase = "video_infer" import numpy as np - preds = _maxpool_predictions([tagger.infer(f) for f in frames]) + preds = _maxpool_predictions( + [tagger.infer(f, store_floor=settings.tagger_store_floor) + for f in frames] + ) embedding = np.mean( [embedder.infer(f) for f in frames], axis=0 ).astype("float32") @@ -136,7 +139,7 @@ def tag_and_embed(self, image_id: int) -> dict: else: phase = "tag" t0 = time.monotonic() - raw = tagger.infer(src) + raw = tagger.infer(src, store_floor=settings.tagger_store_floor) log.info( "tag_and_embed tagged in %.1fs (%d tags): %s", time.monotonic() - t0, len(raw), ctx, diff --git a/frontend/src/components/settings/AllowlistTable.vue b/frontend/src/components/settings/AllowlistTable.vue index 9448125..2ac386d 100644 --- a/frontend/src/components/settings/AllowlistTable.vue +++ b/frontend/src/components/settings/AllowlistTable.vue @@ -10,7 +10,7 @@ @@ -25,23 +25,32 @@ diff --git a/frontend/src/components/settings/SystemActivityTab.vue b/frontend/src/components/settings/SystemActivityTab.vue index 65fd7b7..7342944 100644 --- a/frontend/src/components/settings/SystemActivityTab.vue +++ b/frontend/src/components/settings/SystemActivityTab.vue @@ -21,8 +21,15 @@ - - {{ failureCount }} failures + + + {{ filteredFailures.length }} / {{ failureCount }} @@ -77,6 +84,14 @@ + { const failureCount = computed(() => (store.failures?.recent || []).length) const filteredFailures = computed(() => { - const all = store.failures?.recent || [] - if (!filterErrorType.value) return all - return all.filter(r => r.error_type === filterErrorType.value) + let all = store.failures?.recent || [] + if (filterErrorType.value) { + all = all.filter(r => r.error_type === filterErrorType.value) + } + const q = (failureSearch.value || '').trim().toLowerCase() + if (q) { + all = all.filter(r => + [r.task_name, r.queue, r.target_id, r.error_type] + .some(v => String(v ?? '').toLowerCase().includes(q)), + ) + } + return all }) function toggleErrorTypeFilter(name) { @@ -240,9 +266,23 @@ function toggleErrorTypeFilter(name) { } function onFilterChange() { - store.setFilter({ queue: filterQueue.value, status: filterStatus.value }) + store.setFilter({ + queue: filterQueue.value, + status: filterStatus.value, + task: (filterTask.value || '').trim() || null, + }) store.loadRuns({ reset: true }) } + +// Server-side search hits the full task_run history, not just the loaded +// page — debounce so we don't fire a query per keystroke (matches the +// 300ms idiom in TagsView/AllowlistTable). +let taskSearchDebounce = null +function onTaskSearch() { + if (taskSearchDebounce) clearTimeout(taskSearchDebounce) + taskSearchDebounce = setTimeout(onFilterChange, 300) +} + function onLoadMore() { store.loadRuns({ reset: false }) } function onRefresh() { store.loadRuns({ reset: true }) } diff --git a/frontend/src/stores/systemActivity.js b/frontend/src/stores/systemActivity.js index 44980a6..89ebbab 100644 --- a/frontend/src/stores/systemActivity.js +++ b/frontend/src/stores/systemActivity.js @@ -16,7 +16,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => { const runs = ref([]) const runsCursor = ref(null) const runsHasMore = ref(false) - const runsFilter = ref({ queue: null, status: null, limit: 50 }) + const runsFilter = ref({ queue: null, status: null, task: null, limit: 50 }) const loading = ref({ queues: false, workers: false, runs: false, failures: false }) const lastError = ref(null) @@ -65,6 +65,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => { const params = { limit: runsFilter.value.limit } if (runsFilter.value.queue) params.queue = runsFilter.value.queue if (runsFilter.value.status) params.status = runsFilter.value.status + if (runsFilter.value.task) params.task = runsFilter.value.task if (!reset && runsCursor.value) params.before_id = runsCursor.value const body = await api.get('/api/system/activity/runs', { params }) runs.value = reset ? body.runs : [...runs.value, ...body.runs] diff --git a/tests/test_api_ml_admin.py b/tests/test_api_ml_admin.py index 1692743..76b1002 100644 --- a/tests/test_api_ml_admin.py +++ b/tests/test_api_ml_admin.py @@ -34,6 +34,28 @@ async def test_get_and_patch_settings(client): assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90) +@pytest.mark.asyncio +async def test_tagger_store_floor_default_and_patch(client): + body = await (await client.get("/api/ml/settings")).get_json() + assert body["tagger_store_floor"] == pytest.approx(0.70) + + resp = await client.patch("/api/ml/settings", json={"tagger_store_floor": 0.6}) + assert resp.status_code == 200 + assert (await resp.get_json())["tagger_store_floor"] == pytest.approx(0.6) + + +@pytest.mark.asyncio +async def test_suggestion_threshold_below_store_floor_rejected(client): + # Invariant (#764): a category threshold can't sit below the store floor — + # nothing below the floor is stored, so the gap would surface nothing. + # Floor defaults to 0.70; pushing general down to 0.50 must 400. + resp = await client.patch( + "/api/ml/settings", json={"suggestion_threshold_general": 0.50} + ) + assert resp.status_code == 400 + assert "tagger_store_floor" in (await resp.get_json())["error"] + + @pytest.mark.asyncio async def test_backfill_and_recompute_trigger(client): r1 = await client.post("/api/ml/backfill") diff --git a/tests/test_api_system_activity.py b/tests/test_api_system_activity.py index 8191a8e..1322d92 100644 --- a/tests/test_api_system_activity.py +++ b/tests/test_api_system_activity.py @@ -182,6 +182,25 @@ async def test_runs_filter_by_status(client, _seed_runs): assert len(body["runs"]) == 2 # i=3, i=4 +@pytest.mark.asyncio +async def test_runs_filter_by_task_substring(client, _seed_runs): + # Case-insensitive substring across the full task_name. + resp = await client.get("/api/system/activity/runs?task=FAKE") + body = await resp.get_json() + assert len(body["runs"]) == 5 + assert all("fake" in r["task_name"] for r in body["runs"]) + + +@pytest.mark.asyncio +async def test_runs_filter_by_task_escapes_underscore(client, _seed_runs): + # The literal "_" in "task_3" must match one row, not act as a + # single-char wildcard matching every task_N. + resp = await client.get("/api/system/activity/runs?task=task_3") + body = await resp.get_json() + assert len(body["runs"]) == 1 + assert body["runs"][0]["task_name"].endswith("task_3") + + @pytest.mark.asyncio async def test_runs_keyset_cursor(client, _seed_runs): page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json() diff --git a/tests/test_backup_service.py b/tests/test_backup_service.py index 46b1ffc..cf45fb2 100644 --- a/tests/test_backup_service.py +++ b/tests/test_backup_service.py @@ -69,7 +69,7 @@ def test_backup_db_writes_sql_and_manifest(tmp_path, fake_subprocess): ) sql = Path(result["sql_path"]) manifest = Path(result["manifest_path"]) - assert sql.is_file() and sql.suffix == ".sql" + assert sql.is_file() and sql.suffix == ".dump" assert manifest.is_file() and manifest.suffix == ".json" assert result["kind"] == "db" assert result["tar_path"] is None @@ -92,6 +92,12 @@ def test_backup_db_strips_sqlalchemy_psycopg_driver(tmp_path, fake_subprocess): assert "+psycopg" not in cmd[-1] +def test_backup_db_uses_compressed_custom_format(tmp_path, fake_subprocess): + # -Fc → pg_restore-loadable, compressed dump (#739 backup polish). + backup_service.backup_db(db_url="postgresql://u@h/d", images_root=tmp_path) + assert "-Fc" in fake_subprocess[0] + + def test_backup_db_strips_asyncpg_driver(tmp_path, fake_subprocess): backup_service.backup_db( db_url="postgresql+asyncpg://u:p@h/d", images_root=tmp_path, @@ -132,17 +138,19 @@ def test_backup_images_excludes_backups_and_quarantine(tmp_path, fake_subprocess def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess): - sql_path = tmp_path / "fake.sql" - sql_path.write_text("SELECT 1;") + dump_path = tmp_path / "fake.dump" + dump_path.write_bytes(b"\x00fake custom dump") backup_service.restore_db( - db_url="postgresql://u@h/d", sql_path=sql_path, + db_url="postgresql://u@h/d", sql_path=dump_path, ) - # Two psql calls: one with -c (DROP SCHEMA), one with -f (load). + # Two calls: psql -c (DROP SCHEMA), then pg_restore -d (load the dump). assert len(fake_subprocess) == 2 + assert fake_subprocess[0][0] == "psql" assert "-c" in fake_subprocess[0] assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1] - assert "-f" in fake_subprocess[1] - assert str(sql_path) in fake_subprocess[1] + assert fake_subprocess[1][0] == "pg_restore" + assert "-d" in fake_subprocess[1] + assert str(dump_path) in fake_subprocess[1] # --- restore_images -------------------------------------------------- diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 9873d4d..5f26ba7 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -350,6 +350,62 @@ def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync): assert ml_stale_status == "error" +def test_recover_stalled_task_runs_download_queue_uses_longer_threshold(db_sync): + """download_source legitimately walks 5-25 min (Patreon/gallery-dl). + The 5-min default flagged healthy in-flight walks as phantom + 'RecoverySweep' failures — visible in System Activity but absent from + the Subscriptions view because the download actually finished ok. + The 30-min download override (QUEUE_STUCK_THRESHOLD_MINUTES) must + protect a 10-min-old download row while still flagging a 35-min one. + Audit 2026-06-10.""" + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import recover_stalled_task_runs + + now = datetime.now(UTC) + # 10-min-old download row: stale by the default 5-min rule but fresh + # by the 30-min download override. Must survive the sweep. + dl_fresh_id = _make_task_run( + db_sync, status="running", queue="download", + task_name="backend.app.tasks.download.download_source", + started_at=now - timedelta(minutes=10), + ) + # 35-min-old download row: past even the 30-min override (a genuine + # hard kill). Must be flagged. + dl_stale_id = _make_task_run( + db_sync, status="running", queue="download", + task_name="backend.app.tasks.download.download_source", + started_at=now - timedelta(minutes=35), + ) + db_sync.commit() + + recovered = recover_stalled_task_runs.apply().get() + assert recovered == 1 + + db_sync.expire_all() + dl_fresh_status = db_sync.execute( + select(TaskRun.status).where(TaskRun.id == dl_fresh_id) + ).scalar_one() + dl_stale_status = db_sync.execute( + select(TaskRun.status).where(TaskRun.id == dl_stale_id) + ).scalar_one() + assert dl_fresh_status == "running" + assert dl_stale_status == "error" + + +def test_download_stuck_threshold_exceeds_hard_time_limit(): + """Invariant guard (maintenance.py:112): every queue override MUST be + ≥ the relevant task's hard time_limit, else the sweep flags in-flight + work. download_source is the one that regressed — pin it so a future + DOWNLOAD_HARD_TIME_LIMIT bump can't silently re-break it.""" + from backend.app.tasks.download import DOWNLOAD_HARD_TIME_LIMIT + from backend.app.tasks.maintenance import QUEUE_STUCK_THRESHOLD_MINUTES + + hard_minutes = DOWNLOAD_HARD_TIME_LIMIT / 60 + assert QUEUE_STUCK_THRESHOLD_MINUTES["download"] >= hard_minutes + + def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync): """import_archive_file shares the 'import' queue with fast single-file import_media_file, so it gets a per-task-name override diff --git a/tests/test_ml_allowlist.py b/tests/test_ml_allowlist.py index a4c94bd..ad3298f 100644 --- a/tests/test_ml_allowlist.py +++ b/tests/test_ml_allowlist.py @@ -104,3 +104,17 @@ async def test_update_threshold_and_remove(db): assert abs(row.min_confidence - 0.80) < 1e-6 await svc.remove(tag.id) assert await db.get(TagAllowlist, tag.id) is None + + +@pytest.mark.asyncio +async def test_update_threshold_clamped_to_store_floor(db): + # A min_confidence below the store floor (default 0.70) is clamped up — + # predictions below the floor aren't stored, so a lower threshold can't + # apply more permissively than the floor (#764). + tag = await TagService(db).find_or_create("Lowthr", TagKind.general) + svc = AllowlistService(db) + img = await _make_image(db) + await svc.accept(img.id, tag.id) + await svc.update_threshold(tag.id, 0.30) + row = await db.get(TagAllowlist, tag.id) + assert abs(row.min_confidence - 0.70) < 1e-6 diff --git a/tests/test_ml_tagger.py b/tests/test_ml_tagger.py index c36c64a..369ce62 100644 --- a/tests/test_ml_tagger.py +++ b/tests/test_ml_tagger.py @@ -1,14 +1,14 @@ """Tagger unit tests. The ONNX model isn't available in CI (it's a 1GB -download into /models), so these test the pure-logic surface: STORE_FLOOR -constant, SURFACED_CATEGORIES set, TagPrediction dataclass, and the -load()-missing-file error path. Full inference is exercised by the local -integration suite against a real /models volume. +download into /models), so these test the pure-logic surface: +DEFAULT_STORE_FLOOR constant, SURFACED_CATEGORIES set, TagPrediction +dataclass, and the load()-missing-file error path. Full inference is +exercised by the local integration suite against a real /models volume. """ import pytest from backend.app.services.ml.tagger import ( - STORE_FLOOR, + DEFAULT_STORE_FLOOR, SURFACED_CATEGORIES, Tagger, TagPrediction, @@ -26,8 +26,12 @@ def test_surfaced_categories(): assert "copyright" not in SURFACED_CATEGORIES -def test_store_floor_is_low(): - assert 0 < STORE_FLOOR < 0.2 +def test_default_store_floor(): + # Raised 0.05 → 0.70 (plan-task #764): the suggestion path filters at + # 0.70 and the centroid path covers lower-confidence preferred tags, so + # storing the sub-0.70 tail was redundant (100 GB of TOAST). The live + # value is DB-backed (ml_settings.tagger_store_floor); this is the default. + assert DEFAULT_STORE_FLOOR == 0.70 def test_tag_prediction_dataclass(): 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 ------------------------------------- diff --git a/tests/test_tasks_backup.py b/tests/test_tasks_backup.py index 4c285e3..583a202 100644 --- a/tests/test_tasks_backup.py +++ b/tests/test_tasks_backup.py @@ -93,7 +93,7 @@ async def test_backup_db_task_creates_backup_run_row_status_ok(db_sync): ).one() assert row.kind == "db" assert row.status == "ok" - assert row.sql_path and row.sql_path.endswith(".sql") + assert row.sql_path and row.sql_path.endswith(".dump") assert row.size_bytes is not None and row.size_bytes > 0 assert row.finished_at is not None assert row.error is None