fix(documents): keep module-off accounts from filling the index retry boot

The boot discovery for the documents 0479 marked took the first thousand
by id and skipped those whose account has the documents module off.
Such an account keeps its marks, so with enough of them it held every
place of every boot and the accounts after it never got their retry.

The marks are walked by id past the last row read, an account found
with the module off is left out of the query from then on, and the
boot stops at a thousand documents enqueued rather than read.

Refs #1277

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<string, boolean>();
+  const moduleOff: string[] = [];
+  const moduleOn = new Set<string>();
+  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 };
 }

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);
+  });
 });