diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 1a2e028..4ef590a 100644 --- a/backend/app/tasks/maintenance.py +++ b/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 - files; even a large-video transcode caps at the per-task soft_time_limit - (5 min) defined on the task itself. Anything still 'processing' after - that window is a confirmed crash (worker died, DB disconnect mid-flush, - OOM) and must be recycled. Was 30 min historically; tightened - 2026-05-24 after operator hit a 2224-row zombie pile during the IR - migration scan. + 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. + + 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( - update(ImportTask) - .where(ImportTask.id.in_(stuck_ids)) - .values(status="queued", started_at=None, error="recovered from stuck state") - ) + if stuck_ids: + session.execute( + update(ImportTask) + .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 - for tid in stuck_ids: - import_media_file.delay(tid) + if stuck_ids: + from .import_file import import_media_file + 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") diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 6e08a91..7fc04e9 100644 --- a/tests/test_maintenance.py +++ b/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)