feat: the plugin keeps its own record of runs, and the runner port narrows

apps/dev/src/integration/translator/access-control-queue.int.test.ts

@@ -27,15 +27,13 @@ const withFieldRule = (collections: CollectionConfig[]): CollectionConfig[] =>
     } as CollectionConfig;
   });
 
-const jobsFor = async (documentId: string) => {
+const runsFor = async (documentId: string) => {
   const { docs } = await ctx.payload.find({
-    collection: "payload-jobs" as "pages",
+    collection: "translator-runs" as "pages",
     pagination: false,
-    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+    where: { collectionId: { equals: documentId } } as never,
   });
-  return (docs as Array<{ input?: Record<string, unknown> }>).filter(
-    (j) => j.input?.collection_id === documentId
-  );
+  return docs as Array<{ requesterId?: string | null; requesterCollection?: string | null }>;
 };
 
 type Assignment = { handle: string };
@@ -100,14 +98,14 @@ describe("a queued translation is decided by whoever asked for it", () => {
     await ctx?.cleanup();
   });
 
-  it("records the requester on the job row", async () => {
+  it("records the requester on the run, which is what survives until it runs", async () => {
     const id = await seed("Queued", "Queued line");
 
     await enqueueAs(id, editor);
 
-    const [job] = await jobsFor(id);
-    expect(String(job?.input?.requester_id)).toBe(editor.id);
-    expect(job?.input?.requester_collection).toBe("users");
+    const [run] = await runsFor(id);
+    expect(run?.requesterId).toBe(editor.id);
+    expect(run?.requesterCollection).toBe("users");
   });
 
   // The request that queued this is long gone by the time it runs, so rebuilding from the row is the

apps/dev/src/integration/translator/bootTestPayload.ts

@@ -70,11 +70,14 @@ export type TestPayload = {
  *   publish; the enqueue route still works.
  * @param opts.collections - replaces the shared fixture set entirely (not merged). The set must
  *   still contain a `docs` collection when `autoTranslate` is passed.
+ * @param opts.unmanagedCollections - registered with Payload, but not handed to the plugin.
  * @param opts.failFor - target locales the fake provider should throw for, so a spec can exercise a
  *   partial failure. Every other locale translates normally.
  * @param opts.runner - defaults to the sync runner. `createPayloadJobsRunner({ autoRun: false })`
  *   leaves queued jobs unprocessed in `payload-jobs`, so a spec can read the rows.
  * @param opts.onTranslate - awaited before each provider call, so a spec can hold a locale mid-run.
+ * @param opts.deleteJobOnComplete - Payload's own option; Payload defaults it to `true`, this boot
+ *   to `false` so the status panels have rows to read.
  * @param opts.exclusiveQueue - Payload's `enableConcurrencyControl` for this boot; defaults to
  *   `EXCLUSIVE_QUEUE=1`.
  * @param opts.fallback - localization fallback, off by default: an unwritten locale reads as
@@ -83,7 +86,9 @@ export type TestPayload = {
 export async function bootTestPayload(opts?: {
   autoTranslate?: { targets: string[] };
   collections?: CollectionConfig[];
+  unmanagedCollections?: CollectionConfig[];
   fallback?: boolean;
+  deleteJobOnComplete?: boolean;
   exclusiveQueue?: boolean;
   failFor?: string[];
   onTranslate?: (targetLng: string) => Promise<void> | void;
@@ -146,10 +151,10 @@ export async function bootTestPayload(opts?: {
         { code: "es", label: "Español" },
       ],
     },
-    collections,
+    collections: [...collections, ...(opts?.unmanagedCollections ?? [])],
     jobs: {
       // Payload deletes completed jobs by default, leaving the status panels nothing to read.
-      deleteJobOnComplete: false,
+      deleteJobOnComplete: opts?.deleteJobOnComplete ?? false,
       enableConcurrencyControl: opts?.exclusiveQueue ?? process.env.EXCLUSIVE_QUEUE === "1",
     },
     plugins: [

apps/dev/src/integration/translator/cancel-announcement.int.test.ts

@@ -17,7 +17,7 @@ const runStillOnRecordWhenAnnounced: boolean[] = [];
 const onCancelled = vi.fn(async (task: Seen) => {
   const row = await ctx.payload.find({
     collection: "payload-jobs" as "docs",
-    where: { id: { equals: task.handle } } as never,
+    where: { "input.handle": { equals: task.handle } } as never,
   });
   runStillOnRecordWhenAnnounced.push(row.docs.length === 1);
 });

apps/dev/src/integration/translator/enqueue-handles.int.test.ts

@@ -54,9 +54,9 @@ describe("the enqueue answer names the run that will translate each locale", ()
     const rows = await ctx.payload.find({
       collection: "payload-jobs" as "docs",
       pagination: false,
-      where: { id: { in: assignments.map((j) => j.handle) } } as never,
+      where: { "input.handle": { in: assignments.map((j) => j.handle) } } as never,
     });
-    expect(rows.docs, "every handle answered names a row that is actually there").toHaveLength(
+    expect(rows.docs, "every handle answered names a job that is actually there").toHaveLength(
       new Set(assignments.map((j) => j.handle)).size
     );
   });
@@ -85,19 +85,16 @@ describe("the enqueue answer names the run that will translate each locale", ()
     ).toHaveLength(1);
   });
 
-  it("still names the run for a locale a live run already covers", async () => {
+  it("names no run for a document it is already translating", async () => {
     const id = await create("Asked twice");
 
     const first = await enqueue([id], ["de"]);
     const second = await enqueue([id], ["de", "fr"]);
 
+    expect(first.map((j) => j.target_lng)).toEqual(["de"]);
     expect(
-      second.map((j) => j.target_lng).sort(),
-      "de needed no write, but it is being translated — leaving it out would report nothing for it"
-    ).toEqual(["de", "fr"]);
-    expect(
-      second.find((j) => j.target_lng === "de")?.handle,
-      "and it is the run that already had it"
-    ).toBe(first[0]?.handle);
+      second,
+      "a second run over this document would put the first run's locales back as they were"
+    ).toEqual([]);
   });
 });

apps/dev/src/integration/translator/exclusive-queue.int.test.ts

@@ -89,39 +89,6 @@ const titleIn = async (id: string, locale: string) =>
   ).title;
 
 describe("with the host's concurrency control on", () => {
-  it("holds a second job for the same document, then runs it and loses nothing", async () => {
-    armBarrier();
-    const id = await createDoc("Exclusive source");
-
-    await enqueue(id, ["de"]);
-    const first = runQueue();
-    await held;
-
-    await enqueue(id, ["fr"]);
-
-    // The picker check below also passes when there was nothing to take, so first prove the second
-    // job exists.
-    const queued = await jobsFor(id);
-    expect(queued.length, "the request did not get a job of its own").toBe(2);
-    expect(
-      queued.find((job) => !job.processing)?.input?.targets?.map((t) => t.lng),
-      "the second job should carry only the locale that was asked for"
-    ).toEqual(["fr"]);
-
-    const whileRunning = await runQueue();
-    expect(
-      Object.keys(whileRunning.jobStatus ?? {}),
-      "the picker took a second job for a document already being written"
-    ).toEqual([]);
-
-    release?.();
-    await first;
-    await runQueue();
-
-    expect(await titleIn(id, "de"), "de was lost").toBe("de:Exclusive source");
-    expect(await titleIn(id, "fr"), "fr was lost").toBe("fr:Exclusive source");
-  });
-
   it("still runs jobs for different documents together", async () => {
     const first = await createDoc("Doc one");
     const second = await createDoc("Doc two");

apps/dev/src/integration/translator/field-selection.int.test.ts

@@ -80,10 +80,12 @@ describe("a run carries the fields it was asked for, through the stored job row"
     expect(de.note).toBe(tr("de", "This too"));
   });
 
-  it("keeps one locale's leaf list out of a second request for another locale", async () => {
+  it("keeps one locale's leaf list out of a later request for another locale", async () => {
     const id = await create({ title: "Shared run", note: "Only in French" });
 
+    // The first run must settle before the second is accepted: one run at a time per document.
     await enqueue({ id, target_lng: "de", filters: leaves("title") });
+    await work();
     await enqueue({ id, target_lng: "fr" });
     await work();