From 09e277262859d6764d19775fe3a55a2459ac23df Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 1 Jul 2026 22:52:38 -0400 Subject: [PATCH] =?UTF-8?q?fix(gpu-jobs):=20end=20the=20error-tombstone=20?= =?UTF-8?q?loop=20=E2=80=94=20deliberate=20retry=20semantics=20+=20poison-?= =?UTF-8?q?job=20guards?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hourly ccip backfill's skip-list lacked 'error' (and the daily siglip/embed variants re-gated failures on their missing results), so every permanently-bad file got a fresh doomed job each run — ~24 duplicate error rows/day per file, the perpetual 'unprocessable' flood. An errored job is now a TOMBSTONE: no backfill re-enqueues it; retry is deliberate-only via /retry_errors (an errored back-catalogue needs one button press after a model swap). One shared set of dedupe DELETEs (services/ml/gpu_jobs.error_dedupe_statements) runs before every backfill and inside /retry_errors: error rows made moot by a later pending/leased/done row go first, then older duplicates (newest reason survives) — so the error count reads as distinct failing files and a retry can't fan one file out into duplicate pending jobs. /retry_errors now returns {requeued, pruned} and the toast shows both. Poison-loop guards (release and lease-expiry burn no attempts, so a job that stalls its transfer or crashes the agent every time cycled forever — operator-observed jobs 99044/125288/131594/143131): - agent: 3 in-session transient bounces (fetch or submit) → fail with the real reason instead of another release; strikes never count while stopping, and clear on submit success. Agent build 2026-07-02.3. - server: the 60s orphan sweep (statements shared between the beat task and GpuJobService so they can't drift) converts expired leases with >=5 lease grants and pending jobs with >=10 to 'error', preserving the last stored failure reason. Backstops old agent builds. Tests: tombstone rule across all three backfill variants, moot-row pruning, poison conversions, and the extended /retry_errors dedupe contract. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM --- agent/fc_agent/app.py | 2 +- agent/fc_agent/worker.py | 55 ++++++++ backend/app/api/gpu.py | 14 +- backend/app/services/ml/gpu_jobs.py | 125 ++++++++++++++++-- backend/app/tasks/ml.py | 70 ++++++---- .../src/components/settings/GpuAgentCard.vue | 5 +- tests/test_api_gpu.py | 36 ++++- tests/test_gpu_jobs.py | 87 +++++++++++- 8 files changed, 343 insertions(+), 51 deletions(-) diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 0642b37..cc33aca 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-02.2 · job.error now carries ffmpeg's actual failure reason" +VERSION = "2026-07-02.3 · poison guard: repeated transient failures fail the job (with reason) instead of releasing it forever" logbuf.install() cfg = Config.from_env() diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 3c54b3d..3544797 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -49,6 +49,16 @@ from .detectors import dedupe_crops # restart needed. MAX_BACKOFF_SECONDS = 60.0 +# A job whose fetch dies transiently this many times IN ONE SESSION stops being +# handed back and is failed instead. Transient handbacks (release) burn no +# attempts on the server, so a poisoned transfer — an original that stalls the +# download every single time — would otherwise release→re-lease forever, churning +# bandwidth without ever landing in the error queue (operator-observed jobs +# 99044/125288/131594/143131, 2026-07-01). Failing it lets curator's attempt cap +# tombstone it WITH the real reason. A genuine curator outage is unaffected: +# every job takes at most one strike per outage, and strikes clear on success. +TRANSIENT_JOB_CAP = 3 + def _is_transient(exc: requests.RequestException) -> bool: """A server/transport problem (wait it out) vs a job-specific fault (fail it). @@ -192,6 +202,11 @@ class Worker: # all at once. Add on lease, discard on every terminal client call. self._held: set[int] = set() self._held_lock = threading.Lock() + # job_id → in-session transient-handback count (guarded by _held_lock). + # Cleared on success or terminal fail; KEPT across releases — persisting + # through the release→re-lease cycle is what lets TRANSIENT_JOB_CAP + # catch a poisoned transfer. + self._transient_seen: dict[int, int] = {} self.processed = 0 self.downloaded = 0 # jobs fetched+decoded into the buffer (monotonic); # feeds the server-side downloads/min rate below. @@ -235,6 +250,21 @@ class Worker: with self._held_lock: self._held.discard(job_id) + def _strike(self, jid: int) -> int: + """Record a transient bounce for this job; returns the in-session total + (compared against TRANSIENT_JOB_CAP by both the fetch and submit paths).""" + with self._held_lock: + n = self._transient_seen.get(jid, 0) + 1 + self._transient_seen[jid] = n + return n + + def _forget_strikes(self, jid: int) -> None: + """Terminal outcome (submitted or failed) → this job's strike count no + longer matters. Deliberately NOT called on release: strikes surviving + the release→re-lease cycle is what makes TRANSIENT_JOB_CAP work.""" + with self._held_lock: + self._transient_seen.pop(jid, None) + def _release(self, job_ids: list[int]) -> None: """Hand still-held leases back to curator and drop them from the held set — the single hand-back path for both a downloader exiting (stop/shrink) with @@ -253,6 +283,7 @@ class Worker: log.warning("job %s (image %s) %s: %s", jid, image_id, verb, str(exc)[:200]) self.client.fail(jid, str(exc)[:500]) self._unhold(jid) + self._forget_strikes(jid) def _stopped(self, stop_evt: threading.Event) -> bool: """The shared 'should I bail now?' check — the worker is stopping (global @@ -580,6 +611,20 @@ class Worker: except requests.RequestException as exc: owned.remove(jid) if _is_transient(exc): + # A Stop mid-fetch lands here too (a killed ffmpeg is + # not the job's fault) — no strike for that; only count + # bounces taken while genuinely running. + if (not self._stopped(stop_evt) + and self._strike(jid) >= TRANSIENT_JOB_CAP): + # THIS job keeps dying while others move: a poisoned + # transfer, not a curator outage. Fail it so the + # server's attempt cap tombstones it with the real + # reason instead of cycling it forever. + self._fail( + jid, job.get("image_id"), exc, + verb="gave up after repeated transient failures", + ) + continue # curator down/redeploying or our lease was reclaimed — # NOT the job's fault. Hand back this job + the rest of the # batch and back the whole loop off. @@ -742,6 +787,7 @@ class Worker: with self._timed("submit"): self.client.submit_embedding(jid, vec, embed_version) self._unhold(jid) + self._forget_strikes(jid) return True # task picks what to produce per crop: @@ -854,9 +900,18 @@ class Worker: with self._timed("submit"): self.client.submit(jid, regions, replace_kinds) self._unhold(jid) + self._forget_strikes(jid) return True except requests.RequestException as exc: if _is_transient(exc): + if (not self._stopped(stop_evt) + and self._strike(jid) >= TRANSIENT_JOB_CAP): + # Same poison rationale as the fetch path: a job whose + # submit keeps dying transiently would otherwise re-lease → + # re-download → re-GPU forever. + self._fail(jid, job.get("image_id"), exc, + verb="gave up after repeated transient failures") + return False # curator down/redeploying, a 5xx, or our lease was reclaimed # while we worked. NOT the job's fault — hand it back (best # effort; then the server's orphan-recovery reclaims it if down). diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index 11605e3..9b3c9d5 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -17,7 +17,7 @@ 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 +from ..services.ml.gpu_jobs import GpuJobService, error_dedupe_statements from ..services.ml.regions import RegionService gpu_bp = Blueprint("gpu", __name__, url_prefix="/api/gpu") @@ -115,9 +115,15 @@ async def retry_errors(): 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.""" + 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.""" 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") @@ -127,7 +133,7 @@ async def retry_errors(): ) ) await session.commit() - return jsonify({"requeued": res.rowcount or 0}) + return jsonify({"requeued": res.rowcount or 0, "pruned": pruned}) # --- Agent (bearer token): lease / submit / heartbeat / fail ------------ diff --git a/backend/app/services/ml/gpu_jobs.py b/backend/app/services/ml/gpu_jobs.py index f41086b..5c20892 100644 --- a/backend/app/services/ml/gpu_jobs.py +++ b/backend/app/services/ml/gpu_jobs.py @@ -12,8 +12,9 @@ and the lease itself reclaims expired leases as a final backstop. Result-writing from datetime import UTC, datetime, timedelta -from sqlalchemy import and_, select, update +from sqlalchemy import and_, delete, exists, func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import aliased from ...models import GpuJob @@ -24,6 +25,107 @@ DEFAULT_LEASE_TTL = 180 # seconds an agent holds a job before it can be re-l DEFAULT_BATCH = 8 MAX_ATTEMPTS = 3 +# Poison-loop backstops. `attempts` counts LEASES GRANTED (incremented in +# lease()), but fail()'s MAX_ATTEMPTS cap only fires when the agent reports a +# failure — a job that keeps coming back via release() (transient handback) or +# lease expiry (agent crash/wedge) never gets a verdict and would cycle forever. +# The orphan sweep converts those to 'error': an expired lease that has already +# been granted EXPIRED_POISON_CAP leases is presumed to kill/wedge the agent, +# and a pending job granted PENDING_POISON_CAP leases without ever completing is +# presumed poisoned (e.g. a transfer that stalls every time). Both stay +# resurrectable via /retry_errors, which resets attempts. +EXPIRED_POISON_CAP = MAX_ATTEMPTS + 2 +PENDING_POISON_CAP = 10 + + +def error_dedupe_statements(): + """DELETEs enforcing: at most ONE error row per (image, task), and none that + a live or succeeded row makes moot. The 2026-07-02 tombstone loop (backfill + skip-lists lacked 'error') minted a duplicate error row per bad file per + hour; running these before every backfill and inside /retry_errors keeps the + error count reading as "distinct failing files" and stops a retry fanning + one file out into several duplicate pending jobs. Shared by the sync beat + task and the async API route so both prune by the SAME predicate. + Execution order matters: moot rows first, then older duplicates (the newest + error — the freshest reason — survives).""" + other = aliased(GpuJob) + same_pair = and_( + other.image_record_id == GpuJob.image_record_id, + other.task == GpuJob.task, + ) + moot = ( + delete(GpuJob) + .where( + GpuJob.status == "error", + exists().where( + same_pair, other.status.in_(["pending", "leased", "done"]), + ), + ) + .execution_options(synchronize_session=False) + ) + older_dupe = ( + delete(GpuJob) + .where( + GpuJob.status == "error", + exists().where( + same_pair, + other.status == "error", + or_( + other.updated_at > GpuJob.updated_at, + and_(other.updated_at == GpuJob.updated_at, + other.id > GpuJob.id), + ), + ), + ) + .execution_options(synchronize_session=False) + ) + return [moot, older_dupe] + + +def recover_statements(now: datetime) -> dict: + """UPDATEs for the orphan sweep, keyed by outcome; insertion order IS the + required execution order ('recovered' must run after 'poison_expired', which + claims the crash-loopers out of the same expired-lease pool).""" + expired = and_(GpuJob.status == "leased", GpuJob.lease_expires_at < now) + unlease = {"lease_token": None, "leased_at": None, "lease_expires_at": None, + "updated_at": now} + return { + "poison_expired": ( + update(GpuJob) + .where(expired, GpuJob.attempts >= EXPIRED_POISON_CAP) + .values( + status="error", + # Keep the job's last stored failure reason — it's the triage + # signal for WHY the loop happened. + error=func.concat( + f"poisoned: lease expired after {EXPIRED_POISON_CAP}+ lease " + "attempts (job repeatedly crashes or wedges the agent?); " + "last error: ", + func.coalesce(GpuJob.error, "none"), + ), + **unlease, + ) + ), + "recovered": update(GpuJob).where(expired).values( + status="pending", **unlease, + ), + "poison_pending": ( + update(GpuJob) + .where(GpuJob.status == "pending", + GpuJob.attempts >= PENDING_POISON_CAP) + .values( + status="error", + error=func.concat( + f"poisoned: {PENDING_POISON_CAP}+ lease attempts without " + "ever completing (transfer stalls every time?); " + "last error: ", + func.coalesce(GpuJob.error, "none"), + ), + updated_at=now, + ) + ), + } + class GpuJobService: def __init__(self, session: AsyncSession): @@ -170,16 +272,11 @@ class GpuJobService: async def recover_orphaned(self) -> int: """Reset every expired lease back to pending — catches agents that died - mid-job (no graceful release). Run on a short beat so the queue recovers - + reads honestly even when no worker is actively leasing. Returns rows - recovered.""" - now = datetime.now(UTC) - res = await self.session.execute( - update(GpuJob) - .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now) - .values( - status="pending", lease_token=None, leased_at=None, - lease_expires_at=None, updated_at=now, - ) - ) - return res.rowcount or 0 + mid-job (no graceful release) — and convert poison-loopers to 'error' + (see the *_POISON_CAP rationale above). Run on a short beat so the queue + recovers + reads honestly even when no worker is actively leasing. + Returns rows recovered to pending (poison conversions are extra).""" + counts = {} + for name, stmt in recover_statements(datetime.now(UTC)).items(): + counts[name] = (await self.session.execute(stmt)).rowcount or 0 + return counts["recovered"] diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 3501c64..895f7e2 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -458,15 +458,30 @@ def enqueue_gpu_backfill(task_name: str) -> int: 'siglip' gates on the RESULT (no concept region yet) rather than on a prior job, so it picks up the back-catalogue of images that were CCIP-embedded - before concept crops existed, and retries images whose concept embed failed — - without re-touching their figure/CCIP regions.""" + before concept crops existed — without re-touching their figure/CCIP regions. + + An ERRORED job is a tombstone for its (image, task): no variant re-enqueues + it. Retry is deliberate-only (/retry_errors), which also means an errored + back-catalogue needs one "Retry errored jobs" press after a model swap. + Before the tombstone rule, this loop re-minted a fresh doomed job for every + permanently-bad file each run — ~24 duplicate error rows/day per file (the + 2026-07-02 "unprocessable" flood).""" from sqlalchemy import exists, insert, literal, or_ from sqlalchemy import select as sa_select from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings + from ..services.ml.gpu_jobs import error_dedupe_statements SessionLocal = _sync_session_factory() with SessionLocal() as session: + # Prune stale tombstones first (loop-era duplicates + rows made moot by + # a later success), so 'error' reads as one row per distinct failing + # file and the skip-guards below see a clean picture. + pruned = sum( + session.execute(s).rowcount or 0 for s in error_dedupe_statements() + ) + if pruned: + log.info("gpu backfill: pruned %d stale/duplicate error rows", pruned) cur_version = session.execute( select(MLSettings.embedder_model_version).where(MLSettings.id == 1) ).scalar_one() @@ -478,14 +493,15 @@ def enqueue_gpu_backfill(task_name: str) -> int: ImageRecord.siglip_model_version.is_(None), ImageRecord.siglip_model_version != cur_version, ) - queued = exists().where( + # 'error' blocks too — tombstone rule, see docstring. + blocked = exists().where( GpuJob.image_record_id == ImageRecord.id, GpuJob.task == "embed", - GpuJob.status.in_(["pending", "leased"]), + GpuJob.status.in_(["pending", "leased", "error"]), ) sel = sa_select( ImageRecord.id, literal("embed"), literal("pending") - ).where(stale).where(~queued) + ).where(stale).where(~blocked) elif task_name == "siglip": # Concept-crop re-embed: enqueue when there's no concept region AT THE # CURRENT model version — so a model swap re-triggers crops too, not @@ -495,19 +511,22 @@ def enqueue_gpu_backfill(task_name: str) -> int: ImageRegion.kind == "concept", ImageRegion.embedding_version == cur_version, ) - queued = exists().where( + # 'error' blocks too — tombstone rule, see docstring. + blocked = exists().where( GpuJob.image_record_id == ImageRecord.id, GpuJob.task == "siglip", - GpuJob.status.in_(["pending", "leased"]), + GpuJob.status.in_(["pending", "leased", "error"]), ) sel = sa_select( ImageRecord.id, literal("siglip"), literal("pending") - ).where(~has_current_concept).where(~queued) + ).where(~has_current_concept).where(~blocked) else: + # ANY prior row blocks — including 'error' (tombstone rule, see + # docstring): pre-fix this branch ran HOURLY and was the loop. already = exists().where( GpuJob.image_record_id == ImageRecord.id, GpuJob.task == task_name, - GpuJob.status.in_(["pending", "leased", "done"]), + GpuJob.status.in_(["pending", "leased", "done", "error"]), ) sel = sa_select( ImageRecord.id, literal(task_name), literal("pending") @@ -525,28 +544,29 @@ def enqueue_gpu_backfill(task_name: str) -> int: @celery.task(name="backend.app.tasks.ml.recover_orphaned_gpu_jobs") def recover_orphaned_gpu_jobs() -> int: """Reset expired GPU-job leases back to pending — recovers work orphaned by an - agent that died mid-job (no graceful release). Short beat cadence so orphans - get picked back up quickly + the queue counts read honestly. Returns the - number recovered.""" + agent that died mid-job (no graceful release) — and convert poison-loopers + (release/expiry cycles that never reach fail()'s attempt cap) to 'error'. + Statements are shared with GpuJobService.recover_orphaned so the sweep and + the service can't drift. Short beat cadence so orphans get picked back up + quickly + the queue counts read honestly. Returns the number recovered.""" from datetime import UTC, datetime - from sqlalchemy import update - - from ..models import GpuJob + from ..services.ml.gpu_jobs import recover_statements SessionLocal = _sync_session_factory() with SessionLocal() as session: - now = datetime.now(UTC) - res = session.execute( - update(GpuJob) - .where(GpuJob.status == "leased", GpuJob.lease_expires_at < now) - .values( - status="pending", lease_token=None, leased_at=None, - lease_expires_at=None, updated_at=now, - ) - ) + counts = { + name: session.execute(stmt).rowcount or 0 + for name, stmt in recover_statements(datetime.now(UTC)).items() + } session.commit() - return res.rowcount or 0 + if counts["poison_expired"] or counts["poison_pending"]: + log.warning( + "gpu jobs poisoned -> error: %d crash-loop (expired lease), " + "%d never-complete (pending)", + counts["poison_expired"], counts["poison_pending"], + ) + return counts["recovered"] @celery.task(name="backend.app.tasks.ml.reprocess_gpu_jobs") diff --git a/frontend/src/components/settings/GpuAgentCard.vue b/frontend/src/components/settings/GpuAgentCard.vue index c577937..9f4b0dd 100644 --- a/frontend/src/components/settings/GpuAgentCard.vue +++ b/frontend/src/components/settings/GpuAgentCard.vue @@ -337,8 +337,9 @@ 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' }) + const { requeued, pruned } = await store.retryErrors() + const extra = pruned ? ` (${pruned} stale duplicate${pruned === 1 ? '' : 's'} pruned)` : '' + toast({ text: `Requeued ${requeued} errored job${requeued === 1 ? '' : 's'}${extra} — run the agent to process them`, type: 'success' }) await refreshQueue() } catch (e) { toast({ text: `Could not retry errored jobs: ${e.message}`, type: 'error' }) diff --git a/tests/test_api_gpu.py b/tests/test_api_gpu.py index b714438..ed01efb 100644 --- a/tests/test_api_gpu.py +++ b/tests/test_api_gpu.py @@ -1,6 +1,8 @@ """GPU-job HTTP API (#114): bearer auth + lease/submit round-trip + backfill.""" +from datetime import UTC, datetime, timedelta + import pytest -from sqlalchemy import select +from sqlalchemy import func, select from backend.app.models import GpuJob, ImageRecord from backend.app.services.ml.gpu_jobs import GpuJobService @@ -153,8 +155,11 @@ async def test_release_hands_job_back_to_pending(client, db): @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).""" + """/retry_errors prunes stale tombstones first (older duplicates + rows a + later success made moot), then resets the SURVIVING errored jobs to pending + with a fresh retry budget — and leaves done work untouched (NOT /reprocess). + The prune is what stops one failing file fanning out into duplicate pending + jobs (the 2026-07-02 tombstone loop minted one error row per hour).""" img1 = await _img(db, "1" * 64) img2 = await _img(db, "2" * 64) svc = GpuJobService(db) @@ -165,12 +170,30 @@ async def test_retry_errors_requeues_only_errored(client, db): j_err.status = "error" j_err.attempts = 3 j_err.error = "no frames sampled from video (unprocessable)" + j_err.updated_at = datetime.now(UTC) j_done.status = "done" + # Loop-era leftovers: an OLDER duplicate error row for img1's ccip, and a + # tombstone img2's done row makes moot — both pruned, never requeued. + dup = GpuJob( + image_record_id=img1.id, task="ccip", status="error", + error="older duplicate", updated_at=datetime.now(UTC) - timedelta(hours=1), + ) + moot = GpuJob( + image_record_id=img2.id, task="siglip", status="error", + error="superseded by the done row", + ) + db.add(dup) + db.add(moot) + await db.flush() + dup_id = dup.id + moot_id = moot.id await db.commit() resp = await client.post("/api/gpu/retry_errors") assert resp.status_code == 200 - assert (await resp.get_json())["requeued"] == 1 + body = await resp.get_json() + assert body["requeued"] == 1 + assert body["pruned"] == 2 # Column selects, not ORM refresh — the route wrote via Core DML. row = (await db.execute( @@ -182,6 +205,11 @@ async def test_retry_errors_requeues_only_errored(client, db): select(GpuJob.status).where(GpuJob.id == done_id) ) assert done_status == "done" + survivors = (await db.execute( + select(func.count()).select_from(GpuJob) + .where(GpuJob.id.in_([dup_id, moot_id])) + )).scalar_one() + assert survivors == 0 st = await (await client.get("/api/gpu/status")).get_json() assert st["pending"] == 1 and st["error"] == 0 diff --git a/tests/test_gpu_jobs.py b/tests/test_gpu_jobs.py index 0678dc5..930ea02 100644 --- a/tests/test_gpu_jobs.py +++ b/tests/test_gpu_jobs.py @@ -5,7 +5,11 @@ import pytest from sqlalchemy import func, select from backend.app.models import GpuJob, ImageRecord, ImageRegion -from backend.app.services.ml.gpu_jobs import GpuJobService +from backend.app.services.ml.gpu_jobs import ( + EXPIRED_POISON_CAP, + PENDING_POISON_CAP, + GpuJobService, +) pytestmark = pytest.mark.integration @@ -263,3 +267,84 @@ async def test_recover_orphaned_resets_only_expired(db): await db.commit() assert (await db.get(GpuJob, expired.id)).status == "pending" assert (await db.get(GpuJob, fresh.id)).status == "leased" # untouched + + +@pytest.mark.asyncio +async def test_backfill_skips_errored_images(db): + # An errored job is a TOMBSTONE for its (image, task): no backfill variant + # re-enqueues it — retry is deliberate-only (/retry_errors). Pre-fix, the + # hourly ccip run minted a fresh doomed job per bad file forever. + from backend.app.tasks.ml import enqueue_gpu_backfill + + img = await _img(db, "f1" * 32) + svc = GpuJobService(db) + for task in ("ccip", "siglip", "embed"): + job = await svc.enqueue(img.id, task) + job.status = "error" + job.error = "no frames sampled from video" + await db.commit() + + assert enqueue_gpu_backfill("ccip") == 0 + assert enqueue_gpu_backfill("siglip") == 0 + assert enqueue_gpu_backfill("embed") == 0 + + +@pytest.mark.asyncio +async def test_backfill_prunes_moot_error_tombstones(db): + # Loop-era duplicates: several error rows for one (image, task), all made + # moot by a later done row. The backfill's dedupe pass removes them, and + # the done row still blocks re-enqueue. + from backend.app.tasks.ml import enqueue_gpu_backfill + + img = await _img(db, "f2" * 32) + for i in range(3): + db.add(GpuJob( + image_record_id=img.id, task="ccip", status="error", + error=f"boom {i}", + )) + db.add(GpuJob(image_record_id=img.id, task="ccip", status="done")) + await db.commit() + + assert enqueue_gpu_backfill("ccip") == 0 + + statuses = (await db.execute( + select(GpuJob.status).where( + GpuJob.image_record_id == img.id, GpuJob.task == "ccip" + ) + )).scalars().all() + assert statuses == ["done"] + + +@pytest.mark.asyncio +async def test_recover_poisons_runaway_jobs(db): + # release/expiry loops never reach fail()'s attempt cap — the sweep converts + # them to 'error': an expired lease after EXPIRED_POISON_CAP grants (job + # crashes/wedges the agent every time) and a pending job after + # PENDING_POISON_CAP grants that never completed (transfer stalls forever). + img1 = await _img(db, "06" + "a" * 62) + img2 = await _img(db, "07" + "a" * 62) + svc = GpuJobService(db) + j1 = await svc.enqueue(img1.id, "ccip") + j2 = await svc.enqueue(img2.id, "ccip") + j1.status = "leased" + j1.lease_token = "dead" + j1.lease_expires_at = datetime.now(UTC) - timedelta(minutes=10) + j1.attempts = EXPIRED_POISON_CAP + j2.attempts = PENDING_POISON_CAP + j1_id = j1.id + j2_id = j2.id + await db.commit() + + assert await svc.recover_orphaned() == 0 # nothing recovered — both poisoned + await db.commit() + + # Column selects, not ORM refresh — the sweep wrote via Core DML. + rows = (await db.execute( + select(GpuJob.id, GpuJob.status, GpuJob.error) + .where(GpuJob.id.in_([j1_id, j2_id])) + )).all() + by_id = {r.id: r for r in rows} + assert by_id[j1_id].status == "error" + assert "poisoned" in by_id[j1_id].error + assert by_id[j2_id].status == "error" + assert "poisoned" in by_id[j2_id].error