diff --git a/data/repo-awareness-snapshot.json b/data/repo-awareness-snapshot.json index cbfb6c4d55..908754b513 100644 --- a/data/repo-awareness-snapshot.json +++ b/data/repo-awareness-snapshot.json @@ -1,7 +1,7 @@ { "version": "repo-awareness-snapshot-v3", "captured_revision": { - "committed_at": "2026-10-20" + "committed_at": "2026-10-21" }, "routes": { "modes": [ diff --git a/docs/audit/tenancy-defense-in-depth-review.md b/docs/audit/tenancy-defense-in-depth-review.md index 0c93112d7f..be4badb9a0 100644 --- a/docs/audit/tenancy-defense-in-depth-review.md +++ b/docs/audit/tenancy-defense-in-depth-review.md @@ -452,6 +452,7 @@ guard until it is reviewed. Entries live in `DERIVED_QUERY_INVENTORY` in | `src/app/api/images/[id]/signed-url/route.ts` | `document_images` | `GET` | parent-document-verified | Image row fetched by id, then its parent document resolved through `withOwnerReadScope`; 404 and nothing signed unless the parent is visible. | | `src/app/api/images/signed-urls/route.ts` | `document_images` | `POST` | parent-document-verified | Batch form of the same pattern (`/api/documents/images/batch` is a bare re-export of this module); only images whose parent survived `withOwnerReadScope` are signed. | | `src/app/api/ingestion/jobs/route.ts` | `ingestion_jobs` | `GET` (×2) | documents-inner-join | Page query and active-count query both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain itself. | +| `src/app/api/ingestion/jobs/[id]/retry/route.ts` | `ingestion_jobs` | `POST` (×2) | documents-inner-join | Pre-recovery state read and post-recovery re-read both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain. | | `src/app/api/ingestion/quality/route.ts` | `ingestion_jobs` | `GET` | owner-scoped-id-list (`documentIds`) | `.in("document_id", documentIds)` from the handler's `.eq("owner_id", user.id)` documents query. | | `src/app/api/ingestion/quality/route.ts` | `ingestion_job_stages` | `GET` | owner-scoped-id-list (`documentIds`) | Same owner-scoped id list. | | `src/app/api/ingestion/quality/route.ts` | `document_pages` | `GET` | owner-scoped-id-list (`documentIds`) | Same owner-scoped id list. This one reads page `text`, so it is the highest-value derived read in the handler. | diff --git a/docs/reindex-shadow-harness-design.md b/docs/reindex-shadow-harness-design.md index a79f690468..c472599612 100644 --- a/docs/reindex-shadow-harness-design.md +++ b/docs/reindex-shadow-harness-design.md @@ -6,6 +6,12 @@ The offline decision core (`src/lib/reindex-eval-gate.ts` — `decideReindexGate unit-tested; this doc specifies the remaining live pieces so they are turnkey when the billed OpenAI key + Supabase service secrets are available. +**Until this harness is applied, the eval gate is advisory and by hand** (audit 2026-10-20 ING-A-04). +Nothing calls `decideReindexGate`: the worker refuses only a zero-chunk, zero-image generation, so a +worse re-extraction can replace a good live one unchecked. Before approving any library reload, run a +baseline eval canary first, then after the reload run `npm run reindex:health` and a second canary, +and compare the two. Revert on any recall drop or per-case regression. + ## Goal Build the new (better-chunked, re-enriched) index for a document into a **staged** generation diff --git a/docs/worker-deploy-runbook.md b/docs/worker-deploy-runbook.md index 56184630ed..1d331e3c6e 100644 --- a/docs/worker-deploy-runbook.md +++ b/docs/worker-deploy-runbook.md @@ -725,6 +725,15 @@ The second Gate B caveat is already load-bearing above: docling's eager-mode lat node -v # must report v24.x ``` +- **Pausing the worker on purpose (audit EVAL-03):** the worker has no public + address, so monitoring cannot see that it is paused. Whenever you pause it + (`WORKER_PAUSED=true`, or a placeholder start command), also set the GitHub + repository variable `INGESTION_WORKER_PAUSED=true`; set it to `false` when the + worker runs again. The six-hourly ops digest then shows the worker as paused + rather than healthy, and the ingestion autopilot names the pause as the cause + when it alerts. A pause alone never alerts; a pending upload that nothing claims + for more than 30 minutes does, paused or not. + --- ## Rollback diff --git a/scripts/lib/tenancy-scan.mjs b/scripts/lib/tenancy-scan.mjs index bea3f98d7f..8e860ce76d 100644 --- a/scripts/lib/tenancy-scan.mjs +++ b/scripts/lib/tenancy-scan.mjs @@ -1537,6 +1537,15 @@ export const DERIVED_QUERY_INVENTORY = [ reason: 'The page query, the active-count query and the failed-count query each join `documents!inner` and filter `.eq("documents.owner_id", user.id)`, so ownership rides on every one of the three query chains itself. 2 -> 3 on 2026-09-04 when #L15 added the pre-pagination failed-job head count, built as a copy of the active-count chain precisely so it would carry the same on-chain filter rather than need a weaker proof.', }, + { + file: "src/app/api/ingestion/jobs/[id]/retry/route.ts", + table: "ingestion_jobs", + fn: "POST", + queries: 2, + proof: PROOF_KINDS.DOCUMENTS_INNER_JOIN, + reason: + 'Admin Retry (audit 2026-10-20 DB-01): the pre-recovery state read and the post-recovery re-read both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain itself.', + }, { file: "src/app/api/ingestion/quality/route.ts", table: "ingestion_jobs", diff --git a/scripts/reindex-health.ts b/scripts/reindex-health.ts index fb33d6a20e..4a60a529c3 100644 --- a/scripts/reindex-health.ts +++ b/scripts/reindex-health.ts @@ -126,6 +126,35 @@ async function main() { .order("created_at", { ascending: true }) .limit(25); + // Audit 2026-10-20 ING-A-02 / EVAL-09: the open-jobs preview above is the 25 oldest jobs, so a job + // stranded on its final attempt can sit behind old failed jobs unseen. Count it directly, along + // with indexed documents that have no searchable chunks. (Embeddings are NOT NULL, so no missing-embedding count.) + const nowIso = new Date().toISOString(); + const integrityCounts = await Promise.all([ + safeDerivedCount("jobs_stranded_final_attempt", async () => { + const { data, error } = await supabase + .from("ingestion_jobs") + .select("id,attempt_count,max_attempts") + .eq("status", "processing") + .lt("lease_expires_at", nowIso) + .limit(500); + if (error) throw new Error(error.message); + return (data ?? []).filter((job) => Number(job.attempt_count ?? 0) >= Number(job.max_attempts ?? 0)); + }), + safeCount( + "documents_indexed_without_chunks", + supabase + .from("documents") + .select("id", { count: "exact", head: true }) + .eq("status", "indexed") + .eq("chunk_count", 0), + ), + // document_chunks.embedding is NOT NULL, so a missing-embedding count would always be 0; not checked. + ]); + const integrityProblems = integrityCounts + .filter((item) => Number(item.count ?? 0) > 0) + .map((item) => `${item.label}: ${item.count}`); + const counts = [ indexed, queued, @@ -137,7 +166,7 @@ async function main() { strandedQueued, chunksWithSynopsis, ]; - const countErrors = counts.filter((item) => item.error); + const countErrors = [...counts, ...integrityCounts].filter((item) => item.error); const errors = [ ...countErrors.map((item) => ({ label: item.label, error: unavailableMessage(item.error) ?? item.error })), ...(jobsError ? [{ label: "open_jobs", error: unavailableMessage(jobsError.message) ?? jobsError.message }] : []), @@ -148,7 +177,8 @@ async function main() { { ok: countErrors.length === 0 && !jobsError, generatedAt: new Date().toISOString(), - counts: Object.fromEntries(counts.map((item) => [item.label, item.count])), + counts: Object.fromEntries([...counts, ...integrityCounts].map((item) => [item.label, item.count])), + integrityProblems, errors, openJobs: openJobs ?? [], recommendation: diff --git a/src/app/api/documents/[id]/route.ts b/src/app/api/documents/[id]/route.ts index f2fce382c6..c971ac0e2b 100644 --- a/src/app/api/documents/[id]/route.ts +++ b/src/app/api/documents/[id]/route.ts @@ -240,7 +240,7 @@ export async function DELETE(request: Request, { params }: { params: Promise<{ i } if (result.outcome === "active_job") { throw new PublicApiError( - "Document has pending or processing indexing work. Stop or wait for the worker before deleting.", + "Document has pending or processing indexing work. Wait for it to finish, or recover a job stuck on its last attempt, before deleting.", 409, { code: "document_indexing_active" }, ); diff --git a/src/app/api/ingestion/jobs/[id]/retry/route.ts b/src/app/api/ingestion/jobs/[id]/retry/route.ts index cd116cb8d3..3cca4bda18 100644 --- a/src/app/api/ingestion/jobs/[id]/retry/route.ts +++ b/src/app/api/ingestion/jobs/[id]/retry/route.ts @@ -4,7 +4,11 @@ import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit import { env, isDemoMode } from "@/lib/env"; import { jsonError, publicErrorResponse } from "@/lib/http"; import { ingestionJobRetryRejectionReason } from "@/lib/ingestion"; -import { ingestionRollbackFenceStamp } from "@/lib/ingestion-mutation-safety"; +import { + applyIngestionRecoveryActions, + isFreshProcessingJob, + type IngestionRecoveryJob, +} from "@/lib/ingestion-recovery"; import { createAdminClient } from "@/lib/supabase/admin"; import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth"; import { parseRouteParams } from "@/lib/validation/params"; @@ -14,12 +18,28 @@ export const runtime = "nodejs"; const ingestionRetryRouteParamsSchema = z.object({ id: z.string().uuid(), }); -const ingestionRetryResultSchema = z.discriminatedUnion("outcome", [ - z.object({ outcome: z.literal("not_found") }), - z.object({ outcome: z.literal("completed") }), - z.object({ outcome: z.literal("active_worker") }), - z.object({ outcome: z.literal("queued"), job: z.object({}).passthrough() }), -]); + +const ACTIVE_JOB_MESSAGE = + "This job is still being processed by a worker. Wait for it to finish or go stale before retrying."; + +// Audit 2026-10-20 DB-01: Retry goes through recover_ingestion_job, the same hardened path the +// operator recovery scripts use (document authority lock, managed-evidence hold, sibling-job and +// live-lease checks), guarded on the row state read here. Each refusal is a 409 the panel shows. +const RECOVERY_REFUSALS: Record = { + active_lease: { message: ACTIVE_JOB_MESSAGE, code: "ingestion_job_active" }, + state_changed: { + message: "This job or its document has changed since the list was loaded. Refresh and try again.", + code: "ingestion_job_state_changed", + }, + managed_recovery_required: { + message: "This document has managed index evidence, so it needs operator recovery rather than Retry.", + code: "ingestion_job_managed_recovery_required", + }, + authority_unavailable: { + message: "Another change to this document is in progress. Try again shortly.", + code: "ingestion_job_authority_unavailable", + }, +}; export async function POST(request: Request, { params }: { params: Promise<{ id: string }> }) { try { @@ -44,39 +64,75 @@ export async function POST(request: Request, { params }: { params: Promise<{ id: return rateLimitJsonResponse("Too many ingestion administration requests. Retry shortly.", rateLimit); } - const staleThreshold = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString(); - const resetNextRunAt = ingestionRollbackFenceStamp(); - const { data, error } = await supabase.rpc("retry_ingestion_job_if_idle", { - p_job_id: id, - p_owner_id: user.id, - p_stale_before: staleThreshold, - p_max_attempts: env.WORKER_MAX_ATTEMPTS, - p_next_run_at: resetNextRunAt, - p_document_updated_at: ingestionRollbackFenceStamp(), - }); - if (error) throw new Error(error.message); - - const parsed = ingestionRetryResultSchema.safeParse(data); - if (!parsed.success) throw new Error("retry_ingestion_job_if_idle returned an invalid result."); - if (parsed.data.outcome === "not_found") { + const { data: row, error: readError } = await supabase + .from("ingestion_jobs") + .select( + "id,document_id,status,locked_at,lease_token,lease_expires_at,documents!inner(status,owner_id,chunk_count)", + ) + .eq("id", id) + .eq("documents.owner_id", user.id) + .maybeSingle(); + if (readError) throw new Error(readError.message); + const job = row as unknown as (IngestionRecoveryJob & { status: string | null }) | null; + if (!job || job.documents?.owner_id !== user.id) { return publicErrorResponse("Ingestion job not found.", 404, { code: "ingestion_job_not_found" }); } - if (parsed.data.outcome === "completed") { + if (job.status === "completed") { return publicErrorResponse( ingestionJobRetryRejectionReason("completed") ?? "Completed ingestion jobs cannot be retried.", 409, { code: "ingestion_job_completed" }, ); } - if (parsed.data.outcome === "active_worker") { + if (job.status === "pending") { + return publicErrorResponse("This job is already queued.", 409, { code: "ingestion_job_queued" }); + } + if (isFreshProcessingJob(job, new Date(), env.WORKER_STALE_AFTER_MINUTES)) { + return publicErrorResponse(ACTIVE_JOB_MESSAGE, 409, { code: "ingestion_job_active" }); + } + if (job.documents?.status === "indexed" && Number(job.documents.chunk_count ?? 0) > 0) { return publicErrorResponse( - "This job is still being processed by a worker. Wait for it to finish or go stale before retrying.", + "This document is already indexed, so its old job cannot be retried. Use Re-index to rebuild it.", 409, - { code: "ingestion_job_active" }, + { code: "ingestion_document_indexed" }, ); } - return NextResponse.json({ job: parsed.data.job }); + const [result] = await applyIngestionRecoveryActions(supabase, [ + { + action: { action: "retry", jobId: job.id, documentId: job.document_id, resetDocument: false }, + expected: { + jobStatus: job.status ?? null, + jobLockedAt: job.locked_at ?? null, + jobLeaseToken: job.lease_token ?? null, + jobLeaseExpiresAt: job.lease_expires_at ?? null, + documentStatus: job.documents?.status ?? null, + documentOwnerId: job.documents?.owner_id ?? null, + }, + }, + ]); + if (!result?.applied) { + if (result?.reason === "job_closed") { + return publicErrorResponse("Ingestion job not found.", 404, { code: "ingestion_job_not_found" }); + } + const refusal = RECOVERY_REFUSALS[result?.reason ?? ""]; + if (refusal) return publicErrorResponse(refusal.message, 409, { code: refusal.code }); + throw new Error(`recover_ingestion_job refused the retry: ${result?.reason ?? "no result"}.`); + } + + const { data: queued, error: queuedError } = await supabase + .from("ingestion_jobs") + .select("*, documents!inner(owner_id)") + .eq("id", job.id) + .eq("documents.owner_id", user.id) + .maybeSingle(); + if (queuedError) throw new Error(queuedError.message); + // The owner join is only the tenancy filter; keep the response shape the panel already reads. + const queuedJob = queued ? { ...(queued as Record) } : null; + if (queuedJob) delete queuedJob.documents; + return NextResponse.json({ + job: queuedJob ?? { id: job.id, document_id: job.document_id, status: "pending" }, + }); } catch (error) { if (error instanceof AuthenticationError) return unauthorizedResponse(); return jsonError(error); diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts index 3c04ce4c65..1a2e29f85c 100644 --- a/tests/private-access-routes.test.ts +++ b/tests/private-access-routes.test.ts @@ -2571,95 +2571,127 @@ describe("private document API access", () => { expect(client.from).not.toHaveBeenCalled(); }); - it("refuses to retry a job a live worker still holds (IDX-C3, B6)", async () => { - const client = createSupabaseMock(); - // The route now consumes the ingestion_admin rate-limit bucket (its own - // consume_api_rate_limit RPC call) before retry_ingestion_job_if_idle - // (#L43), so a plain mockResolvedValueOnce would intercept that earlier - // call instead. Target the retry RPC by name and fall through to the - // mock's normal per-name behaviour (including the rate-limit check) for - // everything else. + // Audit 2026-10-20 DB-01: the admin Retry button now goes through recover_ingestion_job, the + // hardened recovery path (authority lock, managed-evidence hold, sibling-job check), instead of + // the older retry_ingestion_job_if_idle, which skipped those checks. + const retryJobId = "99999999-9999-4999-8999-999999999999"; + function retryJobRow(overrides: Record = {}, document: Record = {}) { + return { + id: retryJobId, + document_id: documentId, + status: "failed", + locked_at: null, + lease_token: null, + lease_expires_at: null, + documents: { status: "failed", owner_id: userId, chunk_count: 0, ...document }, + ...overrides, + }; + } + function retryClient(row: unknown, recover: (args?: Record) => QueryResult) { + const client = createSupabaseMock((call) => { + if (call.table === "ingestion_jobs" && call.operation === "select" && call.maybeSingle) { + if (call.selected === "*, documents!inner(owner_id)") + return ok({ ...(row as object), status: "pending", stage: "queued after recovery" }); + return ok(row); + } + return defaultQueryResolver(call); + }); const baseRpc = client.rpc.getMockImplementation()!; client.rpc.mockImplementation(async (name: string, args?: Record) => { - if (name === "retry_ingestion_job_if_idle") return ok({ outcome: "active_worker" }); + if (name === "recover_ingestion_job") return recover(args); return baseRpc(name, args); }); - mockRuntime(client); + return client; + } + async function postRetry() { const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); + return POST(authenticatedRequest(`/api/ingestion/jobs/${retryJobId}/retry`, { method: "POST" }), { + params: Promise.resolve({ id: retryJobId }), + }); + } - const response = await POST( - authenticatedRequest("/api/ingestion/jobs/99999999-9999-4999-8999-999999999999/retry", { method: "POST" }), - { - params: Promise.resolve({ id: "99999999-9999-4999-8999-999999999999" }), + it("re-queues a failed job through the hardened recovery RPC with the observed row state (DB-01)", async () => { + const client = retryClient(retryJobRow(), () => ok({ ok: true, action: "retry", artifacts_preserved: true })); + mockRuntime(client); + + const response = await postRetry(); + + expect(response.status).toBe(200); + expect(await payload(response)).toMatchObject({ job: { id: retryJobId, status: "pending" } }); + expect(client.rpc).toHaveBeenCalledWith("recover_ingestion_job", { + p_job_id: retryJobId, + p_action: "retry", + p_expected: { + jobStatus: "failed", + jobLockedAt: null, + jobLeaseToken: null, + jobLeaseExpiresAt: null, + documentStatus: "failed", + documentOwnerId: userId, }, + }); + expect(client.rpc).not.toHaveBeenCalledWith("retry_ingestion_job_if_idle", expect.anything()); + }); + + it("refuses to retry a job a live worker still holds (IDX-C3, B6)", async () => { + const lease = new Date(Date.now() + 10 * 60_000).toISOString(); + const client = retryClient( + retryJobRow({ status: "processing", locked_at: new Date().toISOString(), lease_expires_at: lease }), + () => ok({ ok: true }), ); + mockRuntime(client); + + const response = await postRetry(); expect(response.status).toBe(409); expect(String((await payload(response)).error)).toContain("still being processed"); - expect(client.rpc).toHaveBeenCalledWith( - "retry_ingestion_job_if_idle", - expect.objectContaining({ - p_job_id: "99999999-9999-4999-8999-999999999999", - p_owner_id: userId, - p_stale_before: expect.any(String), - }), - ); - expect(client.calls).toHaveLength(0); + expect(client.rpc).not.toHaveBeenCalledWith("recover_ingestion_job", expect.anything()); }); - it("re-queues a stale/non-processing job without resetting the live index (IDX-C3, IDX-H1, B6)", async () => { - const client = createSupabaseMock(); - // Same reason as the previous test: target the retry RPC by name so the - // earlier rate-limit consume call keeps its own normal behaviour (#L43). - const baseRpc = client.rpc.getMockImplementation()!; - client.rpc.mockImplementation(async (name: string, args?: Record) => { - if (name === "retry_ingestion_job_if_idle") { - return ok({ - outcome: "queued", - job: { id: "99999999-9999-4999-8999-999999999999", document_id: documentId, status: "pending" }, - }); - } - return baseRpc(name, args); - }); + it.each([ + ["active_lease", "still being processed"], + ["state_changed", "changed since"], + ["managed_recovery_required", "operator recovery"], + ["authority_unavailable", "Try again"], + ])("explains a %s refusal from the recovery RPC as a conflict", async (reason, message) => { + const client = retryClient(retryJobRow(), () => ok({ ok: false, reason })); mockRuntime(client); - const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); - const response = await POST( - authenticatedRequest("/api/ingestion/jobs/99999999-9999-4999-8999-999999999999/retry", { method: "POST" }), - { - params: Promise.resolve({ id: "99999999-9999-4999-8999-999999999999" }), - }, - ); - expect(response.status).toBe(200); - expect(await payload(response)).toMatchObject({ job: { status: "pending", document_id: documentId } }); - expect(client.rpc).toHaveBeenCalledWith( - "retry_ingestion_job_if_idle", - expect.objectContaining({ - p_job_id: "99999999-9999-4999-8999-999999999999", - p_owner_id: userId, - p_max_attempts: expect.any(Number), - p_next_run_at: expect.any(String), - p_document_updated_at: expect.any(String), - }), - ); - expect(client.calls).toHaveLength(0); + const response = await postRetry(); + + expect(response.status).toBe(409); + expect(String((await payload(response)).error)).toContain(message); }); - it("returns failure when the transactional retry RPC rolls back", async () => { - const retryJobId = "99999999-9999-4999-8999-999999999999"; - const client = createSupabaseMock(); - client.rpc.mockResolvedValueOnce(fail("transaction rolled back")); + it("does not re-queue an indexed document's job, which recovery would refuse", async () => { + const client = retryClient(retryJobRow({}, { status: "indexed", chunk_count: 12 }), () => ok({ ok: true })); mockRuntime(client); - const { POST } = await import("../src/app/api/ingestion/jobs/[id]/retry/route"); - const response = await POST(authenticatedRequest(`/api/ingestion/jobs/${retryJobId}/retry`, { method: "POST" }), { - params: Promise.resolve({ id: retryJobId }), - }); + const response = await postRetry(); + + expect(response.status).toBe(409); + expect(String((await payload(response)).error)).toContain("already indexed"); + expect(client.rpc).not.toHaveBeenCalledWith("recover_ingestion_job", expect.anything()); + }); + + it("returns 404 for a job outside the administrator's documents", async () => { + const client = retryClient(null, () => ok({ ok: true })); + mockRuntime(client); + + const response = await postRetry(); + + expect(response.status).toBe(404); + expect(client.rpc).not.toHaveBeenCalledWith("recover_ingestion_job", expect.anything()); + }); + + it("returns failure when the recovery RPC errors", async () => { + const client = retryClient(retryJobRow(), () => fail("transaction rolled back")); + mockRuntime(client); + + const response = await postRetry(); expect(response.status).toBe(500); expect(String((await payload(response)).error)).toBe("Request failed."); - expect(client.rpc).toHaveBeenCalledWith("retry_ingestion_job_if_idle", expect.any(Object)); - expect(client.calls).toHaveLength(0); }); it("runs enrichment-only reindex for owned indexed documents using generic metadata", async () => { @@ -4685,7 +4717,8 @@ describe("private document API access", () => { expect(response.status).toBe(409); expect(await payload(response)).toMatchObject({ - error: "Document has pending or processing indexing work. Stop or wait for the worker before deleting.", + error: + "Document has pending or processing indexing work. Wait for it to finish, or recover a job stuck on its last attempt, before deleting.", }); expect(client.calls.some((call) => call.table === "documents" && call.operation === "delete")).toBe(false); expect(client.storageMocks.remove).not.toHaveBeenCalled(); diff --git a/tests/worker-generation-attempt-control.test.ts b/tests/worker-generation-attempt-control.test.ts index 1e4ad91bca..39249496ec 100644 --- a/tests/worker-generation-attempt-control.test.ts +++ b/tests/worker-generation-attempt-control.test.ts @@ -187,7 +187,7 @@ describe("actual issued attempt-control prerequisite", () => { const worker = readFileSync(new URL("../worker/main.ts", import.meta.url), "utf8"); const process = worker.slice(worker.indexOf("async function processJob(job: JobRow)")); const capture = process.indexOf("const issuedControlSnapshot = captureIssuedAttemptControl(job)"); - const heartbeat = process.indexOf("const heartbeat = setInterval("); + const heartbeat = process.indexOf("const heartbeat = startIngestionHeartbeat("); const lifecycle = process.indexOf("try {"); const inventory = process.indexOf('"read_ingestion_generation_inventory"'); const savingPages = process.indexOf('stage: "saving pages"'); diff --git a/tests/worker-ingestion-attempt.test.ts b/tests/worker-ingestion-attempt.test.ts index 3aecc5c255..b0c69ecaac 100644 --- a/tests/worker-ingestion-attempt.test.ts +++ b/tests/worker-ingestion-attempt.test.ts @@ -7,6 +7,7 @@ import { persistIngestionImage, runIngestionMutation, runIngestionStrictCompletion, + startIngestionHeartbeat, writeIngestionTableFacts, } from "../worker/behavior"; @@ -330,3 +331,96 @@ describe("transient database rejections retry", () => { expect(decision.retry).toBe(false); }); }); + +// Audit 2026-10-20 ING-A-03: one failed heartbeat used to cancel the whole attempt and throw +// away its paid work, though the lease still had most of its 45 minutes left. +describe("heartbeat tolerates brief control-plane failures", () => { + const heartbeatRpc = (outcomes: Array<"ok" | "throw" | "uncertain" | "lease_lost" | "rejected" | "transient">) => { + let call = 0; + return vi.fn(async () => { + const outcome = outcomes[Math.min(call, outcomes.length - 1)]; + call += 1; + if (outcome === "throw") throw new TypeError("fetch failed"); + if (outcome === "uncertain") return { data: null, error: { message: "gateway timeout", code: "PGRST003" } }; + if (outcome === "rejected") return { data: null, error: { message: "permission denied", code: "42501" } }; + if (outcome === "transient") return { data: null, error: { message: "statement timeout", code: "57014" } }; + if (outcome === "lease_lost") return { data: { ok: false, reason: "lease_lost" }, error: null }; + return { data: { ok: true }, error: null }; + }); + }; + + async function runBeats( + outcomes: Parameters[0], + beats: number, + options: { intervalMs?: number; toleranceMs?: number } = {}, + ) { + vi.useFakeTimers(); + try { + const attempt = new IngestionAttempt(fence); + const rpc = heartbeatRpc(outcomes); + const handle = startIngestionHeartbeat( + attempt, + () => runIngestionMutation(attempt, rpc, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }), + { + intervalMs: options.intervalMs ?? 60_000, + toleranceMs: options.toleranceMs ?? 15 * 60_000, + leaseMs: 30 * 60_000, + }, + ); + for (let index = 0; index < beats; index += 1) await vi.advanceTimersByTimeAsync(options.intervalMs ?? 60_000); + clearInterval(handle); + return { attempt, rpc }; + } finally { + vi.useRealTimers(); + } + } + + it("keeps the attempt alive through a failing-then-succeeding heartbeat", async () => { + const { attempt, rpc } = await runBeats(["throw", "uncertain", "transient", "ok", "ok"], 5); + expect(rpc).toHaveBeenCalledTimes(5); + expect(attempt.signal.aborted).toBe(false); + }); + + it("revokes once heartbeats have failed for the whole tolerance window", async () => { + const { attempt } = await runBeats(["ok", "throw"], 6, { toleranceMs: 4 * 60_000 }); + expect(attempt.signal.aborted).toBe(true); + }); + + it("still revokes at once when the database reports the lease lost", async () => { + const { attempt } = await runBeats(["lease_lost", "ok"], 1); + expect(attempt.signal.aborted).toBe(true); + }); + + it("still revokes at once on a confirmed, non-transient database rejection", async () => { + const { attempt } = await runBeats(["rejected", "ok"], 1); + expect(attempt.signal.aborted).toBe(true); + }); + + it("refuses a tolerance that is not strictly shorter than the job lease", () => { + const attempt = new IngestionAttempt(fence); + const beat = async () => undefined; + for (const toleranceMs of [30 * 60_000, 31 * 60_000, 0]) + expect(() => + startIngestionHeartbeat(attempt, beat, { intervalMs: 60_000, toleranceMs, leaseMs: 30 * 60_000 }), + ).toThrow(RangeError); + const handle = startIngestionHeartbeat(attempt, beat, { + intervalMs: 60_000, + toleranceMs: 30 * 60_000 - 1, + leaseMs: 30 * 60_000, + }); + clearInterval(handle); + }); + + it("revokes before the lease ends when every heartbeat fails", async () => { + // Lease 30 min, tolerance 10 min: the attempt must stand down well inside the lease. + const { attempt } = await runBeats(["ok", "throw"], 11, { toleranceMs: 10 * 60_000 }); + expect(attempt.signal.aborted).toBe(true); + }); + + it("leaves every other mutation revoking on an uncertain outcome", async () => { + const attempt = new IngestionAttempt(fence); + const rpc = heartbeatRpc(["throw"]); + await expect(runIngestionMutation(attempt, rpc, "heartbeat_ingestion_job")).rejects.toThrow("fetch failed"); + expect(attempt.signal.aborted).toBe(true); + }); +}); diff --git a/tests/worker-lease-transition-contract.test.ts b/tests/worker-lease-transition-contract.test.ts index cd4694040e..98e3f0601f 100644 --- a/tests/worker-lease-transition-contract.test.ts +++ b/tests/worker-lease-transition-contract.test.ts @@ -107,7 +107,7 @@ describe("worker job transitions have no unfenced fallback", () => { const failOrRetry = sourceSegment(workerSource, "async function failOrRetryJob(", "function workerBackoffMs(", { label: "failOrRetryJob", }); - const mutationAdapter = sourceSegment(workerSource, "const mutation = (attempt:", "async function updateJob(", { + const mutationAdapter = sourceSegment(workerSource, "const mutation = (", "async function updateJob(", { label: "worker mutation adapter", }); const fencedMutation = sourceSegment( @@ -136,9 +136,14 @@ describe("worker job transitions have no unfenced fallback", () => { /\.rpc\(rpcName, payload\)\s*\.abortSignal\(\s*CONTROL_PLANE_RPCS\.has\(rpcName\)\s*\?\s*controlPlaneAbortSignal\(attempt\.signal\)\s*:\s*dataPlaneAbortSignal\(attempt\.signal\),?\s*\)/, ); expect(fencedMutation).toContain("p_fence: attempt.fence"); - expect(fencedMutation).toContain("if (result.error) throw new Error(result.error.message)"); + expect(fencedMutation).toContain("const revokeOnUncertainOutcome = options.revokeOnUncertainOutcome !== false;"); + expect(fencedMutation).toMatch( + /if \(result\.error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw new Error\(result\.error\.message\);/, + ); expect(fencedMutation).toContain("attempt.accept(result.data)"); - expect(fencedMutation).toMatch(/catch \(error\) \{\s*attempt\.cancel\(\);\s*throw error;/); + expect(fencedMutation).toMatch( + /catch \(error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw error;/, + ); }); it("failOrRetryJob refuses to demote a document it may no longer own", () => { diff --git a/tests/worker-shadow-extraction.test.ts b/tests/worker-shadow-extraction.test.ts index fe3a26ce41..bc3e0e7e16 100644 --- a/tests/worker-shadow-extraction.test.ts +++ b/tests/worker-shadow-extraction.test.ts @@ -485,7 +485,13 @@ describe("worker/main.ts + env + image contract (source text)", () => { const shadowRun = workerMain.slice(shadowCallIndex, finalMetadataIndex); expect(shadowRun).toMatch(/\}\s*finally\s*\{\s*attempt\.check\(\);\s*\}/); expect(workerMain).not.toContain("const shadowHeartbeat = setInterval("); - expect(workerMain).toContain('mutation(attempt, "heartbeat_ingestion_job").catch(() => attempt.cancel());'); + // The job-wide heartbeat still revokes on a lost lease; since ING-A-03 it tolerates brief + // uncertain failures inside a bounded window (tests/worker-ingestion-attempt.test.ts). + expect(workerMain).toContain( + '() => mutation(attempt, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }),', + ); + expect(workerMain).toContain("const heartbeat = startIngestionHeartbeat("); + expect(workerMain).toContain("leaseMs: env.WORKER_STALE_AFTER_MINUTES * 60_000,"); // The record travels only inside finalMetadata (deep-merged by apply_document_metadata_patch). expect(workerMain.match(/shadow_extraction: shadowExtraction/g)).toHaveLength(1); expect(workerMain).toContain("...(shadowExtraction ? { shadow_extraction: shadowExtraction } : {})"); diff --git a/tests/worker-visual-capture.test.ts b/tests/worker-visual-capture.test.ts index 27e63805dc..ccc80309ee 100644 --- a/tests/worker-visual-capture.test.ts +++ b/tests/worker-visual-capture.test.ts @@ -65,8 +65,14 @@ describe("worker visual capture hardening", () => { // The commit RPC prunes stale/legacy generation rows and upserts index quality atomically // server-side; the worker must surface any RPC failure (fail closed) rather than run a // client-side fallback that could partially write after an error. - expect(behaviorSource).toContain("if (result.error) throw new Error(result.error.message);"); - expect(behaviorSource).toMatch(/catch \(error\) \{\s*attempt\.cancel\(\);\s*throw error;\s*\}/); + // Every mutation revokes on an uncertain outcome unless it opts out; only the heartbeat does. + expect(behaviorSource).toContain("const revokeOnUncertainOutcome = options.revokeOnUncertainOutcome !== false;"); + expect(behaviorSource).toMatch( + /if \(result\.error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw new Error\(result\.error\.message\);\s*\}/, + ); + expect(behaviorSource).toMatch( + /catch \(error\) \{\s*if \(revokeOnUncertainOutcome\) attempt\.cancel\(\);\s*throw error;\s*\}/, + ); expect(workerSource).not.toContain("deleteStaleIndexGenerationRows"); // Image objects are generation-scoped under the server-prepared, lease-scoped prefix. expect(workerSource).toMatch(/await persistIngestionImage\(\{\s*attempt,\s*generation: indexGenerationId,/); diff --git a/worker/behavior.ts b/worker/behavior.ts index 3c4104f97c..458a24b542 100644 --- a/worker/behavior.ts +++ b/worker/behavior.ts @@ -336,20 +336,26 @@ export async function runIngestionMutation( rpc: (name: string, args: Record) => PromiseLike, name: string, payload: Record = {}, + options: { revokeOnUncertainOutcome?: boolean } = {}, ) { + // Only a call that writes nothing (the heartbeat) may opt out; its caller then owns revocation. + const revokeOnUncertainOutcome = options.revokeOnUncertainOutcome !== false; return attempt.serialize(async () => { attempt.check(); let result: IngestionRpcResult; try { result = await rpc(name, { ...payload, p_fence: attempt.fence }); } catch (error) { - attempt.cancel(); + if (revokeOnUncertainOutcome) attempt.cancel(); throw error; } const rejection = result.error ? confirmedRpcRejection(result.error) : null; if (rejection) throw rejection; + if (result.error) { + if (revokeOnUncertainOutcome) attempt.cancel(); + throw new Error(result.error.message); + } try { - if (result.error) throw new Error(result.error.message); const accepted = name === "commit_document_index_generation" ? attempt.handoff(result.data, payload.p_generation) @@ -363,6 +369,41 @@ export async function runIngestionMutation( }); } +/** + * Audit 2026-10-20 ING-A-03: the heartbeat writes no data, so an uncertain or transiently + * rejected heartbeat need not throw away the attempt's paid work. It is tolerated until no + * heartbeat has been confirmed for `toleranceMs` (a third of the stale window by default, so + * the attempt stands down long before another worker can reclaim the lease). A reported lost + * lease, an invalid response or a confirmed non-transient rejection still revokes at once. + */ +export function startIngestionHeartbeat( + attempt: IngestionAttempt, + beat: () => Promise, + options: { intervalMs: number; toleranceMs: number; leaseMs: number; now?: () => number }, +) { + // The tolerance must end strictly before the lease does, or an attempt could keep working after + // another worker has reclaimed its job. Refuse to start rather than run with a bad configuration. + if (!(options.toleranceMs > 0 && options.toleranceMs < options.leaseMs)) + throw new RangeError( + `Heartbeat tolerance (${options.toleranceMs} ms) must be positive and shorter than the job lease (${options.leaseMs} ms).`, + ); + const now = options.now ?? Date.now; + let lastConfirmedAt = now(); + return setInterval(() => { + beat().then( + () => { + lastConfirmedAt = now(); + }, + (error: unknown) => { + const tolerable = + !(error instanceof IngestionAttemptError) && + (!(error instanceof IngestionRpcRejectedError) || isTransientDatabaseRejection(error)); + if (!tolerable || now() - lastConfirmedAt >= options.toleranceMs) attempt.cancel(); + }, + ); + }, options.intervalMs); +} + export function claimedJobFence( job: { id: string; diff --git a/worker/main.ts b/worker/main.ts index 4ba3f2e8c7..319dba8b50 100644 --- a/worker/main.ts +++ b/worker/main.ts @@ -53,6 +53,7 @@ import { enrichmentRepairDecision, ingestionFailureDecision, runIngestionMutation, + startIngestionHeartbeat, runIngestionPreparedWork, resolveWorkerClaimScopeFilter, isPreClaimCandidateAllowed, @@ -171,7 +172,12 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { attempt.check(); return attempt; } - const mutation = (attempt: IngestionAttempt, name: string, args: Record = {}) => + const mutation = ( + attempt: IngestionAttempt, + name: string, + args: Record = {}, + options?: Parameters[4], + ) => runIngestionMutation( attempt, (rpcName, payload) => @@ -184,6 +190,7 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { ), name, args, + options, ); // The runtime services the extracted ingestion stages use. Every member is the same // closure-owned function or client the stages called when they lived in this closure. @@ -457,11 +464,14 @@ export function createWorkerRuntime(options: WorkerRuntimeOptions) { async function processJob(job: JobRow) { const attempt = attemptForJob(job.id); const issuedControlSnapshot = captureIssuedAttemptControl(job); - const heartbeat = setInterval( - () => { - mutation(attempt, "heartbeat_ingestion_job").catch(() => attempt.cancel()); + const heartbeat = startIngestionHeartbeat( + attempt, + () => mutation(attempt, "heartbeat_ingestion_job", {}, { revokeOnUncertainOutcome: false }), + { + intervalMs: Math.min(60_000, jobLeaseHeartbeatMs), + toleranceMs: jobLeaseHeartbeatMs, + leaseMs: env.WORKER_STALE_AFTER_MINUTES * 60_000, }, - Math.min(60_000, jobLeaseHeartbeatMs), ); let extracted: ExtractedDocument | null = null; const atomicReindex = isAtomicReindexCandidate(job.documents);