From 110c1c0e513fe71fdcfec86d220882ec69bef9fb Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 26 May 2026 11:46:06 -0400 Subject: [PATCH] =?UTF-8?q?fix(maintenance):=20recover=5Finterrupted=5Ftas?= =?UTF-8?q?ks=20=E2=80=94=20fold=20SELECT=20into=20UPDATE=E2=80=A6WHERE?= =?UTF-8?q?=E2=80=A6RETURNING=20so=20the=20IN-list=20no=20longer=20blows?= =?UTF-8?q?=20past=20psycopg's=2065535-parameter=20ceiling=20(operator-hit?= =?UTF-8?q?=202026-05-26=20after=20deep=20scan=20orphan=20pile)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/tasks/maintenance.py | 59 ++++++++++++++++---------------- 1 file changed, 29 insertions(+), 30 deletions(-) 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")