From 32cac46b8b4984ff7f16475efb3cb2ec7763fe4d Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 6 Nov 2026 16:28:04 -0500 Subject: [PATCH] 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 --- backend/app/tasks/maintenance.py | 48 ++++++++++++++++++++++++-------- tests/test_maintenance.py | 39 ++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 11 deletions(-) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index a6651f2..4c4c1a9 100644 --- a/backend/app/tasks/maintenance.py +++ b/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 - 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. + 2. 'pending' > 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. + 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. + + 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", diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 0738d61..9e40150 100644 --- a/tests/test_maintenance.py +++ b/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."""