fix(translator): translate a document's locales in one job, extended rather than replaced

apps/dev/docs/multi-db-verification.md

@@ -54,6 +54,13 @@ is knowingly red (see below) and `&&` would stop the chain before Mongo ever ran
 `bootTestPayload` picks its adapter from `DB_ADAPTER` through `resolveTestDbAdapter()`, the test-side
 sibling of `resolveDbAdapter()` in `src/lib/database/resolveAdapter.ts`.
 
+**Two queue modes, not one.** `EXCLUSIVE_QUEUE=1` turns on Payload's `enableConcurrencyControl` for
+every boot, which is the setting a host enables to have the queue itself hold a second job for a
+document. `test:integration:all` runs the three adapters in both modes — six runs — because the
+plugin behaves differently under it: without the flag a second request extends the live job, with it
+the request gets a job of its own and waits. Two specs pin their own mode rather than following the
+env var (`locale-append` needs it off, `exclusive-queue` needs it on); everything else runs in both.
+
 **Isolation.** Each boot needs a database of its own: the twelve spec files run serially but share
 one server, so without it they would read each other's rows. SQLite gets this for free — a throwaway
 file per boot. Postgres and MongoDB do not, so each boot is given a namespace named by a random run

apps/dev/package.json

@@ -28,8 +28,12 @@
     "test:integration:sqlite": "cross-env DB_ADAPTER=sqlite bun run test:integration",
     "test:integration:postgres": "cross-env DB_ADAPTER=postgres bun run test:integration",
     "test:integration:mongo": "cross-env DB_ADAPTER=mongo bun run test:integration",
