Keeping a page lives in one module with one retry policy

apps/web/lib/job-runner.ts

@@ -0,0 +1,39 @@
+import { claimJob, completeJob, failJob, retryJob, type Job } from "@pluck/db";
+import { log } from "./log";
+
+// The loop that runs queued jobs, apart from the handler table (jobs.ts) so the library can kick it without a cycle.
+export type Handlers = Record<string, (payload: Record<string, unknown>) => Promise<void>>;
+
+const MAX_ATTEMPTS = 5;
+
+// A handler that throws is retried with backoff up to five times.
+async function runOne(handlers: Handlers, job: Job): Promise<void> {
+  const started = Date.now();
+  const handler = handlers[job.queue];
+  if (!handler) return failJob(job.id, `no handler for ${job.queue}`);
+  // A job the platform killed mid-run never threw: its lease ran out and it came back. Stop after five of those.
+  if (job.attempts > MAX_ATTEMPTS) return failJob(job.id, "did not finish in time");
+  try {
+    await handler(job.payload);
+    await completeJob(job.id);
+    log("info", "job done", { jobId: job.id, queue: job.queue, durationMs: Date.now() - started });
+  } catch (e) {
+    const error = e instanceof Error ? e.message : String(e);
+    if (job.attempts >= MAX_ATTEMPTS) await failJob(job.id, error);
+    else await retryJob(job.id, error, 30_000 * 2 ** (job.attempts - 1));
+    log("warn", "job failed", { jobId: job.id, queue: job.queue, attempts: job.attempts, error });
+  }
+}
+
+// Claims and runs jobs (of one queue, if named) until none are visible or the time budget is spent.
+export async function runQueue(handlers: Handlers, { budgetMs = 45_000, max = 500, queue }: { budgetMs?: number; max?: number; queue?: string } = {}): Promise<{ ran: number }> {
+  const deadline = Date.now() + budgetMs;
+  let ran = 0;
+  while (ran < max && Date.now() < deadline) {
+    const job = await claimJob(60_000, queue);
+    if (!job) break;
+    await runOne(handlers, job);
+    ran++;
+  }
+  return { ran };
+}

apps/web/lib/jobs.ts

@@ -1,23 +1,21 @@
-import { claimJob, completeJob, failJob, getItem, hitRateLimit, retryJob, type Job } from "@pluck/db";
-import { extractAttempt } from "./capture";
-import { finishFile, ingestInbound, runExport, runImport, sendDigest, sendWelcome } from "./library";
-import { log } from "./log";
+import { hitRateLimit } from "@pluck/db";
+import { runQueue, type Handlers } from "./job-runner";
+import { ingestInbound, keepFile, keepQueued, runExport, runImport, sendDigest, sendWelcome } from "./library";
 
-const MAX_ATTEMPTS = 5;
 // Queued mail (Monday emails, welcomes) shares one send rate across every instance, under SES's 14 a second,
 // so sign-in links (which skip the queue) always have room.
 const QUEUED_MAIL_PER_SECOND = 8;
 const mailSlot = async () => {
   while (!(await hitRateLimit("send:queued", QUEUED_MAIL_PER_SECOND, 1)).ok) await new Promise((r) => setTimeout(r, 250));
 };
 
