From 686808d3f320f54ea28a64a98ada98522268b401 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 1 Jul 2026 21:09:07 -0400 Subject: [PATCH] =?UTF-8?q?feat(gpu):=20"Retry=20errored=20jobs"=20?= =?UTF-8?q?=E2=80=94=20scoped=20requeue=20of=20errors=20only?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After an agent-side fix (e.g. the short-video sampler), the errored jobs (~2.8k) have exhausted their 3 attempts and stay parked: backfill skips images that already have a job, and /reprocess is the nuclear option (it resets the 179k DONE jobs too). There was no way to re-run just the errors. POST /api/gpu/retry_errors resets every status='error' job (all task types) to pending with attempts=0 and the stored error cleared — a small inline UPDATE that returns {requeued: n} so the UI toast can show the count. UI: a "Retry errored jobs" button on the GPU-agent card, right under the queue tiles; disabled when errored==0. With the agent now logging ffmpeg's stderr on failure, retrying also reveals which errors were real vs victims of the fps-filter bug. Test: retry_errors requeues the errored job (fresh attempts, error cleared) and leaves done work untouched; asserts via column selects (Core-DML gotcha), not ORM refresh. Co-Authored-By: Claude Fable 5 --- backend/app/api/gpu.py | 23 ++++++++++- .../src/components/settings/GpuAgentCard.vue | 25 ++++++++++++ frontend/src/stores/gpu.js | 8 +++- tests/test_api_gpu.py | 39 ++++++++++++++++++- 4 files changed, 92 insertions(+), 3 deletions(-) diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index 26d8377..11605e3 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -11,7 +11,7 @@ homelab admin. import secrets from quart import Blueprint, jsonify, request -from sqlalchemy import func, select +from sqlalchemy import func, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from ..extensions import get_session @@ -109,6 +109,27 @@ async def reprocess(): return jsonify({"celery_task_id": r.id, "task": task}), 202 +@gpu_bp.route("/retry_errors", methods=["POST"]) +async def retry_errors(): + """Requeue every ERRORED job (all task types) back to pending — the scoped + recovery after an agent-side fix (e.g. the short-video sampler), where + /reprocess would needlessly re-run the whole done library too. Attempts and + the stored error reset so each job gets its full retry budget under the + fixed pipeline. Small row count (errors only) → inline UPDATE, and the + response carries the number requeued for the UI toast.""" + async with get_session() as session: + res = await session.execute( + update(GpuJob) + .where(GpuJob.status == "error") + .values( + status="pending", attempts=0, error=None, lease_token=None, + leased_at=None, lease_expires_at=None, updated_at=func.now(), + ) + ) + await session.commit() + return jsonify({"requeued": res.rowcount or 0}) + + # --- Agent (bearer token): lease / submit / heartbeat / fail ------------ @gpu_bp.route("/jobs/lease", methods=["POST"]) diff --git a/frontend/src/components/settings/GpuAgentCard.vue b/frontend/src/components/settings/GpuAgentCard.vue index 37432bb..bbded5e 100644 --- a/frontend/src/components/settings/GpuAgentCard.vue +++ b/frontend/src/components/settings/GpuAgentCard.vue @@ -52,6 +52,17 @@
{{ queue.error }}
errored
+ Retry errored jobs +

+ Errored jobs park after 3 failed attempts. This requeues just those (their + errors cleared, attempts reset) — use after updating the agent, without + re-running the whole done library. +

+ { return await api.post('/api/gpu/reprocess', { body: { task } }) } - return { token, rotateToken, status, backfill, reprocess } + // Requeue ONLY the errored jobs (all task types) — the scoped recovery after + // an agent fix, without re-running the done library. Returns { requeued }. + async function retryErrors() { + return await api.post('/api/gpu/retry_errors') + } + + return { token, rotateToken, status, backfill, reprocess, retryErrors } }) diff --git a/tests/test_api_gpu.py b/tests/test_api_gpu.py index 08fbc5e..b714438 100644 --- a/tests/test_api_gpu.py +++ b/tests/test_api_gpu.py @@ -1,7 +1,8 @@ """GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill.""" import pytest +from sqlalchemy import select -from backend.app.models import ImageRecord +from backend.app.models import GpuJob, ImageRecord from backend.app.services.ml.gpu_jobs import GpuJobService from backend.app.services.ml.regions import RegionService @@ -148,3 +149,39 @@ async def test_release_hands_job_back_to_pending(client, db): assert resp.status_code == 200 and (await resp.get_json())["released"] == 1 st = await (await client.get("/api/gpu/status")).get_json() assert st["pending"] == 1 and st["leased"] == 0 + + +@pytest.mark.asyncio +async def test_retry_errors_requeues_only_errored(client, db): + """/retry_errors resets ERRORED jobs (any task) to pending with a fresh + retry budget — and leaves done work untouched (it is NOT /reprocess).""" + img1 = await _img(db, "1" * 64) + img2 = await _img(db, "2" * 64) + svc = GpuJobService(db) + j_err = await svc.enqueue(img1.id, "ccip") + j_done = await svc.enqueue(img2.id, "siglip") + err_id = j_err.id + done_id = j_done.id + j_err.status = "error" + j_err.attempts = 3 + j_err.error = "no frames sampled from video (unprocessable)" + j_done.status = "done" + await db.commit() + + resp = await client.post("/api/gpu/retry_errors") + assert resp.status_code == 200 + assert (await resp.get_json())["requeued"] == 1 + + # Column selects, not ORM refresh — the route wrote via Core DML. + row = (await db.execute( + select(GpuJob.status, GpuJob.attempts, GpuJob.error) + .where(GpuJob.id == err_id) + )).one() + assert tuple(row) == ("pending", 0, None) + done_status = await db.scalar( + select(GpuJob.status).where(GpuJob.id == done_id) + ) + assert done_status == "done" + + st = await (await client.get("/api/gpu/status")).get_json() + assert st["pending"] == 1 and st["error"] == 0