@@ -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")
|
||||||
@@ -348,3 +348,15 @@ async def trigger_reextract_archives():
|
|||||||
|
|
||||||
async_result = reextract_archive_attachments_task.delay()
|
async_result = reextract_archive_attachments_task.delay()
|
||||||
return jsonify({"task_id": async_result.id, "status": "queued"}), 202
|
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
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ _EDITABLE = (
|
|||||||
"suggestion_threshold_general",
|
"suggestion_threshold_general",
|
||||||
"centroid_similarity_threshold",
|
"centroid_similarity_threshold",
|
||||||
"min_reference_images",
|
"min_reference_images",
|
||||||
|
"tagger_store_floor",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -30,6 +31,7 @@ async def get_settings():
|
|||||||
"suggestion_threshold_general": s.suggestion_threshold_general,
|
"suggestion_threshold_general": s.suggestion_threshold_general,
|
||||||
"centroid_similarity_threshold": s.centroid_similarity_threshold,
|
"centroid_similarity_threshold": s.centroid_similarity_threshold,
|
||||||
"min_reference_images": s.min_reference_images,
|
"min_reference_images": s.min_reference_images,
|
||||||
|
"tagger_store_floor": s.tagger_store_floor,
|
||||||
"tagger_model_version": s.tagger_model_version,
|
"tagger_model_version": s.tagger_model_version,
|
||||||
"embedder_model_version": s.embedder_model_version,
|
"embedder_model_version": s.embedder_model_version,
|
||||||
}
|
}
|
||||||
@@ -47,13 +49,45 @@ async def patch_settings():
|
|||||||
s = (
|
s = (
|
||||||
await session.execute(select(MLSettings).where(MLSettings.id == 1))
|
await session.execute(select(MLSettings).where(MLSettings.id == 1))
|
||||||
).scalar_one()
|
).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:
|
for field in _EDITABLE:
|
||||||
if field in body:
|
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()
|
await session.commit()
|
||||||
return await get_settings()
|
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"])
|
@ml_admin_bp.route("/backfill", methods=["POST"])
|
||||||
async def trigger_backfill():
|
async def trigger_backfill():
|
||||||
from ..tasks.ml import backfill
|
from ..tasks.ml import backfill
|
||||||
|
|||||||
@@ -147,6 +147,7 @@ async def list_runs():
|
|||||||
"""Paginated task_run history. Query params:
|
"""Paginated task_run history. Query params:
|
||||||
queue=<name> filter to one queue
|
queue=<name> filter to one queue
|
||||||
status=<status> filter to one status (running/ok/error/timeout/retry)
|
status=<status> filter to one status (running/ok/error/timeout/retry)
|
||||||
|
task=<substr> case-insensitive substring match on task_name
|
||||||
limit=<int> default 50, max 200
|
limit=<int> default 50, max 200
|
||||||
before_id=<int> cursor for keyset pagination
|
before_id=<int> cursor for keyset pagination
|
||||||
|
|
||||||
@@ -161,6 +162,7 @@ async def list_runs():
|
|||||||
|
|
||||||
queue = request.args.get("queue")
|
queue = request.args.get("queue")
|
||||||
status = request.args.get("status")
|
status = request.args.get("status")
|
||||||
|
task = request.args.get("task")
|
||||||
before_id_raw = request.args.get("before_id")
|
before_id_raw = request.args.get("before_id")
|
||||||
before_id = int(before_id_raw) if before_id_raw else None
|
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)
|
stmt = stmt.where(TaskRun.queue == queue)
|
||||||
if status:
|
if status:
|
||||||
stmt = stmt.where(TaskRun.status == 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:
|
if before_id is not None:
|
||||||
stmt = stmt.where(TaskRun.id < before_id)
|
stmt = stmt.where(TaskRun.id < before_id)
|
||||||
stmt = stmt.limit(limit + 1)
|
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:
|
def _row_to_dict(r: TaskRun) -> dict:
|
||||||
return {
|
return {
|
||||||
"id": r.id,
|
"id": r.id,
|
||||||
|
|||||||
@@ -28,6 +28,15 @@ class MLSettings(Base):
|
|||||||
centroid_similarity_threshold: Mapped[float] = mapped_column(
|
centroid_similarity_threshold: Mapped[float] = mapped_column(
|
||||||
Float, nullable=False, default=0.55
|
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(
|
min_reference_images: Mapped[int] = mapped_column(
|
||||||
Integer, nullable=False, default=5
|
Integer, nullable=False, default=5
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -24,11 +24,17 @@ from pathlib import Path
|
|||||||
|
|
||||||
_BACKUPS_DIRNAME = "_backups"
|
_BACKUPS_DIRNAME = "_backups"
|
||||||
|
|
||||||
# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The
|
# Subprocess-level guardrails BEYOND the Celery soft_time_limit. The Celery
|
||||||
# Celery soft limit signals the Python process; subprocess.Popen in a
|
# soft limit signals the Python process; subprocess.Popen in a blocking syscall
|
||||||
# blocking syscall ignores that signal. These bound the worst case.
|
# ignores that signal, so these bound the worst case directly. Each sits just
|
||||||
_DB_SUBPROCESS_TIMEOUT_S = 12 * 60 # 12 min (Celery soft is 10 min)
|
# UNDER its task's Celery soft_time_limit so the bounded-kill (_run_bounded) is
|
||||||
_IMAGES_SUBPROCESS_TIMEOUT_S = 7 * 60 * 60 # 7 hr (Celery soft is 6 hr)
|
# 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
|
# 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
|
# (an uninterruptible NFS D-state — the failure mode that wedged the
|
||||||
# concurrency-1 maintenance lane for hours, operator-flagged 2026-06-07), we
|
# 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."""
|
to persist into BackupRun. Raises on subprocess failure."""
|
||||||
ts = _now_ts()
|
ts = _now_ts()
|
||||||
out_dir = _backups_dir(images_root)
|
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
|
# 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
|
# 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
|
# killable — instead of an NFS write that can hang uninterruptibly. Only the
|
||||||
# final move touches NFS, and it's a bounded single-file step.
|
# 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)
|
os.close(fd)
|
||||||
tmp_path = Path(tmp_name)
|
tmp_path = Path(tmp_name)
|
||||||
try:
|
try:
|
||||||
_run_bounded(
|
_run_bounded(
|
||||||
[
|
[
|
||||||
"pg_dump", "--no-owner", "--no-acl",
|
"pg_dump", "--no-owner", "--no-acl", "-Fc",
|
||||||
"-f", str(tmp_path), _libpq_url(db_url),
|
"-f", str(tmp_path), _libpq_url(db_url),
|
||||||
],
|
],
|
||||||
_DB_SUBPROCESS_TIMEOUT_S,
|
_DB_SUBPROCESS_TIMEOUT_S,
|
||||||
@@ -184,8 +193,8 @@ def backup_images(
|
|||||||
|
|
||||||
|
|
||||||
def restore_db(*, db_url: str, sql_path: Path) -> None:
|
def restore_db(*, db_url: str, sql_path: Path) -> None:
|
||||||
"""Wipe public schema, then load from .sql. Raises on subprocess
|
"""Wipe public schema, then load from the custom-format dump. Raises on
|
||||||
failure; partial-restore state is the caller's concern."""
|
subprocess failure; partial-restore state is the caller's concern."""
|
||||||
libpq = _libpq_url(db_url)
|
libpq = _libpq_url(db_url)
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
[
|
[
|
||||||
@@ -194,8 +203,9 @@ def restore_db(*, db_url: str, sql_path: Path) -> None:
|
|||||||
],
|
],
|
||||||
capture_output=True, check=True, timeout=120,
|
capture_output=True, check=True, timeout=120,
|
||||||
)
|
)
|
||||||
|
# Custom-format (-Fc) dumps are restored with pg_restore, not psql.
|
||||||
subprocess.run(
|
subprocess.run(
|
||||||
["psql", libpq, "-f", str(sql_path)],
|
["pg_restore", "--no-owner", "--no-acl", "-d", libpq, str(sql_path)],
|
||||||
capture_output=True, check=True,
|
capture_output=True, check=True,
|
||||||
timeout=_DB_SUBPROCESS_TIMEOUT_S,
|
timeout=_DB_SUBPROCESS_TIMEOUT_S,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from sqlalchemy import delete, select
|
|||||||
from sqlalchemy.dialects.postgresql import insert
|
from sqlalchemy.dialects.postgresql import insert
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
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 ...models.tag import image_tag
|
||||||
from .aliases import AliasService
|
from .aliases import AliasService
|
||||||
|
|
||||||
@@ -91,12 +91,25 @@ class AllowlistService:
|
|||||||
)
|
)
|
||||||
await self.dismiss(image_id, tag_id)
|
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(
|
async def update_threshold(
|
||||||
self, tag_id: int, min_confidence: float
|
self, tag_id: int, min_confidence: float
|
||||||
) -> None:
|
) -> None:
|
||||||
row = await self.session.get(TagAllowlist, tag_id)
|
row = await self.session.get(TagAllowlist, tag_id)
|
||||||
if row is not None:
|
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:
|
async def remove(self, tag_id: int) -> None:
|
||||||
await self.session.execute(
|
await self.session.execute(
|
||||||
|
|||||||
@@ -33,8 +33,13 @@ _MODEL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "camie"
|
|||||||
_MODEL_FILE = f"{MODEL_NAME}.onnx"
|
_MODEL_FILE = f"{MODEL_NAME}.onnx"
|
||||||
_METADATA_FILE = f"{MODEL_NAME}-metadata.json"
|
_METADATA_FILE = f"{MODEL_NAME}-metadata.json"
|
||||||
|
|
||||||
# Below this confidence, predictions aren't stored (keeps the JSON compact).
|
# Ingest floor below which predictions aren't stored (keeps the JSON compact).
|
||||||
STORE_FLOOR = float(os.environ.get("TAGGER_STORE_FLOOR", "0.05"))
|
# 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
|
# The categories FC-2b surfaces in the UI. Others (meta/rating/year) are
|
||||||
# still stored but the suggestion service filters them out.
|
# still stored but the suggestion service filters them out.
|
||||||
@@ -145,10 +150,13 @@ class Tagger:
|
|||||||
arr = arr.transpose(2, 0, 1) # HWC -> CHW
|
arr = arr.transpose(2, 0, 1) # HWC -> CHW
|
||||||
return arr[np.newaxis, :, :, :] # NCHW
|
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
|
"""Run Camie v2 on one image. Returns {name: TagPrediction} with
|
||||||
confidence >= STORE_FLOOR (across all categories — the suggestion
|
confidence >= store_floor (across all categories — the suggestion
|
||||||
service does category filtering later).
|
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
|
v2 emits multiple outputs; we use the refined predictions
|
||||||
(output[1] per onnx_inference.py). Sigmoid is applied to raw
|
(output[1] per onnx_inference.py). Sigmoid is applied to raw
|
||||||
@@ -167,7 +175,7 @@ class Tagger:
|
|||||||
cats = self._tag_categories
|
cats = self._tag_categories
|
||||||
for idx, score in enumerate(probs):
|
for idx, score in enumerate(probs):
|
||||||
conf = float(score)
|
conf = float(score)
|
||||||
if conf < STORE_FLOOR:
|
if conf < store_floor:
|
||||||
continue
|
continue
|
||||||
if idx >= len(names):
|
if idx >= len(names):
|
||||||
# Output longer than metadata declared — shouldn't happen but
|
# Output longer than metadata declared — shouldn't happen but
|
||||||
|
|||||||
@@ -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"])
|
rescan_series_suggestions_task.delay(summary["resume_after_id"])
|
||||||
return summary
|
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,
|
||||||
|
}
|
||||||
|
|||||||
@@ -120,6 +120,15 @@ IMPORT_BATCH_KEEP_DAYS = 30
|
|||||||
# files); time_limit=2100.
|
# files); time_limit=2100.
|
||||||
QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
|
QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
|
||||||
"ml": 25,
|
"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
|
# Audit 2026-06-02 — maintenance/scan queues run tasks that
|
||||||
# legitimately exceed the 5-min default (verify_integrity at 70m
|
# legitimately exceed the 5-min default (verify_integrity at 70m
|
||||||
# hard, scan_directory at 70m hard, apply_allowlist_tags /
|
# hard, scan_directory at 70m hard, apply_allowlist_tags /
|
||||||
|
|||||||
@@ -127,7 +127,10 @@ def tag_and_embed(self, image_id: int) -> dict:
|
|||||||
phase = "video_infer"
|
phase = "video_infer"
|
||||||
import numpy as np
|
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(
|
embedding = np.mean(
|
||||||
[embedder.infer(f) for f in frames], axis=0
|
[embedder.infer(f) for f in frames], axis=0
|
||||||
).astype("float32")
|
).astype("float32")
|
||||||
@@ -136,7 +139,7 @@ def tag_and_embed(self, image_id: int) -> dict:
|
|||||||
else:
|
else:
|
||||||
phase = "tag"
|
phase = "tag"
|
||||||
t0 = time.monotonic()
|
t0 = time.monotonic()
|
||||||
raw = tagger.infer(src)
|
raw = tagger.infer(src, store_floor=settings.tagger_store_floor)
|
||||||
log.info(
|
log.info(
|
||||||
"tag_and_embed tagged in %.1fs (%d tags): %s",
|
"tag_and_embed tagged in %.1fs (%d tags): %s",
|
||||||
time.monotonic() - t0, len(raw), ctx,
|
time.monotonic() - t0, len(raw), ctx,
|
||||||
|
|||||||
@@ -10,7 +10,7 @@
|
|||||||
<v-text-field
|
<v-text-field
|
||||||
:model-value="item.min_confidence" type="number"
|
:model-value="item.min_confidence" type="number"
|
||||||
density="compact" hide-details style="max-width: 100px;"
|
density="compact" hide-details style="max-width: 100px;"
|
||||||
min="0.05" max="1" step="0.05"
|
:min="floor" max="1" step="0.05"
|
||||||
@update:model-value="(v) => onThreshold(item.tag_id, v)"
|
@update:model-value="(v) => onThreshold(item.tag_id, v)"
|
||||||
/>
|
/>
|
||||||
</template>
|
</template>
|
||||||
@@ -25,23 +25,32 @@
|
|||||||
</template>
|
</template>
|
||||||
|
|
||||||
<script setup>
|
<script setup>
|
||||||
import { onMounted } from 'vue'
|
import { computed, onMounted } from 'vue'
|
||||||
import { useAllowlistStore } from '../../stores/allowlist.js'
|
import { useAllowlistStore } from '../../stores/allowlist.js'
|
||||||
|
import { useMLStore } from '../../stores/ml.js'
|
||||||
|
|
||||||
const store = useAllowlistStore()
|
const store = useAllowlistStore()
|
||||||
|
const ml = useMLStore()
|
||||||
|
// min_confidence can't be set below the tagger store floor — predictions
|
||||||
|
// below it aren't stored, so a lower threshold would behave identically to
|
||||||
|
// the floor. The backend clamps too (#764).
|
||||||
|
const floor = computed(() => ml.settings?.tagger_store_floor ?? 0.70)
|
||||||
const headers = [
|
const headers = [
|
||||||
{ title: 'Tag', key: 'tag_name', sortable: true },
|
{ title: 'Tag', key: 'tag_name', sortable: true },
|
||||||
{ title: 'Kind', key: 'tag_kind', sortable: true, width: 110 },
|
{ title: 'Kind', key: 'tag_kind', sortable: true, width: 110 },
|
||||||
{ title: 'Min confidence', key: 'min_confidence', sortable: false, width: 140 },
|
{ title: 'Min confidence', key: 'min_confidence', sortable: false, width: 140 },
|
||||||
{ title: '', key: 'actions', sortable: false, width: 60 }
|
{ title: '', key: 'actions', sortable: false, width: 60 }
|
||||||
]
|
]
|
||||||
onMounted(() => store.load())
|
onMounted(() => {
|
||||||
|
store.load()
|
||||||
|
if (!ml.settings) ml.loadSettings()
|
||||||
|
})
|
||||||
|
|
||||||
let debounce = null
|
let debounce = null
|
||||||
function onThreshold(tagId, value) {
|
function onThreshold(tagId, value) {
|
||||||
if (debounce) clearTimeout(debounce)
|
if (debounce) clearTimeout(debounce)
|
||||||
debounce = setTimeout(() => {
|
debounce = setTimeout(() => {
|
||||||
const v = parseFloat(value)
|
const v = Math.max(parseFloat(value), floor.value)
|
||||||
if (v > 0 && v <= 1) store.updateThreshold(tagId, v)
|
if (v > 0 && v <= 1) store.updateThreshold(tagId, v)
|
||||||
}, 500)
|
}, 500)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,9 +6,28 @@
|
|||||||
<v-col cols="12">
|
<v-col cols="12">
|
||||||
<v-slider
|
<v-slider
|
||||||
v-model="local[f.key]" :label="f.label"
|
v-model="local[f.key]" :label="f.label"
|
||||||
|
:min="f.floorMin ? local.tagger_store_floor : 0" max="1" step="0.05"
|
||||||
|
thumb-label hide-details
|
||||||
|
color="accent" @end="save"
|
||||||
|
/>
|
||||||
|
</v-col>
|
||||||
|
</v-row>
|
||||||
|
|
||||||
|
<v-divider class="my-4" />
|
||||||
|
|
||||||
|
<v-row>
|
||||||
|
<v-col cols="12">
|
||||||
|
<v-slider
|
||||||
|
v-model="local.tagger_store_floor" label="Tagger store floor"
|
||||||
min="0" max="1" step="0.05" thumb-label hide-details
|
min="0" max="1" step="0.05" thumb-label hide-details
|
||||||
color="accent" @end="save"
|
color="accent" @end="save"
|
||||||
/>
|
/>
|
||||||
|
<div class="text-caption fc-muted mt-1">
|
||||||
|
Tagger predictions below this confidence aren't stored — raising it
|
||||||
|
keeps the image library lean. Suggestions can't be shown below the
|
||||||
|
floor; lower-confidence tags you actually want still surface through
|
||||||
|
the learned centroid path.
|
||||||
|
</div>
|
||||||
</v-col>
|
</v-col>
|
||||||
</v-row>
|
</v-row>
|
||||||
</v-card-text>
|
</v-card-text>
|
||||||
@@ -24,17 +43,25 @@ import { useMLStore } from '../../stores/ml.js'
|
|||||||
const store = useMLStore()
|
const store = useMLStore()
|
||||||
// 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired as
|
// 'artist' (FC-2d-vii-c) and 'copyright' (2026-06-01) retired as
|
||||||
// suggestion categories; their threshold rows are gone.
|
// suggestion categories; their threshold rows are gone.
|
||||||
|
// floorMin: the per-category suggestion thresholds can't drop below the
|
||||||
|
// tagger store floor (nothing below the floor is stored to surface).
|
||||||
const fields = [
|
const fields = [
|
||||||
{ key: 'suggestion_threshold_character', label: 'Character' },
|
{ key: 'suggestion_threshold_character', label: 'Character', floorMin: true },
|
||||||
{ key: 'suggestion_threshold_general', label: 'General' },
|
{ key: 'suggestion_threshold_general', label: 'General', floorMin: true },
|
||||||
{ key: 'centroid_similarity_threshold', label: 'Centroid similarity' }
|
{ key: 'centroid_similarity_threshold', label: 'Centroid similarity' }
|
||||||
]
|
]
|
||||||
const local = reactive({})
|
const local = reactive({})
|
||||||
watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true })
|
watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true })
|
||||||
|
|
||||||
async function save() {
|
async function save() {
|
||||||
|
// Mirror the server invariant: keep the category thresholds at or above the
|
||||||
|
// store floor so a raised floor doesn't leave a threshold stranded below it.
|
||||||
|
const floor = local.tagger_store_floor
|
||||||
|
local.suggestion_threshold_character = Math.max(local.suggestion_threshold_character, floor)
|
||||||
|
local.suggestion_threshold_general = Math.max(local.suggestion_threshold_general, floor)
|
||||||
const patch = {}
|
const patch = {}
|
||||||
for (const f of fields) patch[f.key] = local[f.key]
|
for (const f of fields) patch[f.key] = local[f.key]
|
||||||
|
patch.tagger_store_floor = local.tagger_store_floor
|
||||||
try { await store.patchSettings(patch) }
|
try { await store.patchSettings(patch) }
|
||||||
catch (e) { toast({ text: e.message, type: 'error' }) }
|
catch (e) { toast({ text: e.message, type: 'error' }) }
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,6 +12,7 @@
|
|||||||
<ThumbnailBackfillCard />
|
<ThumbnailBackfillCard />
|
||||||
</div>
|
</div>
|
||||||
<MLThresholdSliders class="mt-4" />
|
<MLThresholdSliders class="mt-4" />
|
||||||
|
<PrunePredictionsCard class="mt-4" />
|
||||||
<AllowlistTable class="mt-4" />
|
<AllowlistTable class="mt-4" />
|
||||||
<AliasTable class="mt-4" />
|
<AliasTable class="mt-4" />
|
||||||
<DbMaintenanceCard class="mt-6" />
|
<DbMaintenanceCard class="mt-6" />
|
||||||
@@ -31,6 +32,7 @@ import MLBackfillCard from './MLBackfillCard.vue'
|
|||||||
import CentroidRecomputeCard from './CentroidRecomputeCard.vue'
|
import CentroidRecomputeCard from './CentroidRecomputeCard.vue'
|
||||||
import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
|
import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue'
|
||||||
import MLThresholdSliders from './MLThresholdSliders.vue'
|
import MLThresholdSliders from './MLThresholdSliders.vue'
|
||||||
|
import PrunePredictionsCard from './PrunePredictionsCard.vue'
|
||||||
import AllowlistTable from './AllowlistTable.vue'
|
import AllowlistTable from './AllowlistTable.vue'
|
||||||
import AliasTable from './AliasTable.vue'
|
import AliasTable from './AliasTable.vue'
|
||||||
import DbMaintenanceCard from './DbMaintenanceCard.vue'
|
import DbMaintenanceCard from './DbMaintenanceCard.vue'
|
||||||
|
|||||||
@@ -0,0 +1,58 @@
|
|||||||
|
<template>
|
||||||
|
<!-- #764: drop stored tagger predictions below the store floor to shrink
|
||||||
|
image_record's TOAST (the sub-0.70 score tail had grown it to ~100 GB). -->
|
||||||
|
<v-card>
|
||||||
|
<v-card-title>Prune low-confidence predictions</v-card-title>
|
||||||
|
<v-card-text>
|
||||||
|
<p class="text-body-2 mb-3">
|
||||||
|
Removes stored tagger predictions below the current store floor
|
||||||
|
(<strong>{{ floorPct }}</strong>) from every image, and clamps any
|
||||||
|
allowlist threshold below the floor up to it. This is what shrinks the
|
||||||
|
database — the low-confidence tail was the bulk of its size. Idempotent
|
||||||
|
and resumable; safe to run more than once. Afterward, reclaim the freed
|
||||||
|
space with <code>VACUUM FULL</code> / <code>pg_repack</code> on
|
||||||
|
<code>image_record</code>.
|
||||||
|
</p>
|
||||||
|
<v-btn color="primary" rounded="pill" :loading="busy" @click="run">
|
||||||
|
<v-icon start>mdi-database-minus-outline</v-icon> Prune predictions now
|
||||||
|
</v-btn>
|
||||||
|
<span v-if="queued" class="ml-3 text-caption text-success">Queued ✓</span>
|
||||||
|
<QueueStatusBar queue="maintenance_long" queue-label="Maintenance (long)" />
|
||||||
|
</v-card-text>
|
||||||
|
</v-card>
|
||||||
|
</template>
|
||||||
|
|
||||||
|
<script setup>
|
||||||
|
import { computed, onMounted, ref } from 'vue'
|
||||||
|
|
||||||
|
import { useApi } from '../../composables/useApi.js'
|
||||||
|
import { toast } from '../../utils/toast.js'
|
||||||
|
import { useMLStore } from '../../stores/ml.js'
|
||||||
|
import QueueStatusBar from './QueueStatusBar.vue'
|
||||||
|
|
||||||
|
const api = useApi()
|
||||||
|
const ml = useMLStore()
|
||||||
|
const busy = ref(false)
|
||||||
|
const queued = ref(false)
|
||||||
|
|
||||||
|
const floorPct = computed(() => {
|
||||||
|
const f = ml.settings?.tagger_store_floor
|
||||||
|
return f == null ? '—' : `${Math.round(f * 100)}%`
|
||||||
|
})
|
||||||
|
|
||||||
|
onMounted(() => { if (!ml.settings) ml.loadSettings() })
|
||||||
|
|
||||||
|
async function run () {
|
||||||
|
busy.value = true
|
||||||
|
queued.value = false
|
||||||
|
try {
|
||||||
|
await api.post('/api/admin/maintenance/prune-predictions')
|
||||||
|
queued.value = true
|
||||||
|
toast({ text: 'Prediction prune queued', type: 'success' })
|
||||||
|
} catch (e) {
|
||||||
|
toast({ text: e?.body?.detail || e?.message || 'Failed to queue', type: 'error' })
|
||||||
|
} finally {
|
||||||
|
busy.value = false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
</script>
|
||||||
@@ -21,8 +21,15 @@
|
|||||||
<v-card class="mb-4">
|
<v-card class="mb-4">
|
||||||
<CardHeading icon="mdi-alert-circle-outline" title="Recent failures (last 24h)">
|
<CardHeading icon="mdi-alert-circle-outline" title="Recent failures (last 24h)">
|
||||||
<v-spacer />
|
<v-spacer />
|
||||||
<span class="text-caption fc-muted">
|
<v-text-field
|
||||||
{{ failureCount }} failures
|
v-model="failureSearch"
|
||||||
|
placeholder="Search task, queue, target, error…"
|
||||||
|
density="compact" hide-details clearable
|
||||||
|
prepend-inner-icon="mdi-magnify"
|
||||||
|
style="max-width: 280px;"
|
||||||
|
/>
|
||||||
|
<span class="text-caption fc-muted ml-3">
|
||||||
|
{{ filteredFailures.length }} / {{ failureCount }}
|
||||||
</span>
|
</span>
|
||||||
</CardHeading>
|
</CardHeading>
|
||||||
<v-card-text>
|
<v-card-text>
|
||||||
@@ -77,6 +84,14 @@
|
|||||||
<v-card>
|
<v-card>
|
||||||
<CardHeading icon="mdi-format-list-bulleted" title="All recent activity">
|
<CardHeading icon="mdi-format-list-bulleted" title="All recent activity">
|
||||||
<v-spacer />
|
<v-spacer />
|
||||||
|
<v-text-field
|
||||||
|
v-model="filterTask"
|
||||||
|
placeholder="Search task…"
|
||||||
|
density="compact" hide-details clearable
|
||||||
|
prepend-inner-icon="mdi-magnify"
|
||||||
|
style="max-width: 220px;"
|
||||||
|
@update:model-value="onTaskSearch"
|
||||||
|
/>
|
||||||
<v-select
|
<v-select
|
||||||
v-model="filterQueue"
|
v-model="filterQueue"
|
||||||
:items="queueOptions" density="compact" hide-details
|
:items="queueOptions" density="compact" hide-details
|
||||||
@@ -170,6 +185,8 @@ const store = useSystemActivityStore()
|
|||||||
const filterQueue = ref(null)
|
const filterQueue = ref(null)
|
||||||
const filterStatus = ref(null)
|
const filterStatus = ref(null)
|
||||||
const filterErrorType = ref(null)
|
const filterErrorType = ref(null)
|
||||||
|
const filterTask = ref(null) // server-side task-name search (All activity)
|
||||||
|
const failureSearch = ref('') // client-side search over loaded failures
|
||||||
|
|
||||||
const queueOptions = [
|
const queueOptions = [
|
||||||
{ title: 'All queues', value: null },
|
{ title: 'All queues', value: null },
|
||||||
@@ -225,9 +242,18 @@ const errorTypes = computed(() => {
|
|||||||
const failureCount = computed(() => (store.failures?.recent || []).length)
|
const failureCount = computed(() => (store.failures?.recent || []).length)
|
||||||
|
|
||||||
const filteredFailures = computed(() => {
|
const filteredFailures = computed(() => {
|
||||||
const all = store.failures?.recent || []
|
let all = store.failures?.recent || []
|
||||||
if (!filterErrorType.value) return all
|
if (filterErrorType.value) {
|
||||||
return all.filter(r => r.error_type === 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) {
|
function toggleErrorTypeFilter(name) {
|
||||||
@@ -240,9 +266,23 @@ function toggleErrorTypeFilter(name) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function onFilterChange() {
|
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 })
|
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 onLoadMore() { store.loadRuns({ reset: false }) }
|
||||||
function onRefresh() { store.loadRuns({ reset: true }) }
|
function onRefresh() { store.loadRuns({ reset: true }) }
|
||||||
|
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
|
|||||||
const runs = ref([])
|
const runs = ref([])
|
||||||
const runsCursor = ref(null)
|
const runsCursor = ref(null)
|
||||||
const runsHasMore = ref(false)
|
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 loading = ref({ queues: false, workers: false, runs: false, failures: false })
|
||||||
const lastError = ref(null)
|
const lastError = ref(null)
|
||||||
@@ -65,6 +65,7 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
|
|||||||
const params = { limit: runsFilter.value.limit }
|
const params = { limit: runsFilter.value.limit }
|
||||||
if (runsFilter.value.queue) params.queue = runsFilter.value.queue
|
if (runsFilter.value.queue) params.queue = runsFilter.value.queue
|
||||||
if (runsFilter.value.status) params.status = runsFilter.value.status
|
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
|
if (!reset && runsCursor.value) params.before_id = runsCursor.value
|
||||||
const body = await api.get('/api/system/activity/runs', { params })
|
const body = await api.get('/api/system/activity/runs', { params })
|
||||||
runs.value = reset ? body.runs : [...runs.value, ...body.runs]
|
runs.value = reset ? body.runs : [...runs.value, ...body.runs]
|
||||||
|
|||||||
@@ -34,6 +34,28 @@ async def test_get_and_patch_settings(client):
|
|||||||
assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90)
|
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
|
@pytest.mark.asyncio
|
||||||
async def test_backfill_and_recompute_trigger(client):
|
async def test_backfill_and_recompute_trigger(client):
|
||||||
r1 = await client.post("/api/ml/backfill")
|
r1 = await client.post("/api/ml/backfill")
|
||||||
|
|||||||
@@ -182,6 +182,25 @@ async def test_runs_filter_by_status(client, _seed_runs):
|
|||||||
assert len(body["runs"]) == 2 # i=3, i=4
|
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
|
@pytest.mark.asyncio
|
||||||
async def test_runs_keyset_cursor(client, _seed_runs):
|
async def test_runs_keyset_cursor(client, _seed_runs):
|
||||||
page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json()
|
page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json()
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ def test_backup_db_writes_sql_and_manifest(tmp_path, fake_subprocess):
|
|||||||
)
|
)
|
||||||
sql = Path(result["sql_path"])
|
sql = Path(result["sql_path"])
|
||||||
manifest = Path(result["manifest_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 manifest.is_file() and manifest.suffix == ".json"
|
||||||
assert result["kind"] == "db"
|
assert result["kind"] == "db"
|
||||||
assert result["tar_path"] is None
|
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]
|
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):
|
def test_backup_db_strips_asyncpg_driver(tmp_path, fake_subprocess):
|
||||||
backup_service.backup_db(
|
backup_service.backup_db(
|
||||||
db_url="postgresql+asyncpg://u:p@h/d", images_root=tmp_path,
|
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):
|
def test_restore_db_drops_schema_then_loads(tmp_path, fake_subprocess):
|
||||||
sql_path = tmp_path / "fake.sql"
|
dump_path = tmp_path / "fake.dump"
|
||||||
sql_path.write_text("SELECT 1;")
|
dump_path.write_bytes(b"\x00fake custom dump")
|
||||||
backup_service.restore_db(
|
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 len(fake_subprocess) == 2
|
||||||
|
assert fake_subprocess[0][0] == "psql"
|
||||||
assert "-c" in fake_subprocess[0]
|
assert "-c" in fake_subprocess[0]
|
||||||
assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1]
|
assert "DROP SCHEMA IF EXISTS public CASCADE" in fake_subprocess[0][-1]
|
||||||
assert "-f" in fake_subprocess[1]
|
assert fake_subprocess[1][0] == "pg_restore"
|
||||||
assert str(sql_path) in fake_subprocess[1]
|
assert "-d" in fake_subprocess[1]
|
||||||
|
assert str(dump_path) in fake_subprocess[1]
|
||||||
|
|
||||||
|
|
||||||
# --- restore_images --------------------------------------------------
|
# --- restore_images --------------------------------------------------
|
||||||
|
|||||||
@@ -350,6 +350,62 @@ def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync):
|
|||||||
assert ml_stale_status == "error"
|
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):
|
def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
|
||||||
"""import_archive_file shares the 'import' queue with fast
|
"""import_archive_file shares the 'import' queue with fast
|
||||||
single-file import_media_file, so it gets a per-task-name override
|
single-file import_media_file, so it gets a per-task-name override
|
||||||
|
|||||||
@@ -104,3 +104,17 @@ async def test_update_threshold_and_remove(db):
|
|||||||
assert abs(row.min_confidence - 0.80) < 1e-6
|
assert abs(row.min_confidence - 0.80) < 1e-6
|
||||||
await svc.remove(tag.id)
|
await svc.remove(tag.id)
|
||||||
assert await db.get(TagAllowlist, tag.id) is None
|
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
|
||||||
|
|||||||
+11
-7
@@ -1,14 +1,14 @@
|
|||||||
"""Tagger unit tests. The ONNX model isn't available in CI (it's a 1GB
|
"""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
|
download into /models), so these test the pure-logic surface:
|
||||||
constant, SURFACED_CATEGORIES set, TagPrediction dataclass, and the
|
DEFAULT_STORE_FLOOR constant, SURFACED_CATEGORIES set, TagPrediction
|
||||||
load()-missing-file error path. Full inference is exercised by the local
|
dataclass, and the load()-missing-file error path. Full inference is
|
||||||
integration suite against a real /models volume.
|
exercised by the local integration suite against a real /models volume.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from backend.app.services.ml.tagger import (
|
from backend.app.services.ml.tagger import (
|
||||||
STORE_FLOOR,
|
DEFAULT_STORE_FLOOR,
|
||||||
SURFACED_CATEGORIES,
|
SURFACED_CATEGORIES,
|
||||||
Tagger,
|
Tagger,
|
||||||
TagPrediction,
|
TagPrediction,
|
||||||
@@ -26,8 +26,12 @@ def test_surfaced_categories():
|
|||||||
assert "copyright" not in SURFACED_CATEGORIES
|
assert "copyright" not in SURFACED_CATEGORIES
|
||||||
|
|
||||||
|
|
||||||
def test_store_floor_is_low():
|
def test_default_store_floor():
|
||||||
assert 0 < STORE_FLOOR < 0.2
|
# 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():
|
def test_tag_prediction_dataclass():
|
||||||
|
|||||||
@@ -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 -------------------------------------
|
# --- delete_artist_cascade_task -------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -93,7 +93,7 @@ async def test_backup_db_task_creates_backup_run_row_status_ok(db_sync):
|
|||||||
).one()
|
).one()
|
||||||
assert row.kind == "db"
|
assert row.kind == "db"
|
||||||
assert row.status == "ok"
|
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.size_bytes is not None and row.size_bytes > 0
|
||||||
assert row.finished_at is not None
|
assert row.finished_at is not None
|
||||||
assert row.error is None
|
assert row.error is None
|
||||||
|
|||||||
Reference in New Issue
Block a user