From eb446cdd42e02719fc1b1ec86388dd5282148580 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:13:11 +0800 Subject: [PATCH 1/2] fix: surface and recover stranded uploads Detect aged owner-scoped queued documents without open jobs in the scheduled health probe, alert operators without automatic production mutation, and recover approved candidates through the transactional reindex RPC. --- .github/workflows/ingestion-autopilot.yml | 18 ++++--- docs/outstanding-issues.md | 4 +- docs/reindex-runbook.md | 2 +- docs/webhooks.md | 6 ++- scripts/ingestion-autopilot.ts | 23 ++++++++- scripts/reindex-health.ts | 26 +++++++++- src/lib/stranded-queued-recovery.ts | 49 +++++++++++++++---- tests/ingestion-autopilot.test.ts | 23 +++++++++ tests/stranded-queued-recovery.test.ts | 58 ++++++++++++++++------- 9 files changed, 169 insertions(+), 40 deletions(-) diff --git a/.github/workflows/ingestion-autopilot.yml b/.github/workflows/ingestion-autopilot.yml index e94c071fc..55164f672 100644 --- a/.github/workflows/ingestion-autopilot.yml +++ b/.github/workflows/ingestion-autopilot.yml @@ -1,12 +1,14 @@ # Ingestion autopilot. Probes the live ingestion queue (reindex-health) and, when # a stuck-queue signal is present, runs the existing recovery path. Alerts (opens/ -# updates an issue) only when Supabase is unreachable or recovery fails. +# updates an issue) when recoverable work is detected, Supabase is unreachable, +# or an explicitly requested recovery fails. # # ENABLED IN DRY-RUN: the 6-hourly schedule is active, but recovery only APPLIES when # BOTH the dispatch input `apply=true` AND repo variable INGESTION_AUTOPILOT_APPLY == # "true". The variable is intentionally unset, so scheduled runs are read-only probes -# that alert on unreachable Supabase or (with apply) failed recovery. To allow real -# recovery after observing a clean dry-run cycle, set INGESTION_AUTOPILOT_APPLY=true. +# that alert on detected stuck/stranded work, unreachable Supabase, or (with apply) +# failed recovery. To allow real recovery after reviewing a dry-run, set +# INGESTION_AUTOPILOT_APPLY=true and dispatch with apply=true. name: Ingestion Autopilot on: @@ -75,6 +77,9 @@ jobs: if [ "$APPLY_REQUESTED" = "true" ] && [ "$APPLY_ALLOWED" = "true" ]; then echo "Applying recovery when stuck." npm run ingestion:autopilot -- --apply + elif [ "$GITHUB_EVENT_NAME" = "schedule" ]; then + echo "Scheduled dry-run; alert if recoverable work is detected." + npm run ingestion:autopilot -- --alert-on-stuck else echo "Dry-run (set repo var INGESTION_AUTOPILOT_APPLY=true and dispatch apply=true to recover)." npm run ingestion:autopilot @@ -92,9 +97,10 @@ jobs: "", `Run: ${runUrl}`, "", - "Either Supabase was unreachable or recovery failed. Triage:", - "`npm run reindex:health`, then `npm run recover:ingestion -- --apply` if a stuck", - "queue is confirmed. Do not assume corruption — a transient outage looks the same.", + "Either recoverable work was detected, Supabase was unreachable, or recovery failed. Triage:", + "`npm run reindex:health`, then `npm run recover:ingestion -- --include-stranded-queued --apply`", + "if a stuck queue or aged queued-without-job document is confirmed. Do not assume corruption —", + "a transient outage looks the same.", ].join("\n"); const { data: existing } = await github.rest.issues.listForRepo({ owner: context.repo.owner, diff --git a/docs/outstanding-issues.md b/docs/outstanding-issues.md index 3f722303a..ca1982cbc 100644 --- a/docs/outstanding-issues.md +++ b/docs/outstanding-issues.md @@ -151,8 +151,8 @@ Move resolved rows here with the resolution date and a one-line outcome. Keep th | #073 | issue | Presentation compare dock CTA is a self-link no-op | Fixed in PR #1135: dock shows non-link "Comparing (N)" status while already comparing. | 2026-07-24 | | #074 | issue | Mode-action popup hard-reloads internal clinical routes | Fixed in PR #1135: `master-search-header` uses `router.push` for DSM/Specifiers/Formulation actions and mode href fallback. | 2026-07-24 | | #068 | task | Regenerate full drift-manifest snapshot after schema hygiene | Full Docker `npm run drift:manifest` replay succeeded on a Docker-capable host; `supabase/drift-manifest.json` now carries live `def_hash` values for the plpgsql table-facts body (offline generator_note removed). | 2026-07-24 | -| #052 | issue | Reindex can overlap a fresh agent-enrichment pass | Full/retry single and bulk reindex now preflight `listDocumentsWithActiveAgentEnrichment` / `hasActiveAgentEnrichmentJob`; enrichment mode unchanged; stale leases still do not block. Focused lease-gate tests plus private-access reindex coverage prove the 409 path. | 2026-07-24 | -| #062 | issue | Upload crash can strand a queued document without a job | Added stranded-row reproducer and owner-scoped aged `queued`-without-open-job recovery via `recover:ingestion --include-stranded-queued`, reusing idempotent `enqueueDocumentReindexJob`. Hosted apply of any future atomic-enqueue RPC remains optional; sweep is the smallest recovery path. | 2026-07-24 | +| #052 | issue | Reindex can overlap a fresh agent-enrichment pass | PR #1143 retained the friendly full/retry preflight and closed its check-then-enqueue race with an owner-scoped transactional RPC. Reindex enqueue and the agent claim path serialize on the document row; disposable PostgreSQL proved both interleavings, and exact-head migration replay/unit/build/Chromium/policy/security checks passed. | 2026-07-24 | +| #062 | issue | Upload crash can strand a queued document without a job | Aged owner-scoped `queued`-without-open-job rows are detected by `reindex:health`; the six-hour autopilot raises a durable alert, and guarded recovery uses PR #1143's transactional RPC so enqueue is owner-scoped, idempotent and atomic. `recover:ingestion --include-stranded-queued` remains dry-run/confirmation-first; scheduled production mutation is not enabled. | 2026-07-24 | | #060 | issue | Safety Plan Generator contradicted the privacy contract | PR #1119 removed patient identifier entry, leaves the post-export name line blank, and aligned tool, privacy and PIA copy. DOM/privacy tests and Chromium copy/print/network coverage prove working content remains in React memory with no fetch/XHR; hosted Production UI, build, unit, policy, safety, static-analysis and secret checks passed. Support-contact details remain classified as sensitive local-only working content. | 2026-07-24 | | #061 | issue | Missing answer relevance metadata was treated as source-backed | PR #1125 now requires explicit source-backed relevance for trusted/grounded presentation and prevents visual tables, clinical-note sections and quotes, and comparison metadata from bypassing the render model. Three actionable P2 review paths were fixed; focused policy/DOM tests, offline RAG, production-readiness, build, unit, static, security, and Production UI gates passed. No retrieval, ranking, generation, provider, or data behavior changed. | 2026-07-24 | | #034 | issue | Answer cache can serve stale governance metadata | Current-source verification found direct route coverage already asserts RAG-cache invalidation on document PATCH, source review, label, bulk, and reindex mutation paths. The residual test recommendation is already met; changing the protected cache key is unnecessary. | 2026-07-24 | diff --git a/docs/reindex-runbook.md b/docs/reindex-runbook.md index ef308c364..f05cb98af 100644 --- a/docs/reindex-runbook.md +++ b/docs/reindex-runbook.md @@ -56,7 +56,7 @@ If the response includes `safety.safeToRun: false`, do not retry repeatedly. Use - `supabase_unavailable`: pause reindexing and rerun `npm run supabase:recovery-status`. - `active_jobs`: wait for pending or processing jobs to finish before retrying. - `stale_processing_jobs`: run `npm run recover:ingestion -- --apply --limit 20`, then rerun `npm run reindex:health`. -- `queued` documents with no open job (upload crash between document insert and job insert): dry-run `npm run recover:ingestion -- --include-stranded-queued`, then re-run with `--apply` after reviewing the candidate list. Optional `--owner-id` and `--stranded-min-age-minutes` keep the sweep bounded. +- `queued` documents with no open job (upload crash between document insert and job insert): dry-run `npm run recover:ingestion -- --include-stranded-queued`, then re-run with `--apply` after reviewing the candidate list. Optional `--owner-id` and `--stranded-min-age-minutes` keep the sweep bounded. The six-hour ingestion-autopilot probe detects aged owner-scoped rows and opens/updates the operator alert issue; it never applies this recovery unless the guarded manual apply dispatch is used. The preflight response also includes active job counts and job metadata for operator review. diff --git a/docs/webhooks.md b/docs/webhooks.md index 59b448d20..299933c69 100644 --- a/docs/webhooks.md +++ b/docs/webhooks.md @@ -189,8 +189,10 @@ missing or the POST fails (it just does nothing, mirroring the receiver's inert > an orphan, but a process crash/timeout/kill _between_ the two inserts can still > strand the same `queued`-without-job row nothing sweeps. The only true > must-never-drop guarantees are an atomic enqueue (document + job in one -> transaction/RPC) or a scheduled queued-without-job sweep. Operator recovery -> for aged stranded rows is available via +> transaction/RPC) or a scheduled queued-without-job sweep. The six-hour +> ingestion-autopilot probe now detects aged owner-scoped rows and raises a +> durable operator alert without mutating production automatically. Recovery +> uses the transactional reindex RPC and is available via > `npm run recover:ingestion -- --include-stranded-queued` (optional > `--owner-id`, `--stranded-min-age-minutes`, `--apply`). This trigger is a > low-latency optimisation, not a delivery guarantee. diff --git a/scripts/ingestion-autopilot.ts b/scripts/ingestion-autopilot.ts index 120e70b35..b77d534c6 100644 --- a/scripts/ingestion-autopilot.ts +++ b/scripts/ingestion-autopilot.ts @@ -39,6 +39,7 @@ export type IngestionAssessment = { reasons: string[]; failedJobs: number; staleProcessingJobs: number; + strandedQueuedDocuments: number; }; /** @@ -55,7 +56,14 @@ export function assessIngestionHealth( const now = options.now ?? Date.now(); if (health.ok === false || health.status === "supabase_unavailable") { - return { available: false, stuck: false, reasons: ["supabase unavailable"], failedJobs: 0, staleProcessingJobs: 0 }; + return { + available: false, + stuck: false, + reasons: ["supabase unavailable"], + failedJobs: 0, + staleProcessingJobs: 0, + strandedQueuedDocuments: 0, + }; } const reasons: string[] = []; @@ -75,6 +83,10 @@ export function assessIngestionHealth( if (staleProcessingJobs > 0) { reasons.push(`${staleProcessingJobs} processing job(s) locked > ${staleAfterMinutes}m (stale worker)`); } + const strandedQueuedDocuments = Number(health.counts?.documents_stranded_queued ?? 0) || 0; + if (strandedQueuedDocuments > 0) { + reasons.push(`${strandedQueuedDocuments} stranded queued document(s)`); + } return { available: true, @@ -82,6 +94,7 @@ export function assessIngestionHealth( reasons, failedJobs, staleProcessingJobs, + strandedQueuedDocuments, }; } @@ -101,6 +114,7 @@ function runTsxScript(script: string, args: string[] = []) { function main() { const apply = process.argv.includes("--apply"); + const alertOnStuck = process.argv.includes("--alert-on-stuck"); const staleAfterMinutes = parseIntFlag("--stale-after-minutes", 30); const limit = parseIntFlag("--limit", 20); @@ -134,9 +148,13 @@ function main() { if (!apply) { console.log( `[autopilot] DRY RUN — would run: npm run recover:ingestion -- --apply --yes ` + - `--stale-after-minutes ${staleAfterMinutes} --limit ${limit}\n` + + `--include-stranded-queued --stale-after-minutes ${staleAfterMinutes} --limit ${limit}\n` + `[autopilot] pass --apply to recover.`, ); + if (alertOnStuck) { + console.error("[autopilot] scheduled probe detected recoverable ingestion work — alerting."); + process.exitCode = 2; + } return; } @@ -144,6 +162,7 @@ function main() { const recover = runTsxScript("scripts/recover-ingestion-queue.ts", [ "--apply", "--yes", + "--include-stranded-queued", "--stale-after-minutes", String(staleAfterMinutes), "--limit", diff --git a/scripts/reindex-health.ts b/scripts/reindex-health.ts index 316282663..fb33d6a20 100644 --- a/scripts/reindex-health.ts +++ b/scripts/reindex-health.ts @@ -24,10 +24,23 @@ async function safeCount(label: string, query: PromiseLike) { } } +async function safeDerivedCount(label: string, load: () => Promise) { + try { + return { label, count: (await load()).length, error: null }; + } catch (error) { + return { label, count: null, error: error instanceof Error ? error.message : String(error) }; + } +} + async function main() { - const [{ requireServerEnv }, { createAdminClient }] = await Promise.all([ + const [ + { requireServerEnv }, + { createAdminClient }, + { listStrandedQueuedDocuments, STRANDED_QUEUED_DEFAULT_LIMIT, STRANDED_QUEUED_DEFAULT_MIN_AGE_MINUTES }, + ] = await Promise.all([ import("@/lib/env"), import("@/lib/supabase/admin"), + import("@/lib/stranded-queued-recovery"), ]); requireServerEnv(); const supabase = createAdminClient(); @@ -58,6 +71,7 @@ async function main() { pendingJobs, processingJobs, failedJobs, + strandedQueued, chunksWithSynopsis, ] = await Promise.all([ safeCount( @@ -88,6 +102,13 @@ async function main() { "jobs_failed", supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "failed"), ), + safeDerivedCount("documents_stranded_queued", () => + listStrandedQueuedDocuments({ + supabase, + minAgeMinutes: STRANDED_QUEUED_DEFAULT_MIN_AGE_MINUTES, + limit: STRANDED_QUEUED_DEFAULT_LIMIT, + }), + ), safeCount( "chunks_with_retrieval_synopsis", supabase @@ -113,6 +134,7 @@ async function main() { pendingJobs, processingJobs, failedJobs, + strandedQueued, chunksWithSynopsis, ]; const countErrors = counts.filter((item) => item.error); @@ -131,7 +153,7 @@ async function main() { openJobs: openJobs ?? [], recommendation: countErrors.length === 0 && !jobsError - ? "If no jobs are processing unexpectedly, run npm run recover:ingestion -- --apply before resuming conservative worker:once." + ? "If no jobs are processing unexpectedly, run npm run recover:ingestion -- --include-stranded-queued --apply before resuming conservative worker:once." : "Fix the reported read errors before running workers or evals.", }, null, diff --git a/src/lib/stranded-queued-recovery.ts b/src/lib/stranded-queued-recovery.ts index 5d202e4c1..b62f75053 100644 --- a/src/lib/stranded-queued-recovery.ts +++ b/src/lib/stranded-queued-recovery.ts @@ -1,10 +1,18 @@ import "server-only"; -import { enqueueDocumentReindexJob, type EnqueueableDocument } from "@/lib/ingestion-enqueue"; +import { env } from "@/lib/env"; import type { createAdminClient } from "@/lib/supabase/admin"; type AdminClient = ReturnType; -export type StrandedQueuedDocument = EnqueueableDocument & { +export type StrandedQueuedDocument = { + id: string; + owner_id: string | null; + status?: string | null; + error_message?: string | null; + page_count?: number | null; + chunk_count?: number | null; + image_count?: number | null; + import_batch_id?: string | null; created_at: string; updated_at: string; }; @@ -13,6 +21,7 @@ export type StrandedQueuedRecoveryResult = | { documentId: string; outcome: "enqueued"; jobId?: string } | { documentId: string; outcome: "already_active" } | { documentId: string; outcome: "document_deleted" } + | { documentId: string; outcome: "skipped_ownerless" } | { documentId: string; outcome: "skipped_young"; ageMinutes: number } | { documentId: string; outcome: "error"; message: string }; @@ -73,6 +82,7 @@ export async function listStrandedQueuedDocuments(args: { "id,owner_id,status,error_message,page_count,chunk_count,image_count,import_batch_id,created_at,updated_at", ) .eq("status", "queued") + .not("owner_id", "is", null) .lt("updated_at", cutoff) .order("updated_at", { ascending: true }); @@ -131,20 +141,43 @@ export async function recoverStrandedQueuedDocuments(args: { documents: StrandedQueuedDocument[]; }): Promise { const results: StrandedQueuedRecoveryResult[] = []; + const staleBefore = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString(); for (const document of args.documents) { + if (!document.owner_id) { + results.push({ documentId: document.id, outcome: "skipped_ownerless" }); + continue; + } + try { - const result = await enqueueDocumentReindexJob({ - supabase: args.supabase, - document, + const { data, error } = await args.supabase.rpc("request_ingestion_reindex_if_agent_idle", { + p_document_id: document.id, + p_owner_id: document.owner_id, + p_stale_before: staleBefore, + p_max_attempts: env.WORKER_MAX_ATTEMPTS, }); - if (result.outcome === "enqueued") { + if (error) throw new Error(error.message); + if (!data || typeof data !== "object" || !("outcome" in data) || typeof data.outcome !== "string") { + throw new Error("Atomic stranded-queue recovery returned an invalid result."); + } + + if (data.outcome === "queued") { const jobId = - result.job && typeof result.job === "object" && "id" in result.job ? String(result.job.id ?? "") : undefined; + "job" in data && data.job && typeof data.job === "object" && "id" in data.job + ? String(data.job.id ?? "") + : undefined; results.push({ documentId: document.id, outcome: "enqueued", jobId: jobId || undefined }); continue; } - results.push({ documentId: document.id, outcome: result.outcome }); + if (data.outcome === "ingestion_active" || data.outcome === "agent_enrichment_active") { + results.push({ documentId: document.id, outcome: "already_active" }); + continue; + } + if (data.outcome === "not_found") { + results.push({ documentId: document.id, outcome: "document_deleted" }); + continue; + } + throw new Error(`Atomic stranded-queue recovery returned unknown outcome "${data.outcome}".`); } catch (error) { results.push({ documentId: document.id, diff --git a/tests/ingestion-autopilot.test.ts b/tests/ingestion-autopilot.test.ts index a7123e2b7..f8a25cf2e 100644 --- a/tests/ingestion-autopilot.test.ts +++ b/tests/ingestion-autopilot.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it } from "vitest"; +import { readFileSync } from "node:fs"; import { assessIngestionHealth } from "../scripts/ingestion-autopilot"; +const autopilotSource = readFileSync(new URL("../scripts/ingestion-autopilot.ts", import.meta.url), "utf8"); +const healthSource = readFileSync(new URL("../scripts/reindex-health.ts", import.meta.url), "utf8"); +const workflow = readFileSync(new URL("../.github/workflows/ingestion-autopilot.yml", import.meta.url), "utf8"); + const NOW = Date.parse("2026-07-13T12:00:00Z"); const minutesAgo = (m: number) => new Date(NOW - m * 60_000).toISOString(); @@ -54,4 +59,22 @@ describe("assessIngestionHealth", () => { expect(a.stuck).toBe(false); expect(a.staleProcessingJobs).toBe(0); }); + + it("flags aged queued documents without open jobs as stuck", () => { + const a = assessIngestionHealth( + { ok: true, counts: { jobs_failed: 0, documents_stranded_queued: 2 }, openJobs: [] }, + { now: NOW }, + ); + expect(a.stuck).toBe(true); + expect(a.strandedQueuedDocuments).toBe(2); + expect(a.reasons).toContain("2 stranded queued document(s)"); + }); + + it("wires scheduled detection to a durable alert and the existing bounded recovery path", () => { + expect(healthSource).toContain('"documents_stranded_queued"'); + expect(autopilotSource).toContain('"--include-stranded-queued"'); + expect(autopilotSource).toContain('process.argv.includes("--alert-on-stuck")'); + expect(workflow).toContain("npm run ingestion:autopilot -- --alert-on-stuck"); + expect(workflow).toContain("--include-stranded-queued"); + }); }); diff --git a/tests/stranded-queued-recovery.test.ts b/tests/stranded-queued-recovery.test.ts index a1d39853d..5cc236941 100644 --- a/tests/stranded-queued-recovery.test.ts +++ b/tests/stranded-queued-recovery.test.ts @@ -1,10 +1,4 @@ -import { beforeEach, describe, expect, it, vi } from "vitest"; - -const enqueueDocumentReindexJob = vi.hoisted(() => vi.fn()); - -vi.mock("@/lib/ingestion-enqueue", () => ({ - enqueueDocumentReindexJob, -})); +import { describe, expect, it, vi } from "vitest"; import { isStrandedQueuedDocument, @@ -12,10 +6,6 @@ import { recoverStrandedQueuedDocuments, } from "../src/lib/stranded-queued-recovery"; -beforeEach(() => { - enqueueDocumentReindexJob.mockReset(); -}); - describe("stranded queued-without-job recovery (#062)", () => { const now = new Date("2026-07-24T12:00:00.000Z"); @@ -93,6 +83,10 @@ describe("stranded queued-without-job recovery (#062)", () => { filters.push({ column, value }); return documentsQuery; }); + documentsQuery.not = vi.fn((column: string, operator: string, value: unknown) => { + filters.push({ column, value: `${operator}:${String(value)}` }); + return documentsQuery; + }); documentsQuery.lt = vi.fn(() => documentsQuery); documentsQuery.order = vi.fn(() => documentsQuery); documentsQuery.range = vi.fn(async () => ({ data: [stranded, withOpenJob], error: null })); @@ -122,6 +116,7 @@ describe("stranded queued-without-job recovery (#062)", () => { expect(filters).toContainEqual({ column: "owner_id", value: ownerId }); expect(filters).toContainEqual({ column: "status", value: "queued" }); + expect(filters).toContainEqual({ column: "owner_id", value: "is:null" }); expect(listed.map((document) => document.id)).toEqual([stranded.id]); }); @@ -154,6 +149,7 @@ describe("stranded queued-without-job recovery (#062)", () => { const documentsQuery: Record = {}; documentsQuery.select = vi.fn(() => documentsQuery); documentsQuery.eq = vi.fn(() => documentsQuery); + documentsQuery.not = vi.fn(() => documentsQuery); documentsQuery.lt = vi.fn(() => documentsQuery); documentsQuery.order = vi.fn(() => documentsQuery); documentsQuery.range = vi.fn(async () => { @@ -193,10 +189,11 @@ describe("stranded queued-without-job recovery (#062)", () => { expect(listed.map((document) => document.id)).toEqual([stranded.id]); }); - it("recovers a stranded row by enqueueing exactly one job and treats already-active as idempotent", async () => { - enqueueDocumentReindexJob - .mockResolvedValueOnce({ outcome: "enqueued", job: { id: "job-1" } }) - .mockResolvedValueOnce({ outcome: "already_active" }); + it("recovers with the atomic owner-scoped RPC and treats an existing job as idempotent", async () => { + const rpc = vi + .fn() + .mockResolvedValueOnce({ data: { outcome: "queued", job: { id: "job-1" } }, error: null }) + .mockResolvedValueOnce({ data: { outcome: "ingestion_active" }, error: null }); const docs = [ { @@ -225,11 +222,38 @@ describe("stranded queued-without-job recovery (#062)", () => { }, ]; - const results = await recoverStrandedQueuedDocuments({ supabase: {} as never, documents: docs }); + const results = await recoverStrandedQueuedDocuments({ supabase: { rpc } as never, documents: docs }); expect(results).toEqual([ { documentId: docs[0].id, outcome: "enqueued", jobId: "job-1" }, { documentId: docs[1].id, outcome: "already_active" }, ]); - expect(enqueueDocumentReindexJob).toHaveBeenCalledTimes(2); + expect(rpc).toHaveBeenCalledTimes(2); + expect(rpc).toHaveBeenNthCalledWith(1, "request_ingestion_reindex_if_agent_idle", { + p_document_id: docs[0].id, + p_owner_id: docs[0].owner_id, + p_stale_before: expect.any(String), + p_max_attempts: expect.any(Number), + }); + }); + + it("fails closed for an ownerless candidate instead of issuing an unscoped recovery mutation", async () => { + const rpc = vi.fn(); + const document = { + id: "11111111-1111-4111-8111-111111111111", + owner_id: null, + status: "queued", + error_message: null, + page_count: 0, + chunk_count: 0, + image_count: 0, + import_batch_id: null, + created_at: "2026-07-24T11:00:00.000Z", + updated_at: "2026-07-24T11:00:00.000Z", + }; + + await expect( + recoverStrandedQueuedDocuments({ supabase: { rpc } as never, documents: [document] }), + ).resolves.toEqual([{ documentId: document.id, outcome: "skipped_ownerless" }]); + expect(rpc).not.toHaveBeenCalled(); }); }); From ff3cc60cbfe028e4a8c0a47141e2cca48da7063e Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:21:06 +0800 Subject: [PATCH 2/2] fix: keep failed jobs in recovery plan Exclude queued documents with failed ingestion jobs from the stranded sweep so ordinary failed-job recovery retains the single action limit. --- src/lib/stranded-queued-recovery.ts | 11 ++++++----- tests/stranded-queued-recovery.test.ts | 5 ++++- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/src/lib/stranded-queued-recovery.ts b/src/lib/stranded-queued-recovery.ts index b62f75053..c6e9756ed 100644 --- a/src/lib/stranded-queued-recovery.ts +++ b/src/lib/stranded-queued-recovery.ts @@ -38,7 +38,8 @@ function ageMinutes(iso: string, nowMs: number) { // Pure predicate used by the reproducer and recovery planner: a document is // stranded when it is still `queued`, old enough that a mid-upload crash is -// plausible, and has no open pending/processing ingestion job. +// plausible, and has no pending/processing job or failed job already handled by +// the ordinary ingestion-recovery plan. export function isStrandedQueuedDocument(args: { document: { status: string | null; created_at: string; updated_at?: string | null }; openJobCount: number; @@ -97,15 +98,15 @@ export async function listStrandedQueuedDocuments(args: { if (candidates.length === 0) break; const documentIds = candidates.map((document) => document.id); - const { data: openJobs, error: openJobsError } = await args.supabase + const { data: recoveryJobs, error: recoveryJobsError } = await args.supabase .from("ingestion_jobs") .select("document_id") .in("document_id", documentIds) - .in("status", ["pending", "processing"]); - if (openJobsError) throw new Error(openJobsError.message); + .in("status", ["pending", "processing", "failed"]); + if (recoveryJobsError) throw new Error(recoveryJobsError.message); const openJobCounts = new Map(); - for (const job of openJobs ?? []) { + for (const job of recoveryJobs ?? []) { const documentId = job.document_id; if (!documentId) continue; openJobCounts.set(documentId, (openJobCounts.get(documentId) ?? 0) + 1); diff --git a/tests/stranded-queued-recovery.test.ts b/tests/stranded-queued-recovery.test.ts index 5cc236941..b5e10d0ce 100644 --- a/tests/stranded-queued-recovery.test.ts +++ b/tests/stranded-queued-recovery.test.ts @@ -92,11 +92,13 @@ describe("stranded queued-without-job recovery (#062)", () => { documentsQuery.range = vi.fn(async () => ({ data: [stranded, withOpenJob], error: null })); let jobInCalls = 0; + let recoveryJobStatuses: string[] = []; const jobsQuery: Record = {}; jobsQuery.select = vi.fn(() => jobsQuery); - jobsQuery.in = vi.fn(() => { + jobsQuery.in = vi.fn((column: string, values: string[]) => { jobInCalls += 1; if (jobInCalls >= 2) { + if (column === "status") recoveryJobStatuses = [...values]; return Promise.resolve({ data: [{ document_id: withOpenJob.id }], error: null }); } return jobsQuery; @@ -117,6 +119,7 @@ describe("stranded queued-without-job recovery (#062)", () => { expect(filters).toContainEqual({ column: "owner_id", value: ownerId }); expect(filters).toContainEqual({ column: "status", value: "queued" }); expect(filters).toContainEqual({ column: "owner_id", value: "is:null" }); + expect(recoveryJobStatuses).toEqual(["pending", "processing", "failed"]); expect(listed.map((document) => document.id)).toEqual([stranded.id]); });