fix(workers): worker-level recovery — autoretry on transient errors + tightened sweep (5min) + import_media_file body-wrap so no path leaves rows stuck in 'processing'

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-24 11:46:36 -04:00
parent dd4874ae8d
commit be0f472894
5 changed files with 195 additions and 96 deletions
@@ -4,6 +4,7 @@ import asyncio
 from pathlib import Path

 from sqlalchemy import select
+from sqlalchemy.exc import DBAPIError, OperationalError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

 from ..celery_app import celery
@@ -27,7 +28,18 @@ def _async_session_factory():
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine


-@celery.task(name="backend.app.tasks.download.download_source", bind=True, acks_late=True)
+@celery.task(
+    name="backend.app.tasks.download.download_source",
+    bind=True,
+    acks_late=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=10,
+    retry_backoff_max=120,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=900,
+    time_limit=1200,
+)
 def download_source(self, source_id: int) -> int:
    """Returns the DownloadEvent.id."""

@@ -1,11 +1,21 @@
 """import_media_file task: imports one file via the Importer service and
 updates the ImportTask state machine + ImportBatch counters atomically.
+
+Resilience contract (2026-05-24, operator-mandated): once a row has been
+flipped to 'processing' inside this task, EVERY exit path MUST flip it
+to a terminal state (complete / skipped / failed) or rely on Celery's
+autoretry to attempt the work again. No exit path is allowed to leave
+the row stuck in 'processing'. The `recover_interrupted_tasks`
+maintenance sweep is the safety net (5 min threshold) for the case
+where even the failure-marking commit can't be written.
 """

 from datetime import UTC, datetime
 from pathlib import Path

+from celery.exceptions import SoftTimeLimitExceeded
 from sqlalchemy import select, update
+from sqlalchemy.exc import DBAPIError, OperationalError

 from ..celery_app import celery
 from ..models import ImportBatch, ImportSettings, ImportTask
@@ -29,9 +39,54 @@ def _map_result_to_status(result):
    return ("failed", False)


-@celery.task(name="backend.app.tasks.import_file.import_media_file", bind=True)
+def _mark_failed(session, task, error_msg: str) -> None:
+    """Best-effort flip of a 'processing' row to 'failed' + batch counter
+    increment. Wrapped in its own try because if the DB is what just
+    broke, this commit will also fail — that's why the maintenance sweep
+    exists as a backstop."""
+    try:
+        task.status = "failed"
+        task.error = error_msg
+        task.finished_at = datetime.now(UTC)
+        session.add(task)
+        session.execute(
+            update(ImportBatch)
+            .where(ImportBatch.id == task.batch_id)
+            .values(failed=ImportBatch.failed + 1)
+        )
+        session.commit()
+    except Exception:  # noqa: BLE001 — best-effort, sweep catches the rest
+        try:
+            session.rollback()
+        except Exception:  # noqa: BLE001
+            pass
+
+
+@celery.task(
+    name="backend.app.tasks.import_file.import_media_file",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=300,
+    time_limit=360,
+)
 def import_media_file(self, import_task_id: int) -> dict:
-    """Returns a dict so the eager-mode tests can assert without DB."""
+    """Returns a dict so the eager-mode tests can assert without DB.
+
+    Decorator notes:
+      - autoretry_for: transient DB / filesystem errors retry with
+        exponential backoff (5s base, jitter, max 3 attempts). On final
+        give-up the task raises and acks_late=True (set globally on the
+        Celery app) does NOT redeliver — the recovery sweep catches the
+        row instead.
+      - soft_time_limit (300s) raises SoftTimeLimitExceeded in this
+        process so the task can mark its row failed before being killed.
+      - time_limit (360s) is the hard cap; SIGKILL if the soft signal
+        was swallowed.
+    """
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session:
        task = session.get(ImportTask, import_task_id)
@@ -43,99 +98,104 @@ def import_media_file(self, import_task_id: int) -> dict:
        session.add(task)
        session.commit()

-        settings = session.execute(
-            select(ImportSettings).where(ImportSettings.id == 1)
-        ).scalar_one()
-        import_root = Path(settings.import_scan_path)
-        batch = session.get(ImportBatch, task.batch_id)
-        deep = bool(batch and batch.scan_mode == "deep")
-        importer = Importer(
-            session=session,
-            images_root=IMAGES_ROOT,
-            import_root=import_root,
-            thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
-            settings=settings,
-            deep=deep,
-        )
-
        try:
