diff --git a/next.config.ts b/next.config.ts index 3a9b959db..013e81c7a 100644 --- a/next.config.ts +++ b/next.config.ts @@ -302,11 +302,13 @@ const nextConfig: NextConfig = { outputFileTracingIncludes: { "*": [ "./src/lib/ai/prompts/safety-contracts.*.yaml", - // PDFs are read and rendered in a child process with a memory ceiling - // and a time limit (`src/lib/documents/pdf-isolation.ts`). The child is - // started from these two files by path, which the tracer cannot follow. + // PDFs are read and rendered, and uploaded images decoded into their + // thumbnails, in a child process with a memory ceiling and a time limit + // (`src/lib/documents/pdf-isolation.ts`). The child is started from + // these files by path, which the tracer cannot follow. "./src/lib/documents/pdf-child.mjs", "./src/lib/documents/pdf-engine.mjs", + "./src/lib/documents/image-engine.mjs", // Document AI local text extraction and rasterization both run the // app's one pdfjs-dist (`src/lib/documents/pdfjs.ts`) server-side; the // text read is the provider-free fallback that keeps content search diff --git a/src/lib/documents/__tests__/image-engine.test.ts b/src/lib/documents/__tests__/image-engine.test.ts new file mode 100644 index 000000000..95237b1a8 --- /dev/null +++ b/src/lib/documents/__tests__/image-engine.test.ts @@ -0,0 +1,54 @@ +import { Buffer } from "node:buffer"; + +import { createCanvas } from "@napi-rs/canvas"; +import { describe, expect, it, vi } from "vitest"; + +import { renderImageThumbnail } from "../image-engine.mjs"; + +/** + * What runs in the reader's child for an image upload: decode, check the + * decoded size once more, draw onto a fresh JPEG. The header walk in the + * server is a reading of the file; the decoder has the final word, so the + * decoded size is checked before a canvas is drawn from it. + */ + +const OPTIONS = { longEdge: 320, quality: 70, maxPixels: 60_000_000 }; + +function png(width: number, height: number): Buffer { + const canvas = createCanvas(width, height); + const ctx = canvas.getContext("2d"); + ctx.fillStyle = "#3366cc"; + ctx.fillRect(0, 0, width, height); + return canvas.toBuffer("image/png"); +} + +describe("renderImageThumbnail", () => { + it("draws a decoded image onto a JPEG no longer than the long edge", async () => { + const outcome = await renderImageThumbnail(png(1200, 800), OPTIONS); + expect(outcome).toMatchObject({ ok: true, width: 320, height: 213 }); + if (!outcome.ok) return; + expect(Buffer.from(outcome.jpeg).subarray(0, 2).toString("hex")).toBe( + "ffd8", + ); + }); + + it("refuses a decoded image larger than the header said, before drawing it", async () => { + const createCanvasSpy = vi.fn(); + const outcome = await renderImageThumbnail( + Buffer.from("anything"), + OPTIONS, + { + loadImage: async () => ({ width: 30_000, height: 30_000 }), + createCanvas: createCanvasSpy, + }, + ); + expect(outcome).toEqual({ ok: false, reason: "pixel-cap" }); + expect(createCanvasSpy).not.toHaveBeenCalled(); + }); + + it("throws what the decoder threw", async () => { + await expect( + renderImageThumbnail(Buffer.from("not an image"), OPTIONS), + ).rejects.toThrow(); + }); +}); diff --git a/src/lib/documents/__tests__/local-extract-resource-limit.test.ts b/src/lib/documents/__tests__/local-extract-resource-limit.test.ts index 22663188f..05fb4f0d4 100644 --- a/src/lib/documents/__tests__/local-extract-resource-limit.test.ts +++ b/src/lib/documents/__tests__/local-extract-resource-limit.test.ts @@ -19,11 +19,11 @@ vi.mock("../pdfjs", async (importOriginal) => ({ readPdfText, })); -import { PdfResourceLimitError } from "../pdfjs"; +import { PdfReaderUnavailableError, PdfResourceLimitError } from "../pdfjs"; import { extractPdfText, localExtractText } from "../local-extract"; describe("a PDF past the reader's limits", () => { - it.each(["memory", "time", "exit"] as const)( + it.each(["memory", "time"] as const)( "answers resource-limit for the %s bound, without the error log a broken file gets", async (limit) => { readPdfText.mockRejectedValueOnce(new PdfResourceLimitError(limit)); @@ -42,6 +42,18 @@ describe("a PDF past the reader's limits", () => { }, ); + it.each(["busy", "interrupted"] as const)( + "passes a reader that was %s up to the job instead of recording an outcome", + async (kind) => { + readPdfText.mockRejectedValueOnce(new PdfReaderUnavailableError(kind)); + logCaught.mockClear(); + await expect( + extractPdfText(Buffer.from("%PDF-1.7 large")), + ).rejects.toBeInstanceOf(PdfReaderUnavailableError); + expect(logCaught).not.toHaveBeenCalled(); + }, + ); + it("still answers error for anything else the reader throws", async () => { const broken = new Error("Invalid PDF structure"); broken.name = "InvalidPDFException"; diff --git a/src/lib/documents/__tests__/pdf-isolation.test.ts b/src/lib/documents/__tests__/pdf-isolation.test.ts index 28ba074a5..4bf197d77 100644 --- a/src/lib/documents/__tests__/pdf-isolation.test.ts +++ b/src/lib/documents/__tests__/pdf-isolation.test.ts @@ -1,5 +1,6 @@ import { Buffer } from "node:buffer"; -import { readFileSync } from "node:fs"; +import { mkdtempSync, readFileSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; import { join } from "node:path"; import { createDeflate } from "node:zlib"; @@ -8,7 +9,15 @@ import { describe, expect, it, vi } from "vitest"; const { annotate } = vi.hoisted(() => ({ annotate: vi.fn() })); vi.mock("@/lib/logging/context", () => ({ annotate })); -import { runPdfChild } from "../pdf-isolation"; +import { gif } from "../../__tests__/gif-fixture"; +import { + MAX_QUEUED_PER_OWNER, + runOnce, + runPdfChild, + withPdfWork, + type PdfChildRequest, + type PdfChildResult, +} from "../pdf-isolation"; import { readPdfText } from "../pdfjs"; import { rasterizePdf } from "../rasterize-pdf"; import { generateThumbnail } from "../thumbnail"; @@ -187,3 +196,108 @@ describe("PDF reading runs in a bounded child process", () => { expect(result).toMatchObject({ ok: false, reason: "error" }); }); }); + +describe("an image upload is decoded in the bounded child", () => { + it("ends the child, not the server, when a decoder allocates past the ceiling", async () => { + // The header walk refuses this file; here it goes to the decoder + // directly, as a decoder surprise the walk did not foresee would. + const bomb = gif({ width: 10, height: 10 }, [ + { width: 20_000, height: 20_000 }, + ]); + const before = process.memoryUsage().rss; + const result = await runPdfChild( + { + op: "image-thumbnail", + data: bomb, + options: { longEdge: 320, quality: 70, maxPixels: 1e12 }, + }, + { timeoutMs: 60_000, memoryMb: 300 }, + ); + expect(result).toEqual({ ok: false, reason: "limit", limit: "memory" }); + expect(process.memoryUsage().rss - before).toBeLessThan(200 * 1024 ** 2); + }, 90_000); +}); + +describe("a child that ends without an answer", () => { + const dir = mkdtempSync(join(tmpdir(), "reader-child-")); + const script = (name: string, body: string) => { + const path = join(dir, name); + writeFileSync(path, body); + return path; + }; + const request = { op: "text", data: Buffer.from("x"), maxPages: 1 } as const; + + it("is not a final verdict when nothing this module measured ended it", async () => { + // A kill from outside: the kernel's out-of-memory killer acting for the + // whole container looks like this to the parent. + const killed = script( + "killed-child", + 'process.kill(process.pid, "SIGKILL");\n', + ); + await expect( + runOnce(killed, request, { timeoutMs: 10_000 }), + ).resolves.toEqual({ ok: false, reason: "interrupted" }); + }); + + it("is the document's doing when V8 reached the heap cap the child was given", async () => { + const heap = script( + "heap-child", + 'process.stderr.write("FATAL ERROR: Reached heap limit Allocation failed - JavaScript heap out of memory\\n", () => process.exit(134));\n', + ); + await expect( + runOnce(heap, request, { timeoutMs: 10_000 }), + ).resolves.toEqual({ ok: false, reason: "limit", limit: "memory" }); + }); + + it("is run once more, and answered interrupted when it ends that way again", async () => { + const interrupted = vi.fn( + async (): Promise> => ({ + ok: false, + reason: "interrupted", + }), + ); + await expect( + runPdfChild(request, { timeoutMs: 1_000, run: interrupted }), + ).resolves.toEqual({ ok: false, reason: "interrupted" }); + expect(interrupted).toHaveBeenCalledTimes(2); + + const recovers = vi + .fn<() => Promise>>() + .mockResolvedValueOnce({ ok: false, reason: "interrupted" }) + .mockResolvedValueOnce({ ok: true, value: "text" }); + await expect( + runPdfChild(request, { timeoutMs: 1_000, run: recovers }), + ).resolves.toEqual({ ok: true, value: "text" }); + expect(recovers).toHaveBeenCalledTimes(2); + }); +}); + +describe("the reader's queue", () => { + it("answers one account's work past its waiting cap busy, at once", async () => { + const request = { + op: "text", + data: Buffer.from("x"), + maxPages: 1, + } as const; + let open: () => void = () => {}; + const gate = new Promise((resolve) => (open = resolve)); + const run = vi.fn(async (): Promise> => { + await gate; + return { ok: true, value: "text" }; + }); + const asA = () => + withPdfWork({ owner: "account-a", lane: "interactive" }, () => + runPdfChild(request, { timeoutMs: 1_000, run }), + ); + + const running = asA(); + const waiting = Array.from({ length: MAX_QUEUED_PER_OWNER }, asA); + await expect(asA()).resolves.toMatchObject({ ok: false, reason: "busy" }); + + open(); + await expect(running).resolves.toMatchObject({ ok: true }); + for (const item of waiting) { + await expect(item).resolves.toMatchObject({ ok: true }); + } + }); +}); diff --git a/src/lib/documents/__tests__/reader-queue.test.ts b/src/lib/documents/__tests__/reader-queue.test.ts new file mode 100644 index 000000000..ea4f96c7c --- /dev/null +++ b/src/lib/documents/__tests__/reader-queue.test.ts @@ -0,0 +1,138 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { createReaderQueue, type ReaderTicket } from "../reader-queue"; + +/** + * The document reader has one slot. It used to go to whoever asked first, + * and a waiting item waited without end: a few heavy PDFs from one account + * held every other account's lab scan behind them. These pin the order the + * slot is handed out in, the cap on what one account may queue, and the + * deadline a waiting item gets. + */ + +const WAIT = { interactive: 1_000, background: 5_000 }; + +function queue(maxQueuedPerOwner = 10) { + return createReaderQueue({ slots: 1, maxQueuedPerOwner, waitMs: WAIT }); +} + +/** Acquire, and note the order in which tickets are granted. */ +function recorder() { + const order: string[] = []; + const tickets = new Map(); + const track = (label: string, pending: Promise) => + pending.then((ticket) => { + if (ticket.ok) order.push(label); + tickets.set(label, ticket); + return ticket; + }); + const release = (label: string) => { + const ticket = tickets.get(label); + if (ticket?.ok) ticket.release(); + }; + return { order, track, release }; +} + +const flush = () => new Promise((resolve) => setImmediate(resolve)); + +describe("the reader queue", () => { + it("takes accounts in turn instead of first come, first served", async () => { + const q = queue(); + const r = recorder(); + void r.track("a1", q.acquire("a", "background")); + void r.track("a2", q.acquire("a", "background")); + void r.track("a3", q.acquire("a", "background")); + void r.track("b1", q.acquire("b", "background")); + void r.track("c1", q.acquire("c", "background")); + await flush(); + for (const label of ["a1", "a2", "b1", "c1", "a3"]) { + r.release(label); + await flush(); + } + // a1 held the slot. Of the waiting items a2 came first, then b and c + // each get their turn before a's third item comes round. + expect(r.order).toEqual(["a1", "a2", "b1", "c1", "a3"]); + }); + + it("serves a read a person is waiting on before background work", async () => { + const q = queue(); + const r = recorder(); + void r.track("bg-running", q.acquire("a", "background")); + void r.track("bg-a", q.acquire("a", "background")); + void r.track("bg-b", q.acquire("b", "background")); + void r.track("lab-scan", q.acquire("c", "interactive")); + await flush(); + r.release("bg-running"); + await flush(); + expect(r.order).toEqual(["bg-running", "lab-scan"]); + }); + + it("refuses an account's item past its waiting cap at once, without touching others", async () => { + const q = queue(2); + const r = recorder(); + await r.track("a-running", q.acquire("a", "interactive")); + void r.track("a-1", q.acquire("a", "interactive")); + void r.track("a-2", q.acquire("a", "background")); + await expect(q.acquire("a", "interactive")).resolves.toMatchObject({ + ok: false, + reason: "queue-full", + }); + // Another account still gets in line, and its turn comes before the + // capped account's second waiting item. + void r.track("b-1", q.acquire("b", "interactive")); + expect(q.snapshot().waiting).toEqual({ interactive: 2, background: 1 }); + for (const label of ["a-running", "a-1"]) { + r.release(label); + await flush(); + } + expect(r.order).toEqual(["a-running", "a-1", "b-1"]); + r.release("b-1"); + await flush(); + expect(r.order.at(-1)).toBe("a-2"); + r.release("a-2"); + }); + + describe("with a deadline", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("answers a waiting item busy when its lane's deadline passes, and keeps the slot for the next", async () => { + const q = queue(); + const held = await q.acquire("a", "background"); + const waiting = q.acquire("b", "interactive"); + const later = q.acquire("c", "background"); + + await vi.advanceTimersByTimeAsync(WAIT.interactive + 1); + await expect(waiting).resolves.toEqual({ + ok: false, + reason: "wait-expired", + retryAfterMs: WAIT.interactive, + }); + expect(q.snapshot()).toEqual({ + running: 1, + waiting: { interactive: 0, background: 1 }, + }); + + // The expired item left the queue: releasing hands the slot to the + // next one in line, not to nobody. + if (held.ok) held.release(); + await expect(later).resolves.toMatchObject({ ok: true }); + }); + + it("releases a slot once, however often release is called", async () => { + const q = queue(); + const first = await q.acquire("a", "background"); + if (!first.ok) throw new Error("expected the slot"); + first.release(); + first.release(); + expect(q.snapshot().running).toBe(0); + const a = await q.acquire("a", "background"); + expect(a.ok).toBe(true); + expect(q.snapshot().running).toBe(1); + }); + }); +}); diff --git a/src/lib/documents/__tests__/thumbnail.test.ts b/src/lib/documents/__tests__/thumbnail.test.ts index 7620df07e..6a28f1aa4 100644 --- a/src/lib/documents/__tests__/thumbnail.test.ts +++ b/src/lib/documents/__tests__/thumbnail.test.ts @@ -5,10 +5,9 @@ import { describe, expect, it, vi } from "vitest"; vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); -// Partial mock of the native canvas module: `loadImage` delegates to the real -// decoder by default (so every real-fixture test below is unaffected) and can -// be overridden per-test via `loadImageMock` to simulate a decoded image whose -// dimensions the header sniff could not have known. +// Partial mock of the native canvas module in this process: `loadImage` +// delegates to the real decoder and records every call, so a test can tell +// whether an upload was decoded here rather than in the reader's child. const { loadImageMock } = vi.hoisted(() => ({ loadImageMock: vi.fn() })); vi.mock("@napi-rs/canvas", async (importActual) => { const actual = await importActual(); @@ -166,14 +165,16 @@ describe("generateThumbnail", () => { }); }); - it("refuses an oversize image the header sniff missed (post-decode cap)", async () => { - // A readable header declaring a small image whose decoder is forced to - // return 30000×30000 = 900 MP: only the post-decode backstop can catch - // it. - loadImageMock.mockResolvedValueOnce({ width: 30_000, height: 30_000 }); - await expect( - generateThumbnail(sourceImage(16, 16, "image/jpeg"), "image/jpeg"), - ).resolves.toEqual({ ok: false, reason: "pixel-cap" }); + it("decodes an uploaded image in the reader's child process, not in this one", async () => { + // The decoder allocates what the file makes it allocate; in the child + // that ends the child. Nothing of an image upload is decoded here. + loadImageMock.mockClear(); + const result = await generateThumbnail( + sourceImage(640, 480, "image/png"), + "image/png", + ); + expect(result.ok).toBe(true); + expect(loadImageMock).not.toHaveBeenCalled(); }); /** diff --git a/src/lib/documents/image-engine.mjs b/src/lib/documents/image-engine.mjs new file mode 100644 index 000000000..1012ea2dd --- /dev/null +++ b/src/lib/documents/image-engine.mjs @@ -0,0 +1,81 @@ +// @ts-check +/** + * Decoding an uploaded image into the vault's preview thumbnail. + * + * The server never runs this in its own process. An image decoder allocates + * the whole bitmap before anything can look at it, and the size it allocates + * is whatever the file declares in a place the decoder chooses: a GIF frame + * descriptor once reached a 1.6 GB decode behind a header check that only + * read the logical screen. The header check (`../image-dimensions.ts`) still + * refuses what it can see before a child is started; this module is what + * runs in that child (`./pdf-child.mjs`, under the memory ceiling and time + * limit of `./pdf-isolation.ts`), so a decoder surprise ends the child and + * not the server. + * + * Plain JavaScript for the same reason as `./pdf-engine.mjs`: the child runs + * it with plain Node from the checkout and the standalone image alike. + */ +import { Buffer } from "node:buffer"; + +/** + * @typedef {object} ImageThumbnailOptions + * @property {number} longEdge The preview's longest side; never upscaled. + * @property {number} quality JPEG quality, 0 to 100. + * @property {number} maxPixels The most pixels a decoded source may have. + */ + +/** + * @typedef {{ ok: true; jpeg: Uint8Array; width: number; height: number } + * | { ok: false; reason: "pixel-cap" | "empty-render" }} ImageThumbnailOutcome + */ + +/** + * The slice of `@napi-rs/canvas` used here; the suites pass a stand-in. + * + * @typedef {object} CanvasModule + * @property {(source: Buffer) => Promise<{ width: number; height: number }>} loadImage + * @property {(width: number, height: number) => { + * getContext(type: "2d"): { drawImage(image: unknown, dx: number, dy: number, dw: number, dh: number): void }; + * toBuffer(mime: "image/jpeg", quality?: number): Buffer; + * }} createCanvas + */ + +/** + * Decode an image and draw it onto a fresh JPEG no longer than + * `options.longEdge` on its long side. The re-encode carries none of the + * source's metadata. Throws what the decoder threw. + * + * @param {Uint8Array} data + * @param {ImageThumbnailOptions} options + * @param {CanvasModule} [canvasModule] + * @returns {Promise} + */ +export async function renderImageThumbnail(data, options, canvasModule) { + const napi = + canvasModule ?? + /** @type {CanvasModule} */ ( + /** @type {unknown} */ (await import("@napi-rs/canvas")) + ); + const image = await napi.loadImage( + Buffer.from(data.buffer, data.byteOffset, data.byteLength), + ); + // The decoded size again, before a canvas of that size is drawn from: the + // header check is a reading of the file, the decoder had the final word. + if (image.width * image.height > options.maxPixels) { + return { ok: false, reason: "pixel-cap" }; + } + if (image.width <= 0 || image.height <= 0) { + return { ok: false, reason: "empty-render" }; + } + const scale = Math.min( + 1, + options.longEdge / Math.max(image.width, image.height), + ); + const width = Math.max(1, Math.round(image.width * scale)); + const height = Math.max(1, Math.round(image.height * scale)); + const canvas = napi.createCanvas(width, height); + canvas.getContext("2d").drawImage(image, 0, 0, width, height); + const jpeg = canvas.toBuffer("image/jpeg", options.quality); + if (jpeg.byteLength === 0) return { ok: false, reason: "empty-render" }; + return { ok: true, jpeg: new Uint8Array(jpeg), width, height }; +} diff --git a/src/lib/documents/local-extract.ts b/src/lib/documents/local-extract.ts index 8469dc7c6..9507e89a9 100644 --- a/src/lib/documents/local-extract.ts +++ b/src/lib/documents/local-extract.ts @@ -32,7 +32,11 @@ import { Buffer } from "node:buffer"; import { annotate } from "@/lib/logging/context"; import { logCaught } from "@/lib/logging/signal"; -import { PdfResourceLimitError, readPdfText } from "./pdfjs"; +import { + PdfReaderUnavailableError, + PdfResourceLimitError, + readPdfText, +} from "./pdfjs"; /** * Minimum joined text length for a PDF to count as a real text-layer document. @@ -70,9 +74,11 @@ export type LocalExtractResult = /** * Extract a text-layer PDF's embedded text server-side. Returns `empty` for a * scanned/no-text-layer PDF, `resource-limit` for one past the reader's memory - * or time bound, and `error` for a malformed/encrypted one — never - * throws, so a bad document can never abort the caller (the upload/backfill job - * just leaves it un-indexed). An `error` leaves a `warn` line naming what + * or time bound, and `error` for a malformed/encrypted one, so a bad document + * can never abort the caller (the upload/backfill job just leaves it + * un-indexed). The one throw is `PdfReaderUnavailableError`: the reader was + * busy or its process ended for no measured reason, which says nothing about + * the document, so the job runs again instead of recording an outcome. An `error` leaves a `warn` line naming what * pdf.js threw, because the outcome alone does not say why a document stayed * out of the index. */ @@ -84,6 +90,7 @@ export async function extractPdfText( // Bound the page walk so a pathological many-page PDF costs no more. text = (await readPdfText(buffer, LOCAL_MAX_PAGES)).trim(); } catch (err) { + if (err instanceof PdfReaderUnavailableError) throw err; if (err instanceof PdfResourceLimitError) { // Not a fault of the file's bytes being unreadable: they are readable, // only not within the bounds this server gives one document. diff --git a/src/lib/documents/pdf-child.mjs b/src/lib/documents/pdf-child.mjs index d143ad503..28b4f9429 100644 --- a/src/lib/documents/pdf-child.mjs +++ b/src/lib/documents/pdf-child.mjs @@ -1,9 +1,10 @@ // @ts-check /** - * The process a PDF is read and rendered in. `./pdf-isolation.ts` starts it - * for one request, hands it the bytes over the IPC channel, and takes back - * either the result or the fact that it died. It holds no secrets: it is - * started with an almost empty environment and never opens the database. + * The process a PDF is read and rendered in, and an uploaded image is decoded + * into its thumbnail in. `./pdf-isolation.ts` starts it for one request, + * hands it the bytes over the IPC channel, and takes back either the result + * or the fact that it died. It holds no secrets: it is started with an almost + * empty environment and never opens the database. * * On Linux it volunteers itself to the kernel's out-of-memory killer first, * so when a document outruns the memory check between two samples and the @@ -12,6 +13,7 @@ */ import { writeFileSync } from "node:fs"; +import { renderImageThumbnail } from "./image-engine.mjs"; import { readTextLayer, renderPdfPages } from "./pdf-engine.mjs"; try { @@ -22,7 +24,8 @@ try { /** * @typedef {{ op: "text"; data: Uint8Array; maxPages: number } - * | { op: "render"; data: Uint8Array; options: import("./pdf-engine.mjs").RenderOptions }} PdfChildRequest + * | { op: "render"; data: Uint8Array; options: import("./pdf-engine.mjs").RenderOptions } + * | { op: "image-thumbnail"; data: Uint8Array; options: import("./image-engine.mjs").ImageThumbnailOptions }} PdfChildRequest */ /** @@ -52,6 +55,12 @@ async function handle(request) { document: await renderPdfPages(data, request.options), }; } + if (request.op === "image-thumbnail") { + return { + ok: true, + thumbnail: await renderImageThumbnail(data, request.options), + }; + } return failure(new Error("unknown request")); } catch (err) { return failure(err); diff --git a/src/lib/documents/pdf-isolation.ts b/src/lib/documents/pdf-isolation.ts index 5149f9b3e..c0a2ea8dc 100644 --- a/src/lib/documents/pdf-isolation.ts +++ b/src/lib/documents/pdf-isolation.ts @@ -1,5 +1,6 @@ /** - * Runs pdf.js in a child process with a memory ceiling and a time limit. + * Runs pdf.js, and the image decoder behind the vault's thumbnails, in a child + * process with a memory ceiling and a time limit. * * A PDF's cost is set by whoever wrote it. A 400 KB file can declare an image * of twenty thousand pixels square, and pdf.js allocated about 7 GB decoding @@ -20,20 +21,34 @@ * - it starts with an almost empty environment, so it holds no key, token * or database address; * - one runs at a time (`MAX_PARALLEL_CHILDREN`), so a burst of uploads - * cannot multiply the ceiling. + * cannot multiply the ceiling. Who runs next is decided by + * `./reader-queue.ts`: a read a person is waiting on before background + * work, accounts in turn, a few waiting items per account, and a + * deadline after which a waiting item is answered `busy`. * - * A document that crosses a bound fails with `limit` and is not something a - * retry will fix; callers record it as a final failure. + * A document that crosses a bound this module measured (the sampled memory, + * the time limit, or the child's own heap cap) fails with `limit` and is not + * something a retry will fix; callers record it as a final failure. A child + * that is gone without an answer and without a measured cause is not that: + * the kernel may have taken it because the container as a whole ran short, + * the server's own memory included. It is run once more, and a second such + * exit answers `interrupted`, which callers treat as retryable. * * This module logs nothing and imports nothing from the app, so the * end-to-end suite can read report text through `./pdfjs.ts` as well. */ +import { AsyncLocalStorage } from "node:async_hooks"; import { execFile, fork, type ChildProcess } from "node:child_process"; import { existsSync } from "node:fs"; import { readFile } from "node:fs/promises"; import { join } from "node:path"; +import type { + ImageThumbnailOptions, + ImageThumbnailOutcome, +} from "./image-engine.mjs"; import type { RenderOptions, RenderedDocument } from "./pdf-engine.mjs"; +import { createReaderQueue, type ReaderLane } from "./reader-queue"; /** Resident memory a child may reach before it is killed. */ export const PDF_CHILD_MEMORY_MB = 640; @@ -61,18 +76,31 @@ const CHILD_PATH = ["src", "lib", "documents", "pdf-child.mjs"]; export type PdfChildRequest = | { op: "text"; data: Uint8Array; maxPages: number } - | { op: "render"; data: Uint8Array; options: RenderOptions }; + | { op: "render"; data: Uint8Array; options: RenderOptions } + | { op: "image-thumbnail"; data: Uint8Array; options: ImageThumbnailOptions }; type PdfChildValue = R extends { op: "text" } ? string - : RenderedDocument; + : R extends { op: "render" } + ? RenderedDocument + : ImageThumbnailOutcome; /** Why a child gave no result. */ export type PdfChildFailure = /** The document crossed the memory ceiling or the time limit. */ - | { ok: false; reason: "limit"; limit: "memory" | "time" | "exit" } - /** pdf.js refused the document: malformed, encrypted, unreadable. */ - | { ok: false; reason: "error"; name: string; message: string }; + | { ok: false; reason: "limit"; limit: "memory" | "time" } + /** pdf.js or the decoder refused the input: malformed, encrypted, unreadable. */ + | { ok: false; reason: "error"; name: string; message: string } + /** + * The child ended without an answer and without a cause this module + * measured, twice. Not the document's verdict; try again later. + */ + | { ok: false; reason: "interrupted" } + /** + * The reader was taken for longer than the wait this work may spend, or + * this account already has as much waiting as it may. Try again later. + */ + | { ok: false; reason: "busy"; retryAfterMs: number }; export type PdfChildResult = { ok: true; value: PdfChildValue } | PdfChildFailure; @@ -81,27 +109,67 @@ export interface PdfChildLimits { timeoutMs: number; /** Overrides `PDF_CHILD_MEMORY_MB`; the suites use a small one. */ memoryMb?: number; + /** Replaces the child start; the suites use it to stand in for a child. */ + run?: ( + script: string, + request: PdfChildRequest, + limits: PdfChildLimits, + ) => Promise>; } -// ── Concurrency ──────────────────────────────────────────────────────────── +// ── Who runs next ────────────────────────────────────────────────────────── -let running = 0; -const waiting: Array<() => void> = []; +/** Items one account may have waiting for the reader, across both lanes. */ +export const MAX_QUEUED_PER_OWNER = 4; -async function acquireSlot(): Promise { - if (running < MAX_PARALLEL_CHILDREN) { - running++; - return; - } - await new Promise((resolve) => waiting.push(resolve)); +/** + * How long an item may wait for the reader. A person waiting on a read gets + * an answer within a minute; background work can wait longer, and its job is + * retried when it does not get a turn. + */ +export const READER_WAIT_MS: Record = { + interactive: 60_000, + background: 5 * 60_000, +}; + +const queue = createReaderQueue({ + slots: MAX_PARALLEL_CHILDREN, + maxQueuedPerOwner: MAX_QUEUED_PER_OWNER, + waitMs: READER_WAIT_MS, +}); + +/** Whose work this is and whether a person is waiting on it. */ +export interface PdfWork { + /** The account the work is done for; the queue takes accounts in turn. */ + owner: string; + lane: ReaderLane; } -function releaseSlot(): void { - const next = waiting.shift(); - if (next) next(); - else running--; +const workContext = new AsyncLocalStorage(); + +/** + * Run `fn` with every reader call inside it counted as `work`. The entry + * points set it: the background AI run worker (interactive) and the document + * jobs (background). A call outside any of them is background work of an + * unattributed owner. + */ +export function withPdfWork( + work: PdfWork, + fn: () => Promise, +): Promise { + return workContext.run(work, fn); } +const UNATTRIBUTED: PdfWork = { owner: "unattributed", lane: "background" }; + +/** The work the reader would count a call made here as. */ +export function currentPdfWork(): PdfWork { + return workContext.getStore() ?? UNATTRIBUTED; +} + +/** How many times a child that vanished without a measured cause is run. */ +const MAX_ATTEMPTS = 2; + // ── Resident memory of another process ───────────────────────────────────── /** The child's resident set in MB, or null where it cannot be read. */ @@ -132,6 +200,7 @@ interface ChildReply { ok: boolean; text?: string; document?: RenderedDocument; + thumbnail?: ImageThumbnailOutcome; error?: { name: string; message: string }; } @@ -150,9 +219,36 @@ function decodeDocument(document: RenderedDocument): RenderedDocument { }; } +function decodeValue( + request: R, + reply: ChildReply, +): PdfChildValue { + if (request.op === "text") return reply.text as PdfChildValue; + if (request.op === "render") { + return decodeDocument( + reply.document as RenderedDocument, + ) as PdfChildValue; + } + const thumbnail = reply.thumbnail as ImageThumbnailOutcome; + return ( + thumbnail.ok + ? { + ...thumbnail, + jpeg: Buffer.from( + thumbnail.jpeg.buffer, + thumbnail.jpeg.byteOffset, + thumbnail.jpeg.byteLength, + ), + } + : thumbnail + ) as PdfChildValue; +} + /** - * Read or render one PDF in a child process. Never throws: a document that - * crosses a bound answers `limit`, one pdf.js refuses answers `error`. + * Read or render one PDF, or decode one image, in a child process. Never + * throws: a document that crosses a bound answers `limit`, one the reader + * refuses answers `error`, a reader that is taken answers `busy`, and a + * child that twice ended for no reason it could see answers `interrupted`. */ export async function runPdfChild( request: R, @@ -168,15 +264,26 @@ export async function runPdfChild( }; } - await acquireSlot(); + const work = currentPdfWork(); + const ticket = await queue.acquire(work.owner, work.lane); + if (!ticket.ok) { + return { ok: false, reason: "busy", retryAfterMs: ticket.retryAfterMs }; + } try { - return await runOnce(script, request, limits); + for (let attempt = 1; ; attempt++) { + const result = limits.run + ? ((await limits.run(script, request, limits)) as PdfChildResult) + : await runOnce(script, request, limits); + if (result.ok || result.reason !== "interrupted") return result; + if (attempt >= MAX_ATTEMPTS) return result; + } } finally { - releaseSlot(); + ticket.release(); } } -function runOnce( +/** Start one child for `request` and wait for its answer or its end. */ +export function runOnce( script: string, request: R, limits: PdfChildLimits, @@ -244,11 +351,7 @@ function runOnce( child.on("message", (raw) => { const reply = raw as ChildReply; if (reply.ok) { - const value = - request.op === "text" - ? reply.text - : decodeDocument(reply.document as RenderedDocument); - finish({ ok: true, value: value as PdfChildValue }); + finish({ ok: true, value: decodeValue(request, reply) }); } else { finish({ ok: false, @@ -259,13 +362,10 @@ function runOnce( } }); - child.on("error", (err) => { - finish({ - ok: false, - reason: "error", - name: err.name, - message: err.message.slice(0, 300), - }); + // The process could not be started or signalled. What pdf.js or the + // decoder thought of the document arrives as a reply, never here. + child.on("error", () => { + finish({ ok: false, reason: "interrupted" }); }); child.on("close", () => { @@ -273,21 +373,24 @@ function runOnce( finish({ ok: false, reason: "limit", limit: killedFor }); return; } - // Gone without an answer: V8 hit its heap ceiling, the kernel took it - // for memory, or it crashed in native code. Any of them is the - // document's doing. - const heap = /heap limit|heap out of memory/i.test(stderrTail); - finish({ ok: false, reason: "limit", limit: heap ? "memory" : "exit" }); + // Gone without an answer. V8 reaching the heap cap this module gave + // the child is the document's doing, whatever else the host runs. Any + // other end (the kernel's out-of-memory killer, which also acts when + // the server itself grew, a signal from outside, a native crash) says + // nothing certain about the document, so it is not a final verdict. + if (/heap limit/i.test(stderrTail)) { + finish({ ok: false, reason: "limit", limit: "memory" }); + return; + } + finish({ ok: false, reason: "interrupted" }); }); child.send(request, (err) => { - if (err) { - finish({ - ok: false, - reason: "error", - name: err.name, - message: err.message.slice(0, 300), - }); + // The bytes never reached the child, mostly because it was already + // gone. That says nothing about the document either: end the child and + // let its close answer `interrupted`. + if (err && child.exitCode === null && child.signalCode === null) { + child.kill("SIGKILL"); } }); }); diff --git a/src/lib/documents/pdfjs.ts b/src/lib/documents/pdfjs.ts index e106ff129..7a55c75ef 100644 --- a/src/lib/documents/pdfjs.ts +++ b/src/lib/documents/pdfjs.ts @@ -21,16 +21,33 @@ export const PDF_TEXT_TIMEOUT_MS = 60_000; /** A document that crossed the memory ceiling or the time limit. */ export class PdfResourceLimitError extends Error { override name = "PdfResourceLimitError"; - constructor(readonly limit: "memory" | "time" | "exit") { + constructor(readonly limit: "memory" | "time") { super(`the PDF crossed the reader's ${limit} limit`); } } +/** + * The reader could not take the document now: it was busy past the wait this + * work may spend, or its process ended for a reason that says nothing about + * the document. Not a verdict on the file; the work is tried again later. + */ +export class PdfReaderUnavailableError extends Error { + override name = "PdfReaderUnavailableError"; + constructor(readonly kind: "busy" | "interrupted") { + super( + kind === "busy" + ? "the document reader is busy" + : "the document reader stopped before it answered", + ); + } +} + /** * The text layer of the first `maxPages` pages, one page after another. * Throws what pdf.js threw (a malformed file, a password, a page that will - * not parse) under its own name, or `PdfResourceLimitError` when the - * document crossed a bound. + * not parse) under its own name, `PdfResourceLimitError` when the document + * crossed a bound, or `PdfReaderUnavailableError` when the reader could not + * take it now. */ export async function readPdfText( data: Uint8Array, @@ -43,6 +60,9 @@ export async function readPdfText( ); if (result.ok) return result.value; if (result.reason === "limit") throw new PdfResourceLimitError(result.limit); + if (result.reason === "busy" || result.reason === "interrupted") { + throw new PdfReaderUnavailableError(result.reason); + } const err = new Error(result.message); err.name = result.name; throw err; diff --git a/src/lib/documents/rasterize-pdf.ts b/src/lib/documents/rasterize-pdf.ts index e410ad6e6..632bbb8dd 100644 --- a/src/lib/documents/rasterize-pdf.ts +++ b/src/lib/documents/rasterize-pdf.ts @@ -107,7 +107,20 @@ export type RasterResult = */ texts?: string[]; } - | { ok: false; reason: "unsupported" | "render-failed" | "resource-limit" }; + | { + ok: false; + reason: + | "unsupported" + | "render-failed" + | "resource-limit" + /** + * Not an answer about this document: the reader was busy past the + * wait this work may spend, or its process ended for no reason it + * measured. The caller tries again later. + */ + | "busy" + | "interrupted"; + }; export { pageText, @@ -166,6 +179,13 @@ export async function rasterizePdf( ); if (!result.ok) { + if (result.reason === "busy" || result.reason === "interrupted") { + annotate({ + action: { name: "documents.rasterize.deferred" }, + meta: { reason: result.reason }, + }); + return { ok: false, reason: result.reason }; + } if (result.reason === "limit") { annotate({ action: { name: "documents.rasterize.failed" }, diff --git a/src/lib/documents/reader-queue.ts b/src/lib/documents/reader-queue.ts new file mode 100644 index 000000000..6c05de942 --- /dev/null +++ b/src/lib/documents/reader-queue.ts @@ -0,0 +1,163 @@ +/** + * Who gets the document reader next. + * + * The reader (`./pdf-isolation.ts`) runs one child process at a time, because + * the server and one child at its memory ceiling are what fit the default + * container. It used to hand that one slot out first come, first served, + * with no limit on how long anyone waited: a few heavy PDFs from one account, + * each allowed up to a minute, put every other person's lab scan behind them + * for as long as the uploads kept coming. + * + * Now the slot goes: + * + * - to interactive work first: a read a person started and is waiting on + * (a lab scan, a medication label, a document read) is served before + * background work (thumbnails, indexing, summaries) that nobody watches; + * - round robin across accounts within each lane, so one account's queue + * takes one turn and then waits behind everyone else's next item; + * - to at most `maxQueuedPerOwner` waiting items per account; one more is + * refused at once rather than queued; + * - within a deadline per lane: an item still waiting when it passes leaves + * the queue with a retryable "busy" answer instead of waiting on. + * + * Nothing here knows what a PDF is, so the order can be tested on its own. + */ + +export type ReaderLane = "interactive" | "background"; + +/** Interactive before background. */ +const LANE_ORDER: readonly ReaderLane[] = ["interactive", "background"]; + +export interface ReaderQueueOptions { + /** Work running at once. */ + slots: number; + /** Items one owner may have waiting, across both lanes. */ + maxQueuedPerOwner: number; + /** How long an item may wait in each lane before it is answered busy. */ + waitMs: Record; +} + +/** The slot, or why there is none. `release` must be called exactly once. */ +export type ReaderTicket = + | { ok: true; release: () => void } + | { ok: false; reason: "queue-full" | "wait-expired"; retryAfterMs: number }; + +interface Waiter { + owner: string; + grant: () => void; + timer: ReturnType; +} + +export interface ReaderQueue { + acquire(owner: string, lane: ReaderLane): Promise; + /** Items running and waiting, for the suites and diagnostics. */ + snapshot(): { running: number; waiting: Record }; +} + +export function createReaderQueue(options: ReaderQueueOptions): ReaderQueue { + let running = 0; + // Per lane, one FIFO per owner. A Map iterates in insertion order, so the + // first key is the owner whose turn it is; after a turn the owner moves to + // the back. + const lanes: Record> = { + interactive: new Map(), + background: new Map(), + }; + + const waitingFor = (owner: string): number => + LANE_ORDER.reduce( + (sum, lane) => sum + (lanes[lane].get(owner)?.length ?? 0), + 0, + ); + + const remove = (lane: ReaderLane, waiter: Waiter): void => { + const queue = lanes[lane].get(waiter.owner); + if (!queue) return; + const at = queue.indexOf(waiter); + if (at !== -1) queue.splice(at, 1); + if (queue.length === 0) lanes[lane].delete(waiter.owner); + }; + + const next = (): Waiter | null => { + for (const lane of LANE_ORDER) { + const owners = lanes[lane]; + const first = owners.keys().next(); + if (first.done) continue; + const owner = first.value; + const queue = owners.get(owner)!; + const waiter = queue.shift()!; + owners.delete(owner); + if (queue.length > 0) owners.set(owner, queue); + return waiter; + } + return null; + }; + + const release = (): void => { + const waiter = next(); + if (waiter) { + clearTimeout(waiter.timer); + waiter.grant(); + } else { + running--; + } + }; + + const ticket = (): ReaderTicket => { + let released = false; + return { + ok: true, + release: () => { + if (released) return; + released = true; + release(); + }, + }; + }; + + return { + acquire(owner, lane) { + if (running < options.slots) { + running++; + return Promise.resolve(ticket()); + } + if (waitingFor(owner) >= options.maxQueuedPerOwner) { + return Promise.resolve({ + ok: false, + reason: "queue-full", + retryAfterMs: options.waitMs[lane], + }); + } + return new Promise((resolve) => { + const waiter: Waiter = { + owner, + // The slot passes straight from the releasing item to this one; + // `running` stays as it is. + grant: () => resolve(ticket()), + timer: setTimeout(() => { + remove(lane, waiter); + resolve({ + ok: false, + reason: "wait-expired", + retryAfterMs: options.waitMs[lane], + }); + }, options.waitMs[lane]), + }; + const queue = lanes[lane].get(owner); + if (queue) queue.push(waiter); + else lanes[lane].set(owner, [waiter]); + }); + }, + snapshot() { + const count = (lane: ReaderLane) => + [...lanes[lane].values()].reduce((sum, q) => sum + q.length, 0); + return { + running, + waiting: { + interactive: count("interactive"), + background: count("background"), + }, + }; + }, + }; +} diff --git a/src/lib/documents/thumbnail.ts b/src/lib/documents/thumbnail.ts index 7cc504e42..cb6d4c6ed 100644 --- a/src/lib/documents/thumbnail.ts +++ b/src/lib/documents/thumbnail.ts @@ -6,6 +6,12 @@ * rasterised (via `rasterizePdf`, bounded to one page) then downscaled; every * other MIME type has no preview (the card keeps its kind icon). * + * Both untrusted decodes happen in the reader's child process + * (`./pdf-isolation.ts`) under its memory ceiling and time limit: the image + * through `./image-engine.mjs`, the PDF page through `./pdf-engine.mjs`. An + * image's declared size is still checked here first, so a bomb the header + * shows is refused without starting a child. + * * `@napi-rs/canvas` + pdfjs are pulled LAZILY via runtime `import()` — pdfjs's * module top-level references the browser-only `DOMMatrix` global, so a STATIC * import would make the Turbopack server chunk evaluate that reference at @@ -24,12 +30,14 @@ * binary, or any render error resolves to `{ ok: false }`, and the caller * leaves the document without a thumbnail (the card falls back to its kind * icon). This preserves the "a bad document never aborts an upload/backfill" - * contract. + * contract. `reader-busy` and `interrupted` are the two answers that say + * nothing about the document; the job tries those again. */ import { Buffer } from "node:buffer"; import { nativeCanvasSupported } from "./native-canvas-support"; +import { runPdfChild } from "@/lib/documents/pdf-isolation"; import { rasterizePdf } from "@/lib/documents/rasterize-pdf"; import { readImageDimensions, @@ -96,6 +104,10 @@ export type ThumbnailSkipReason = | "raster-failed" | "resource-limit" | "empty-render" + /** The reader was busy; the job is retried. */ + | "reader-busy" + /** The reader's process ended for no measured reason; tried again later. */ + | "interrupted" | "error"; /** Best-effort generation outcome — never an exception. */ @@ -152,6 +164,76 @@ function drawScaledJpeg( return { jpeg, width, height }; } +/** How long decoding one image into its preview may take. */ +export const IMAGE_THUMBNAIL_TIMEOUT_MS = 30_000; + +/** + * Decode and downscale an image in the reader's child process, under its + * memory ceiling and time limit. The decoder allocates whatever the file + * makes it allocate; in the child that ends the child, not the server. + */ +async function imageThumbnail(bytes: Buffer): Promise { + const result = await runPdfChild( + { + op: "image-thumbnail", + data: bytes, + options: { + longEdge: THUMB_LONG_EDGE, + quality: THUMB_QUALITY, + maxPixels: MAX_SOURCE_PIXELS, + }, + }, + { timeoutMs: IMAGE_THUMBNAIL_TIMEOUT_MS }, + ); + if (!result.ok) { + if (result.reason === "busy") return { ok: false, reason: "reader-busy" }; + if (result.reason === "interrupted") { + return { ok: false, reason: "interrupted" }; + } + if (result.reason === "limit") { + annotate({ + action: { name: "documents.thumbnail.rejected" }, + meta: { + source: "image", + reason: "resource_limit", + limit: result.limit, + }, + }); + return { ok: false, reason: "resource-limit" }; + } + annotate({ + action: { name: "documents.thumbnail.failed" }, + meta: { + mimeType: "image", + reason: result.name, + message: result.message.slice(0, 300), + }, + }); + return { ok: false, reason: "error" }; + } + const outcome = result.value; + if (!outcome.ok) { + if (outcome.reason === "pixel-cap") { + // The decoder found a larger image than the header walk did. + annotate({ + action: { name: "documents.thumbnail.rejected" }, + meta: { source: "image", reason: "pixel_cap_post_decode" }, + }); + } + return { ok: false, reason: outcome.reason }; + } + const thumb: Thumbnail = { + jpeg: Buffer.from(outcome.jpeg), + width: outcome.width, + height: outcome.height, + }; + annotate({ + action: { name: "documents.thumbnail.ok" }, + meta: { source: "image", width: thumb.width, height: thumb.height }, + }); + return { ok: true, thumbnail: thumb }; +} + /** * Generate a small JPEG preview for a stored document, or `{ ok: false }` when * the type is unsupported or rendering fails. Never throws. @@ -173,15 +255,10 @@ export async function generateThumbnail( }); return { ok: false, reason: "native-canvas-unsupported" }; } - // Lazy import: keeps @napi-rs/canvas (+ transitively pdfjs on the PDF path) - // out of the eager server chunk. Resolved on first generation, then cached. - const napi = - (await import("@napi-rs/canvas")) as unknown as NapiCanvasModule; - if (THUMBNAILABLE_IMAGE_MIMES.has(mimeType)) { - // Refuse a decompression bomb BEFORE `loadImage` allocates the bitmap, - // and refuse an image whose size cannot be read: that one would reach - // the decoder with no ceiling at all. + // Refuse a decompression bomb before a decoder is started at all, and + // refuse an image whose size cannot be read: that one would reach the + // decoder with no ceiling but the child's. const pixels = sniffSourcePixels(bytes, mimeType); if (pixels === null) { annotate({ @@ -197,28 +274,7 @@ export async function generateThumbnail( }); return { ok: false, reason: "pixel-cap" }; } - const image = await napi.loadImage(bytes); - // Post-decode backstop: should a decoder ever read a larger frame than - // the header walk did, the decoded size is checked again before the - // bitmap is drawn. - if (image.width * image.height > MAX_SOURCE_PIXELS) { - annotate({ - action: { name: "documents.thumbnail.rejected" }, - meta: { - source: "image", - reason: "pixel_cap_post_decode", - pixels: image.width * image.height, - }, - }); - return { ok: false, reason: "pixel-cap" }; - } - const thumb = drawScaledJpeg(napi, image); - if (!thumb) return { ok: false, reason: "empty-render" }; - annotate({ - action: { name: "documents.thumbnail.ok" }, - meta: { source: "image", width: thumb.width, height: thumb.height }, - }); - return { ok: true, thumbnail: thumb }; + return await imageThumbnail(bytes); } if (mimeType === "application/pdf") { @@ -229,9 +285,21 @@ export async function generateThumbnail( // bytes will again, so the job records it as final. return { ok: false, reason: "resource-limit" }; } + if (!raster.ok && raster.reason === "busy") { + return { ok: false, reason: "reader-busy" }; + } + if (!raster.ok && raster.reason === "interrupted") { + return { ok: false, reason: "interrupted" }; + } if (!raster.ok || raster.images.length === 0) { return { ok: false, reason: "raster-failed" }; } + // The page is a JPEG the child rendered, no longer than + // `RASTER_TARGET_LONG_EDGE` on its long side: the one decode left in + // this process is of the server's own output at a bounded size. + // Lazy import: keeps @napi-rs/canvas out of the eager server chunk. + const napi = + (await import("@napi-rs/canvas")) as unknown as NapiCanvasModule; const pageJpeg = Buffer.from(raster.images[0]!.dataBase64, "base64"); const image = await napi.loadImage(pageJpeg); const thumb = drawScaledJpeg(napi, image); diff --git a/src/lib/jobs/__tests__/document-thumbnail.test.ts b/src/lib/jobs/__tests__/document-thumbnail.test.ts index 4ae6ef3e3..c2b6e918f 100644 --- a/src/lib/jobs/__tests__/document-thumbnail.test.ts +++ b/src/lib/jobs/__tests__/document-thumbnail.test.ts @@ -177,4 +177,47 @@ describe("thumbnailRetryDue", () => { thumbnailRetryDue({ reason: "error", attempts: 3, failedAt: at }, before), ).toBe(false); }); + + it("gives a renderer that ended for no measured reason the ordinary attempts, not a final verdict", async () => { + const { thumbnailRetryDue } = + await import("../document-thumbnail-backfill"); + expect( + thumbnailRetryDue( + { reason: "interrupted", attempts: 1, failedAt: at }, + before, + ), + ).toBe(true); + expect( + thumbnailRetryDue( + { reason: "interrupted", attempts: 3, failedAt: at }, + before, + ), + ).toBe(false); + expect( + thumbnailRetryDue( + { reason: "resource-limit", attempts: 1, failedAt: at }, + before, + ), + ).toBe(false); + }); +}); + +describe("runDocumentThumbnail when the reader cannot take the document now", () => { + it("throws for a busy reader so the job runs again, and records nothing", async () => { + generateThumbnail.mockResolvedValue({ ok: false, reason: "reader-busy" }); + await expect( + runDocumentThumbnail({ userId: "user-1", documentId: "doc-1" }), + ).rejects.toMatchObject({ name: "PdfReaderUnavailableError" }); + expect(failureUpsert).not.toHaveBeenCalled(); + }); + + it("records an interrupted render as a failure the backfill tries again", async () => { + generateThumbnail.mockResolvedValue({ ok: false, reason: "interrupted" }); + await runDocumentThumbnail({ userId: "user-1", documentId: "doc-1" }); + expect(failureUpsert).toHaveBeenCalledWith( + expect.objectContaining({ + create: expect.objectContaining({ reason: "interrupted" }), + }), + ); + }); }); diff --git a/src/lib/jobs/document-thumbnail-backfill.ts b/src/lib/jobs/document-thumbnail-backfill.ts index 4da644497..f0c87426e 100644 --- a/src/lib/jobs/document-thumbnail-backfill.ts +++ b/src/lib/jobs/document-thumbnail-backfill.ts @@ -51,9 +51,15 @@ const THUMBNAILABLE_MIMES = [ /** * Render refusals that the same bytes will always get again: the first page * renders blank, the render comes out empty, the image is past the pixel cap - * or its size cannot be read, the PDF crossed the renderer's memory or time bound. One attempt settles - * them; queueing a document that exhausts the renderer again at every boot - * would only repeat the exhaustion. + * or its size cannot be read, the document crossed a bound the renderer + * measured (its memory ceiling, its time limit, its heap cap). One attempt + * settles them; queueing a document that exhausts the renderer again at + * every boot would only repeat the exhaustion. + * + * `interrupted` is deliberately not here: a renderer that ended without a + * measured cause may have been taken by the kernel because the whole + * container ran short, which says nothing about the document. It gets the + * ordinary bounded attempts. */ const FINAL_THUMBNAIL_FAILURES = [ "raster-failed", diff --git a/src/lib/jobs/document-thumbnail.ts b/src/lib/jobs/document-thumbnail.ts index 4195820ee..0ead396c8 100644 --- a/src/lib/jobs/document-thumbnail.ts +++ b/src/lib/jobs/document-thumbnail.ts @@ -29,6 +29,8 @@ import { decryptDocumentContent, encryptThumbnail, } from "@/lib/documents/store"; +import { withPdfWork } from "@/lib/documents/pdf-isolation"; +import { PdfReaderUnavailableError } from "@/lib/documents/pdfjs"; import { generateThumbnail } from "@/lib/documents/thumbnail"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { withBackgroundEvent } from "@/lib/logging/background"; @@ -55,10 +57,12 @@ export async function runDocumentThumbnail( if (!userId || !documentId) return; await withBackgroundEvent("job.document_thumbnail", async (evt) => { evt.addMeta("document_id", documentId); - await renderThumbnail(userId, documentId, (reason) => { - evt.addMeta("thumbnail_missing_reason", reason); - evt.elevateLevel("warn"); - }); + await withPdfWork({ owner: userId, lane: "background" }, () => + renderThumbnail(userId, documentId, (reason) => { + evt.addMeta("thumbnail_missing_reason", reason); + evt.elevateLevel("warn"); + }), + ); }); } @@ -107,6 +111,11 @@ async function renderThumbnail( } const result = await generateThumbnail(bytes, document.mimeType); + if (!result.ok && result.reason === "reader-busy") { + // The reader was taken by other work for longer than a background item + // waits. Nothing is wrong with the document: pg-boss runs the job again. + throw new PdfReaderUnavailableError("busy"); + } if (!result.ok) { annotate({ action: { name: "documents.thumbnail.run" },