From b0bfbc585a1cd7d5e885fc2ed7c0544c8372ea0c Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Tue, 26 May 2026 11:47:40 -0400
Subject: [PATCH] =?UTF-8?q?fix(import-admin):=20retry-failed=20+=20clear-s?=
 =?UTF-8?q?tuck=20=E2=80=94=20same=20UPDATE=E2=80=A6WHERE=20pattern=20as?=
 =?UTF-8?q?=20the=20maintenance=20sweep,=20so=20neither=20endpoint=20can?=
 =?UTF-8?q?=20hit=20psycopg's=2065535-parameter=20ceiling=20once=20accumul?=
 =?UTF-8?q?ated=20row=20counts=20exceed=20~65k?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 backend/app/api/import_admin.py | 61 +++++++++++++++++----------------
 1 file changed, 32 insertions(+), 29 deletions(-)

diff --git a/backend/app/api/import_admin.py b/backend/app/api/import_admin.py
index 367305b..02f65d2 100644
--- a/backend/app/api/import_admin.py
+++ b/backend/app/api/import_admin.py
@@ -103,17 +103,22 @@ async def list_tasks():
 
 @import_admin_bp.route("/retry-failed", methods=["POST"])
 async def retry_failed():
+    # Fold SELECT into UPDATE…WHERE…RETURNING — the prior SELECT-then-
+    # UPDATE-WHERE-id-IN pattern blew past psycopg's 65535-parameter
+    # ceiling once failed_ids exceeded ~65k rows.
     async with get_session() as session:
-        failed_ids = (
-            await session.execute(select(ImportTask.id).where(ImportTask.status == "failed"))
-        ).scalars().all()
+        result = await session.execute(
+            update(ImportTask)
+            .where(ImportTask.status == "failed")
+            .values(
+                status="queued", error=None,
+                started_at=None, finished_at=None,
+            )
+            .returning(ImportTask.id)
+        )
+        failed_ids = [row[0] for row in result.all()]
         if not failed_ids:
             return jsonify({"retried": 0})
-        await session.execute(
-            update(ImportTask)
-            .where(ImportTask.id.in_(failed_ids))
-            .values(status="queued", error=None, started_at=None, finished_at=None)
-        )
         await session.commit()
 
     from ..tasks.import_file import import_media_file
@@ -138,28 +143,26 @@ async def clear_stuck():
     autoretry-looped for 2 days after a corrupt-data PIL OSError.
     """
     async with get_session() as session:
-        stuck_ids = (
-            await session.execute(
-                select(ImportTask.id).where(
-                    ImportTask.status.in_(["pending", "queued", "processing"])
-                )
+        # Fold SELECT into UPDATE…WHERE — see /retry-failed for the
+        # 65535-parameter ceiling rationale. rowcount is enough here
+        # because we don't need the ids afterward (no .delay()).
+        clear_result = await session.execute(
+            update(ImportTask)
+            .where(
+                ImportTask.status.in_(["pending", "queued", "processing"])
             )
-        ).scalars().all()
-        if stuck_ids:
-            await session.execute(
-                update(ImportTask)
-                .where(ImportTask.id.in_(stuck_ids))
-                .values(
-                    status="failed",
-                    finished_at=datetime.now(UTC),
-                    error=(
-                        "manually cleared via /api/import/clear-stuck "
-                        "— stuck in non-terminal state; retry once "
-                        "underlying cause (corrupt file, missing model, "
-                        "etc.) is resolved"
-                    ),
-                )
+            .values(
+                status="failed",
+                finished_at=datetime.now(UTC),
+                error=(
+                    "manually cleared via /api/import/clear-stuck "
+                    "— stuck in non-terminal state; retry once "
+                    "underlying cause (corrupt file, missing model, "
+                    "etc.) is resolved"
+                ),
             )
+        )
+        tasks_failed = clear_result.rowcount or 0
 
         # Finalize any 'running' ImportBatch that no longer has any
         # active children. The "Scanning..." banner is driven by
@@ -195,7 +198,7 @@ async def clear_stuck():
         await session.commit()
 
     return jsonify({
-        "tasks_failed": len(stuck_ids),
+        "tasks_failed": tasks_failed,
         "batches_finalized": finalized_batches,
     })