-            result = importer.import_one(Path(task.source_path))
-        except Exception as exc:  # pragma: no cover — pipeline crash
-            task.status = "failed"
-            task.error = f"{type(exc).__name__}: {exc}"
-            task.finished_at = datetime.now(UTC)
-            session.execute(
-                update(ImportBatch)
-                .where(ImportBatch.id == task.batch_id)
-                .values(failed=ImportBatch.failed + 1)
-            )
-            session.add(task)
-            session.commit()
+            return _do_import(session, task, import_task_id)
+        except SoftTimeLimitExceeded:
+            _mark_failed(session, task, "soft_time_limit exceeded (>300s)")
+            raise
+        except (OperationalError, DBAPIError, OSError):
+            # Retryable per the decorator; do NOT mark failed (let
+            # autoretry have a clean go at it). If autoretry exhausts,
+            # the row stays 'processing' and the maintenance sweep
+            # flips it within 5 min.
+            raise
+        except Exception as exc:  # noqa: BLE001 — pipeline crash, mark + re-raise
+            _mark_failed(session, task, f"{type(exc).__name__}: {exc}")
            raise

-        if result.status in ("imported", "superseded"):
-            task.status = "complete"
-            task.result_image_id = result.image_id
-            counter_col_name = "imported"
-            counter_col = ImportBatch.imported
-        elif result.status == "attached":
-            task.status = "complete"
-            counter_col_name = "attachments"
-            counter_col = ImportBatch.attachments
-        elif result.status == "skipped":
-            task.status = "skipped"
-            task.error = (
-                f"{result.skip_reason.value}: {result.error}"
-                if result.skip_reason
-                else result.error
-            )
-            task.result_image_id = result.image_id
-            counter_col_name = "skipped"
-            counter_col = ImportBatch.skipped
-        else:
-            task.status = "failed"
-            task.error = result.error
-            counter_col_name = "failed"
-            counter_col = ImportBatch.failed

-        task.finished_at = datetime.now(UTC)
+def _do_import(session, task, import_task_id: int) -> dict:
+    """Actual work, called from inside the resilience wrapper."""
+    settings = session.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    import_root = Path(settings.import_scan_path)
+    batch = session.get(ImportBatch, task.batch_id)
+    deep = bool(batch and batch.scan_mode == "deep")
+    importer = Importer(
+        session=session,
+        images_root=IMAGES_ROOT,
+        import_root=import_root,
+        thumbnailer=Thumbnailer(images_root=IMAGES_ROOT),
+        settings=settings,
+        deep=deep,
+    )
+
+    result = importer.import_one(Path(task.source_path))
+
+    if result.status in ("imported", "superseded"):
+        task.status = "complete"
+        task.result_image_id = result.image_id
+        counter_col_name = "imported"
+        counter_col = ImportBatch.imported
+    elif result.status == "attached":
+        task.status = "complete"
+        counter_col_name = "attachments"
+        counter_col = ImportBatch.attachments
+    elif result.status == "skipped":
+        task.status = "skipped"
+        task.error = (
+            f"{result.skip_reason.value}: {result.error}"
+            if result.skip_reason
+            else result.error
+        )
+        task.result_image_id = result.image_id
+        counter_col_name = "skipped"
+        counter_col = ImportBatch.skipped
+    else:
+        task.status = "failed"
+        task.error = result.error
+        counter_col_name = "failed"
+        counter_col = ImportBatch.failed
+
+    task.finished_at = datetime.now(UTC)
+    session.execute(
+        update(ImportBatch)
+        .where(ImportBatch.id == task.batch_id)
+        .values({counter_col_name: counter_col + 1})
+    )
+    session.add(task)
+    session.commit()
+
+    # Enqueue thumbnail + ML for newly imported AND superseded images
+    # (a superseded row has cleared ML + no thumbnail).
+    if result.status in ("imported", "superseded"):
+        from .ml import tag_and_embed
+        from .thumbnail import generate_thumbnail
+
+        ids = list(result.member_image_ids)
+        if result.image_id is not None and result.image_id not in ids:
+            ids.append(result.image_id)
+        for img_id in ids:
+            generate_thumbnail.delay(img_id)
+            tag_and_embed.delay(img_id)
+
+    # If this was the last task in the batch, mark the batch complete.
+    remaining = session.execute(
+        select(ImportTask.id)
+        .where(ImportTask.batch_id == task.batch_id)
+        .where(ImportTask.status.in_(["pending", "queued", "processing"]))
+        .limit(1)
+    ).scalar_one_or_none()
+    if remaining is None:
        session.execute(
            update(ImportBatch)
            .where(ImportBatch.id == task.batch_id)
-            .values({counter_col_name: counter_col + 1})
+            .values(
+                status="complete",
+                finished_at=datetime.now(UTC),
+            )
        )