-// Worker job handlers. A handler that throws is retried with backoff up to five times.
-const handlers: Record<string, (payload: Record<string, unknown>) => Promise<void>> = {
+// Worker job handlers, by queue. Keeping a page has its own retry policy (library/keeping.ts).
+const handlers: Handlers = {
   async extract(p) {
-    const userId = String(p.userId);
-    const item = await getItem(userId, String(p.itemId));
-    if (!item || item.extraction.state !== "pending") return; // done, failed, re-plucked or gone
-    await extractAttempt(userId, item, { timeoutMs: 20_000 }); // records the attempt and queues the next one itself
+    await keepQueued(String(p.userId), String(p.itemId));
+  },
+  async file(p) {
+    await keepFile(String(p.userId), String(p.itemId), String(p.key), String(p.contentType), String(p.name));
   },
   async digest(p) {
     await mailSlot();
@@ -33,41 +31,10 @@ const handlers: Record<string, (payload: Record<string, unknown>) => Promise<voi
   async import(p) {
     await runImport(String(p.userId), String(p.importId));
   },
-  async file(p) {
-    await finishFile(String(p.userId), String(p.itemId), String(p.key), String(p.contentType), String(p.name));
-  },
   async export(p) {
     await runExport(String(p.userId), String(p.exportId));
   },
 };
 
-async function runOne(job: Job): Promise<void> {
-  const started = Date.now();
-  const handler = handlers[job.queue];
-  if (!handler) return failJob(job.id, `no handler for ${job.queue}`);
-  // A job the platform killed mid-run never threw: its lease ran out and it came back. Stop after five of those.
-  if (job.attempts > MAX_ATTEMPTS) return failJob(job.id, "did not finish in time");
-  try {
-    await handler(job.payload);
-    await completeJob(job.id);
-    log("info", "job done", { jobId: job.id, queue: job.queue, durationMs: Date.now() - started });
-  } catch (e) {
-    const error = e instanceof Error ? e.message : String(e);
-    if (job.attempts >= MAX_ATTEMPTS) await failJob(job.id, error);
-    else await retryJob(job.id, error, 30_000 * 2 ** (job.attempts - 1));
-    log("warn", "job failed", { jobId: job.id, queue: job.queue, attempts: job.attempts, error });
-  }
-}
-
-// Claims and runs jobs until none are visible or the time budget is spent. Used by the cron tick and the worker loop.
-export async function runJobs({ budgetMs = 45_000, max = 500 } = {}): Promise<{ ran: number }> {
-  const deadline = Date.now() + budgetMs;
-  let ran = 0;
-  while (ran < max && Date.now() < deadline) {
-    const job = await claimJob(60_000);
-    if (!job) break;
-    await runOne(job);
-    ran++;
-  }
-  return { ran };
-}
+// Used by the cron tick and the worker loop.
+export const runJobs = (opts?: { budgetMs?: number; max?: number }) => runQueue(handlers, opts);

apps/web/lib/library/account.ts

@@ -1,16 +1,17 @@
 import { TooLarge, deleteObject, deletePrefix, getObject, putObject, signedPutUrl, signedUrl } from "@pluck/blob";
 import {
-  aiConnections, applyExtraction, claimOnce, countItems, createApiKey, disconnectAi, claimImportStart, createImport, deleteAccountData, enqueue, exportData, finishExport, getImport, getMe, importLink,
+  aiConnections, claimOnce, countItems, createApiKey, disconnectAi, claimImportStart, createImport, deleteAccountData, enqueue, exportData, finishExport, getImport, getMe, importLink,
   latestExport, recountTags, listApiKeys, revokeApiKey, startExport, updateImport,
 } from "@pluck/db";
-import { extractFromHtml, parseImport, type ImportSource } from "@pluck/extract";
+import { parseImport, type ImportSource } from "@pluck/extract";
 import { MAX_IMPORT_BYTES, normalizeUrl, type AiConnection, type ApiKey, type ImportStatus } from "@pluck/shared-schemas";
 import { strToU8, zipSync } from "fflate";
 import { sendEmail } from "../email";
 import { exportEmail, welcomeEmail } from "../emails";
 import { ApiError } from "../errors";
 import { log } from "../log";
 import { ORIGIN } from "../origin";
+import { keepImported } from "./keeping";
 
 // ---- Welcome -------------------------------------------------------------------------------------
 
@@ -105,18 +106,8 @@ export async function runImport(userId: string, importId: string): Promise<void>
     }
     const { item, created, tagIds } = await importLink(userId, url, e);
     touchedTags.push(...tagIds);
-    if (created) {
-      const page = e.html ? await extractFromHtml(e.html(), url).catch(() => null) : null;
-      if (page && page.wordCount > 0) {
-        await applyExtraction(userId, item.id, {
-          kind: page.kind, preview: page.preview, facts: page.facts, title: e.title ?? page.title, siteName: page.siteName, author: page.author, excerpt: page.excerpt,
-          canonicalUrl: page.canonicalUrl, lang: page.lang, wordCount: page.wordCount, markdown: page.markdown, extractor: page.extractor,
-        });
-      } else {
-        // Spread the fetching out: a thousand imports should not hit the network at once.
-        await enqueue("extract", { userId, itemId: item.id }, { delayMs: (done % 1000) * 3000 });
-      }
-    }
+    // Spread the fetching out: a thousand imports should not hit the network at once.
+    if (created) await keepImported(userId, item, { html: e.html, title: e.title, delayMs: (done % 1000) * 3000 });
     done++;
   }
   await recountTags(userId, touchedTags);

apps/web/lib/library/files.test.ts

@@ -1,7 +1,8 @@
 import { readFileSync } from "node:fs";
+import { pendingJobCount } from "@pluck/db";
 import { twoUsers } from "@pluck/db/testing";
 import { expect, test } from "vitest";
-import { beginFile, finishFile, readItem, readText } from "./index";
+import { beginFile, keepFile, readItem, readText } from "./index";
 
 const PDF = readFileSync(new URL("../../../../packages/extract/fixtures/pages/doc.pdf", import.meta.url));
 
