diff --git a/src/lib/jobs/document-index-retry.ts b/src/lib/jobs/document-index-retry.ts index 4df8c08e0..a6809953f 100644 --- a/src/lib/jobs/document-index-retry.ts +++ b/src/lib/jobs/document-index-retry.ts @@ -26,37 +26,65 @@ import { isModuleEnabled } from "@/lib/modules/gate"; /** Documents enqueued per boot; the rest wait for the next one. */ const MAX_ENQUEUES_PER_BOOT = 1000; +/** Marked documents read per query. */ +const PAGE_SIZE = 500; + +/** + * The marks are walked by id with a cursor, and an account found with the + * module off is left out of every later page. Taking the first thousand by id + * once let one such account's marks fill the whole boot, every boot, while + * the accounts after it never got their retry. + */ export async function enqueueBootTimeIndexRetry(): Promise<{ enqueued: number; }> { - const rows = await prisma.inboundDocument.findMany({ - where: { - deletedAt: null, - lastIndexAttemptAt: null, - lastIndexOutcome: { not: null }, - contentIndex: { is: null }, - }, - select: { id: true, userId: true }, - orderBy: { id: "asc" }, - take: MAX_ENQUEUES_PER_BOOT, - }); - - const moduleOn = new Map(); + const moduleOff: string[] = []; + const moduleOn = new Set(); + let marked = 0; let enqueued = 0; - for (const { id, userId } of rows) { - let on = moduleOn.get(userId); - if (on === undefined) { - on = await isModuleEnabled(userId, "inboundDocuments"); - moduleOn.set(userId, on); + let cursor: string | null = null; + + walk: for (;;) { + const rows: { id: string; userId: string }[] = + await prisma.inboundDocument.findMany({ + where: { + deletedAt: null, + lastIndexAttemptAt: null, + lastIndexOutcome: { not: null }, + contentIndex: { is: null }, + ...(moduleOff.length > 0 ? { userId: { notIn: moduleOff } } : {}), + // Past the last row read. Not Prisma's `cursor` + `skip`: the + // cursor row may belong to an account the filter now leaves out, + // and `skip` would then drop a row that was never read. + ...(cursor ? { id: { gt: cursor } } : {}), + }, + select: { id: true, userId: true }, + orderBy: { id: "asc" }, + take: PAGE_SIZE, + }); + if (rows.length === 0) break; + marked += rows.length; + + for (const { id, userId } of rows) { + if (moduleOff.includes(userId)) continue; + if (!moduleOn.has(userId)) { + if (!(await isModuleEnabled(userId, "inboundDocuments"))) { + moduleOff.push(userId); + continue; + } + moduleOn.add(userId); + } + const { enqueued: ok } = await enqueueDocumentIndex(userId, id); + if (ok) enqueued += 1; + if (enqueued >= MAX_ENQUEUES_PER_BOOT) break walk; } - if (!on) continue; - const { enqueued: ok } = await enqueueDocumentIndex(userId, id); - if (ok) enqueued += 1; + cursor = rows[rows.length - 1]!.id; + if (rows.length < PAGE_SIZE) break; } annotate({ action: { name: "documents.contentIndex.retry.boot" }, - meta: { marked: rows.length, enqueued }, + meta: { marked, enqueued, moduleOffAccounts: moduleOff.length }, }); return { enqueued }; } diff --git a/tests/integration/pdfjs-clash-recovery.test.ts b/tests/integration/pdfjs-clash-recovery.test.ts index dd9da8f93..e64fd5ca9 100644 --- a/tests/integration/pdfjs-clash-recovery.test.ts +++ b/tests/integration/pdfjs-clash-recovery.test.ts @@ -289,4 +289,41 @@ describe("0479: PDFs whose local text read failed", () => { }), ).toEqual({ lastIndexAttemptAt: null }); }); + + it("does not let an account with the module off hold every place of a boot", async () => { + // An account whose documents module is off, with more marked documents + // than one boot enqueues, all ordered before the next account's. + const OFF = "pdfjs-clash-module-off"; + await prisma.user.create({ + data: { + id: OFF, + username: OFF, + modulePreferencesJson: { inboundDocuments: false }, + }, + }); + const { content, codec } = encryptDocumentContent(textPdf()); + const marked = 1000; + await prisma.inboundDocument.createMany({ + data: Array.from({ length: marked }, (_, i) => ({ + id: `a-off-${String(i).padStart(4, "0")}`, + userId: OFF, + mimeType: "application/pdf", + byteSize: 1, + contentEncrypted: content, + contentCodec: codec, + lastIndexOutcome: "decrypt-error", + })), + }); + await seedPdf("b-on", textPdf(), { lastIndexOutcome: "decrypt-error" }); + + expect((await enqueueBootTimeIndexRetry()).enqueued).toBe(1); + const jobs = takeSends("document-index") as DocumentIndexPayload[]; + expect(jobs.map((j) => j.documentId)).toEqual(["b-on"]); + // The module-off account keeps its marks for the boot after it is on. + expect( + await prisma.inboundDocument.count({ + where: { userId: OFF, lastIndexAttemptAt: null }, + }), + ).toBe(marked); + }); });