From 407de18ff61b687a7658adf7aac82c661714ae00 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 27 May 2026 22:23:35 -0400 Subject: [PATCH] fix(ml): video branch needs longer time limits; recovery sweep is now per-queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Operator-flagged 2026-05-28: tag_and_embed on image 6288 (an mp4) was marked failed by recover_stalled_task_runs at the 5-min sweep tick while still legitimately running. The error_type='RecoverySweep' / "no completion signal received within 5 min" message was misleading — the worker was busy, not stuck. Root cause is two interacting limits, both undersized for video work: tag_and_embed: soft_time_limit=300, time_limit=420 (sized for the image branch, ≈2 GPU ops) recovery sweep: STUCK_THRESHOLD_MINUTES = 5 across all queues The video branch samples 10 frames via ffmpeg, then runs tagger + embedder on EACH frame — ~20 GPU ops vs 2 for an image. A loaded ml-worker can take 5-10 min on a long video, which trips both limits well before the task naturally finishes. **Two-part fix** 1. `tag_and_embed` time limits bumped to soft=900 (15 min) / time=1200 (20 min). Sized for the video path's worst case; image runs return in seconds and don't care. 2. New `QUEUE_STUCK_THRESHOLD_MINUTES` override dict in maintenance.py. Queues with legitimately-long-running tasks (currently just `ml` at 25 min — 5-min buffer past the new hard kill) get their own threshold; queues not in the dict use the default 5 min. The sweep now issues one UPDATE per distinct threshold value, with `queue.notin_(override_queues)` on the default pass so each row is touched at most once. Tests: - _make_task_run helper accepts `queue=` (defaults to "default") so existing tests use the default-threshold path. - New test `test_recover_stalled_task_runs_ml_queue_uses_longer_threshold` pins both directions: a 10-min-old ml row survives (fresh by 25-min override), a 30-min-old ml row gets flagged. After deploy, operator's mp4 ML jobs run to completion without spurious RecoverySweep failures. --- backend/app/tasks/maintenance.py | 73 ++++++++++++++++++++++++++++---- backend/app/tasks/ml.py | 11 ++++- tests/test_maintenance.py | 45 +++++++++++++++++++- 3 files changed, 116 insertions(+), 13 deletions(-) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index df854a3..c5bb617 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -24,6 +24,21 @@ FFPROBE_TIMEOUT_SECONDS = 10 TASK_RUN_KEEP_OK_SECONDS = 24 * 3600 # 24 h TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600 # 7 days +# Per-queue overrides for recover_stalled_task_runs. Queues whose +# tasks can legitimately run longer than the default 5-min threshold +# need their own larger value, otherwise the sweep marks in-flight +# tasks 'error' before they get a chance to finish. The dict's value +# MUST be ≥ the longest task.time_limit on the queue + a small buffer. +# +# ml: tag_and_embed video branch samples 10 frames, runs tagger + +# embedder on each — soft_time_limit=900 / time_limit=1200; sweep +# at 25 min gives a 5-min buffer past the hard kill. +# Operator-flagged 2026-05-28 (image 6288, an mp4, marked failed +# at the 5-min sweep tick while still processing). +QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { + "ml": 25, +} + @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") def recover_interrupted_tasks() -> int: @@ -121,21 +136,35 @@ def cleanup_old_tasks() -> int: @celery.task(name="backend.app.tasks.maintenance.recover_stalled_task_runs") def recover_stalled_task_runs() -> int: - """Flip task_run rows stuck in 'running' for >STUCK_THRESHOLD_MINUTES - to 'error'. FC-3i. + """Flip task_run rows stuck in 'running' past their queue-specific + threshold to 'error'. FC-3i. A row gets stuck when the worker dies without emitting task_postrun / task_failure (e.g. OOM, container restart between - signals, signal handler raised+logged). Shares the 5-min threshold - with recover_interrupted_tasks for consistency. + signals, signal handler raised+logged). The default 5-min threshold + fits short-lived queues (import/thumbnail/download); queues that + legitimately run longer tasks (ml-video, deep scans) get their + own larger threshold via QUEUE_STUCK_THRESHOLD_MINUTES so the + sweep doesn't preempt them. + + Runs once per distinct threshold value: each pass updates rows + whose queue maps to that threshold. """ SessionLocal = _sync_session_factory() - cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + now = datetime.now(UTC) + # Group queues by their threshold value so we issue one UPDATE + # per distinct threshold. Queues NOT in the override dict use the + # default; their UPDATE excludes the override queues so each row + # is touched at most once. + override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys()) + total = 0 with SessionLocal() as session: - result = session.execute( + # Default-threshold pass — all queues except the overridden ones. + default_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + default_stmt = ( update(TaskRun) .where(TaskRun.status == "running") - .where(TaskRun.started_at < cutoff) + .where(TaskRun.started_at < default_cutoff) .values( status="error", error_type="RecoverySweep", @@ -143,11 +172,37 @@ def recover_stalled_task_runs() -> int: f"no completion signal received within " f"{STUCK_THRESHOLD_MINUTES} min" ), - finished_at=datetime.now(UTC), + finished_at=now, ) ) + if override_queues: + default_stmt = default_stmt.where( + TaskRun.queue.notin_(override_queues) + ) + total += session.execute(default_stmt).rowcount or 0 + + # Per-queue override passes. + for queue, minutes in QUEUE_STUCK_THRESHOLD_MINUTES.items(): + cutoff = now - timedelta(minutes=minutes) + stmt = ( + update(TaskRun) + .where(TaskRun.status == "running") + .where(TaskRun.queue == queue) + .where(TaskRun.started_at < cutoff) + .values( + status="error", + error_type="RecoverySweep", + error_message=( + f"no completion signal received within " + f"{minutes} min" + ), + finished_at=now, + ) + ) + total += session.execute(stmt).rowcount or 0 + session.commit() - return result.rowcount or 0 + return total @celery.task(name="backend.app.tasks.maintenance.prune_task_runs") diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 75de41d..c4b8ddc 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -31,8 +31,15 @@ def _is_video(path: Path) -> bool: retry_backoff_max=60, retry_jitter=True, max_retries=3, - soft_time_limit=300, - time_limit=420, + # Sized for the video branch: sample 10 frames, run tagger + + # embedder on each (≈20 GPU ops vs 2 for an image). A loaded + # ml-worker can take 5-10 min on a long video; bumped from + # 5min/7min on 2026-05-28 after operator-flagged image 6288 (a + # .mp4) hit the recovery sweep at 5 min while still legitimately + # processing. Image runs return in seconds; the bump doesn't + # affect their UX. + soft_time_limit=900, # 15 min + time_limit=1200, # 20 min hard ) def tag_and_embed(self, image_id: int) -> dict: """Run Camie + SigLIP on one image; store predictions + embedding; diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 7fc04e9..9f2803e 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -195,11 +195,11 @@ def test_cleanup_old_deletes_finished_old(db_sync): def _make_task_run(db_sync, *, status, started_at, finished_at=None, - error_type=None): + error_type=None, queue="default"): from backend.app.models import TaskRun row = TaskRun( celery_task_id="x", - queue="ml", + queue=queue, task_name="backend.app.tasks.fake.t", target_id=1, started_at=started_at, @@ -255,6 +255,47 @@ def test_recover_stalled_task_runs_skips_fresh_running(db_sync): recovered = recover_stalled_task_runs.apply().get() assert recovered == 0 + +def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync): + """ml-queue tasks (tag_and_embed video branch) legitimately run + past the default 5-min threshold. The sweep must NOT flag an + ml-queue task that's only been running 10 min — the override + threshold (25 min via QUEUE_STUCK_THRESHOLD_MINUTES) protects + in-flight video tagging. Operator-flagged 2026-05-28 after + image 6288 (mp4) was marked failed at the 5-min tick mid-run.""" + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import recover_stalled_task_runs + + now = datetime.now(UTC) + # 10-min-old ml-queue row: stale by the default 5-min rule but + # fresh by the 25-min ml override. Must survive the sweep. + ml_fresh_id = _make_task_run( + db_sync, status="running", queue="ml", + started_at=now - timedelta(minutes=10), + ) + # 30-min-old ml-queue row: past even the ml override. Must be + # flagged. + ml_stale_id = _make_task_run( + db_sync, status="running", queue="ml", + started_at=now - timedelta(minutes=30), + ) + db_sync.commit() + + recovered = recover_stalled_task_runs.apply().get() + assert recovered == 1 + + db_sync.expire_all() + ml_fresh_status = db_sync.execute( + select(TaskRun.status).where(TaskRun.id == ml_fresh_id) + ).scalar_one() + ml_stale_status = db_sync.execute( + select(TaskRun.status).where(TaskRun.id == ml_stale_id) + ).scalar_one() + assert ml_fresh_status == "running" + assert ml_stale_status == "error" + db_sync.expire_all() status = db_sync.execute( select(TaskRun.status).where(TaskRun.id == fresh_id)