From b1778ca9f2a91c73f106ee86c0e463bc08912a7b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 8 Jun 2026 08:49:37 -0400 Subject: [PATCH] obs(ml): tag_and_embed logs file + phase + timing; failures name them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The task logged nothing and SoftTimeLimitExceeded stringifies to empty, so a timeout surfaced as a bare 'SoftTimeLimitExceeded()' with no clue which file or why (operator-flagged 2026-06-08). - Log start (id/path/mime/bytes/video?), per-phase timing (load_models, video probe/sample/infer, tag, embed, persist), and a success summary. - Track a + file ; on SoftTimeLimitExceeded log it and re-raise SoftTimeLimitExceeded WITH that context (keeps the 'timeout' task_run status but gives the activity a real error_message: which file, which phase, elapsed). - On other exceptions, log context then re-raise the ORIGINAL (preserves autoretry for OSError/DBAPIError/OperationalError). Now a stuck run names the culprit — most likely a slow video (frame sampling is up to 10x60s ffmpeg) or a huge image; the phase log will say which. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/tasks/ml.py | 173 ++++++++++++++++++++++++++++------------ 1 file changed, 123 insertions(+), 50 deletions(-) diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 8867cee..3be9f7b 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -6,8 +6,10 @@ apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions (Celery workers are sync processes), same pattern as FC-2a tasks. """ +import logging from pathlib import Path +from celery.exceptions import SoftTimeLimitExceeded from sqlalchemy import select from sqlalchemy.exc import DBAPIError, OperationalError @@ -15,6 +17,8 @@ from ..celery_app import celery from ..models import ImageRecord, MLSettings from ._sync_engine import sync_session_factory as _sync_session_factory +log = logging.getLogger(__name__) + IMAGES_ROOT = Path("/images") VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"} @@ -50,67 +54,136 @@ def tag_and_embed(self, image_id: int) -> dict: SigLIP embeddings. On no-frames returns status='no_frames' (not an error). """ import os + import time from ..services.ml.embedder import get_embedder from ..services.ml.tagger import get_tagger - SessionLocal = _sync_session_factory() - with SessionLocal() as session: - record = session.get(ImageRecord, image_id) - if record is None: - return {"status": "missing", "image_id": image_id} - settings = session.execute( - select(MLSettings).where(MLSettings.id == 1) - ).scalar_one() + # Phase + file context, so a timeout/crash names WHICH file and WHERE it + # died instead of a bare SoftTimeLimitExceeded() (operator-flagged 2026-06-08: + # the activity told them nothing about the file or why). `ctx` is enriched + # once the record is loaded; both feed the worker log AND the re-raised + # exception message (which becomes the activity's error_message). + started = time.monotonic() + phase = "open_session" + ctx = f"image_id={image_id}" - src = Path(record.path) - if not src.is_file(): - return {"status": "file_missing", "image_id": image_id} + def _elapsed() -> float: + return time.monotonic() - started - tagger = get_tagger() - embedder = get_embedder() + try: + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + record = session.get(ImageRecord, image_id) + if record is None: + return {"status": "missing", "image_id": image_id} + settings = session.execute( + select(MLSettings).where(MLSettings.id == 1) + ).scalar_one() - if _is_video(src): - # Layer-3 isolation: ffprobe (a separate process) validates - # the container before we burn ~20 GPU ops sampling frames - # from it. A corrupt video that would crash the frame - # decoder is rejected cleanly here instead of taking down - # the ml-worker. Operator-flagged 2026-05-28. - from ..utils import safe_probe - vprobe = safe_probe.probe_video(src) - if not vprobe.ok: - return { - "status": "bad_video", "image_id": image_id, - "reason": vprobe.reason, - } - frames = _sample_video_frames( - src, int(os.environ.get("VIDEO_ML_FRAMES", "10")) + src = Path(record.path) + is_vid = _is_video(src) + ctx = ( + f"image_id={image_id} path={record.path} mime={record.mime} " + f"bytes={record.size_bytes} video={is_vid}" ) - if not frames: - return {"status": "no_frames", "image_id": image_id} - preds = _maxpool_predictions([tagger.infer(f) for f in frames]) - import numpy as np + log.info("tag_and_embed start: %s", ctx) + if not src.is_file(): + log.warning("tag_and_embed file missing on disk: %s", ctx) + return {"status": "file_missing", "image_id": image_id} - embedding = np.mean( - [embedder.infer(f) for f in frames], axis=0 - ).astype("float32") - for f in frames: - f.unlink(missing_ok=True) - else: - raw = tagger.infer(src) - preds = { - name: {"category": p.category, "confidence": p.confidence} - for name, p in raw.items() - } - embedding = embedder.infer(src) + phase = "load_models" + tagger = get_tagger() + embedder = get_embedder() - record.tagger_predictions = preds - record.tagger_model_version = settings.tagger_model_version - record.siglip_embedding = embedding.tolist() - record.siglip_model_version = settings.embedder_model_version - session.add(record) - session.commit() + if is_vid: + # Layer-3 isolation: ffprobe (a separate process) validates + # the container before we burn ~20 GPU ops sampling frames + # from it. A corrupt video that would crash the frame + # decoder is rejected cleanly here instead of taking down + # the ml-worker. Operator-flagged 2026-05-28. + phase = "video_probe" + from ..utils import safe_probe + vprobe = safe_probe.probe_video(src) + if not vprobe.ok: + log.warning( + "tag_and_embed bad video (%s): %s", vprobe.reason, ctx + ) + return { + "status": "bad_video", "image_id": image_id, + "reason": vprobe.reason, + } + phase = "video_sample_frames" + t0 = time.monotonic() + frames = _sample_video_frames( + src, int(os.environ.get("VIDEO_ML_FRAMES", "10")) + ) + log.info( + "tag_and_embed sampled %d frame(s) in %.1fs: %s", + len(frames), time.monotonic() - t0, ctx, + ) + if not frames: + return {"status": "no_frames", "image_id": image_id} + phase = "video_infer" + import numpy as np + preds = _maxpool_predictions([tagger.infer(f) for f in frames]) + embedding = np.mean( + [embedder.infer(f) for f in frames], axis=0 + ).astype("float32") + for f in frames: + f.unlink(missing_ok=True) + else: + phase = "tag" + t0 = time.monotonic() + raw = tagger.infer(src) + log.info( + "tag_and_embed tagged in %.1fs (%d tags): %s", + time.monotonic() - t0, len(raw), ctx, + ) + preds = { + name: {"category": p.category, "confidence": p.confidence} + for name, p in raw.items() + } + phase = "embed" + t0 = time.monotonic() + embedding = embedder.infer(src) + log.info( + "tag_and_embed embedded in %.1fs: %s", + time.monotonic() - t0, ctx, + ) + + phase = "persist" + record.tagger_predictions = preds + record.tagger_model_version = settings.tagger_model_version + record.siglip_embedding = embedding.tolist() + record.siglip_model_version = settings.embedder_model_version + session.add(record) + session.commit() + except SoftTimeLimitExceeded: + log.error( + "tag_and_embed TIMED OUT after %.0fs in phase=%s: %s", + _elapsed(), phase, ctx, + ) + # Re-raise as SoftTimeLimitExceeded (preserves the 'timeout' status in + # the task_run signal) but WITH context, so the activity error_message + # names the file + phase instead of being empty. + raise SoftTimeLimitExceeded( + f"timed out in phase={phase} after {_elapsed():.0f}s ({ctx})" + ) from None + except Exception: + # OSError/DBAPIError/OperationalError are autoretried — re-raise the + # ORIGINAL so the type is preserved; just make sure it's logged with + # context first. + log.exception( + "tag_and_embed FAILED in phase=%s after %.0fs: %s", + phase, _elapsed(), ctx, + ) + raise + + log.info( + "tag_and_embed ok in %.1fs (%d tags): %s", _elapsed(), len(preds), ctx + ) apply_allowlist_tags.delay(image_id=image_id) return {"status": "ok", "image_id": image_id, "tags": len(preds)}