-    "test:integration:all": "bun run test:integration:sqlite; bun run test:integration:postgres; bun run test:integration:mongo",
-    "check-types": "tsgo --noEmit -p tsconfig.check.json"
+    "test:integration:all": "bun run test:integration:sqlite; bun run test:integration:postgres; bun run test:integration:mongo; bun run test:integration:exclusive:sqlite; bun run test:integration:exclusive:postgres; bun run test:integration:exclusive:mongo",
+    "check-types": "tsgo --noEmit -p tsconfig.check.json",
+    "test:integration:exclusive": "cross-env EXCLUSIVE_QUEUE=1 bun run test:integration",
+    "test:integration:exclusive:sqlite": "cross-env DB_ADAPTER=sqlite bun run test:integration:exclusive",
+    "test:integration:exclusive:postgres": "cross-env DB_ADAPTER=postgres bun run test:integration:exclusive",
+    "test:integration:exclusive:mongo": "cross-env DB_ADAPTER=mongo bun run test:integration:exclusive"
   },
   "dependencies": {
     "@focus-reactive/payload-plugin-ab": "workspace:*",

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

@@ -22,6 +22,9 @@ import { createTestDatabase } from "../../lib/database/resolveAdapter";
 import { reverseComplete } from "../../lib/translator/fakeComplete";
 import { buildTestCollections } from "./testCollections";
 
+/** Payload's `autoRun.limit` default — these specs reproduce the cron's batching, not a run of one. */
+export const CRON_BATCH_LIMIT = 50;
+
 /**
  * A booted test Payload plus the throwaway resources to tear down after the suite.
  */
@@ -49,6 +52,8 @@ export type TestPayload = {
  *   `runId` (see `resolveTestDbAdapter`), so schema `push` is a clean CREATE with no data-loss branch
  *   — Payload never drops to the interactive "accept data loss?" prompt that would hang an
  *   unattended/headless run. `cleanup()` drops the namespace and removes the temp dir even on failure.
+ * - **One boot per process:** `getPayload` caches, so a second `bootTestPayload` in the same spec
+ *   file returns the first — a case that needs its own boot needs its own file.
  * - **Sync runner:** a translation runs INLINE inside the triggering `afterChange`, so it is complete
  *   when the awaited `payload.update`/`create` resolves — no job autorun, no polling, no async race
  *   in the specs.
@@ -60,15 +65,23 @@ 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.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.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
  *   empty, not as the default locale's text. Localization-level, so it applies to the whole boot.
  */
 export async function bootTestPayload(opts?: {
   autoTranslate?: { targets: string[]; strategy?: "overwrite" | "skip_existing" };
   collections?: CollectionConfig[];
   fallback?: boolean;
+  exclusiveQueue?: boolean;
+  failFor?: string[];
+  onTranslate?: (targetLng: string) => Promise<void> | void;
   runner?: TaskRunnerProvider;
 }): Promise<TestPayload> {
   const dir = mkdtempSync(join(tmpdir(), "translator-int-"));
@@ -84,11 +97,14 @@ export async function bootTestPayload(opts?: {
     : collections;
 
   const baseProvider = createTranslationProvider({ complete: reverseComplete });
+  const failFor = new Set(opts?.failFor);
   let translateCalls = 0;
   const countingProvider: TranslationProvider = {
-    translate: (input, sourceLng, targetLng) => {
+    translate: async (input, sourceLng, targetLng) => {
       translateCalls += 1;
-      return baseProvider.translate(input, sourceLng, targetLng);
+      await opts?.onTranslate?.(targetLng);
+      if (failFor.has(targetLng)) throw new Error(`provider unavailable for ${targetLng}`);
+      return await baseProvider.translate(input, sourceLng, targetLng);
     },
   };
 
@@ -107,9 +123,17 @@ export async function bootTestPayload(opts?: {
         { code: "en", label: "English" },
         { code: "de", label: "Deutsch" },
         { code: "fr", label: "Français" },
+        // Three targets, not two: with two, the failing locale is always the last and "stopped at
+        // the failure" is unobservable.
+        { code: "es", label: "Español" },
       ],
     },
     collections,
+    jobs: {
+      // Payload deletes completed jobs by default, leaving the status panels nothing to read.
+      deleteJobOnComplete: false,
+      enableConcurrencyControl: opts?.exclusiveQueue ?? process.env.EXCLUSIVE_QUEUE === "1",
+    },
     plugins: [
       translatorPlugin({
         collections: managed,

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

@@ -0,0 +1,138 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+// Its own file: the setting is fixed at boot, and a boot is per process (see `bootTestPayload`).
+
+type RunResult = { jobStatus?: Record<string, unknown> };
+
+type Job = {
+  processing?: boolean;
+  input?: { collection_id?: string; target_lngs?: string[] };
+};
+
+let ctx: TestPayload;
+let release: (() => void) | undefined;
+let reached: (() => void) | undefined;
+let held: Promise<void>;
+
+const armBarrier = () => {
+  held = new Promise<void>((resolve) => {
+    reached = resolve;
+  });
+};
+
+beforeAll(async () => {
+  armBarrier();
+  ctx = await bootTestPayload({
+    exclusiveQueue: true,
+    runner: createPayloadJobsRunner({ autoRun: false }),
+    onTranslate: async (targetLng) => {
+      if (targetLng !== "de") return;
+      reached?.();
+      await new Promise<void>((resolve) => {
+        release = resolve;
+      });
+    },
+  });
+});
+afterAll(async () => {
+  release?.();
+  await ctx?.cleanup();
+});
+
+const createDoc = async (title: string) => {
+  const doc = await ctx.payload.create({
+    collection: "docs" as "pages",
+    locale: "en",
+    data: { title, _status: "published" } as never,
+  });
+  return String(doc.id);
+};
+
+const enqueue = (id: string, targets: string[]) =>
+  callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+    body: {
+      source_lng: "en",
+      target_lng: targets,
+      collection_slug: "docs",
+      collection_id: [id],
+      strategy: "overwrite",
+      publish_on_translation: false,
+    },
+  });
+
+const jobsFor = async (documentId: string) => {
+  const { docs } = await ctx.payload.find({
+    collection: "payload-jobs" as "pages",
+    pagination: false,
+    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
+  });
+  return (docs as Job[]).filter((job) => job.input?.collection_id === documentId);
+};
+
+const runQueue = () =>
+  ctx.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT }) as Promise<RunResult>;
+
+const titleIn = async (id: string, locale: string) =>
+  (
+    (await ctx.payload.findByID({
+      collection: "docs" as "pages",
+      id,
+      locale: locale as "en",
+      fallbackLocale: false,
+      draft: true,
+    })) as Record<string, unknown>
+  ).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?.target_lngs,
+      "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("ecruos evisulcxE");
+    expect(await titleIn(id, "fr"), "fr was lost").toBe("ecruos evisulcxE");
+  });
+
+  it("still runs jobs for different documents together", async () => {
+    const first = await createDoc("Doc one");
+    const second = await createDoc("Doc two");
+    await enqueue(first, ["fr"]);
+    await enqueue(second, ["fr"]);
+
+    const batch = await runQueue();
+
+    expect(
+      Object.keys(batch.jobStatus ?? {}).length,
+      "two documents were serialized against each other"
+    ).toBe(2);
+  });
+});

apps/dev/src/integration/translator/job-supersede.int.test.ts → apps/dev/src/integration/translator/job-extend.int.test.ts

@@ -1,43 +1,40 @@
 import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
-import { bootTestPayload } from "./bootTestPayload";
+import { bootTestPayload, CRON_BATCH_LIMIT } from "./bootTestPayload";
 import type { TestPayload } from "./bootTestPayload";
 import { callEndpoint } from "./callEndpoint";
 
-// Rows are counted, never identified by id: SQLite reuses the rowid of a deleted row (integer primary
-// key, no AUTOINCREMENT), so a replacement job can arrive carrying the deleted job's id and an
-// id-based assertion would pass for the wrong reason.
-
 type Job = {
   id: string | number;
   completedAt?: string | null;
-  input?: { target_lng?: string; collection_id?: string };
+  processing?: boolean;
+  input?: { collection_id?: string; target_lngs?: string[] };
 };
 
 let ctx: TestPayload;
 
-const enqueue = async (id: string, target = "de") => {
+const enqueue = async (id: string, targets: string[] = ["de"]) => {
   const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
     body: {
       source_lng: "en",
-      target_lng: target,
+      target_lng: targets,
       collection_slug: "docs",
       collection_id: [id],
       strategy: "overwrite",
       publish_on_translation: false,
     },
   });
   expect(res.status, "the enqueue endpoint rejected the request").toBe(200);
-  return (res.data as { data: { queued: number } }).data.queued;
+  const body = res.data as { data: { queued: number } };
+  return body.data.queued;
 };
 
