fix(maintenance): recover_interrupted_tasks — fold SELECT into UPDATE…WHERE…RETURNING so the IN-list no longer blows past psycopg's 65535-parameter ceiling (operator-hit 2026-05-26 after deep scan orphan pile)

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

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")