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.