-// The cases share one boot, so the table also holds every earlier case's jobs.
 const jobs = async (documentId: string): Promise<Job[]> => {
   const { docs } = await ctx.payload.find({
     collection: "payload-jobs" as "pages",
     pagination: false,
-    where: { taskSlug: { equals: "translate_document" } } as never,
+    where: { workflowSlug: { equals: "translate_document_locales" } } as never,
   });
   return (docs as Job[]).filter((j) => j.input?.collection_id === documentId);
 };
@@ -65,44 +62,60 @@ afterAll(async () => {
   await ctx?.cleanup();
 });
 
-describe("re-enqueue supersedes unfinished work, not finished work", () => {
-  it("keeps a finished job when the same locale is translated again", async () => {
+describe("a second request extends the live job rather than replacing it", () => {
+  it("adds its locales to a job that has not started", async () => {
     const id = await createDoc();
-    expect(await enqueue(id), "fixture: the first enqueue queued a job").toBe(1);
-
-    const [first] = await jobs(id);
-    await markFinished(first.id);
-
-    expect(await enqueue(id), "the re-enqueue queued nothing").toBe(1);
+    await enqueue(id, ["de", "fr"]);
+    await enqueue(id, ["es"]);
 
-    const after = await jobs(id);
-    expect(after.filter((j) => j.completedAt).length, "the finished job was deleted").toBe(1);
-    expect(after.length, "finished job plus the new one").toBe(2);
+    const live = await jobs(id);
+    expect(live.length, "the second request queued a job of its own").toBe(1);
+    expect(live[0].input?.target_lngs, "the locales already owed were dropped").toEqual([
+      "de",
+      "fr",
+      "es",
+    ]);
   });
 
-  it("still supersedes an unfinished job for the same locale", async () => {
+  it("translates every locale the extended job accumulated", async () => {
     const id = await createDoc();
-    await enqueue(id);
-    expect((await jobs(id)).length, "fixture").toBe(1);
+    await enqueue(id, ["de", "fr"]);
+    await enqueue(id, ["es"]);
+
+    await ctx.payload.jobs.run({ queue: "translations", limit: CRON_BATCH_LIMIT });
+
+    for (const locale of ["de", "fr", "es"]) {
+      const doc = (await ctx.payload.findByID({
+        collection: "docs" as "pages",
+        id,
+        locale: locale as "en",
+        fallbackLocale: false,
+        draft: true,
+      })) as Record<string, unknown>;
+      expect(doc.title, `${locale} was not translated`).toBe("crS");
+    }
+  });
 
-    expect(await enqueue(id), "the re-enqueue queued nothing").toBe(1);
+  it("keeps a locale the live job already owes from being listed twice", async () => {
+    const id = await createDoc();
+    await enqueue(id, ["de"]);
+    await enqueue(id, ["de"]);
 
-    expect((await jobs(id)).length, "the unfinished job was not superseded").toBe(1);
+    const live = await jobs(id);
+    expect(live.length).toBe(1);
+    expect(live[0].input?.target_lngs, "the locale was queued a second time").toEqual(["de"]);
   });
 
-  it("leaves another locale's unfinished job alone", async () => {
+  it("leaves a finished job alone and gives the new request its own", async () => {
     const id = await createDoc();
-    await enqueue(id, "de");
-    await enqueue(id, "fr");
-    expect((await jobs(id)).length, "fixture: one job per locale").toBe(2);
+    await enqueue(id, ["de"]);
+    const [first] = await jobs(id);
+    await markFinished(first.id);
 
-    expect(await enqueue(id, "de"), "the re-enqueue queued nothing").toBe(1);
+    await enqueue(id, ["fr"]);
 
     const after = await jobs(id);
-    expect(after.length, "the fr job was collateral damage").toBe(2);
-    expect(after.map((j) => j.input?.target_lng).sort(), "one job per locale, still").toEqual([
-      "de",
-      "fr",
-    ]);
+    expect(after.filter((j) => j.completedAt).length, "the finished job was touched").toBe(1);
+    expect(after.length, "finished job plus the new one").toBe(2);
   });
 });