From be0f472894a300e063aaa16e2c246aa11a13ed4b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 24 May 2026 11:46:36 -0400 Subject: [PATCH] =?UTF-8?q?fix(workers):=20worker-level=20recovery=20?= =?UTF-8?q?=E2=80=94=20autoretry=20on=20transient=20errors=20+=20tightened?= =?UTF-8?q?=20sweep=20(5min)=20+=20import=5Fmedia=5Ffile=20body-wrap=20so?= =?UTF-8?q?=20no=20path=20leaves=20rows=20stuck=20in=20'processing'?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/tasks/download.py | 14 +- backend/app/tasks/import_file.py | 236 +++++++++++++++++++------------ backend/app/tasks/maintenance.py | 14 +- backend/app/tasks/ml.py | 13 +- backend/app/tasks/thumbnail.py | 14 +- 5 files changed, 195 insertions(+), 96 deletions(-) diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py index a6416ab..8801a56 100644 --- a/backend/app/tasks/download.py +++ b/backend/app/tasks/download.py @@ -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.""" diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index 562e7e7..7fd1066 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -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} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 4b23ac0..2a50e86 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -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) diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index a208376..75de41d 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -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. diff --git a/backend/app/tasks/thumbnail.py b/backend/app/tasks/thumbnail.py index 61bed80..dec644f 100644 --- a/backend/app/tasks/thumbnail.py +++ b/backend/app/tasks/thumbnail.py @@ -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: