fix(maintenance): recover_interrupted_tasks also sweeps pending/queued orphans (>30 min) to failed

scan_directory creates ImportTask rows with status='pending' (commit) then
in a second pass transitions to 'queued' + .delay() (commit). Crashes in
that window leave rows orphaned with no recovery path. Operator hit 5490
such rows 2026-05-25; the existing sweep only handled 'processing'.
Flipping to 'failed' (not re-enqueue) lets the operator drain via the
existing /api/import/retry-failed endpoint at their own pace.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/tasks/maintenance.py

@@ -16,6 +16,7 @@ from ._sync_engine import sync_session_factory as _sync_session_factory
 log = logging.getLogger(__name__)
 
 STUCK_THRESHOLD_MINUTES = 5
+ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
 VERIFY_PAGE = 200
@@ -26,40 +27,77 @@ TASK_RUN_KEEP_FAILURE_SECONDS = 7 * 24 * 3600  # 7 days
 
 @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
-    """Find ImportTask rows stuck in 'processing' for >5 min and re-queue them.
+    """Recover stuck ImportTask rows. Two distinct stuck states:
 
-    Why 5 min: import_media_file is sub-second for the vast majority of
+    1. 'processing' > 5 min — worker crash mid-import. Re-queue via
-    files; even a large-video transcode caps at the per-task soft_time_limit
+       .delay() and let the import retry. Was 30 min historically;
-    (5 min) defined on the task itself. Anything still 'processing' after
+       tightened 2026-05-24 after operator hit a 2224-row zombie pile.
-    that window is a confirmed crash (worker died, DB disconnect mid-flush,
+       import_media_file is sub-second for the vast majority of files and
-    OOM) and must be recycled. Was 30 min historically; tightened
+       capped at the per-task soft_time_limit (5 min), so anything still
-    2026-05-24 after operator hit a 2224-row zombie pile during the IR
+       'processing' after that window is a confirmed crash.
-    migration scan.
+
+    2. 'pending' or 'queued' > 30 min — enqueue-phase crash. scan_directory
+       creates rows with status='pending' (commit), then in a second pass
+       transitions to 'queued' and calls .delay() (commit). If the scanner
+       crashes between those two commits, rows are orphaned in 'pending'
+       (never enqueued) with no recovery path — invisible to the
+       'processing' sweep above. Flagged 2026-05-25 by operator hitting a
+       5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so
+       the operator drains them via /api/import/retry-failed at their own
+       pace; bulk-re-enqueueing 5000+ rows would thundering-herd the
+       import worker.
+
+    Returns total rows touched (recovered + marked failed).
     """
     SessionLocal = _sync_session_factory()
-    cutoff = datetime.now(UTC) - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    now = datetime.now(UTC)
+    processing_cutoff = now - timedelta(minutes=STUCK_THRESHOLD_MINUTES)
+    orphan_cutoff = now - timedelta(minutes=ORPHAN_PENDING_THRESHOLD_MINUTES)
     with SessionLocal() as session:
         stuck_ids = session.execute(
             select(ImportTask.id)
             .where(ImportTask.status == "processing")
-            .where(ImportTask.started_at < cutoff)
+            .where(ImportTask.started_at < processing_cutoff)
         ).scalars().all()
 
-        if not stuck_ids:
+        orphan_ids = session.execute(
+            select(ImportTask.id)
+            .where(ImportTask.status.in_(["pending", "queued"]))
+            .where(ImportTask.created_at < orphan_cutoff)
+        ).scalars().all()
+
+        if not stuck_ids and not orphan_ids:
             return 0
 
-        session.execute(
+        if stuck_ids:
-            update(ImportTask)
+            session.execute(
-            .where(ImportTask.id.in_(stuck_ids))
+                update(ImportTask)
-            .values(status="queued", started_at=None, error="recovered from stuck state")
+                .where(ImportTask.id.in_(stuck_ids))
-        )
+                .values(status="queued", started_at=None, error="recovered from stuck state")
+            )
+
+        if orphan_ids:
+            session.execute(
+                update(ImportTask)
+                .where(ImportTask.id.in_(orphan_ids))
+                .values(
+                    status="failed",
+                    error=(
+                        "orphan pending/queued swept by recover_interrupted_tasks "
+                        "(scanner likely crashed mid-enqueue); retry via "
+                        "/api/import/retry-failed"
+                    ),
+                )
+            )
+
         session.commit()
 
-        from .import_file import import_media_file
+        if stuck_ids:
-        for tid in stuck_ids:
+            from .import_file import import_media_file
-            import_media_file.delay(tid)
+            for tid in stuck_ids:
+                import_media_file.delay(tid)
 
-        return len(stuck_ids)
+        return len(stuck_ids) + len(orphan_ids)
 
 
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")

tests/test_maintenance.py

@@ -68,6 +68,102 @@ def test_recover_interrupted_only_old(db_sync, monkeypatch):
     assert dispatched == [stale.id]
 
 
+def test_recover_interrupted_sweeps_pending_orphans_to_failed(db_sync, monkeypatch):
+    """A scan that creates ImportTask rows but crashes before the second
+    pass (transition to 'queued' + .delay()) leaves rows orphaned at
+    status='pending'. The sweep flips them to 'failed' so the operator
+    can drain via /api/import/retry-failed without thundering-herding.
+    Banked 2026-05-25 after operator hit 5490 stuck pending rows.
+    """
+    from backend.app.tasks import import_file
+    monkeypatch.setattr(import_file.import_media_file, "delay", lambda *_: None)
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    fresh_pending = ImportTask(
+        batch_id=batch_id, source_path="/import/fresh.jpg", task_type="media",
+        status="pending",
+    )
+    db_sync.add(fresh_pending)
+    db_sync.flush()
+    # created_at defaults to now() server-side; fresh row stays untouched.
+
+    # Two stale rows simulating the orphan pile: one 'pending', one
+    # 'queued' (scanner crashed AFTER transitioning some rows but
+    # before all). Both should sweep.
+    stale_pending = ImportTask(
+        batch_id=batch_id, source_path="/import/stale1.jpg", task_type="media",
+        status="pending",
+    )
+    stale_queued = ImportTask(
+        batch_id=batch_id, source_path="/import/stale2.jpg", task_type="media",
+        status="queued",
+    )
+    db_sync.add_all([stale_pending, stale_queued])
+    db_sync.flush()
+    # Backdate created_at past the orphan cutoff (30 min).
+    from sqlalchemy import update as _upd
+    db_sync.execute(
+        _upd(ImportTask)
+        .where(ImportTask.id.in_([stale_pending.id, stale_queued.id]))
+        .values(created_at=now - timedelta(hours=2))
+    )
+    db_sync.commit()
+
+    from backend.app.tasks.maintenance import recover_interrupted_tasks
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2
+
+    db_sync.refresh(fresh_pending)
+    db_sync.refresh(stale_pending)
+    db_sync.refresh(stale_queued)
+    assert fresh_pending.status == "pending"  # fresh row untouched
+    assert stale_pending.status == "failed"
+    assert stale_queued.status == "failed"
+    assert "orphan" in (stale_pending.error or "")
+
+
+def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch):
+    """One sweep tick handles both 'processing' crashes AND
+    'pending'/'queued' orphans in a single pass."""
+    from backend.app.tasks import import_file
+    dispatched: list[int] = []
+    monkeypatch.setattr(
+        import_file.import_media_file, "delay", dispatched.append
+    )
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    stuck = ImportTask(
+        batch_id=batch_id, source_path="/import/stuck.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+    )
+    orphan = ImportTask(
+        batch_id=batch_id, source_path="/import/orphan.jpg", task_type="media",
+        status="pending",
+    )
+    db_sync.add_all([stuck, orphan])
+    db_sync.flush()
+    from sqlalchemy import update as _upd
+    db_sync.execute(
+        _upd(ImportTask).where(ImportTask.id == orphan.id)
+        .values(created_at=now - timedelta(hours=2))
+    )
+    db_sync.commit()
+
+    from backend.app.tasks.maintenance import recover_interrupted_tasks
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2
+
+    db_sync.refresh(stuck)
+    db_sync.refresh(orphan)
+    assert stuck.status == "queued"
+    assert orphan.status == "failed"
+    assert dispatched == [stuck.id]  # stuck rows re-enqueue; orphans don't
+
+
 def test_cleanup_old_deletes_finished_old(db_sync):
     batch_id = _make_batch(db_sync)
     now = datetime.now(UTC)