diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index a484a9a..51de0b6 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -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 # 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.) -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() cfg = Config.from_env() diff --git a/agent/fc_agent/media.py b/agent/fc_agent/media.py index ba70754..48c9ed5 100644 --- a/agent/fc_agent/media.py +++ b/agent/fc_agent/media.py @@ -2,6 +2,7 @@ (ffmpeg) at the cadence FC sends — so a video becomes a bag of per-frame instances, each with a timestamp.""" import io +import logging import os import subprocess import tempfile @@ -9,6 +10,8 @@ import time from PIL import Image, ImageFile +log = logging.getLogger("fc_agent.media") + # Load slightly-truncated images (a few missing trailing bytes) instead of # 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). @@ -134,29 +137,62 @@ def sample_frames_from_url( interval = max(0.5, float(interval_seconds or 4.0)) cap = max(1, int(max_frames or 64)) 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: pattern = os.path.join(tmp, "f_%05d.jpg") cmd = ["ffmpeg", "-nostdin", "-loglevel", "error", *_RECONNECT, *hdr, - "-i", url, "-vf", f"fps=1/{interval}", "-frames:v", str(cap), - "-q:v", "3", pattern] + "-i", url, "-vf", vf, "-fps_mode", "vfr", + "-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: - proc = subprocess.Popen( - cmd, stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - ) + with open(errpath, "wb") as errf: + proc = subprocess.Popen( + 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): return [] - # 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: - if (should_stop and should_stop()) or (time.monotonic() - start > timeout): - _terminate(proc) - return [] - if proc.returncode != 0: - return [] - return _collect_frames(tmp, interval, cap) + frames = _collect_frames(tmp, interval, cap) + if not frames: + log.warning("ffmpeg produced no frames (exit %s) for %s — stderr: %s", + proc.returncode, url, _tail(errpath)) + return frames + + +def _tail(path: str, limit: int = 300) -> str: + """Last `limit` chars of a (stderr) file, flattened — for failure logs.""" + try: + with open(path, "rb") as f: + f.seek(0, os.SEEK_END) + f.seek(max(0, f.tell() - limit)) + return f.read().decode("utf-8", "replace").replace("\n", " ").strip() + except OSError: + return "?" 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