diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 4ef590a..df854a3 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -54,41 +54,40 @@ def recover_interrupted_tasks() -> int: 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) + # Both sweeps used to be SELECT ids → UPDATE WHERE id IN (...) which + # blew past psycopg's 65535-parameter ceiling once a sweep covered + # 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 + # .delay() each one. + stuck_result = session.execute( + update(ImportTask) .where(ImportTask.status == "processing") .where(ImportTask.started_at < processing_cutoff) - ).scalars().all() + .values( + status="queued", + started_at=None, + error="recovered from stuck state", + ) + .returning(ImportTask.id) + ) + stuck_ids = [row[0] for row in stuck_result.all()] - orphan_ids = session.execute( - select(ImportTask.id) + orphan_result = session.execute( + update(ImportTask) .where(ImportTask.status.in_(["pending", "queued"])) .where(ImportTask.created_at < orphan_cutoff) - ).scalars().all() - - if not stuck_ids and not orphan_ids: - return 0 - - 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" - ), - ) + .values( + status="failed", + error=( + "orphan pending/queued swept by recover_interrupted_tasks " + "(scanner likely crashed mid-enqueue); retry via " + "/api/import/retry-failed" + ), ) + ) + orphan_count = orphan_result.rowcount or 0 session.commit() @@ -97,7 +96,7 @@ def recover_interrupted_tasks() -> int: for tid in stuck_ids: import_media_file.delay(tid) - return len(stuck_ids) + len(orphan_ids) + return len(stuck_ids) + orphan_count @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")