From 8414ae527beceab6dbf72754908978e4d485c264 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:07:38 +0000 Subject: [PATCH 1/5] Initial plan From 91d54c6c5ab8c0daa0b7d00efddd76284faacd3b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:20:09 +0000 Subject: [PATCH 2/5] fix: harden reindex and retry rollback guards --- src/app/api/documents/[id]/reindex/route.ts | 25 ++- src/app/api/documents/bulk/reindex/route.ts | 25 ++- .../api/ingestion/jobs/[id]/retry/route.ts | 34 +++- tests/private-access-routes.test.ts | 156 ++++++++++++++++++ 4 files changed, 233 insertions(+), 7 deletions(-) diff --git a/src/app/api/documents/[id]/reindex/route.ts b/src/app/api/documents/[id]/reindex/route.ts index 12957ff401..0c8c957c63 100644 --- a/src/app/api/documents/[id]/reindex/route.ts +++ b/src/app/api/documents/[id]/reindex/route.ts @@ -106,7 +106,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: const { data: document, error: documentError } = await supabase .from("documents") - .select("id,owner_id,title,file_name,source_path,import_batch_id,status,metadata") + .select("id,owner_id,title,file_name,source_path,import_batch_id,status,page_count,chunk_count,image_count,error_message,metadata") .eq("id", id) .eq("owner_id", user.id) .maybeSingle(); @@ -199,7 +199,28 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: .select() .single(); - if (jobError) throw new Error(jobError.message); + if (jobError) { + if (!atomicReindex) { + const { error: rollbackError } = await supabase + .from("documents") + .update({ + status: document.status, + error_message: document.error_message, + page_count: document.page_count, + chunk_count: document.chunk_count, + image_count: document.image_count, + }) + .eq("id", id) + .eq("owner_id", user.id) + .eq("status", "queued") + .is("error_message", null) + .eq("page_count", 0) + .eq("chunk_count", 0) + .eq("image_count", 0); + if (rollbackError) throw new Error(rollbackError.message); + } + throw new Error(jobError.message); + } return NextResponse.json({ job }, { status: 201 }); } catch (error) { if (error instanceof AuthenticationError) return unauthorizedResponse(); diff --git a/src/app/api/documents/bulk/reindex/route.ts b/src/app/api/documents/bulk/reindex/route.ts index d7428783f7..724140e80f 100644 --- a/src/app/api/documents/bulk/reindex/route.ts +++ b/src/app/api/documents/bulk/reindex/route.ts @@ -96,7 +96,7 @@ export async function POST(request: Request) { const documentIds = Array.from(new Set(parsed.documentIds)); const { data: documents, error: documentError } = await supabase .from("documents") - .select("id,owner_id,title,file_name,source_path,import_batch_id,status,metadata") + .select("id,owner_id,title,file_name,source_path,import_batch_id,status,page_count,chunk_count,image_count,error_message,metadata") .eq("owner_id", user.id) .in("id", documentIds); if (documentError) throw new Error(documentError.message); @@ -193,7 +193,28 @@ export async function POST(request: Request) { }) .select("id") .single(); - if (jobError) throw new Error(jobError.message); + if (jobError) { + if (!atomicReindex) { + const { error: rollbackError } = await supabase + .from("documents") + .update({ + status: document.status, + error_message: document.error_message, + page_count: document.page_count, + chunk_count: document.chunk_count, + image_count: document.image_count, + }) + .eq("id", document.id) + .eq("owner_id", user.id) + .eq("status", "queued") + .is("error_message", null) + .eq("page_count", 0) + .eq("chunk_count", 0) + .eq("image_count", 0); + if (rollbackError) throw new Error(rollbackError.message); + } + throw new Error(jobError.message); + } results.push({ documentId: document.id, mode: parsed.mode, ok: true, jobId: job.id }); } catch (error) { results.push({ diff --git a/src/app/api/ingestion/jobs/[id]/retry/route.ts b/src/app/api/ingestion/jobs/[id]/retry/route.ts index 5510fd1b24..e2221c2109 100644 --- a/src/app/api/ingestion/jobs/[id]/retry/route.ts +++ b/src/app/api/ingestion/jobs/[id]/retry/route.ts @@ -16,7 +16,9 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: const { data: job, error: jobError } = await supabase .from("ingestion_jobs") - .select("id,document_id,batch_id,status,locked_at,documents!inner(owner_id)") + .select( + "id,document_id,batch_id,status,stage,progress,error_message,attempt_count,max_attempts,locked_at,locked_by,next_run_at,completed_at,documents!inner(owner_id)", + ) .eq("id", id) .eq("documents.owner_id", user.id) .maybeSingle(); @@ -34,6 +36,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: // job is NOT processing, OR its lock is already stale, OR it has no lock. const staleThreshold = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString(); + const nextRunAt = new Date().toISOString(); const { data, error } = await supabase .from("ingestion_jobs") .update({ @@ -45,7 +48,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: max_attempts: env.WORKER_MAX_ATTEMPTS, locked_at: null, locked_by: null, - next_run_at: new Date().toISOString(), + next_run_at: nextRunAt, completed_at: null, }) .eq("id", id) @@ -74,7 +77,32 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: .update({ status: "queued", error_message: null }) .eq("id", job.document_id) .eq("owner_id", user.id); - if (documentError) throw new Error(documentError.message); + if (documentError) { + const { error: rollbackError } = await supabase + .from("ingestion_jobs") + .update({ + status: job.status, + stage: job.stage, + progress: job.progress, + error_message: job.error_message, + attempt_count: job.attempt_count, + max_attempts: job.max_attempts, + locked_at: job.locked_at, + locked_by: job.locked_by, + next_run_at: job.next_run_at, + completed_at: job.completed_at, + }) + .eq("id", id) + .eq("status", "pending") + .eq("stage", "queued") + .eq("progress", 0) + .eq("attempt_count", 0) + .is("locked_at", null) + .is("locked_by", null) + .eq("next_run_at", nextRunAt); + if (rollbackError) throw new Error(rollbackError.message); + throw new Error(documentError.message); + } return NextResponse.json({ job: data }); } catch (error) { diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index ce2291fc7d..746b31aba5 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -1047,6 +1047,50 @@ describe("private document API access", () => { expect(client.rpc).not.toHaveBeenCalled(); }); + it("rolls back retry reset only when this retry's next_run_at is still current", async () => { + const client = createSupabaseMock((call) => { + if (call.table === "ingestion_jobs" && call.operation === "select") { + return ok({ + id: "job-1", + document_id: documentId, + batch_id: null, + status: "failed", + stage: "failed", + progress: 100, + error_message: "old failure", + attempt_count: 2, + max_attempts: 3, + locked_at: null, + locked_by: null, + next_run_at: "2026-01-01T00:00:00.000Z", + completed_at: "2026-01-01T00:05:00.000Z", + }); + } + if (call.table === "ingestion_jobs" && call.operation === "update") { + return ok({ id: "job-1", ...(call.updatePayload as Record) }); + } + if (call.table === "documents" && call.operation === "update") return fail("document update failed"); + return ok([]); + }); + mockRuntime(client); + const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); + + const response = await POST(authenticatedRequest(`/api/ingestion/jobs/job-1/retry`, { method: "POST" }), { + params: Promise.resolve({ id: "job-1" }), + }); + const jobUpdates = client.calls.filter((call) => call.table === "ingestion_jobs" && call.operation === "update"); + const resetUpdate = jobUpdates[0]; + const rollbackUpdate = jobUpdates[1]; + const nextRunAt = (resetUpdate?.updatePayload as { next_run_at?: string } | undefined)?.next_run_at; + + expect(response.status).toBe(400); + expect(await payload(response)).toEqual({ error: "Request could not be completed." }); + expect(jobUpdates).toHaveLength(2); + expect(rollbackUpdate?.filters).toContainEqual({ column: "status", value: "pending" }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "stage", value: "queued" }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "next_run_at", value: nextRunAt }); + }); + it("runs enrichment-only reindex for owned indexed documents using generic metadata", async () => { const document = { id: documentId, @@ -1472,6 +1516,58 @@ describe("private document API access", () => { expect(client.calls.some((call) => call.table === "documents" && call.operation === "update")).toBe(false); }); + it("rolls back single full reindex only if the document still matches this queued reset", async () => { + const document = { + id: documentId, + owner_id: userId, + title: "Failed Protocol", + file_name: "failed.pdf", + source_path: null, + import_batch_id: null, + status: "failed", + page_count: 12, + chunk_count: 34, + image_count: 2, + error_message: "prior failure", + metadata: {}, + }; + const client = createSupabaseMock((call) => { + if (call.table === "documents" && call.operation === "select") return ok(document); + if (call.table === "import_batches") return ok([]); + if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]); + if (call.table === "documents" && call.operation === "update") return ok([]); + if (call.table === "ingestion_jobs" && call.operation === "insert") return fail("job insert failed"); + return ok([]); + }); + mockRuntime(client); + const { POST } = await import("../src/app/api/documents/[id]/reindex/route"); + + const response = await POST( + authenticatedRequest(`/api/documents/${documentId}/reindex`, { + method: "POST", + body: JSON.stringify({ mode: "full" }), + }), + { params: Promise.resolve({ id: documentId }) }, + ); + const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update"); + const rollbackUpdate = documentUpdates[1]; + + expect(response.status).toBe(400); + expect(await payload(response)).toEqual({ error: "Request could not be completed." }); + expect(documentUpdates).toHaveLength(2); + expect(rollbackUpdate?.filters).toContainEqual({ column: "status", value: "queued" }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "page_count", value: 0 }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "chunk_count", value: 0 }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "image_count", value: 0 }); + expect(rollbackUpdate?.updatePayload).toEqual({ + status: "failed", + error_message: "prior failure", + page_count: 12, + chunk_count: 34, + image_count: 2, + }); + }); + it("cleans up uploaded storage when document insert fails", async () => { const client = createSupabaseMock((call) => call.table === "documents" && call.operation === "insert" ? fail("document insert failed") : ok([]), @@ -2773,6 +2869,66 @@ describe("private document API access", () => { expect(client.calls.some((call) => call.table === "documents")).toBe(false); }); + it("rolls back bulk full reindex only if a document still matches this queued reset", async () => { + const document = { + id: documentId, + owner_id: userId, + title: "Bulk Failed Protocol", + file_name: "bulk-failed.pdf", + source_path: null, + import_batch_id: null, + status: "failed", + page_count: 5, + chunk_count: 9, + image_count: 1, + error_message: "bulk prior failure", + metadata: {}, + }; + const client = createSupabaseMock((call) => { + if (call.table === "documents" && call.operation === "select") return ok([document]); + if (call.table === "import_batches") return ok([]); + if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]); + if (call.table === "documents" && call.operation === "update") return ok([]); + if (call.table === "ingestion_jobs" && call.operation === "insert") return fail("job insert failed"); + return ok([]); + }); + mockRuntime(client, { invalidateRagCachesForOwner: vi.fn() }); + const { POST } = await import("../src/app/api/documents/bulk/reindex/route"); + + const response = await POST( + authenticatedRequest("/api/documents/bulk/reindex", { + method: "POST", + body: JSON.stringify({ documentIds: [documentId], mode: "full" }), + }), + ); + const body = await payload(response); + const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update"); + const rollbackUpdate = documentUpdates[1]; + + expect(response.status).toBe(200); + expect(body.ok).toBe(false); + expect(body.results).toEqual([ + { + documentId, + mode: "full", + ok: false, + error: "job insert failed", + }, + ]); + expect(documentUpdates).toHaveLength(2); + expect(rollbackUpdate?.filters).toContainEqual({ column: "status", value: "queued" }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "page_count", value: 0 }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "chunk_count", value: 0 }); + expect(rollbackUpdate?.filters).toContainEqual({ column: "image_count", value: 0 }); + expect(rollbackUpdate?.updatePayload).toEqual({ + status: "failed", + error_message: "bulk prior failure", + page_count: 5, + chunk_count: 9, + image_count: 1, + }); + }); + it("returns a generic not found response when summarizing an unowned document", async () => { const summarizeDocument = vi.fn(async () => { throw new Error("Document not found."); From e0b19a61171c38b0c16a5631630d98d950097b39 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:21:47 +0000 Subject: [PATCH 3/5] fix: use persisted next_run_at for retry rollback guard --- src/app/api/ingestion/jobs/[id]/retry/route.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/app/api/ingestion/jobs/[id]/retry/route.ts b/src/app/api/ingestion/jobs/[id]/retry/route.ts index e2221c2109..d070be02f8 100644 --- a/src/app/api/ingestion/jobs/[id]/retry/route.ts +++ b/src/app/api/ingestion/jobs/[id]/retry/route.ts @@ -67,6 +67,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: { status: 409 }, ); } + const resetNextRunAt = data.next_run_at ?? nextRunAt; // IDX-H1: do NOT reset the document index here. The worker calls resetDocumentIndex at // job start (worker/main.ts), so resetting before enqueue would leave a previously-good @@ -99,7 +100,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: .eq("attempt_count", 0) .is("locked_at", null) .is("locked_by", null) - .eq("next_run_at", nextRunAt); + .eq("next_run_at", resetNextRunAt); if (rollbackError) throw new Error(rollbackError.message); throw new Error(documentError.message); } From 4302eca356615bd9778de2a1198da8d7298b70c4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 10:23:06 +0000 Subject: [PATCH 4/5] chore: clarify retry rollback timestamp naming --- src/app/api/ingestion/jobs/[id]/retry/route.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/app/api/ingestion/jobs/[id]/retry/route.ts b/src/app/api/ingestion/jobs/[id]/retry/route.ts index d070be02f8..2f7e8253c5 100644 --- a/src/app/api/ingestion/jobs/[id]/retry/route.ts +++ b/src/app/api/ingestion/jobs/[id]/retry/route.ts @@ -36,7 +36,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: // job is NOT processing, OR its lock is already stale, OR it has no lock. const staleThreshold = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString(); - const nextRunAt = new Date().toISOString(); + const retryScheduledAt = new Date().toISOString(); const { data, error } = await supabase .from("ingestion_jobs") .update({ @@ -48,7 +48,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: max_attempts: env.WORKER_MAX_ATTEMPTS, locked_at: null, locked_by: null, - next_run_at: nextRunAt, + next_run_at: retryScheduledAt, completed_at: null, }) .eq("id", id) @@ -67,7 +67,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: { status: 409 }, ); } - const resetNextRunAt = data.next_run_at ?? nextRunAt; + const appliedRetryScheduledAt = data.next_run_at ?? retryScheduledAt; // IDX-H1: do NOT reset the document index here. The worker calls resetDocumentIndex at // job start (worker/main.ts), so resetting before enqueue would leave a previously-good @@ -100,7 +100,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: .eq("attempt_count", 0) .is("locked_at", null) .is("locked_by", null) - .eq("next_run_at", resetNextRunAt); + .eq("next_run_at", appliedRetryScheduledAt); if (rollbackError) throw new Error(rollbackError.message); throw new Error(documentError.message); } From e7172c9f5ee9b8330b72fec52f7f0ef7e63e2c4c Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Thu, 2 Jul 2026 20:32:06 +0800 Subject: [PATCH 5/5] test: align rollback-guard tests with current error contract Two fixes for CI failures after main moved: - use a valid UUID job id so the retry test reaches the rollback path instead of being rejected by route param validation (400 Invalid ingestion job id.) - expect 500 / 'Request failed.' for downstream update failures, matching the jsonError contract from the API-semantics fix on main Also merges origin/main into the branch. Co-Authored-By: Claude Fable 5 --- tests/private-access-routes.test.ts | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 2271800129..c4be26dbd2 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -1066,10 +1066,11 @@ describe("private document API access", () => { }); it("rolls back retry reset only when this retry's next_run_at is still current", async () => { + const jobId = "11111111-1111-4111-8111-111111111111"; const client = createSupabaseMock((call) => { if (call.table === "ingestion_jobs" && call.operation === "select") { return ok({ - id: "job-1", + id: jobId, document_id: documentId, batch_id: null, status: "failed", @@ -1085,7 +1086,7 @@ describe("private document API access", () => { }); } if (call.table === "ingestion_jobs" && call.operation === "update") { - return ok({ id: "job-1", ...(call.updatePayload as Record) }); + return ok({ id: jobId, ...(call.updatePayload as Record) }); } if (call.table === "documents" && call.operation === "update") return fail("document update failed"); return ok([]); @@ -1093,16 +1094,16 @@ describe("private document API access", () => { mockRuntime(client); const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); - const response = await POST(authenticatedRequest(`/api/ingestion/jobs/job-1/retry`, { method: "POST" }), { - params: Promise.resolve({ id: "job-1" }), + const response = await POST(authenticatedRequest(`/api/ingestion/jobs/${jobId}/retry`, { method: "POST" }), { + params: Promise.resolve({ id: jobId }), }); const jobUpdates = client.calls.filter((call) => call.table === "ingestion_jobs" && call.operation === "update"); const resetUpdate = jobUpdates[0]; const rollbackUpdate = jobUpdates[1]; const nextRunAt = (resetUpdate?.updatePayload as { next_run_at?: string } | undefined)?.next_run_at; - expect(response.status).toBe(400); - expect(await payload(response)).toEqual({ error: "Request could not be completed." }); + expect(response.status).toBe(500); + expect(await payload(response)).toEqual({ error: "Request failed." }); expect(jobUpdates).toHaveLength(2); expect(rollbackUpdate?.filters).toContainEqual({ column: "status", value: "pending" }); expect(rollbackUpdate?.filters).toContainEqual({ column: "stage", value: "queued" }); @@ -1570,8 +1571,8 @@ describe("private document API access", () => { const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update"); const rollbackUpdate = documentUpdates[1]; - expect(response.status).toBe(400); - expect(await payload(response)).toEqual({ error: "Request could not be completed." }); + expect(response.status).toBe(500); + expect(await payload(response)).toEqual({ error: "Request failed." }); expect(documentUpdates).toHaveLength(2); expect(rollbackUpdate?.filters).toContainEqual({ column: "status", value: "queued" }); expect(rollbackUpdate?.filters).toContainEqual({ column: "page_count", value: 0 });