Compare commits

...

3 Commits

Author SHA1 Message Date
bvandeusen d9354ac1e1 Merge pull request 'Agent: short-video sampling fix + "Retry errored jobs" recovery action' (#185) from dev into main
Build images / sign-extension (push) Successful in 3s
CI / lint (push) Successful in 3s
Build images / build-agent (push) Successful in 8s
Build images / build-ml (push) Successful in 8s
Build images / build-web (push) Successful in 10s
CI / frontend-build (push) Successful in 22s
CI / backend-lint-and-test (push) Successful in 27s
CI / integration (push) Successful in 3m26s
2026-07-01 21:13:48 -04:00
bvandeusen 686808d3f3 feat(gpu): "Retry errored jobs" — scoped requeue of errors only
CI / lint (push) Successful in 2s
CI / frontend-build (push) Successful in 20s
CI / backend-lint-and-test (push) Successful in 25s
CI / integration (push) Successful in 3m26s
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>
2026-07-01 21:09:07 -04:00
bvandeusen 3a683d7feb fix(agent): short videos failed as "unprocessable" — fps filter emits 0 frames
CI / lint (push) Successful in 2s
CI / frontend-build (push) Successful in 18s
CI / backend-lint-and-test (push) Successful in 28s
CI / integration (push) Successful in 3m25s
Root cause of the "no frames sampled from video (unprocessable)" flood
(operator-flagged 2026-07-02, whole 62k-70k image block + others): the
sampler used `-vf fps=1/4`, and ffmpeg's fps filter emits round(duration/4)
frames — which is ZERO for any clip shorter than ~2s. Short animation loops
(0.5s, 1.75s — verified against two originals from different artists) are
complete, valid h264 videos; ffmpeg decoded them fine, emitted no frames,
exited 0, and the agent failed the job as unprocessable. Long videos worked,
so only the short-clip class flooded.

Fix: sample with select ("first frame always, then one per interval of
timestamp") + -fps_mode vfr, and scale=out_range=full so limited-range
yuv420p sources don't trip the mjpeg encoder's full-range strictness
(secondary failure observed on a 4440x2760 clip). Verified locally against
both failing originals (frames extracted, PIL-clean) and a synthetic 15s
video (4 frames at t=0/4/8/12 — long-video behavior unchanged).

Observability (why this hid for weeks): ffmpeg's stderr was discarded, so
every failure logged only "no frames sampled". stderr now goes to a temp
file and its tail is logged on any produced-no-frames/timeout failure — the
log names the actual ffmpeg reason from now on. Also: frames written before
a mid-stream ffmpeg error are now kept (partial > nothing).

VERSION → 2026-07-02.1.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-01 21:00:29 -04:00
6 changed files with 149 additions and 24 deletions
+1 -1
View File
@@ -21,7 +21,7 @@ log = logging.getLogger("fc_agent.app")
# Bump on every agent change. The page embeds this and /status reports it; the UI # Bump on every agent change. The page embeds this and /status reports it; the UI
# warns to reload when they differ — so a stale browser-cached page can't be # warns to reload when they differ — so a stale browser-cached page can't be
# mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.) # mistaken for "the new image didn't deploy". (Belt-and-braces with no-store.)
VERSION = "2026-07-01.10 · fix Status freeze (conchint.textContent destroyed #capn)" VERSION = "2026-07-02.1 · short videos sample again (select, not fps) + ffmpeg errors logged"
logbuf.install() logbuf.install()
cfg = Config.from_env() cfg = Config.from_env()
+56 -20
View File
@@ -2,6 +2,7 @@
(ffmpeg) at the cadence FC sends — so a video becomes a bag of per-frame (ffmpeg) at the cadence FC sends — so a video becomes a bag of per-frame
instances, each with a timestamp.""" instances, each with a timestamp."""
import io import io
import logging
import os import os
import subprocess import subprocess
import tempfile import tempfile
@@ -9,6 +10,8 @@ import time
from PIL import Image, ImageFile from PIL import Image, ImageFile
log = logging.getLogger("fc_agent.media")
# Load slightly-truncated images (a few missing trailing bytes) instead of # Load slightly-truncated images (a few missing trailing bytes) instead of
# raising — matches the server embedder. These are common in scraped libraries # raising — matches the server embedder. These are common in scraped libraries
# and would otherwise fail the job 3× then error (operator-flagged 2026-06-30). # and would otherwise fail the job 3× then error (operator-flagged 2026-06-30).
@@ -134,29 +137,62 @@ def sample_frames_from_url(
interval = max(0.5, float(interval_seconds or 4.0)) interval = max(0.5, float(interval_seconds or 4.0))
cap = max(1, int(max_frames or 64)) cap = max(1, int(max_frames or 64))
hdr = ["-headers", headers] if headers else [] hdr = ["-headers", headers] if headers else []
# select (NOT the fps filter): always keep the FIRST frame, then one per
# `interval` seconds of timestamp. fps=1/N emits round(duration/N) frames,
# which is ZERO for any clip shorter than ~N/2 seconds — a whole class of
# short animation loops failed as "unprocessable" that way (operator-flagged
# 2026-07-02: 0.5s/1.75s clips). scale=out_range=full converts limited-range
# yuv420p to full range so the mjpeg (jpg) encoder accepts it at default
# strictness instead of erroring on "non full-range YUV".
vf = (
f"select='isnan(prev_selected_t)+gte(t-prev_selected_t\\,{interval})',"
"scale=out_range=full"
)
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
pattern = os.path.join(tmp, "f_%05d.jpg") pattern = os.path.join(tmp, "f_%05d.jpg")
cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr, cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr,
"-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap), "-i", url, "-vf", vf, "-fps_mode", "vfr",
"-q:v", "3", pattern] "-frames:v", str(cap), "-q:v", "3", pattern]
# ffmpeg's stderr goes to a file (not a PIPE, which could fill and
# deadlock; not DEVNULL, which is how a filter bug hid as "unprocessable"
# for weeks) — on failure its tail is logged so the operator can see WHY.
errpath = os.path.join(tmp, "stderr.txt")
try: try:
proc = subprocess.Popen( with open(errpath, "wb") as errf:
cmd, stdin=subprocess.DEVNULL, proc = subprocess.Popen(
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cmd, stdin=subprocess.DEVNULL,
) stdout=subprocess.DEVNULL, stderr=errf,
)
# Poll rather than block, so a Stop (or the per-video timeout) can
# kill a slow/wedged ffmpeg promptly instead of waiting it out.
start = time.monotonic()
while True:
try:
proc.wait(timeout=0.5)
break
except subprocess.TimeoutExpired:
stopped = should_stop and should_stop()
if stopped or (time.monotonic() - start > timeout):
_terminate(proc)
if not stopped:
log.warning("ffmpeg timed out after %.0fs: %s",
timeout, url)
return []
except (OSError, ValueError): except (OSError, ValueError):
return [] return []
# Poll rather than block, so a Stop (or the per-video timeout) can kill a frames = _collect_frames(tmp, interval, cap)
# slow/wedged ffmpeg promptly instead of waiting it out. if not frames:
start = time.monotonic() log.warning("ffmpeg produced no frames (exit %s) for %s — stderr: %s",
while True: proc.returncode, url, _tail(errpath))
try: return frames
proc.wait(timeout=0.5)
break
except subprocess.TimeoutExpired: def _tail(path: str, limit: int = 300) -> str:
if (should_stop and should_stop()) or (time.monotonic() - start > timeout): """Last `limit` chars of a (stderr) file, flattened — for failure logs."""
_terminate(proc) try:
return [] with open(path, "rb") as f:
if proc.returncode != 0: f.seek(0, os.SEEK_END)
return [] f.seek(max(0, f.tell() - limit))
return _collect_frames(tmp, interval, cap) return f.read().decode("utf-8", "replace").replace("\n", " ").strip()
except OSError:
return "?"
+22 -1
View File
@@ -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
+7 -1
View File
@@ -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
View File
@@ -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