diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py index 02f65d2..2ebeb9b 100644 --- a/backend/app/api/import_admin.py +++ b/backend/app/api/import_admin.py @@ -114,18 +114,18 @@ async def retry_failed(): status="queued", error=None, started_at=None, finished_at=None, ) - .returning(ImportTask.id) + .returning(ImportTask.id, ImportTask.task_type) ) - failed_ids = [row[0] for row in result.all()] - if not failed_ids: + failed = result.all() + if not failed: return jsonify({"retried": 0}) await session.commit() - from ..tasks.import_file import import_media_file - for tid in failed_ids: - import_media_file.delay(tid) + from ..tasks.import_file import enqueue_import + for tid, task_type in failed: + enqueue_import(tid, task_type) - return jsonify({"retried": len(failed_ids)}) + return jsonify({"retried": len(failed)}) @import_admin_bp.route("/clear-stuck", methods=["POST"]) diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index e0752e8..4a124d1 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -64,30 +64,13 @@ def _mark_failed(session, task, error_msg: str) -> None: 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. - - 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. +def _run_import_task(import_task_id: int) -> dict: + """Shared body for import_media_file + import_archive_file. The two + tasks differ ONLY in their Celery time limits (a single media file + is sub-second; an archive runs the full per-member pipeline inline + for every member and can take many minutes). Both flip the row to + 'processing', dispatch to `_do_import`, and honor the + flip-to-terminal resilience contract. """ SessionLocal = _sync_session_factory() with SessionLocal() as session: @@ -103,19 +86,85 @@ def import_media_file(self, import_task_id: int) -> dict: try: return _do_import(session, task, import_task_id) except SoftTimeLimitExceeded: - _mark_failed(session, task, "soft_time_limit exceeded (>300s)") + _mark_failed(session, task, "soft_time_limit exceeded") 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. + # flips it. raise except Exception as exc: # noqa: BLE001 — pipeline crash, mark + re-raise _mark_failed(session, task, f"{type(exc).__name__}: {exc}") raise +@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: + """Import ONE media file (or non-media → PostAttachment). Sub-second + for the common case; the tight 5-min soft limit keeps a genuinely + stuck single-file import detectable fast. + + 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-process + so the task can mark its row failed before being killed. + - time_limit (360s) is the hard SIGKILL cap. + """ + return _run_import_task(import_task_id) + + +@celery.task( + name="backend.app.tasks.import_file.import_archive_file", + bind=True, + autoretry_for=(OperationalError, DBAPIError, OSError), + retry_backoff=5, + retry_backoff_max=60, + retry_jitter=True, + max_retries=3, + # Archives run the full per-member pipeline (sha256 + pHash + dedup + # query + copy + provenance) for EVERY media member inline, under a + # single task budget. A multi-hundred-member archive blows the + # 5-min media limit. soft=30min / hard=35min sizes for a large + # archive. Operator-flagged 2026-05-28 (target 1645019 hit the old + # shared 300s soft limit). The recovery sweep gives this task its + # own 40-min threshold via maintenance.TASK_STUCK_THRESHOLD_MINUTES + # so it isn't preempted while legitimately grinding through members. + soft_time_limit=1800, + time_limit=2100, +) +def import_archive_file(self, import_task_id: int) -> dict: + """Import an archive: extract + run the per-member media pipeline for + every member inline, then preserve the archive as a PostAttachment. + Same body as import_media_file (dispatch is by file kind inside + Importer.import_one); split out purely for the larger time budget.""" + return _run_import_task(import_task_id) + + +def enqueue_import(task_id: int, task_type: str) -> None: + """Route an ImportTask to the right Celery task by its task_type. + Single source of truth for the media-vs-archive dispatch so the + scan, retry, and recovery-requeue paths stay in sync.""" + if task_type == "archive": + import_archive_file.delay(task_id) + else: + import_media_file.delay(task_id) + + def _do_import(session, task, import_task_id: int) -> dict: """Actual work, called from inside the resilience wrapper.""" settings = session.execute( diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index c5bb617..797d6f2 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -16,6 +16,13 @@ from ._sync_engine import sync_session_factory as _sync_session_factory log = logging.getLogger(__name__) STUCK_THRESHOLD_MINUTES = 5 +# Archive ImportTasks run the per-member pipeline inline for every +# member (import_archive_file: soft=30min/hard=35min). The ImportTask +# 'processing' recovery sweep must give them a longer threshold or it +# re-queues a legitimately-running archive mid-import (double-process). +# 40 min = 5-min buffer past the archive task's hard kill. +# Operator-flagged 2026-05-28 (target 1645019, a big archive). +ARCHIVE_STUCK_THRESHOLD_MINUTES = 40 ORPHAN_PENDING_THRESHOLD_MINUTES = 30 OLD_TASK_DAYS = 7 PHASH_PAGE = 500 @@ -24,32 +31,40 @@ 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. +# Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep). +# Tasks/queues that legitimately run longer than the default 5-min +# threshold need their own larger value, else the sweep marks in-flight +# work 'error' before it finishes. Each value MUST be ≥ the relevant +# task.time_limit + a small buffer. task_name overrides take precedence +# over queue overrides. # -# 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). +# ml queue: tag_and_embed video branch (≈20 GPU ops); time_limit=1200. +# import_archive_file: shares the 'import' queue with the fast +# single-file import_media_file, so it needs a task-name override +# (the import queue itself stays at the 5-min default for single +# files); time_limit=2100. QUEUE_STUCK_THRESHOLD_MINUTES: dict[str, int] = { "ml": 25, } +TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = { + "backend.app.tasks.import_file.import_archive_file": 40, +} @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks") def recover_interrupted_tasks() -> int: """Recover stuck ImportTask rows. Two distinct stuck states: - 1. 'processing' > 5 min — worker crash mid-import. Re-queue via - .delay() and let the import retry. Was 30 min historically; - tightened 2026-05-24 after operator hit a 2224-row zombie pile. - import_media_file is sub-second for the vast majority of files and - capped at the per-task soft_time_limit (5 min), so anything still - 'processing' after that window is a confirmed crash. + 1. 'processing' too long — worker crash mid-import. Re-queue via + enqueue_import (routing media vs archive) and let the import + retry. Threshold is task-type-aware: media files are sub-second + and capped at the 5-min soft limit, so STUCK_THRESHOLD_MINUTES + (5) means a confirmed crash; archives run the per-member + pipeline inline (import_archive_file, 35-min hard limit) so they + get ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid re-queueing a + still-running archive. (Media was tightened from 30 min to 5 + 2026-05-24 after a 2224-row zombie pile; archive split out + 2026-05-28.) 2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory creates rows with status='pending' (commit), then in a second pass @@ -66,7 +81,8 @@ def recover_interrupted_tasks() -> int: """ SessionLocal = _sync_session_factory() now = datetime.now(UTC) - processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + media_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + archive_cutoff = now - timedelta(minutes=ARCHIVE_STUCK_THRESHOLD_MINUTES) orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES) with SessionLocal() as session: # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which @@ -74,20 +90,31 @@ def recover_interrupted_tasks() -> int: # tens of thousands of rows (operator hit it 2026-05-26 after the # /import deep scan piled up orphans). Folding the SELECT into the # UPDATE eliminates the IN-list entirely. RETURNING gives us back - # exactly the ids that flipped so the stuck sweep can still - # .delay() each one. + # exactly the (id, task_type) pairs that flipped so the requeue + # can route media vs archive correctly. + # + # Media + archive get separate cutoffs: a single media file is + # sub-second so 5 min means crash; an archive runs the per-member + # pipeline inline and can legitimately take up to its 35-min hard + # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid + # re-queueing a still-running archive. stuck_result = session.execute( update(ImportTask) .where(ImportTask.status == "processing") - .where(ImportTask.started_at < processing_cutoff) + .where( + ((ImportTask.task_type != "archive") + & (ImportTask.started_at < media_cutoff)) + | ((ImportTask.task_type == "archive") + & (ImportTask.started_at < archive_cutoff)) + ) .values( status="queued", started_at=None, error="recovered from stuck state", ) - .returning(ImportTask.id) + .returning(ImportTask.id, ImportTask.task_type) ) - stuck_ids = [row[0] for row in stuck_result.all()] + stuck = stuck_result.all() orphan_result = session.execute( update(ImportTask) @@ -106,12 +133,12 @@ def recover_interrupted_tasks() -> int: session.commit() - if stuck_ids: - from .import_file import import_media_file - for tid in stuck_ids: - import_media_file.delay(tid) + if stuck: + from .import_file import enqueue_import + for tid, task_type in stuck: + enqueue_import(tid, task_type) - return len(stuck_ids) + orphan_count + return len(stuck) + orphan_count @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks") @@ -152,54 +179,53 @@ def recover_stalled_task_runs() -> int: """ SessionLocal = _sync_session_factory() 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_tasks = set(TASK_STUCK_THRESHOLD_MINUTES.keys()) override_queues = set(QUEUE_STUCK_THRESHOLD_MINUTES.keys()) total = 0 - with SessionLocal() as session: - # Default-threshold pass — all queues except the overridden ones. - default_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES) - default_stmt = ( + + def _flag(minutes, *extra_where): + cutoff = now - timedelta(minutes=minutes) + stmt = ( update(TaskRun) .where(TaskRun.status == "running") - .where(TaskRun.started_at < default_cutoff) + .where(TaskRun.started_at < cutoff) .values( status="error", error_type="RecoverySweep", error_message=( - f"no completion signal received within " - f"{STUCK_THRESHOLD_MINUTES} min" + f"no completion signal received within {minutes} min" ), finished_at=now, ) ) - if override_queues: - default_stmt = default_stmt.where( - TaskRun.queue.notin_(override_queues) - ) - total += session.execute(default_stmt).rowcount or 0 + for w in extra_where: + stmt = stmt.where(w) + return session.execute(stmt).rowcount or 0 - # Per-queue override passes. + with SessionLocal() as session: + # Precedence: task_name override → queue override → default. + # Each pass excludes rows claimed by a higher-precedence pass so + # every row is touched at most once. + + # 1. Per-task-name overrides (e.g. import_archive_file, which + # shares the 'import' queue with fast single-file imports). + for task_name, minutes in TASK_STUCK_THRESHOLD_MINUTES.items(): + total += _flag(minutes, TaskRun.task_name == task_name) + + # 2. Per-queue overrides, excluding the override task-names. 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 + wheres = [TaskRun.queue == queue] + if override_tasks: + wheres.append(TaskRun.task_name.notin_(override_tasks)) + total += _flag(minutes, *wheres) + + # 3. Default — everything not claimed above. + default_wheres = [] + if override_queues: + default_wheres.append(TaskRun.queue.notin_(override_queues)) + if override_tasks: + default_wheres.append(TaskRun.task_name.notin_(override_tasks)) + total += _flag(STUCK_THRESHOLD_MINUTES, *default_wheres) session.commit() return total diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 2e5b612..0451e49 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -17,6 +17,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn from ..celery_app import celery from ..config import get_config from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask +from ..services.archive_extractor import is_archive from ..services.scheduler_service import select_due_sources from ._sync_engine import sync_session_factory as _sync_session_factory @@ -96,7 +97,9 @@ def scan_directory(self, triggered_by: str = "manual", task = ImportTask( batch_id=batch_id, source_path=entry_str, - task_type="media", + # Archives route to import_archive_file (larger time + # budget) — they run the per-member pipeline inline. + task_type="archive" if is_archive(entry) else "media", status="pending", size_bytes=size, ) @@ -115,15 +118,16 @@ def scan_directory(self, triggered_by: str = "manual", batch.finished_at = datetime.now(UTC) session.commit() - # Now enqueue import_media_file for each pending task. + # Now enqueue each pending task on the right Celery task + # (media vs archive) via the shared router. + from .import_file import enqueue_import + for task in session.execute( select(ImportTask).where(ImportTask.batch_id == batch_id) ).scalars(): task.status = "queued" session.add(task) - from .import_file import import_media_file - - import_media_file.delay(task.id) + enqueue_import(task.id, task.task_type) session.commit() if mode == "deep": diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 9f2803e..42d1871 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -195,12 +195,13 @@ def test_cleanup_old_deletes_finished_old(db_sync): def _make_task_run(db_sync, *, status, started_at, finished_at=None, - error_type=None, queue="default"): + error_type=None, queue="default", + task_name="backend.app.tasks.fake.t"): from backend.app.models import TaskRun row = TaskRun( celery_task_id="x", queue=queue, - task_name="backend.app.tasks.fake.t", + task_name=task_name, target_id=1, started_at=started_at, finished_at=finished_at, @@ -296,6 +297,53 @@ def test_recover_stalled_task_runs_ml_queue_uses_longer_threshold(db_sync): assert ml_fresh_status == "running" assert ml_stale_status == "error" + +def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync): + """import_archive_file shares the 'import' queue with fast + single-file import_media_file, so it gets a per-task-name override + (40 min) while the import queue stays at the 5-min default. A + 10-min-old archive task-run must survive; a 50-min-old one is + flagged. Operator-flagged 2026-05-28.""" + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import recover_stalled_task_runs + + archive_name = "backend.app.tasks.import_file.import_archive_file" + now = datetime.now(UTC) + # Fast single-file import on the same queue, 10 min old → flagged + # by the default 5-min rule. + media_id = _make_task_run( + db_sync, status="running", queue="import", + task_name="backend.app.tasks.import_file.import_media_file", + started_at=now - timedelta(minutes=10), + ) + # Archive on the same queue, 10 min old → survives (40-min override). + archive_fresh_id = _make_task_run( + db_sync, status="running", queue="import", + task_name=archive_name, + started_at=now - timedelta(minutes=10), + ) + # Archive 50 min old → past even the 40-min override → flagged. + archive_stale_id = _make_task_run( + db_sync, status="running", queue="import", + task_name=archive_name, + started_at=now - timedelta(minutes=50), + ) + db_sync.commit() + + recovered = recover_stalled_task_runs.apply().get() + assert recovered == 2 # media + stale archive + + db_sync.expire_all() + def _status(_id): + return db_sync.execute( + select(TaskRun.status).where(TaskRun.id == _id) + ).scalar_one() + assert _status(media_id) == "error" + assert _status(archive_fresh_id) == "running" + assert _status(archive_stale_id) == "error" + db_sync.expire_all() status = db_sync.execute( select(TaskRun.status).where(TaskRun.id == fresh_id) diff --git a/tests/test_tasks_register.py b/tests/test_tasks_register.py index 11f457c..2e5414f 100644 --- a/tests/test_tasks_register.py +++ b/tests/test_tasks_register.py @@ -21,6 +21,10 @@ def test_import_media_file_registered(): assert "backend.app.tasks.import_file.import_media_file" in celery.tasks +def test_import_archive_file_registered(): + assert "backend.app.tasks.import_file.import_archive_file" in celery.tasks + + def test_generate_thumbnail_registered(): assert "backend.app.tasks.thumbnail.generate_thumbnail" in celery.tasks