feat(gpu): "Retry errored jobs" — scoped requeue of errors only
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 <noreply@anthropic.com>
This commit is contained in:
+22
-1
@@ -11,7 +11,7 @@ homelab admin.
|
|||||||
import secrets
|
import secrets
|
||||||
|
|
||||||
from quart import Blueprint, jsonify, request
|
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 sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||||
|
|
||||||
from ..extensions import get_session
|
from ..extensions import get_session
|
||||||
@@ -109,6 +109,27 @@ async def reprocess():
|
|||||||
return jsonify({"celery_task_id": r.id, "task": task}), 202
|
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 ------------
|
# --- Agent (bearer token): lease / submit / heartbeat / fail ------------
|
||||||
|
|
||||||
@gpu_bp.route("/jobs/lease", methods=["POST"])
|
@gpu_bp.route("/jobs/lease", methods=["POST"])
|
||||||
|
|||||||
@@ -52,6 +52,17 @@
|
|||||||
<div class="fc-q"><div class="fc-q__n" :class="queue.error ? 'fc-weak' : ''">{{ queue.error }}</div><div class="fc-q__l">errored</div></div>
|
<div class="fc-q"><div class="fc-q__n" :class="queue.error ? 'fc-weak' : ''">{{ queue.error }}</div><div class="fc-q__l">errored</div></div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<v-btn
|
||||||
|
class="mt-3" color="accent" variant="tonal" rounded="pill" size="small"
|
||||||
|
prepend-icon="mdi-restart-alert" :loading="retrying"
|
||||||
|
:disabled="!queue.error" @click="onRetryErrors"
|
||||||
|
>Retry errored jobs</v-btn>
|
||||||
|
<p class="fc-muted text-caption mt-2 mb-0">
|
||||||
|
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.
|
||||||
|
</p>
|
||||||
|
|
||||||
<v-btn
|
<v-btn
|
||||||
class="mt-4" color="accent" variant="tonal" rounded="pill" size="small"
|
class="mt-4" color="accent" variant="tonal" rounded="pill" size="small"
|
||||||
prepend-icon="mdi-account-box-multiple" :loading="backfilling" @click="onBackfill"
|
prepend-icon="mdi-account-box-multiple" :loading="backfilling" @click="onBackfill"
|
||||||
@@ -164,6 +175,7 @@ const rotating = ref(false)
|
|||||||
const backfilling = ref(false)
|
const backfilling = ref(false)
|
||||||
const backfillingSiglip = ref(false)
|
const backfillingSiglip = ref(false)
|
||||||
const reprocessing = ref(false)
|
const reprocessing = ref(false)
|
||||||
|
const retrying = ref(false)
|
||||||
const threshold = ref(0.85)
|
const threshold = ref(0.85)
|
||||||
const savingThreshold = ref(false)
|
const savingThreshold = ref(false)
|
||||||
const autoApply = ref(true)
|
const autoApply = ref(true)
|
||||||
@@ -323,6 +335,19 @@ async function onBackfillSiglip() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async function onRetryErrors() {
|
||||||
|
retrying.value = true
|
||||||
|
try {
|
||||||
|
const { requeued } = await store.retryErrors()
|
||||||
|
toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'} — run the agent to process them`, type: 'success' })
|
||||||
|
await refreshQueue()
|
||||||
|
} catch (e) {
|
||||||
|
toast({ text: `Could not retry errored jobs: ${e.message}`, type: 'error' })
|
||||||
|
} finally {
|
||||||
|
retrying.value = false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async function onReprocess() {
|
async function onReprocess() {
|
||||||
if (!window.confirm('Re-process the ENTIRE library (re-detect + re-crop every image)? This is heavy and runs on the GPU agent.')) return
|
if (!window.confirm('Re-process the ENTIRE library (re-detect + re-crop every image)? This is heavy and runs on the GPU agent.')) return
|
||||||
reprocessing.value = true
|
reprocessing.value = true
|
||||||
|
|||||||
@@ -35,5 +35,11 @@ export const useGpuStore = defineStore('gpu', () => {
|
|||||||
return await api.post('/api/gpu/reprocess', { body: { task } })
|
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 }
|
||||||
})
|
})
|
||||||
|
|||||||
+38
-1
@@ -1,7 +1,8 @@
|
|||||||
"""GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill."""
|
"""GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill."""
|
||||||
import pytest
|
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.gpu_jobs import GpuJobService
|
||||||
from backend.app.services.ml.regions import RegionService
|
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
|
assert resp.status_code == 200 and (await resp.get_json())["released"] == 1
|
||||||
st = await (await client.get("/api/gpu/status")).get_json()
|
st = await (await client.get("/api/gpu/status")).get_json()
|
||||||
assert st["pending"] == 1 and st["leased"] == 0
|
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
|
||||||
|
|||||||
Reference in New Issue
Block a user