@@ -11,13 +12,15 @@ test("a PDF shared from the phone: the item at once, the bytes by signed upload,
   expect(item).toMatchObject({ kind: "pdf", url: null, source: "ios", extraction: { state: "pending" } });
 
   const key = new URL(uploadUrl).pathname.split("/").slice(2).join("/");
-  await expect(finishFile(a, item.id, key, "application/pdf", "Estimates.pdf")).rejects.toThrow(); // not uploaded yet: the job retries
+  await keepFile(a, item.id, key, "application/pdf", "Estimates.pdf"); // not uploaded yet: a failed try, the next one queued
+  expect((await readItem(a, item.id)).extraction).toEqual({ state: "pending", attempts: 1, lastError: "The file has not arrived." });
+  expect(await pendingJobCount("file")).toBe(2); // beginFile's, and the retry
   const put = await fetch(uploadUrl, { method: "PUT", headers: { "content-type": "application/pdf" }, body: PDF });
   expect(put.ok).toBe(true);
 
-  await finishFile(b, item.id, key, "application/pdf", "Estimates.pdf"); // someone else's item: nothing happens
+  await keepFile(b, item.id, key, "application/pdf", "Estimates.pdf"); // someone else's item: nothing happens
   expect((await readItem(a, item.id)).extraction.state).toBe("pending");
-  await finishFile(a, item.id, key, "application/pdf", "Estimates.pdf");
+  await keepFile(a, item.id, key, "application/pdf", "Estimates.pdf");
   expect((await readItem(a, item.id)).extraction.state).toBe("done");
   expect((await readText(a, item.id)).length).toBeGreaterThan(20);
 });

apps/web/lib/library/files.ts

@@ -1,39 +1,16 @@
-import { TooLarge, deleteObject, getObject, keys, signedPutUrl } from "@pluck/blob";
-import { applyExtraction, createFileItem, enqueue, getItem, recordExtractionFailure, setBlobKeys } from "@pluck/db";
-import { extractFile } from "@pluck/extract";
-import { FILE_TYPES, MAX_FILE_BYTES, type CreateFileItemBody, type Item } from "@pluck/shared-schemas";
+import { keys, signedPutUrl } from "@pluck/blob";
+import { createFileItem, enqueue } from "@pluck/db";
+import { FILE_TYPES, type CreateFileItemBody, type Item } from "@pluck/shared-schemas";
 
 const keyFor = (userId: string, itemId: string, contentType: CreateFileItemBody["contentType"]) =>
   contentType === "application/pdf" ? keys.file(userId, itemId, "pdf") : keys.file(userId, itemId, FILE_TYPES[contentType]);
 
 // An Image or PDF from the phone's share sheet. The item lands at once; the bytes go straight to blob storage
-// (a background upload that may finish after the extension has gone), and the worker reads them once they are there.
+// (a background upload that may finish after the extension has gone), and keepFile reads them once they are there.
 export async function beginFile(userId: string, body: CreateFileItemBody): Promise<{ item: Item; uploadUrl: string }> {
   const kind = body.contentType === "application/pdf" ? "pdf" : "image";
   const item = await createFileItem(userId, { kind, title: body.name, source: body.source ?? "ios" });
   const key = keyFor(userId, item.id, body.contentType);
   await enqueue("file", { userId, itemId: item.id, key, contentType: body.contentType, name: body.name }, { delayMs: 15_000 });
   return { item, uploadUrl: await signedPutUrl(key, body.contentType) };
 }
-
-// The worker's half. Throws while the upload has not arrived, so the job retries with backoff.
-export async function finishFile(userId: string, itemId: string, key: string, contentType: string, name: string): Promise<void> {
-  const item = await getItem(userId, itemId);
-  if (!item || item.extraction.state !== "pending") return;
-  // shortcut: an upload that never arrives leaves the item pending once the job gives up (five tries, about 15 minutes); mark it failed if that shows up.
-  let body: Buffer;
-  try {
-    body = await getObject(key, { maxBytes: MAX_FILE_BYTES }); // missing yet: throws, and the job tries again later
-  } catch (e) {
-    if (!(e instanceof TooLarge)) throw e;
-    await recordExtractionFailure(userId, itemId, "That file is over 50 MB.", true);
-    await deleteObject(key);
-    return;
-  }
-  const x = await extractFile({ body, contentType, fileName: name });
-  if (x.kind === "image") await setBlobKeys(userId, itemId, { coverKey: key });
-  await applyExtraction(userId, itemId, {
-    kind: x.kind, preview: x.preview, facts: x.facts, title: x.title, siteName: x.siteName, author: x.author, excerpt: x.excerpt, canonicalUrl: null, lang: x.lang,
-    wordCount: x.wordCount, markdown: x.markdown, extractor: x.extractor,
-  });
-}

apps/web/lib/library/index.ts

@@ -12,3 +12,4 @@ export * from "./email-in";
 export * from "./account";
 export * from "./tries";
 export * from "./files";
+export * from "./keeping";