From 86c2cb336043e6b27db9b8b9bec9edd2d4d80699 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 24 Oct 2026 21:22:01 +0200 Subject: [PATCH] fix(documents): stop the index-all walk when the provider keeps failing The content-index backfill capped a run at 200 documents, but counted only the ones it indexed. Against a provider that failed every call (a model that reasons and never answers) the cap was never reached, so the walk transcribed every un-indexed document in turn, one provider call each, until the job's time ran out after more than two hours, and the refunded failures never reached the day's token ceiling. The cap now counts provider calls, the walk stops with reason provider-failing after three failed calls in a row (one document the model cannot read is still walked past), and a failed read charges what its reply reported spending. The per-document read moves into its own function. Refs #1279 --- ...ent-content-index-backfill-failing.test.ts | 144 ++++++++++++++++++ .../jobs/document-content-index-backfill.ts | 124 ++++++++++----- 2 files changed, 232 insertions(+), 36 deletions(-) create mode 100644 src/lib/jobs/__tests__/document-content-index-backfill-failing.test.ts diff --git a/src/lib/jobs/__tests__/document-content-index-backfill-failing.test.ts b/src/lib/jobs/__tests__/document-content-index-backfill-failing.test.ts new file mode 100644 index 000000000..dcdd6126e --- /dev/null +++ b/src/lib/jobs/__tests__/document-content-index-backfill-failing.test.ts @@ -0,0 +1,144 @@ +/** + * #1279 — the content-index backfill against a provider that fails every + * call (a model that reasons and never answers). Its cap counted indexed + * documents only, so a provider that never answered never reached it, and + * the walk transcribed every un-indexed document in turn until the job's + * time ran out. It now stops after a few failures in a row, counts calls + * against the cap, and charges what the failed replies reported spending. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const findMany = vi.fn(); +const transcribeDocument = vi.fn(); +const reconcileSpend = vi.fn(); + +vi.mock("@/lib/db", () => ({ + prisma: { + inboundDocument: { + findMany: (...a: unknown[]) => findMany(...a), + findFirst: async (args: { where: { id: string } }) => ({ + id: args.where.id, + kind: "OTHER", + contentEncrypted: new Uint8Array([1, 2, 3]), + contentCodec: "binary2", + mimeType: "image/png", + status: "STORED", + }), + }, + documentContentIndex: { + upsert: async () => ({}), + findMany: async () => [], + update: async () => ({}), + }, + }, +})); +vi.mock("@/lib/documents/store", () => ({ + decryptDocumentContent: () => Buffer.from([1, 2, 3]), +})); +vi.mock("@/lib/labs/ocr-upload", () => ({ + detectOcrMimeType: () => "image/png", +})); +vi.mock("@/lib/documents/rasterize-pdf", () => ({ rasterizePdf: vi.fn() })); +vi.mock("@/lib/documents/describe", () => ({ + transcribeDocument: (...a: unknown[]) => transcribeDocument(...a), + DocumentDescribeError: class DocumentDescribeError extends Error {}, +})); +vi.mock("@/lib/documents/content-index", () => ({ + upsertContentIndex: async () => ({ tokenCount: 3 }), + CONTENT_TOKENIZER_VERSION: "1", + decryptIndexText: () => "decrypted index text", + tokeniseAndHash: () => ["hash-a"], +})); +vi.mock("@/lib/documents/provider-order", () => ({ + resolveDocumentVisionProvider: async () => ({ + chain: [{ providerType: "openai-compatible", instance: {} }], + withheld: null, + pick: { + entry: { providerType: "openai-compatible", instance: {} }, + providerType: "openai-compatible", + pdfSupported: true, + }, + }), +})); +vi.mock("@/lib/ai/capabilities/gate", () => ({ + aiCapabilityForJob: async () => ({ + available: true, + reason: null, + onDeviceAllowed: true, + }), + aiCapabilityForRecord: vi.fn(), +})); +vi.mock("@/lib/ai/coach/budget", () => ({ + buildDateKey: () => "2026-10-24", + reserveBudget: async () => ({ allowed: true, reserved: 3000, owner: "user" }), + reconcileSpend: (...a: unknown[]) => reconcileSpend(...a), + resolveDailyCap: () => 1_000_000, + resolveDailyCapFor: () => 1_000_000, + resolveCostOwner: () => "user", +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() })); + +import { + BACKFILL_FAILURES_IN_A_ROW, + runContentIndexBackfillForUser, +} from "../document-content-index-backfill"; + +/** 120 un-indexed images, served 50 to a page like the real walk asks. */ +function seedDocuments(count: number) { + const ids = Array.from({ length: count }, (_, i) => ({ + id: `d${String(i).padStart(4, "0")}`, + })); + findMany.mockImplementation( + async (args: { take: number; cursor?: { id: string } }) => { + const start = args.cursor + ? ids.findIndex((d) => d.id === args.cursor!.id) + 1 + : 0; + return ids.slice(start, start + args.take); + }, + ); +} + +beforeEach(() => { + vi.resetAllMocks(); + reconcileSpend.mockResolvedValue(undefined); +}); + +describe("#1279: the content-index backfill and a provider that never answers", () => { + it("stops after a few failures in a row instead of walking every document", async () => { + seedDocuments(120); + transcribeDocument.mockRejectedValue( + Object.assign(new Error("returned empty content"), { + httpStatus: 0, + kind: "reasoning_only", + tokensUsed: 2700, + }), + ); + + const result = await runContentIndexBackfillForUser("user-1"); + + expect(transcribeDocument).toHaveBeenCalledTimes( + BACKFILL_FAILURES_IN_A_ROW, + ); + expect(result).toMatchObject({ + indexed: 0, + failed: BACKFILL_FAILURES_IN_A_ROW, + reason: "provider-failing", + }); + // What each failed reply reported spending stays on the day's ledger. + for (const call of reconcileSpend.mock.calls) { + expect(call[2]).toBe(2700); + } + }); + + it("keeps walking past one document the model cannot read", async () => { + seedDocuments(5); + transcribeDocument + .mockRejectedValueOnce(new Error("unreadable")) + .mockResolvedValue({ text: "glucose 90" }); + + const result = await runContentIndexBackfillForUser("user-1"); + + expect(transcribeDocument).toHaveBeenCalledTimes(5); + expect(result).toMatchObject({ indexed: 4, failed: 1, reason: "ok" }); + }); +}); diff --git a/src/lib/jobs/document-content-index-backfill.ts b/src/lib/jobs/document-content-index-backfill.ts index 2402e7149..e6e59197a 100644 --- a/src/lib/jobs/document-content-index-backfill.ts +++ b/src/lib/jobs/document-content-index-backfill.ts @@ -39,7 +39,9 @@ import { reserveBudget, resolveCostOwner, resolveDailyCapFor, + type BudgetCostOwner, } from "@/lib/ai/coach/budget"; +import type { ProviderChainType } from "@/lib/ai/provider-chain"; import { prisma } from "@/lib/db"; import { loadOwnedDocument, @@ -55,6 +57,7 @@ import { transcribeDocument } from "@/lib/documents/describe"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { resolveDocumentVisionProvider } from "@/lib/documents/provider-order"; import { annotate } from "@/lib/logging/context"; +import { spentTokensOf } from "@/lib/ai/reasoning/answerless"; import { CONTENT_INDEX_BACKFILL_QUEUE } from "./document-content-index-backfill-queue"; /** Discovery page size (id-only; blobs are loaded one at a time). */ @@ -68,9 +71,26 @@ const PAGE_SIZE = 50; */ export const CONTENT_INDEX_BACKFILL_EXPIRE_SECONDS = 3 * 60 * 60; -/** Max documents one job indexes before yielding (bounds spend + runtime). */ +/** + * Max provider calls one job makes before yielding (bounds spend + runtime). + * #1279 — calls, not successes: counted on indexed documents only, a + * provider that failed every call never reached it, and the walk went on + * until the job's time ran out. + */ const MAX_DOCS_PER_RUN = 200; +/** + * #1279 — provider calls in a row that may fail before the walk stops. One + * document the model cannot read is that document; several in a row is the + * provider, and the rest of the walk would only repeat it. + */ +export const BACKFILL_FAILURES_IN_A_ROW = 3; + +/** The provider the document pick chose. */ +type VisionPick = NonNullable< + Awaited>["pick"] +>; + /** Image MIME types the vision path can always index. */ const IMAGE_MIMES = ["image/jpeg", "image/png", "image/webp"] as const; @@ -100,7 +120,13 @@ export interface ContentIndexBackfillSummary { */ failed: number; reason: - "ok" | "no-provider" | "no-consent" | "budget-reached" | "unavailable"; + | "ok" + | "no-provider" + | "no-consent" + | "budget-reached" + | "unavailable" + /** #1279 — `BACKFILL_FAILURES_IN_A_ROW` provider calls failed in a row. */ + | "provider-failing"; /** * The job's time budget ran out before the walk did. The documents not * reached stay un-indexed for the next run. @@ -234,12 +260,14 @@ export async function runContentIndexBackfillForUser( let indexed = 0; let skipped = 0; let failed = stale.failed; + let calls = 0; + let failedInARow = 0; let reason: ContentIndexBackfillSummary["reason"] = "ok"; let cursor: string | null = null; let stoppedEarly = false; outer: for (;;) { - if (indexed >= MAX_DOCS_PER_RUN) break; + if (calls >= MAX_DOCS_PER_RUN) break; const rows: { id: string }[] = await prisma.inboundDocument.findMany({ where: { userId, @@ -258,7 +286,11 @@ export async function runContentIndexBackfillForUser( if (rows.length === 0) break; for (const { id } of rows) { - if (indexed >= MAX_DOCS_PER_RUN) break outer; + if (calls >= MAX_DOCS_PER_RUN) break outer; + if (failedInARow >= BACKFILL_FAILURES_IN_A_ROW) { + reason = "provider-failing"; + break outer; + } if (shouldStop()) { stoppedEarly = true; break outer; @@ -301,40 +333,18 @@ export async function runContentIndexBackfillForUser( continue; } - try { - const { text } = await transcribeDocument({ - provider: pick.entry.instance, - providerType: pick.providerType, - images: vision.images, - documents: vision.documents, - }); - await reconcileSpend( - userId, - reservation.reserved, - reservation.reserved, - dateKey, - 0, - { - servedBy: pick.entry.providerType, - reservedOwner: reservation.owner, - }, - ); - await upsertContentIndex({ - userId, - documentId: document.id, - text, - source: "vision", - providerType: pick.providerType, - }); + calls += 1; + const read = await transcribeAndIndex(userId, document.id, pick, vision, { + reserved: reservation.reserved, + owner: reservation.owner, + dateKey, + }); + if (read === "indexed") { indexed += 1; - } catch { - // A provider miss on one document must not abort the batch — refund and - // leave it un-indexed (a later run retries it). - await reconcileSpend(userId, reservation.reserved, 0, dateKey, 0, { - servedBy: null, - reservedOwner: reservation.owner, - }); + failedInARow = 0; + } else { failed += 1; + failedInARow += 1; } } @@ -363,6 +373,48 @@ export async function runContentIndexBackfillForUser( }; } +/** + * One document's provider read and its index row, with the reservation + * settled either way. A provider miss on one document must not abort the + * batch: it is refunded and left un-indexed (a later run retries it). + * #1279 — what the failed reply reported spending stays charged, and the + * caller stops the walk after several misses in a row. + */ +async function transcribeAndIndex( + userId: string, + documentId: string, + pick: VisionPick, + vision: Extract>, { ok: true }>, + budget: { reserved: number; owner: BudgetCostOwner; dateKey: string }, +): Promise<"indexed" | { error: unknown }> { + const settle = (spent: number, servedBy: ProviderChainType | null) => + reconcileSpend(userId, budget.reserved, spent, budget.dateKey, 0, { + servedBy, + reservedOwner: budget.owner, + }); + try { + const { text } = await transcribeDocument({ + provider: pick.entry.instance, + providerType: pick.providerType, + images: vision.images, + documents: vision.documents, + }); + await settle(budget.reserved, pick.entry.providerType); + await upsertContentIndex({ + userId, + documentId, + text, + source: "vision", + providerType: pick.providerType, + }); + return "indexed"; + } catch (err) { + await settle(spentTokensOf(err) ?? 0, null); + // Counted into the run's `failed` tally by the caller. + return { error: err }; + } +} + /** * Enqueue a per-user content-index backfill. Coalesced by `singletonKey` so a * user cannot pile up parallel runs; returns whether a job was created.