From a7abcc41ca130fdf0e9185076f52b46955f91823 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 2 Jul 2026 12:36:02 -0400 Subject: [PATCH] =?UTF-8?q?feat(triage):=20failed-processing=20triage=20?= =?UTF-8?q?=E2=80=94=20probe=20errored=20files,=20flag=20defects,=20recove?= =?UTF-8?q?r=20(#125=20C1-C3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An errored GPU job's stored reason is a suspicion; the file probe is the verdict. A 15-min beat sweep (triage_gpu_errors) runs verify_integrity's own probe (sha256 + decode) on each errored image ONCE and writes both verdicts: ImageRecord.integrity_status and the new GpuJob.triage_status ('defect' | 'file_ok', migration 0072). Every classification logs at WARNING so it surfaces in Logs/System Activity. - 'defect' rows are excluded from /retry_errors (re-running a known-bad file burns agent time re-minting the tombstone); response now reports defects_kept and the GpuAgentCard toast says so. - GET /api/gpu/errors: triage view — reason buckets (classify_reason), probe verdicts, per-job detail. POST /errors/triage runs the sweep now. - POST /api/gpu/errors//recover: reuses the Layer-2 refetch pattern — delete the defective copy + record (full cascade takes the tombstones too) and re-poll its subscription Source so a fresh copy re-imports and re-enters the pipeline; 'no_source' when nothing pollable resolves. - New 'Failed processing' card (GpuTriageCard) in Maintenance: verdict counts, reason summary, probe-now, defect list with thumbnails + per-image Recover. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- .../versions/0072_gpu_job_triage_status.py | 32 +++ backend/app/api/gpu.py | 118 ++++++++++- backend/app/celery_app.py | 4 + backend/app/models/gpu_job.py | 5 + backend/app/services/ml/gpu_triage.py | 156 +++++++++++++++ backend/app/tasks/maintenance.py | 22 ++ .../src/components/settings/GpuAgentCard.vue | 7 +- .../src/components/settings/GpuTriageCard.vue | 189 ++++++++++++++++++ .../components/settings/MaintenancePanel.vue | 2 + frontend/src/stores/gpu.js | 26 ++- tests/test_api_gpu.py | 47 +++++ tests/test_gpu_triage.py | 165 +++++++++++++++ 12 files changed, 763 insertions(+), 10 deletions(-) create mode 100644 alembic/versions/0072_gpu_job_triage_status.py create mode 100644 backend/app/services/ml/gpu_triage.py create mode 100644 frontend/src/components/settings/GpuTriageCard.vue create mode 100644 tests/test_gpu_triage.py diff --git a/alembic/versions/0072_gpu_job_triage_status.py b/alembic/versions/0072_gpu_job_triage_status.py new file mode 100644 index 0000000..1dce875 --- /dev/null +++ b/alembic/versions/0072_gpu_job_triage_status.py @@ -0,0 +1,32 @@ +"""gpu_job.triage_status — the probe's verdict on an errored job's FILE + +Failure triage (#125): a periodic sweep probes each errored image's file +(sha256 + decode, verify_integrity's machinery) exactly once and stores the +verdict here — 'defect' (the file is bad: recovery material, excluded from +/retry_errors) or 'file_ok' (failure was operational, safe to retry). NULL +means not yet probed; selecting on NULL is what makes the sweep resumable. +No index: the errored slice the sweep scans is tiny by design (tombstones). + +Revision ID: 0072 +Revises: 0071 +Create Date: 2026-07-02 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0072" +down_revision: Union[str, None] = "0071" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "gpu_job", sa.Column("triage_status", sa.String(16), nullable=True) + ) + + +def downgrade() -> None: + op.drop_column("gpu_job", "triage_status") diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index 9b3c9d5..bbbbecd 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -9,19 +9,25 @@ homelab admin. """ import secrets +from pathlib import Path from quart import Blueprint, jsonify, request -from sqlalchemy import func, select, update +from sqlalchemy import func, or_, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from ..extensions import get_session from ..models import AppSetting, GpuJob, ImageRecord, MLSettings from ..services.gallery_service import image_url from ..services.ml.gpu_jobs import GpuJobService, error_dedupe_statements +from ..services.ml.gpu_triage import classify_reason, recover_defective_image from ..services.ml.regions import RegionService gpu_bp = Blueprint("gpu", __name__, url_prefix="/api/gpu") +# Same container mount the maintenance tasks use (tasks/admin.py) — recovery +# deletes the defective original + thumbnail under it. +_IMAGES_ROOT = Path("/images") + _TOKEN_KEY = "gpu_agent_token" @@ -118,22 +124,122 @@ async def retry_errors(): fixed pipeline. Stale tombstones are pruned FIRST (loop-era duplicates and rows a later success made moot — the same statements the backfills run), so one failing file requeues as ONE job, never a fan-out of duplicates. Small - row count (errors only) → inline statements; the response carries both - counts for the UI toast.""" + row count (errors only) → inline statements; the response carries the + counts for the UI toast. Triage-confirmed defects are NOT requeued (see + the WHERE below) — they stay on the recovery surface.""" async with get_session() as session: pruned = 0 for stmt in error_dedupe_statements(): pruned += (await session.execute(stmt)).rowcount or 0 res = await session.execute( update(GpuJob) - .where(GpuJob.status == "error") + .where( + GpuJob.status == "error", + # Triage-confirmed DEFECTS stay errored: the integrity probe + # already proved the FILE is bad, so re-running the job just + # burns agent time re-minting the same tombstone — those go + # through /errors//recover instead. + or_(GpuJob.triage_status.is_(None), + GpuJob.triage_status != "defect"), + ) .values( status="pending", attempts=0, error=None, lease_token=None, - leased_at=None, lease_expires_at=None, updated_at=func.now(), + leased_at=None, lease_expires_at=None, triage_status=None, + updated_at=func.now(), ) ) + kept = ( + await session.execute( + select(func.count()).select_from(GpuJob) + .where(GpuJob.status == "error") + ) + ).scalar_one() await session.commit() - return jsonify({"requeued": res.rowcount or 0, "pruned": pruned}) + return jsonify({ + "requeued": res.rowcount or 0, "pruned": pruned, "defects_kept": kept, + }) + + +# --- Failure triage + recovery (#125) ------------------------------------ + +@gpu_bp.route("/errors", methods=["GET"]) +async def errors(): + """The triage view of the error tombstones: every errored job joined with + its image's integrity verdict, bucketed by reason for the overview. The + probe sweep (triage_gpu_errors, 15-min beat) fills triage_status; 'defect' + rows are the recovery surface's list.""" + async with get_session() as session: + rows = ( + await session.execute( + select( + GpuJob.id, GpuJob.image_record_id, GpuJob.task, + GpuJob.error, GpuJob.triage_status, GpuJob.updated_at, + ImageRecord.integrity_status, ImageRecord.mime, + ImageRecord.path, ImageRecord.thumbnail_path, + ) + .join(ImageRecord, ImageRecord.id == GpuJob.image_record_id) + .where(GpuJob.status == "error") + .order_by(GpuJob.updated_at.desc()) + .limit(500) + ) + ).all() + total = ( + await session.execute( + select(func.count()).select_from(GpuJob) + .where(GpuJob.status == "error") + ) + ).scalar_one() + by_class: dict[str, int] = {} + triage = {"defect": 0, "file_ok": 0, "unclassified": 0} + items = [] + for r in rows: + cls = classify_reason(r.error) + by_class[cls] = by_class.get(cls, 0) + 1 + bucket = r.triage_status or "unclassified" + triage[bucket] = triage.get(bucket, 0) + 1 + items.append({ + "job_id": r.id, + "image_id": r.image_record_id, + "task": r.task, + "error": r.error, + "reason_class": cls, + "triage_status": r.triage_status, + "integrity_status": r.integrity_status, + "mime": r.mime, + "image_url": image_url(r.path), + "thumbnail_url": ( + image_url(r.thumbnail_path) if r.thumbnail_path else None + ), + "updated_at": r.updated_at.isoformat() if r.updated_at else None, + }) + return jsonify({ + "total": total, "by_class": by_class, "triage": triage, "items": items, + }) + + +@gpu_bp.route("/errors/triage", methods=["POST"]) +async def errors_triage(): + """Run the probe sweep NOW (the card's button) instead of waiting out the + 15-minute beat cadence.""" + from ..tasks.maintenance import triage_gpu_errors + + r = triage_gpu_errors.delay() + return jsonify({"celery_task_id": r.id}), 202 + + +@gpu_bp.route("/errors//recover", methods=["POST"]) +async def errors_recover(image_id: int): + """Recover a defect-triaged original: delete the bad copy + record and + re-poll its subscription Source (a fresh fetch re-imports the file, which + re-enters the GPU pipeline). Returns status 'no_source' when nothing + pollable resolves — the file needs manual replacement there.""" + async with get_session() as session: + result = await session.run_sync( + lambda s: recover_defective_image( + s, image_id, images_root=_IMAGES_ROOT, + ) + ) + return jsonify(result) # --- Agent (bearer token): lease / submit / heartbeat / fail ------------ diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index e75c626..5cb1ab1 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -109,6 +109,10 @@ def make_celery() -> Celery: "task": "backend.app.tasks.ml.recover_orphaned_gpu_jobs", "schedule": 60.0, # quick pickup of work a dead agent orphaned }, + "triage-gpu-errors": { + "task": "backend.app.tasks.maintenance.triage_gpu_errors", + "schedule": 900.0, # probe errored jobs' files → defect/file_ok + }, "enqueue-ccip-backfill-hourly": { "task": "backend.app.tasks.ml.enqueue_gpu_backfill", "schedule": 3600.0, # auto-feed NEW images; errored are diff --git a/backend/app/models/gpu_job.py b/backend/app/models/gpu_job.py index 5e14e2d..dba5997 100644 --- a/backend/app/models/gpu_job.py +++ b/backend/app/models/gpu_job.py @@ -62,6 +62,11 @@ class GpuJob(Base): ) attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) error: Mapped[str | None] = mapped_column(Text, nullable=True) + # Triage verdict for an ERRORED job (#125): NULL = not yet probed; + # 'defect' = the integrity probe says the FILE itself is bad (surfaced for + # recovery, excluded from /retry_errors); 'file_ok' = the file passes — + # the failure was operational (timeout/transient), safe to retry. + triage_status: Mapped[str | None] = mapped_column(String(16), nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) diff --git a/backend/app/services/ml/gpu_triage.py b/backend/app/services/ml/gpu_triage.py new file mode 100644 index 0000000..dddd095 --- /dev/null +++ b/backend/app/services/ml/gpu_triage.py @@ -0,0 +1,156 @@ +"""GPU-failure triage (#125): classify errored jobs, PROBE the file, recover. + +An errored GPU job is a tombstone with a stored reason, but the reason alone is +a suspicion, not a verdict — a timeout can hit a perfectly fine file, and +"moov atom not found" can mean a truncated download OR a one-off transfer +fault. So triage EVALUATES: it runs the real integrity probe (sha256 recompute ++ PIL/ffprobe — verify_integrity's own machinery) on each errored image ONCE +and records both verdicts: + + ImageRecord.integrity_status <- file-level verdict (ok / corrupt / ...) + GpuJob.triage_status <- 'defect' (file is bad: recovery material, + excluded from /retry_errors) + 'file_ok' (file passes: the failure was + operational, safe to retry) + +Recovery reuses established primitives: delete the defective copy + record +(cleanup_service.delete_images — full cascade) and re-poll the image's +subscription Source (the Layer-2 refetch pattern: gallery-dl re-fetches the +now-absent file on the next source check). Images without a pollable Source +report 'no_source' — manual remediation. Every classification is logged at +WARNING so the operator notices in Logs / System Activity. +""" +import logging +import time +from datetime import UTC, datetime +from pathlib import Path + +from sqlalchemy import select, update +from sqlalchemy.orm import Session + +from ...models import GpuJob, ImageProvenance, ImageRecord, Source +from ..cleanup_service import delete_images + +log = logging.getLogger(__name__) + +# Reason buckets for the triage overview (reporting only — the PROBE decides +# 'defect', never the string). Ordered: first match wins. +_REASON_BUCKETS = ( + ("poisoned", ("poisoned:",)), + ("transient", ("gave up after repeated transient", "curator unreachable", + "connection", "read timed out")), + ("timeout", ("timed out", "timeout")), + ("truncated_or_corrupt", ("moov atom", "invalid data", "end of file", + "header missing", "error reading header", + "truncated", "premature", "corrupt", + "no frames sampled")), + ("decode", ("cannot identify", "decompression", "broken data stream", + "unrecognized data")), +) + + +def classify_reason(error: str | None) -> str: + """Bucket a stored job-error string for the overview table.""" + text = (error or "").lower() + if not text: + return "other" + for bucket, needles in _REASON_BUCKETS: + if any(n in text for n in needles): + return bucket + return "other" + + +def triage_errored_jobs( + session: Session, *, time_budget_seconds: float = 300.0, +) -> dict: + """Probe every not-yet-triaged errored image and write both verdicts. + + Time-boxed (sha256 of a large original over NFS can take tens of seconds) + and inherently resumable: rows are selected by `triage_status IS NULL`, so + the next sweep continues exactly where a budget cut stopped. Commits per + image so a mid-run crash keeps completed verdicts.""" + image_ids = session.execute( + select(GpuJob.image_record_id) + .where(GpuJob.status == "error", GpuJob.triage_status.is_(None)) + .group_by(GpuJob.image_record_id) + .order_by(GpuJob.image_record_id) + ).scalars().all() + counts = {"probed": 0, "defect": 0, "file_ok": 0, "partial": False} + if not image_ids: + return counts + # Lazy imports: the probe helper lives in the maintenance task module and + # the hasher in the importer — importing either at module load would pull + # celery into every service consumer. + from ...tasks.maintenance import _verify_one + from ..importer import _sha256_of + + started = time.monotonic() + for image_id in image_ids: + if time.monotonic() - started > time_budget_seconds: + counts["partial"] = True + break + rec = session.get(ImageRecord, image_id) + if rec is None: # record deleted since the job errored + continue + verdict = _verify_one(Path(rec.path), rec.sha256, rec.mime, _sha256_of) + # 'ok' means the failure was operational; anything else (corrupt / + # failed_verification = missing/unreadable) makes the file itself the + # problem — recovery material. + triage = "file_ok" if verdict == "ok" else "defect" + reason = session.execute( + select(GpuJob.error) + .where(GpuJob.image_record_id == image_id, GpuJob.status == "error") + .limit(1) + ).scalar_one_or_none() + rec.integrity_status = verdict + session.execute( + update(GpuJob) + .where(GpuJob.image_record_id == image_id, GpuJob.status == "error") + .values(triage_status=triage, updated_at=datetime.now(UTC)) + ) + session.commit() + counts["probed"] += 1 + counts[triage] += 1 + log.warning( + "gpu triage: image %s (%s) job error %r -> integrity probe %r -> %s", + image_id, rec.path, (reason or "")[:120], verdict, triage, + ) + return counts + + +def recover_defective_image( + session: Session, image_id: int, *, images_root: Path, +) -> dict: + """Delete the defective copy + record and re-poll its subscription Source. + + Mirrors the Layer-2 import refetch: with the bad file gone, the source's + next gallery-dl run re-fetches a fresh copy, which re-imports as a new + record and re-enters the GPU pipeline. The record delete cascades the + error tombstones with it. 'no_source' when no enabled, real-URL Source is + reachable via the image's provenance — manual remediation there.""" + rec = session.get(ImageRecord, image_id) + if rec is None: + return {"status": "not_found"} + src_id = session.execute( + select(Source.id) + .join(ImageProvenance, ImageProvenance.source_id == Source.id) + .where( + ImageProvenance.image_record_id == image_id, + Source.enabled.is_(True), + ~Source.url.like("sidecar:%"), # synthetic anchor — not pollable + ) + .order_by(Source.id.asc()) + ).scalars().first() + if src_id is None: + return {"status": "no_source"} + path = rec.path + summary = delete_images(session, image_ids=[image_id], images_root=images_root) + # Lazy import (services -> tasks would cycle at module load). + from ...tasks.download import download_source + + download_source.delay(src_id) + log.warning( + "gpu triage recovery: deleted defective image %s (%s) and queued a " + "re-check of source %s to re-fetch it", image_id, path, src_id, + ) + return {"status": "refetch_queued", "source_id": src_id, **summary} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index e9e334e..842df0f 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -582,6 +582,28 @@ def verify_integrity() -> int: return total +@celery.task( + name="backend.app.tasks.maintenance.triage_gpu_errors", + # Bounded small-set probe (only errored images, once each), but a single + # large original's sha256 over NFS can run tens of seconds — same quick-lane + # tolerance rationale as verify_integrity above. + soft_time_limit=600, time_limit=900, +) +def triage_gpu_errors() -> dict: + """Failure triage (#125): probe each errored GPU job's file once and write + the verdicts (ImageRecord.integrity_status + GpuJob.triage_status) — see + services/ml/gpu_triage.py. Time-boxed + resumable; no-op when every errored + job is already triaged.""" + from ..services.ml.gpu_triage import triage_errored_jobs + + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + summary = triage_errored_jobs(session, time_budget_seconds=300.0) + if summary["probed"]: + log.info("triage_gpu_errors: %s", summary) + return summary + + @celery.task(name="backend.app.tasks.maintenance.recover_stalled_download_events") def recover_stalled_download_events() -> int: """Recover DownloadEvent rows stuck pending/running past the worker hard kill. diff --git a/frontend/src/components/settings/GpuAgentCard.vue b/frontend/src/components/settings/GpuAgentCard.vue index 9f4b0dd..b9a8492 100644 --- a/frontend/src/components/settings/GpuAgentCard.vue +++ b/frontend/src/components/settings/GpuAgentCard.vue @@ -337,8 +337,11 @@ async function onBackfillSiglip() { async function onRetryErrors() { retrying.value = true try { - const { requeued, pruned } = await store.retryErrors() - const extra = pruned ? ` (${pruned} stale duplicate${pruned === 1 ? '' : 's'} pruned)` : '' + const { requeued, pruned, defects_kept: kept } = await store.retryErrors() + const extras = [] + if (pruned) extras.push(`${pruned} stale duplicate${pruned === 1 ? '' : 's'} pruned`) + if (kept) extras.push(`${kept} known-bad file${kept === 1 ? '' : 's'} kept for recovery`) + const extra = extras.length ? ` (${extras.join(', ')})` : '' toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'}${extra} — run the agent to process them`, type: 'success' }) await refreshQueue() } catch (e) { diff --git a/frontend/src/components/settings/GpuTriageCard.vue b/frontend/src/components/settings/GpuTriageCard.vue new file mode 100644 index 0000000..a2ab8ac --- /dev/null +++ b/frontend/src/components/settings/GpuTriageCard.vue @@ -0,0 +1,189 @@ + + + + + diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 55d5f75..8957496 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -13,6 +13,7 @@ + @@ -48,6 +49,7 @@ import MLBackfillCard from './MLBackfillCard.vue' import ThumbnailBackfillCard from './ThumbnailBackfillCard.vue' import ArchiveReextractCard from './ArchiveReextractCard.vue' import MissingFileRepairCard from './MissingFileRepairCard.vue' +import GpuTriageCard from './GpuTriageCard.vue' import DbMaintenanceCard from './DbMaintenanceCard.vue' import MLThresholdSliders from './MLThresholdSliders.vue' import HeadsCard from './HeadsCard.vue' diff --git a/frontend/src/stores/gpu.js b/frontend/src/stores/gpu.js index bff702a..79825e4 100644 --- a/frontend/src/stores/gpu.js +++ b/frontend/src/stores/gpu.js @@ -36,10 +36,32 @@ export const useGpuStore = defineStore('gpu', () => { } // Requeue ONLY the errored jobs (all task types) — the scoped recovery after - // an agent fix, without re-running the done library. Returns { requeued }. + // an agent fix, without re-running the done library. Triage-confirmed + // defects stay put (they recover via recoverImage instead). + // Returns { requeued, pruned, defects_kept }. async function retryErrors() { return await api.post('/api/gpu/retry_errors') } - return { token, rotateToken, status, backfill, reprocess, retryErrors } + // Failure-triage view (#125): errored jobs joined with integrity verdicts. + // Returns { total, by_class, triage, items }. + async function errors() { + return await api.get('/api/gpu/errors') + } + + // Run the file-probe sweep now instead of waiting out the 15-min beat. + async function triageErrors() { + return await api.post('/api/gpu/errors/triage') + } + + // Recover a defective original: delete the bad copy + record, re-poll its + // Source. Returns { status: 'refetch_queued'|'no_source'|'not_found', ... }. + async function recoverImage(imageId) { + return await api.post(`/api/gpu/errors/${imageId}/recover`) + } + + return { + token, rotateToken, status, backfill, reprocess, retryErrors, + errors, triageErrors, recoverImage, + } }) diff --git a/tests/test_api_gpu.py b/tests/test_api_gpu.py index ed01efb..4c29727 100644 --- a/tests/test_api_gpu.py +++ b/tests/test_api_gpu.py @@ -213,3 +213,50 @@ async def test_retry_errors_requeues_only_errored(client, db): st = await (await client.get("/api/gpu/status")).get_json() assert st["pending"] == 1 and st["error"] == 0 + + +@pytest.mark.asyncio +async def test_retry_errors_keeps_triaged_defects(client, db): + """A probe-confirmed DEFECT is a bad FILE — requeueing it just burns agent + time re-minting the tombstone, so /retry_errors leaves it for the recovery + surface and reports it as defects_kept.""" + img1 = await _img(db, "4" * 64) + img2 = await _img(db, "5" * 64) + db.add(GpuJob(image_record_id=img1.id, task="ccip", status="error", + attempts=3, error="moov atom not found", + triage_status="defect")) + db.add(GpuJob(image_record_id=img2.id, task="ccip", status="error", + attempts=3, error="ffmpeg timed out after 1200s")) + await db.commit() + + body = await (await client.post("/api/gpu/retry_errors")).get_json() + assert body["requeued"] == 1 + assert body["defects_kept"] == 1 + + rows = dict((await db.execute( + select(GpuJob.image_record_id, GpuJob.status) + )).all()) + assert rows[img1.id] == "error" # defect stays tombstoned + assert rows[img2.id] == "pending" # operational failure requeued + + +@pytest.mark.asyncio +async def test_errors_endpoint_reports_triage_view(client, db): + img = await _img(db, "6" * 64) + db.add(GpuJob(image_record_id=img.id, task="ccip", status="error", + attempts=3, + error="no frames sampled from video — moov atom not found")) + await db.commit() + + resp = await client.get("/api/gpu/errors") + assert resp.status_code == 200 + body = await resp.get_json() + assert body["total"] == 1 + assert body["by_class"] == {"truncated_or_corrupt": 1} + assert body["triage"]["unclassified"] == 1 + item = body["items"][0] + assert item["image_id"] == img.id + assert item["task"] == "ccip" + assert item["reason_class"] == "truncated_or_corrupt" + assert item["triage_status"] is None + assert item["image_url"].startswith("/images/") diff --git a/tests/test_gpu_triage.py b/tests/test_gpu_triage.py new file mode 100644 index 0000000..6bb1b61 --- /dev/null +++ b/tests/test_gpu_triage.py @@ -0,0 +1,165 @@ +"""Failure triage (#125): probe errored jobs' files, flag verdicts, recover. + +The probe is the arbiter: reason strings only bucket the overview. A file that +passes checksum+decode is 'file_ok' (operational failure); anything else is a +'defect' — surfaced for recovery and excluded from /retry_errors. +""" +import hashlib + +import pytest +from PIL import Image as PILImage +from sqlalchemy import select + +from backend.app.models import ( + Artist, + GpuJob, + ImageProvenance, + ImageRecord, + Post, + Source, +) +from backend.app.services.ml.gpu_triage import ( + classify_reason, + recover_defective_image, + triage_errored_jobs, +) + +pytestmark = pytest.mark.integration + + +def test_classify_reason_buckets(): + assert classify_reason( + "no frames sampled from video — moov atom not found" + ) == "truncated_or_corrupt" + assert classify_reason("ffmpeg timed out after 1200s") == "timeout" + assert classify_reason( + "gave up after repeated transient failures: HTTPConnectionPool read timed out" + ) == "transient" + assert classify_reason( + "poisoned: 10+ lease attempts without ever completing" + ) == "poisoned" + assert classify_reason("cannot identify image file") == "decode" + assert classify_reason("something novel") == "other" + assert classify_reason(None) == "other" + + +async def _errored_image(db, tmp_path, *, name, sha, content: bytes | None, + error="no frames sampled from video — moov atom not found"): + """An ImageRecord (file written iff content is not None) + an errored job.""" + path = tmp_path / name + if content is not None: + path.write_bytes(content) + img = ImageRecord( + path=str(path), sha256=sha, size_bytes=1, mime="image/png", + width=1, height=1, origin="imported_filesystem", + integrity_status="unknown", + ) + db.add(img) + await db.flush() + db.add(GpuJob(image_record_id=img.id, task="ccip", status="error", + error=error, attempts=3)) + await db.flush() + return img + + +@pytest.mark.asyncio +async def test_triage_probes_and_splits_defect_vs_file_ok(db, tmp_path): + # Healthy: a real PNG whose recorded sha matches its bytes → file_ok. + ok_path = tmp_path / "fine.png" + PILImage.new("RGB", (4, 4), (200, 30, 30)).save(ok_path) + ok_sha = hashlib.sha256(ok_path.read_bytes()).hexdigest() + ok = await _errored_image(db, tmp_path, name="fine.png", sha=ok_sha, + content=None, error="ffmpeg timed out after 1200s") + # Corrupt: bytes don't match the recorded sha → defect. + bad = await _errored_image(db, tmp_path, name="bad.png", sha="0" * 64, + content=b"not a real png") + # Missing: no file on disk at all → defect (failed_verification). + gone = await _errored_image(db, tmp_path, name="gone.png", sha="1" * 64, + content=None) + await db.commit() + + summary = await db.run_sync(lambda s: triage_errored_jobs(s)) + assert summary["probed"] == 3 + assert summary["defect"] == 2 + assert summary["file_ok"] == 1 + assert summary["partial"] is False + + # Column selects, not ORM refresh — the sweep wrote via Core DML. + rows = dict((await db.execute( + select(GpuJob.image_record_id, GpuJob.triage_status) + .where(GpuJob.status == "error") + )).all()) + assert rows[ok.id] == "file_ok" + assert rows[bad.id] == "defect" + assert rows[gone.id] == "defect" + verdicts = dict((await db.execute( + select(ImageRecord.id, ImageRecord.integrity_status) + .where(ImageRecord.id.in_([ok.id, bad.id, gone.id])) + )).all()) + assert verdicts[ok.id] == "ok" + assert verdicts[bad.id] == "corrupt" + assert verdicts[gone.id] == "failed_verification" + + # Idempotent: everything already triaged → no re-probe. + again = await db.run_sync(lambda s: triage_errored_jobs(s)) + assert again["probed"] == 0 + + +@pytest.mark.asyncio +async def test_recover_without_pollable_source_reports_no_source(db, tmp_path): + img = await _errored_image(db, tmp_path, name="orphan.png", sha="2" * 64, + content=b"x") + await db.commit() + res = await db.run_sync( + lambda s: recover_defective_image(s, img.id, images_root=tmp_path) + ) + assert res["status"] == "no_source" + still_there = (await db.execute( + select(ImageRecord.id).where(ImageRecord.id == img.id) + )).scalar_one_or_none() + assert still_there == img.id + + +@pytest.mark.asyncio +async def test_recover_deletes_record_and_requeues_source( + client, db, tmp_path, monkeypatch, +): + img = await _errored_image(db, tmp_path, name="fixme.png", sha="3" * 64, + content=b"x") + artist = Artist(name="Recov", slug="recov") + db.add(artist) + await db.flush() + src = Source(artist_id=artist.id, platform="patreon", + url="https://www.patreon.com/recov", enabled=True) + db.add(src) + await db.flush() + post = Post(artist_id=artist.id, source_id=src.id, external_post_id="p1") + db.add(post) + await db.flush() + db.add(ImageProvenance(image_record_id=img.id, post_id=post.id, + source_id=src.id)) + img_id, src_id = img.id, src.id + await db.commit() + + queued = [] + monkeypatch.setattr( + "backend.app.tasks.download.download_source.delay", + lambda sid: queued.append(sid), + ) + + resp = await client.post(f"/api/gpu/errors/{img_id}/recover") + assert resp.status_code == 200 + body = await resp.get_json() + assert body["status"] == "refetch_queued" + assert body["source_id"] == src_id + assert queued == [src_id] + + # Record gone — the error tombstones cascade away with it. + remaining = (await db.execute( + select(ImageRecord.id).where(ImageRecord.id == img_id) + )).scalar_one_or_none() + assert remaining is None + jobs_left = (await db.execute( + select(GpuJob.id).where(GpuJob.image_record_id == img_id) + )).scalars().all() + assert jobs_left == []