-        session.add(task)
        session.commit()

-        # Enqueue thumbnail + ML for newly imported AND superseded images
-        # (a superseded row has cleared ML + no thumbnail).
-        if result.status in ("imported", "superseded"):
-            from .ml import tag_and_embed
-            from .thumbnail import generate_thumbnail
-
-            ids = list(result.member_image_ids)
-            if result.image_id is not None and result.image_id not in ids:
-                ids.append(result.image_id)
-            for img_id in ids:
-                generate_thumbnail.delay(img_id)
-                tag_and_embed.delay(img_id)
-
-        # If this was the last task in the batch, mark the batch complete.
-        remaining = session.execute(
-            select(ImportTask.id)
-            .where(ImportTask.batch_id == task.batch_id)
-            .where(ImportTask.status.in_(["pending", "queued", "processing"]))
-            .limit(1)
-        ).scalar_one_or_none()
-        if remaining is None:
-            session.execute(
-                update(ImportBatch)
-                .where(ImportBatch.id == task.batch_id)
-                .values(
-                    status="complete",
-                    finished_at=datetime.now(UTC),
-                )
-            )
-            session.commit()
-
-        return {"task_id": import_task_id, "status": task.status}
+    return {"task_id": import_task_id, "status": task.status}
@@ -15,7 +15,7 @@ from ._sync_engine import sync_session_factory as _sync_session_factory

 log = logging.getLogger(__name__)

-STUCK_THRESHOLD_MINUTES = 30
+STUCK_THRESHOLD_MINUTES = 5
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
 VERIFY_PAGE = 200
@@ -24,11 +24,15 @@ FFPROBE_TIMEOUT_SECONDS = 10

@celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
-    """Find ImportTask rows stuck in 'processing' for >30 min and re-queue them.
+    """Find ImportTask rows stuck in 'processing' for >5 min and re-queue them.

-    Why 30 min: large videos can legitimately take many minutes to import;
-    30 is a safe gate that catches actual crashes (which leave the row stuck
-    forever) without resetting slow-but-still-running jobs.
+    Why 5 min: import_media_file is sub-second for the vast majority of
+    files; even a large-video transcode caps at the per-task soft_time_limit
+    (5 min) defined on the task itself. Anything still 'processing' after
+    that window is a confirmed crash (worker died, DB disconnect mid-flush,
+    OOM) and must be recycled. Was 30 min historically; tightened
+    2026-05-24 after operator hit a 2224-row zombie pile during the IR
+    migration scan.
    """
    SessionLocal = _sync_session_factory()
    cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
@@ -9,6 +9,7 @@ apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions
 from pathlib import Path

 from sqlalchemy import select
+from sqlalchemy.exc import DBAPIError, OperationalError

 from ..celery_app import celery
 from ..models import ImageRecord, MLSettings
@@ -22,7 +23,17 @@ def _is_video(path: Path) -> bool:
    return path.suffix.lower() in VIDEO_EXTS


-@celery.task(name="backend.app.tasks.ml.tag_and_embed", bind=True)
+@celery.task(
+    name="backend.app.tasks.ml.tag_and_embed",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=300,
+    time_limit=420,
+)
 def tag_and_embed(self, image_id: int) -> dict:
    """Run Camie + SigLIP on one image; store predictions + embedding;
    then enqueue per-image allowlist application.
@@ -7,6 +7,8 @@ so they deserve their own queue lane.

 from pathlib import Path

+from sqlalchemy.exc import DBAPIError, OperationalError
+
 from ..celery_app import celery
 from ..models import ImageRecord
 from ..services.importer import is_video
@@ -16,7 +18,17 @@ from ._sync_engine import sync_session_factory as _sync_session_factory
 IMAGES_ROOT = Path("/images")


-@celery.task(name="backend.app.tasks.thumbnail.generate_thumbnail", bind=True)
+@celery.task(
+    name="backend.app.tasks.thumbnail.generate_thumbnail",
+    bind=True,
+    autoretry_for=(OperationalError, DBAPIError, OSError),
+    retry_backoff=5,
+    retry_backoff_max=60,
+    retry_jitter=True,
+    max_retries=3,
+    soft_time_limit=120,
+    time_limit=180,
+)
 def generate_thumbnail(self, image_id: int) -> dict:
    SessionLocal = _sync_session_factory()
    with SessionLocal() as session: