fix(import): split archive imports into their own task + budget; archive-aware recovery sweeps

Operator-flagged 2026-05-28: import_media_file on target 1645019 hit
SoftTimeLimitExceeded at exactly 5.0 min. Their diagnosis was correct —
the timeout covered the WHOLE archive, not per object. Importer._import_archive
(importer.py:409) runs the full per-member pipeline (sha256 + pHash +
dedup query + copy + provenance) for EVERY media member inline, all
under import_media_file's single 300s soft limit. A single media file
is sub-second; a multi-hundred-member archive blows the budget. They
shared one task name and one timeout.

**Split archive into its own task**

- New `import_archive_file` task: same body as import_media_file
  (dispatch is by file-kind inside Importer.import_one) but
  soft=30min / hard=35min. Shared `_run_import_task` helper holds the
  flip-to-processing + resilience-contract wrapper; both tasks call it.
- New `enqueue_import(task_id, task_type)` router — single source of
  truth for media-vs-archive dispatch. Used by all three enqueue sites:
  scan_directory, /api/import/retry-failed, recover_interrupted_tasks.
- scan_directory now sets ImportTask.task_type = "archive" when
  is_archive(entry) (the model field already existed, anticipating
  this; scan was hardcoding "media").
- import_archive_file routes to the existing 'import' queue via the
  task_routes `import_file.*` wildcard — no worker config change.

**Archive-aware recovery sweeps**

Both sweeps would otherwise preempt a legitimately-running archive:

- recover_interrupted_tasks (ImportTask 'processing' sweep): now
  task-type-aware. Media stays at STUCK_THRESHOLD_MINUTES (5); archives
  get ARCHIVE_STUCK_THRESHOLD_MINUTES (40 = 5-min buffer past the
  35-min hard limit). Single UPDATE with an OR predicate over the two
  (task_type, cutoff) pairs; requeue routes via enqueue_import.
- recover_stalled_task_runs (TaskRun 'running' sweep): now supports
  per-task-name overrides (TASK_STUCK_THRESHOLD_MINUTES) layered above
  the per-queue overrides added for ml. import_archive_file gets 40 min
  while the 'import' queue stays at the 5-min default for single-file
  imports. Precedence: task_name → queue → default, each pass excluding
  rows claimed by a higher-precedence pass so every row is touched once.

**Tests**

- test_import_archive_file_registered
- test_recover_stalled_task_runs_archive_task_uses_longer_threshold —
  pins that a 10-min archive task-run survives, a 50-min one is flagged,
  and a same-queue 10-min media import is flagged at the default.
- _make_task_run gains queue= + task_name= params.

After deploy: archive imports get a 30-min budget and aren't preempted
by either sweep; single-file imports keep their tight 5-min detection.

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()]
+        failed = result.all()
-        if not failed_ids:
+        if not failed:
             return jsonify({"retried": 0})
         await session.commit()
 
-    from ..tasks.import_file import import_media_file
+    from ..tasks.import_file import enqueue_import
-    for tid in failed_ids:
+    for tid, task_type in failed:
-        import_media_file.delay(tid)
+        enqueue_import(tid, task_type)
 
-    return jsonify({"retried": len(failed_ids)})
+    return jsonify({"retried": len(failed)})
 
 
 @import_admin_bp.route("/clear-stuck", methods=["POST"])

backend/app/tasks/import_file.py

@@ -64,30 +64,13 @@ def _mark_failed(session, task, error_msg: str) -> None:
             pass
 
 
-@celery.task(
+def _run_import_task(import_task_id: int) -> dict:
-    name="backend.app.tasks.import_file.import_media_file",
+    """Shared body for import_media_file + import_archive_file. The two
-    bind=True,
+    tasks differ ONLY in their Celery time limits (a single media file
-    autoretry_for=(OperationalError, DBAPIError, OSError),
+    is sub-second; an archive runs the full per-member pipeline inline
-    retry_backoff=5,
+    for every member and can take many minutes). Both flip the row to
-    retry_backoff_max=60,
+    'processing', dispatch to `_do_import`, and honor the
-    retry_jitter=True,
+    flip-to-terminal resilience contract.
-    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.
     """
     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(

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
+# Overrides for recover_stalled_task_runs (the TaskRun 'running' sweep).
-# tasks can legitimately run longer than the default 5-min threshold
+# Tasks/queues that legitimately run longer than the default 5-min
-# need their own larger value, otherwise the sweep marks in-flight
+# threshold need their own larger value, else the sweep marks in-flight
-# tasks 'error' before they get a chance to finish. The dict's value
+# work 'error' before it finishes. Each value MUST be ≥ the relevant
-# MUST be ≥ the longest task.time_limit on the queue + a small buffer.
+# 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 +
+# ml queue: tag_and_embed video branch (≈20 GPU ops); time_limit=1200.
-#     embedder on each — soft_time_limit=900 / time_limit=1200; sweep
+# import_archive_file: shares the 'import' queue with the fast
-#     at 25 min gives a 5-min buffer past the hard kill.
+#   single-file import_media_file, so it needs a task-name override
-#     Operator-flagged 2026-05-28 (image 6288, an mp4, marked failed
+#   (the import queue itself stays at the 5-min default for single
-#     at the 5-min sweep tick while still processing).
+#   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
+    1. 'processing' too long — worker crash mid-import. Re-queue via
-       .delay() and let the import retry. Was 30 min historically;
+       enqueue_import (routing media vs archive) and let the import
-       tightened 2026-05-24 after operator hit a 2224-row zombie pile.
+       retry. Threshold is task-type-aware: media files are sub-second
-       import_media_file is sub-second for the vast majority of files and
+       and capped at the 5-min soft limit, so STUCK_THRESHOLD_MINUTES
-       capped at the per-task soft_time_limit (5 min), so anything still
+       (5) means a confirmed crash; archives run the per-member
-       'processing' after that window is a confirmed crash.
+       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
+        # exactly the (id, task_type) pairs that flipped so the requeue
-        # .delay() each one.
+        # 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:
+        if stuck:
-            from .import_file import import_media_file
+            from .import_file import enqueue_import
-            for tid in stuck_ids:
+            for tid, task_type in stuck:
-                import_media_file.delay(tid)
+                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
+    override_tasks = set(TASK_STUCK_THRESHOLD_MINUTES.keys())
-    # 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:
+
-        # Default-threshold pass — all queues except the overridden ones.
+    def _flag(minutes, *extra_where):
-        default_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+        cutoff = now - timedelta(minutes=minutes)
-        default_stmt = (
+        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"no completion signal received within {minutes} min"
-                    f"{STUCK_THRESHOLD_MINUTES} min"
                 ),
                 finished_at=now,
             )
         )
-        if override_queues:
+        for w in extra_where:
-            default_stmt = default_stmt.where(
+            stmt = stmt.where(w)
-                TaskRun.queue.notin_(override_queues)
+        return session.execute(stmt).rowcount or 0
-            )
-        total += session.execute(default_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)
+            wheres = [TaskRun.queue == queue]
-            stmt = (
+            if override_tasks:
-                update(TaskRun)
+                wheres.append(TaskRun.task_name.notin_(override_tasks))
-                .where(TaskRun.status == "running")
+            total += _flag(minutes, *wheres)
-                .where(TaskRun.queue == queue)
+
-                .where(TaskRun.started_at < cutoff)
+        # 3. Default — everything not claimed above.
-                .values(
+        default_wheres = []
-                    status="error",
+        if override_queues:
-                    error_type="RecoverySweep",
+            default_wheres.append(TaskRun.queue.notin_(override_queues))
-                    error_message=(
+        if override_tasks:
-                        f"no completion signal received within "
+            default_wheres.append(TaskRun.task_name.notin_(override_tasks))
-                        f"{minutes} min"
+        total += _flag(STUCK_THRESHOLD_MINUTES, *default_wheres)
-                    ),
-                    finished_at=now,
-                )
-            )
-            total += session.execute(stmt).rowcount or 0
 
         session.commit()
         return total

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
+            enqueue_import(task.id, task.task_type)
-
-            import_media_file.delay(task.id)
         session.commit()
 
         if mode == "deep":

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)

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