fix: the import recovery sweep no longer fails queued files that are waiting their turn

The orphan sweep in recover_interrupted_tasks failed every 'queued' row
older than 30 minutes. A queued row has its broker message, so in a batch
bigger than half an hour of work the sweep failed the whole tail: a
library import of 8,135 files lost 3,114 that way. Never-enqueued
'pending' rows are still swept after 30 minutes; a 'queued' row is swept
only when the broker's import queue is empty, and is kept when the queue
can't be read.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/app/tasks/maintenance.py

@@ -200,6 +200,23 @@ TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
 }
 
 
+def _import_queue_depth() -> int | None:
+    """Messages waiting in the broker's import queue, or None when the broker
+    can't be read."""
+    try:
+        import redis
+
+        from ..config import get_config
+
+        client = redis.Redis.from_url(
+            get_config().celery_broker_url, socket_timeout=5, socket_connect_timeout=5,
+        )
+        return int(client.llen("import"))
+    except Exception:  # noqa: BLE001 — an unreadable broker means "don't know"
+        log.warning("recover_interrupted_tasks: import queue depth unreadable", exc_info=True)
+        return None
+
+
 @celery.task(name="backend.app.tasks.maintenance.recover_interrupted_tasks")
 def recover_interrupted_tasks() -> int:
     """Recover stuck ImportTask rows. Two distinct stuck states:
@@ -215,16 +232,22 @@ def recover_interrupted_tasks() -> int:
        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
+    2. 'pending' > 30 min — enqueue-phase crash. scan_directory creates rows
-       creates rows with status='pending' (commit), then in a second pass
+       with status='pending' (commit), then in a second pass transitions to
-       transitions to 'queued' and calls .delay() (commit). If the scanner
+       'queued' and calls .delay() (commit). If the scanner crashes between
-       crashes between those two commits, rows are orphaned in 'pending'
+       those two commits, rows are orphaned in 'pending' (never enqueued)
-       (never enqueued) with no recovery path — invisible to the
+       with no recovery path — invisible to the 'processing' sweep above.
-       'processing' sweep above. Flagged 2026-05-25 by operator hitting a
+       Flip these to 'failed' (not re-enqueue) so the operator drains them
-       5490-row orphan pile. Flip these to 'failed' (not re-enqueue) so
+       via /api/import/retry-failed at their own pace; bulk-re-enqueueing
-       the operator drains them via /api/import/retry-failed at their own
+       5000+ rows would thundering-herd the import worker.
-       pace; bulk-re-enqueueing 5000+ rows would thundering-herd the
+
-       import worker.
+       A 'queued' row has a broker message, and in a batch larger than half
+       an hour of work it is simply waiting its turn, so its age says
+       nothing. It counts as orphaned only when the broker's import queue
+       is empty: then its message is gone (a broker flush or restart) and
+       nothing will ever run it. When the queue can't be read, queued rows
+       are left alone — a queued row costs nothing to keep, and failing a
+       live one loses its import.
 
     Returns total rows touched (recovered + marked failed).
     """
@@ -301,9 +324,12 @@ def recover_interrupted_tasks() -> int:
         )
         stuck = stuck_result.all()
 
+        orphan_statuses = ["pending"]
+        if _import_queue_depth() == 0:
+            orphan_statuses.append("queued")
         orphan_result = session.execute(
             update(ImportTask)
-            .where(ImportTask.status.in_(["pending", "queued"]))
+            .where(ImportTask.status.in_(orphan_statuses))
             .where(ImportTask.created_at < orphan_cutoff)
             .values(
                 status="failed",

tests/test_maintenance.py

@@ -122,7 +122,10 @@ def test_recover_interrupted_sweeps_pending_orphans_to_failed(db_sync, monkeypat
     Banked 2026-05-25 after operator hit 5490 stuck pending rows.
     """
     from backend.app.tasks import import_file
+    from backend.app.tasks import maintenance as m
     monkeypatch.setattr(import_file.import_media_file, "delay", lambda *_: None)
+    # The broker's import queue is empty, so a stale 'queued' row's message is gone.
+    monkeypatch.setattr(m, "_import_queue_depth", lambda: 0)
 
     batch_id = _make_batch(db_sync)
     now = datetime.now(UTC)
@@ -170,6 +173,42 @@ def test_recover_interrupted_sweeps_pending_orphans_to_failed(db_sync, monkeypat
     assert "orphan" in (stale_pending.error or "")
 
 
+@pytest.mark.parametrize("depth", [7, None])
+def test_recover_interrupted_keeps_queued_rows_while_the_import_queue_holds_work(
+    db_sync, monkeypatch, depth,
+):
+    """A queued row in a batch bigger than half an hour of work is waiting its
+    turn, not orphaned: while the import queue holds messages (or can't be
+    read), only never-enqueued 'pending' rows are swept."""
+    from sqlalchemy import update as _upd
+
+    from backend.app.tasks import maintenance as m
+
+    monkeypatch.setattr(m, "_import_queue_depth", lambda: depth)
+    batch_id = _make_batch(db_sync)
+    waiting = ImportTask(
+        batch_id=batch_id, source_path="/import/waiting.jpg", task_type="media",
+        status="queued",
+    )
+    never_sent = ImportTask(
+        batch_id=batch_id, source_path="/import/never.jpg", task_type="media",
+        status="pending",
+    )
+    db_sync.add_all([waiting, never_sent])
+    db_sync.flush()
+    db_sync.execute(
+        _upd(ImportTask).where(ImportTask.id.in_([waiting.id, never_sent.id]))
+        .values(created_at=datetime.now(UTC) - timedelta(hours=2))
+    )
+    db_sync.commit()
+
+    assert m.recover_interrupted_tasks.apply().get() == 1
+    db_sync.refresh(waiting)
+    db_sync.refresh(never_sent)
+    assert waiting.status == "queued"
+    assert never_sent.status == "failed"
+
+
 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."""