From af32ef9792ccaa6b09464cfa6fcf6764d23f4b7c Mon Sep 17 00:00:00 2001 From: Siarhei Date: Tue, 6 Oct 2026 14:24:20 +0200 Subject: [PATCH 1/6] fix(translator): stop hosting a request on a job that will never run again A job that ran out of retries keeps `hasError` and never gets a completion date, so it stayed in the plugin's live set while Payload's own picker excluded it for good. Asking again for a locale whose job had given up was therefore absorbed onto that job and silently produced no work at all. `canStillRun` is now the single predicate behind host selection and the re-read after a write, and the enqueue scan filters the same rows in the database rather than loading them. Deliberately narrower than Payload's picker: `processing` and a future `waitUntil` mean "not right now", which is exactly the job a new request may still join. --- .../retranslate-after-giving-up.int.test.ts | 88 +++++++++++++++++++ .../__tests__/planEnqueue.test.ts | 85 +++++++++++++++--- .../payload-jobs-runner/normalizeJob.ts | 13 ++- .../payload-jobs-runner/planEnqueue.ts | 15 ++-- .../task-runner/payload-jobs-runner/types.ts | 5 +- 5 files changed, 186 insertions(+), 20 deletions(-) create mode 100644 apps/dev/src/integration/translator/retranslate-after-giving-up.int.test.ts diff --git a/apps/dev/src/integration/translator/retranslate-after-giving-up.int.test.ts b/apps/dev/src/integration/translator/retranslate-after-giving-up.int.test.ts new file mode 100644 index 000000000..d0b247c3d --- /dev/null +++ b/apps/dev/src/integration/translator/retranslate-after-giving-up.int.test.ts @@ -0,0 +1,88 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +const RETRIES = 1; + +let ctx: TestPayload; +let providerIsDown = true; +let docId = ""; +let firstJobId: string | undefined; +let secondJobId: string | undefined; + +const enqueue = async (): Promise => { + const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: "de", + collection_slug: "docs", + collection_id: [docId], + strategy: "overwrite", + publish_on_translation: true, + }, + }); + return (res.data as { data: { jobs: { job_id: string }[] } }).data.jobs[0]?.job_id; +}; + +describe("asking again for a locale whose job already gave up", () => { + beforeAll(async () => { + ctx = await bootTestPayload({ + runner: createPayloadJobsRunner({ + autoRun: false, + retries: { attempts: RETRIES, backoff: { type: "fixed", delay: 0 } }, + }), + onTranslate: () => { + if (providerIsDown) throw new Error("provider down"); + }, + }); + + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title: "Abandoned", note: "A note" }, + }); + docId = String(made.id); + + firstJobId = await enqueue(); + for (let run = 0; run <= RETRIES; run++) { + await ctx.payload.jobs.run({ queue: "translations" }); + } + + providerIsDown = false; + secondJobId = await enqueue(); + await ctx.payload.jobs.run({ queue: "translations" }); + }); + + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("leaves the first job marked as having given up, with no completion date", async () => { + const dead = await ctx.payload.findByID({ collection: "payload-jobs", id: firstJobId ?? "" }); + + expect( + (dead as { hasError?: boolean }).hasError, + "this is the row Payload's picker will never take again" + ).toBe(true); + expect( + (dead as { completedAt?: string | null }).completedAt ?? null, + "and it stays in the plugin's live set, which is what made it a tempting host" + ).toBeNull(); + }); + + it("answers with a new job rather than the one that gave up", () => { + expect(secondJobId).toBeTruthy(); + expect(secondJobId, "naming the abandoned job would promise work nothing runs").not.toBe( + firstJobId + ); + }); + + it("actually translates the locale on the second ask", async () => { + const doc = await ctx.payload.findByID({ collection: "docs", id: docId, locale: "de" }); + + expect((doc as { title?: string }).title).toBe("de:Abandoned"); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/planEnqueue.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/planEnqueue.test.ts index 3e6acc29f..5b4675abd 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/planEnqueue.test.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/planEnqueue.test.ts @@ -36,34 +36,54 @@ const plan = (live: PayloadJob[], requested: string[], exclusiveQueue = false) = describe("planEnqueue", () => { it("queues everything when the document has no live job", () => { - expect(plan([], ["de", "fr"])).toEqual({ host: null, append: [], queue: ["de", "fr"] }); + expect(plan([], ["de", "fr"])).toEqual({ + host: null, + extend: false, + append: [], + queue: ["de", "fr"], + }); }); it("adds locales the live job does not carry yet", () => { const live = job(); - expect(plan([live], ["fr", "es"])).toEqual({ host: live, append: ["fr", "es"], queue: [] }); + expect(plan([live], ["fr", "es"])).toEqual({ + host: live, + extend: true, + append: ["fr", "es"], + queue: [], + }); }); it("does nothing for a locale the live job already owes", () => { const live = job(); - expect(plan([live], ["de"])).toEqual({ host: live, append: [], queue: [] }); + expect(plan([live], ["de"])).toEqual({ host: live, extend: true, append: [], queue: [] }); }); it("gives a locale the live job has already translated a job of its own", () => { const live = job({ log: [{ state: "succeeded", input: { target_lng: "de" } }], }); - expect(plan([live], ["de"])).toEqual({ host: live, append: [], queue: ["de"] }); + expect(plan([live], ["de"])).toEqual({ host: live, extend: true, append: [], queue: ["de"] }); }); it("splits a request across both when it mixes the two", () => { const live = job({ log: [{ state: "succeeded", input: { target_lng: "de" } }] }); - expect(plan([live], ["de", "fr"])).toEqual({ host: live, append: ["fr"], queue: ["de"] }); + expect(plan([live], ["de", "fr"])).toEqual({ + host: live, + extend: true, + append: ["fr"], + queue: ["de"], + }); }); it("ignores duplicates in the request", () => { const live = job(); - expect(plan([live], ["fr", "fr"])).toEqual({ host: live, append: ["fr"], queue: [] }); + expect(plan([live], ["fr", "fr"])).toEqual({ + host: live, + extend: true, + append: ["fr"], + queue: [], + }); }); it.each([ @@ -96,18 +116,41 @@ describe("planEnqueue", () => { const legacy = job({ input: { ...storedInput, target_lngs: undefined, target_lng: "de" }, }); - expect(plan([legacy], ["fr"])).toEqual({ host: null, append: [], queue: ["fr"] }); + expect(plan([legacy], ["fr"])).toEqual({ + host: null, + extend: false, + append: [], + queue: ["fr"], + }); }); it("does not extend a cancelled job", () => { const cancelled = job({ error: { cancelled: true } }); - expect(plan([cancelled], ["fr"])).toEqual({ host: null, append: [], queue: ["fr"] }); + expect(plan([cancelled], ["fr"])).toEqual({ + host: null, + extend: false, + append: [], + queue: ["fr"], + }); + }); + + it("does not extend a job that ran out of retries", () => { + const exhausted = job({ hasError: true, error: { message: "provider down" } }); + expect( + plan([exhausted], ["fr"]), + "Payload's picker skips a row with hasError, so extending it would queue work nothing runs" + ).toEqual({ host: null, extend: false, append: [], queue: ["fr"] }); }); describe("when the host enabled Payload's concurrency control", () => { it("queues alongside a running job instead of extending it", () => { const running = job({ processing: true }); - expect(plan([running], ["fr"], true)).toEqual({ host: null, append: [], queue: ["fr"] }); + expect(plan([running], ["fr"], true)).toEqual({ + host: running, + extend: false, + append: [], + queue: ["fr"], + }); }); it("merges an already-translated locale into the alongside job too", () => { @@ -116,7 +159,8 @@ describe("planEnqueue", () => { log: [{ state: "succeeded", input: { target_lng: "de" } }], }); expect(plan([running], ["de", "fr"], true)).toEqual({ - host: null, + host: running, + extend: false, append: [], queue: ["fr", "de"], }); @@ -124,13 +168,23 @@ describe("planEnqueue", () => { it("still extends a job that has not started", () => { const pending = job({ processing: false }); - expect(plan([pending], ["fr"], true)).toEqual({ host: pending, append: ["fr"], queue: [] }); + expect(plan([pending], ["fr"], true)).toEqual({ + host: pending, + extend: true, + append: ["fr"], + queue: [], + }); }); }); it("extends a running job when the host has NOT enabled concurrency control", () => { const running = job({ processing: true }); - expect(plan([running], ["fr"])).toEqual({ host: running, append: ["fr"], queue: [] }); + expect(plan([running], ["fr"])).toEqual({ + host: running, + extend: true, + append: ["fr"], + queue: [], + }); }); describe("a job can only take locales from a request it matches", () => { @@ -140,7 +194,12 @@ describe("planEnqueue", () => { ["a different publish flag", { publish_on_translation: true }], ])("starts its own job for %s", (_label, storedOverride) => { const live = job({ input: { ...storedInput, ...storedOverride } }); - expect(plan([live], ["fr"])).toEqual({ host: null, append: [], queue: ["fr"] }); + expect(plan([live], ["fr"])).toEqual({ + host: null, + extend: false, + append: [], + queue: ["fr"], + }); }); }); }); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts index 18cbf4bff..7c0abe85c 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/normalizeJob.ts @@ -31,11 +31,22 @@ export function isCancelled(error: unknown): boolean { ); } +/** + * Whether this row can ever run again: no completion date, no `hasError` (a row that ran out of + * retries keeps it for good), not cancelled. + * + * Narrower than Payload's picker on purpose — `processing` and a future `waitUntil` only mean "not + * right now", and a job mid-run or debouncing is exactly the one a new request may join. + */ +export function canStillRun(job: PayloadJob): boolean { + return !job.completedAt && job.hasError !== true && !isCancelled(job.error); +} + export function normalizeJob(job: PayloadJob): Task { const { collectionSlug, collectionId } = readCollectionRef(job.input); return { - id: job.id, + id: String(job.id), status: getJobStatus(job), input: { collectionSlug, diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts index dffe36d0c..57e169325 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/planEnqueue.ts @@ -1,6 +1,6 @@ import type { CollectionSlug } from "payload"; -import { isCancelled, latestLogByLocale } from "./normalizeJob.js"; +import { canStillRun, latestLogByLocale } from "./normalizeJob.js"; import type { PayloadJob } from "./types.js"; export type RequestShape = { @@ -19,6 +19,11 @@ export type RequestShape = { */ export type EnqueuePlan = { host: PayloadJob | null; + /** + * False when {@link host} runs under an exclusive queue: the host still runs the locales it + * already lists — so its id is still owed to the caller — it just cannot be given new ones. + */ + extend: boolean; append: string[]; queue: string[]; }; @@ -38,7 +43,7 @@ export function planEnqueue(args: { }): EnqueuePlan { const requested = [...new Set(args.requested)]; const host = pickHost(args.live, args.request); - if (!host) return { host: null, append: [], queue: requested }; + if (!host) return { host: null, extend: false, append: [], queue: requested }; // Already-succeeded locales need a fresh job: the workflow passes the locale as the task id, so // Payload's restoration would skip them (see the workflow handler in PayloadJobsRunnerProvider). @@ -48,9 +53,9 @@ export function planEnqueue(args: { const missing = requested.filter((locale) => !listed.has(locale)); if (args.exclusiveQueue && host.processing) { - return { host: null, append: [], queue: [...missing, ...done] }; + return { host, extend: false, append: [], queue: [...missing, ...done] }; } - return { host, append: missing, queue: done }; + return { host, extend: true, append: missing, queue: done }; } /** @@ -61,7 +66,7 @@ function pickHost(live: PayloadJob[], request: RequestShape): PayloadJob | null const usable = live.filter( (job) => Array.isArray(job.input?.target_lngs) && - !isCancelled(job.error) && + canStillRun(job) && job.input?.source_lng === request.sourceLng && job.input?.strategy === request.strategy && (job.input?.publish_on_translation ?? false) === request.publishOnTranslation && diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts index 8b8e3f0d6..09bfbbb89 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/types.ts @@ -45,11 +45,14 @@ export type PayloadJobsRunnerConfig = { export type PayloadJob = { log?: JobLogEntry[]; - id: string; + /** Whatever the host's primary key is — a uuid on some databases, an autoincrement number on others. */ + id: string | number; completedAt?: string | null; createdAt: string; updatedAt: string; error?: unknown; + /** Set by Payload when a task runs out of retries. Its picker skips such a row for good. */ + hasError?: boolean | null; processing?: boolean | null; waitUntil?: string | null; input?: { From a65238ae67603581c77376d8d79facdf9f7b30bb Mon Sep 17 00:00:00 2001 From: Siarhei Date: Tue, 6 Oct 2026 14:24:55 +0200 Subject: [PATCH 2/6] feat(translator): return the ids of the jobs an enqueue created MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #107. One entry per requested target locale, since a single job row covers a document's whole locale list: two locales of one document name the same job, and a request spanning several documents still reads unambiguously. `TaskRunner.enqueue` widens its return rather than changing it — the old `void` answer still satisfies the contract, now deprecated for the next major. An overload pair cannot be used here: an implementation must satisfy every overload, so it would break every third-party runner. `TaskRunner` and `EnqueueResult` are exported so an implementation outside this package can name what it has always had to satisfy. --- .../integration/translator/bootTestPayload.ts | 4 + .../integration/translator/callEndpoint.ts | 2 +- .../translator/enqueue-job-ids.int.test.ts | 80 +++ .../translator/extend-live-job.int.test.ts | 71 +++ .../docs/DEPRECATIONS.md | 24 + .../2026-10-05-run-identity-for-lifecycle.md | 511 ++++++++++++++++++ .../__tests__/handler.test.ts | 30 +- .../features/enqueue-translation/handler.ts | 10 +- .../features/enqueue-translation/model.ts | 25 + .../task-runner/TaskRunner.interface.ts | 41 +- .../enqueueResult.conformance.types.ts | 29 + .../src/server/modules/task-runner/index.ts | 2 +- .../PayloadJobsTaskRunner.ts | 99 +++- ...ayloadJobsTaskRunner.enqueueResult.test.ts | 215 ++++++++ .../__tests__/PayloadJobsTaskRunner.test.ts | 11 +- .../task-runner/sync-runner/SyncTaskRunner.ts | 18 +- .../SyncTaskRunner.enqueueResult.test.ts | 95 ++++ .../__tests__/SyncTaskRunner.test.ts | 17 +- 18 files changed, 1237 insertions(+), 47 deletions(-) create mode 100644 apps/dev/src/integration/translator/enqueue-job-ids.int.test.ts create mode 100644 apps/dev/src/integration/translator/extend-live-job.int.test.ts create mode 100644 packages/payload-plugin-translator/docs/plans/2026-10-05-run-identity-for-lifecycle.md create mode 100644 packages/payload-plugin-translator/src/server/modules/task-runner/__tests__/enqueueResult.conformance.types.ts create mode 100644 packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.enqueueResult.test.ts create mode 100644 packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.enqueueResult.test.ts diff --git a/apps/dev/src/integration/translator/bootTestPayload.ts b/apps/dev/src/integration/translator/bootTestPayload.ts index 50c478e55..82afb0cf8 100644 --- a/apps/dev/src/integration/translator/bootTestPayload.ts +++ b/apps/dev/src/integration/translator/bootTestPayload.ts @@ -14,6 +14,7 @@ import { } from "@focus-reactive/payload-plugin-translator"; import type { TaskRunnerProvider, + TranslationLifecycleCallbacks, TranslationProvider, } from "@focus-reactive/payload-plugin-translator"; import { buildConfig } from "payload"; @@ -97,6 +98,8 @@ export async function bootTestPayload(opts?: { fake?: FakeTranslationOptions; /** Also register the synchronous per-field surface, `POST {basePath}/field`. */ fieldSurface?: boolean; + /** Omitted by default, so a boot without it leaves the runner undecorated. */ + lifecycle?: TranslationLifecycleCallbacks; }): Promise { const dir = mkdtempSync(join(tmpdir(), "translator-int-")); const { db, drop } = createTestDatabase(join(dir, "test.db")); @@ -161,6 +164,7 @@ export async function bootTestPayload(opts?: { runner: opts?.runner ?? createSyncRunner(), levels: opts?.fieldSurface ? [documentLevel(), fieldLevel()] : [documentLevel()], provenance: true, + ...(opts?.lifecycle ? { lifecycle: opts.lifecycle } : {}), ...(opts?.inlineMarks ? { experimental: { inlineMarks: true } } : {}), }), ], diff --git a/apps/dev/src/integration/translator/callEndpoint.ts b/apps/dev/src/integration/translator/callEndpoint.ts index 39544841e..76a1cf884 100644 --- a/apps/dev/src/integration/translator/callEndpoint.ts +++ b/apps/dev/src/integration/translator/callEndpoint.ts @@ -14,7 +14,7 @@ import type { Payload, PayloadRequest } from "payload"; */ export async function callEndpoint( payload: Payload, - method: "get" | "post", + method: "get" | "post" | "delete", path: string, opts?: { routeParams?: Record; diff --git a/apps/dev/src/integration/translator/enqueue-job-ids.int.test.ts b/apps/dev/src/integration/translator/enqueue-job-ids.int.test.ts new file mode 100644 index 000000000..d47f8f4e8 --- /dev/null +++ b/apps/dev/src/integration/translator/enqueue-job-ids.int.test.ts @@ -0,0 +1,80 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +type QueuedJob = { + collection_slug: string; + collection_id: string; + target_lng: string; + job_id: string; +}; + +let ctx: TestPayload; + +const enqueue = (collectionId: string[], targetLng: string | string[]) => + callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: targetLng, + collection_slug: "docs", + collection_id: collectionId, + strategy: "overwrite", + publish_on_translation: true, + }, + }); + +const bodyOf = (res: { data: unknown }) => + (res.data as { data: { queued: number; jobs?: QueuedJob[] } }).data; + +const jobsOf = (res: { data: unknown }) => bodyOf(res).jobs ?? []; + +const create = async (title: string): Promise => { + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title, note: "A note" }, + }); + return String(made.id); +}; + +describe("the enqueue response names the job that will run each locale", () => { + beforeAll(async () => { + ctx = await bootTestPayload(); + }); + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("answers one entry per target locale", async () => { + const id = await create("Three locales"); + + const jobs = jobsOf(await enqueue([id], ["de", "fr", "es"])); + + expect(jobs.map((job) => job.target_lng).sort()).toEqual(["de", "es", "fr"]); + expect(jobs.every((job) => job.job_id.length > 0)).toBe(true); + }); + + it("names the document each entry belongs to, so a bulk request stays unambiguous", async () => { + const first = await create("First"); + const second = await create("Second"); + + const jobs = jobsOf(await enqueue([first, second], "de")); + + expect( + jobs.map((job) => job.collection_id).sort(), + "a locale-keyed answer could not tell these two apart" + ).toEqual([first, second].sort()); + expect(jobs.every((job) => job.collection_slug === "docs")).toBe(true); + }); + + it("still reports how many it queued", async () => { + const id = await create("Counted"); + + const res = await enqueue([id], ["de", "fr"]); + + expect(bodyOf(res).queued, "the count that was already there must survive").toBe(2); + expect(jobsOf(res)).toHaveLength(2); + }); +}); diff --git a/apps/dev/src/integration/translator/extend-live-job.int.test.ts b/apps/dev/src/integration/translator/extend-live-job.int.test.ts new file mode 100644 index 000000000..35c795572 --- /dev/null +++ b/apps/dev/src/integration/translator/extend-live-job.int.test.ts @@ -0,0 +1,71 @@ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +type Seen = { targetLng: string; jobId?: string; attempt?: number }; + +const onCompleted = vi.fn<(task: Seen) => void>(); +const onFailed = vi.fn<(task: Seen) => void>(); + +let ctx: TestPayload; + +const create = async (title: string): Promise => { + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title, note: "A note" }, + }); + return String(made.id); +}; + +const enqueue = (id: string, targetLng: string) => + callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: targetLng, + collection_slug: "docs", + collection_id: [id], + strategy: "overwrite", + publish_on_translation: true, + }, + }); + +const jobsOf = (res: { data: unknown }) => + (res.data as { data: { jobs: { target_lng: string; job_id: string }[] } }).data.jobs; + +const callbackJobIds = (calls: [Seen][]) => + Object.fromEntries(calls.map(([task]) => [task.targetLng, task.jobId])); + +describe("a locale added to a job already in flight", () => { + beforeAll(async () => { + ctx = await bootTestPayload({ + runner: createPayloadJobsRunner({ autoRun: false }), + lifecycle: { onCompleted, onFailed }, + }); + }); + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("reports the live job in the response, and the same id in the callback, with no cross-talk", async () => { + onCompleted.mockClear(); + const id = await create("Extended"); + + const first = jobsOf(await enqueue(id, "de")); + const second = jobsOf(await enqueue(id, "fr")); + + expect( + second[0]?.job_id, + "fr joined the job de already had — a new id here would be a lie" + ).toBe(first[0]?.job_id); + + await ctx.payload.jobs.run({ queue: "translations" }); + + const byLocale = callbackJobIds(onCompleted.mock.calls); + expect(byLocale.de, "the first run's locale keeps its own reporting").toBe(first[0]?.job_id); + expect(byLocale.fr).toBe(second[0]?.job_id); + }); +}); diff --git a/packages/payload-plugin-translator/docs/DEPRECATIONS.md b/packages/payload-plugin-translator/docs/DEPRECATIONS.md index e729f4abd..e450aaeb3 100644 --- a/packages/payload-plugin-translator/docs/DEPRECATIONS.md +++ b/packages/payload-plugin-translator/docs/DEPRECATIONS.md @@ -100,6 +100,30 @@ the single source of truth — code annotations link here by anchor instead of d - `src/server/modules/task-runner/sync-runner/SyncTaskRunner.ts` - `src/server/features/get-document-status/handler.ts` (a remaining array-form caller) +### enqueue-void-return + +- **What:** a `TaskRunner.enqueue` implementation that returns `Promise` instead of an + `EnqueueResult`. +- **Status:** live +- **Deprecated:** 2026-10-05 / PR #TBD +- **Replacement:** return one `{ collectionSlug, collectionId, targetLng, jobId }` entry per requested + target locale. +- **Remove in:** next major +- **Why:** the ids exist at enqueue — Payload's `jobs.queue` hands back the row it created, and the + sync runner mints its own — and were discarded at the point of creation. Without them a host cannot + connect its own record to the work: it cannot store anything to cancel later, and it cannot match a + lifecycle callback to the request that caused it. The return widened rather than changed so that an + implementation written against the old signature keeps compiling. +- **Migration:** the return is a union, so both forms satisfy the interface. A runner that still + answers `void` queues exactly as before; the plugin degrades to reporting no ids — the enqueue + response omits them and the lifecycle callbacks carry none. Adopting it is a matter of returning + what the runner already has in hand. +- **Code refs:** + - `src/server/modules/task-runner/TaskRunner.interface.ts` (the union, `EnqueueResult`) + - `src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts` + - `src/server/modules/task-runner/sync-runner/SyncTaskRunner.ts` + - `src/server/features/enqueue-translation/handler.ts` (the caller that reports them) + ### cancel-by-collection-route - **What:** `POST {basePath}/cancel-by-collection` endpoint. diff --git a/packages/payload-plugin-translator/docs/plans/2026-10-05-run-identity-for-lifecycle.md b/packages/payload-plugin-translator/docs/plans/2026-10-05-run-identity-for-lifecycle.md new file mode 100644 index 000000000..516b80ab8 --- /dev/null +++ b/packages/payload-plugin-translator/docs/plans/2026-10-05-run-identity-for-lifecycle.md @@ -0,0 +1,511 @@ +# Research: run identity across the translation lifecycle (#107, #110, #106, #103) + +Date: 2026-10-05. Read against **0.14.0** (`package.json:3`), with #168 (0.15.0) merged to `main`. + +## Input + +Four issues filed 2026-08-28 by one external consumer building an editorial workflow on top of the +plugin, plus one adjacent issue considered for inclusion: + +| Issue | Ask | +|---|---| +| #107 | `enqueue` should return the job ids it created | +| #110 | Pass run context into lifecycle callbacks | +| #106 | An `onCancelled` lifecycle callback | +| #103 | `get-collection-status` should return document id and target locale, not just job id | +| #123 | Status endpoints still load a collection's whole job history (considered, see *Size check*) | + +## Parsing summary + +- Source type: four GitHub issues, each written as a problem statement plus a suggested shape. +- Stakeholder: a single consumer. All four describe one workflow and cite each other (#107 cites #110 + and #108; #110 cites the constraint #107 forces). +- Explicit request: give a host a durable way to connect its own editorial record to one translation + run, from enqueue through completion or cancellation. +- Implicit expectation: that a "job" corresponds to one document × one target locale. **This is the + load-bearing assumption and it is no longer true** — see *Codebase map*. +- Constraint stated by the consumer: they currently enforce "one open run per document × target + locale" as a workaround, and must run with `retries: { attempts: 0 }`. +- Contradiction found: #106 describes two cancellation paths; one of them no longer exists. + +## Codebase map + +### Prior art + +- `docs/plans/2026-09-04-locale-workflow.task.md` — replaced per-locale jobs with one workflow per + document. Landed `3ecc2c89` (2026-09-04). +- `docs/plans/2026-09-08-one-live-job-per-document.task.md` — **removed supersession entirely**; D1: + "a locale is added to the live job's stored list; no cancel, no delete". Landed `69274e01` + (2026-09-08). +- `docs/plans/2026-09-04-job-scan-bounds.task.md` — `TaskFilter`, and D4's reasoning that `TaskRunner` + is reachable through the exported `TaskRunnerProvider.create()` and therefore cannot be changed + freely. +- `docs/plans/2026-09-04-job-history-pruning.task.md` — retention is the host's policy, not the + plugin's; nothing about the job table is exported. +- Epic #52 — "Grouping translation variants into editorial sets … belongs in the consuming app and + would build on the per-locale status + staleness primitives rather than in the plugin." The four + issues are consistent with that: they ask for better primitives, not for the workflow itself. + +No prior plan covers run identity. No previous attempt in `git log`. + +### Relevant files + +**Runner contracts** +- `src/server/modules/task-runner/TaskRunner.interface.ts` — `enqueue(tasks, scope?): Promise` + (`:22`); `cancel(taskIds)` (`:27`); `findByCollection` overloads (`:39-44`); `TaskFilter` (`:53-61`). + The docblock at `:8-13` still claims cancellation-before-enqueue is encapsulated here; that comment + is stale. +- `src/server/modules/task-runner/types.ts` — `TaskInput` (`:22-36`), `Task` (`:41-50`), + `TaskStatus` (`:17`). +- `src/server/modules/task-runner/TaskRunnerProvider.interface.ts` — `TaskHandlerInput` (`:10-17`), + `TaskHandler` (`:22-26`), `TaskRunnerProvider` (`:56-75`). + +**Payload Jobs runner** +- `payload-jobs-runner/PayloadJobsTaskRunner.ts` — `enqueue` (`:51-77`), `serve` (`:79-99`), + `extendJob` (`:101-135`), `queueWorkflow` (`:137-162`), `cancel` (`:164-176`), `run` (`:178-202`), + `ownJobs` (`:266-273`). +- `payload-jobs-runner/planEnqueue.ts` — `planEnqueue` (`:33-54`), `pickHost` (`:60-73`). +- `payload-jobs-runner/types.ts` — `StoredWorkflowInput` (`:77-91`), `PayloadJob` (`:46-74`). +- `payload-jobs-runner/normalizeJob.ts` — `normalizeJob` (`:34-54`), `normalizeJobLocales` (`:60-80`). +- `payload-jobs-runner/PayloadJobsRunnerProvider.ts` — retry defaults (`:28-41`), task registration + (`:148-162`), workflow registration (`:183`). + +**Sync runner** +- `sync-runner/SyncTaskRunner.ts` — in-memory ids (`:29`), inline execution (`:47-68`), `cancel` + no-op (`:72-74`). +- `sync-runner/SyncRunnerProvider.ts` — `configure` is identity (`:38-40`). + +**Lifecycle** +- `src/server/modules/lifecycle/types.ts` — `TranslationTask` (`:9-20`), + `TranslationLifecycleCallbacks` (`:34-46`), per-attempt semantics (`:22-33`). +- `src/server/modules/lifecycle/LifecycleNotifier.ts` — `queued` (`:21-24`), `completed` (`:26-29`), + `failed` (`:31-44`), `safe` (`:46-53`). +- `src/server/modules/lifecycle/withQueuedNotification.ts` — the only decorator (`:12-27`). +- `src/server/modules/lifecycle/taskMapping.ts` — `taskFromInput` (`:6-12`), + `taskFromHandlerInput` (`:15-21`). +- `src/server/features/translate-document/wireTranslateRunner.ts` — the only assembly point + (`:84-90`), `failed` (`:76`), `completed` (`:79`). + +**Endpoints** +- `src/server/features/enqueue-translation/handler.ts` — response at `:98`. +- `src/server/features/get-collection-status/handler.ts` — mapping at `:39-43`. +- `src/server/features/get-document-status/handler.ts` — mapping at `:42`; `model.ts:74-81`, + `:86-104`. +- `src/server/features/cancel/handler.ts` — `:14-25`; route is **DELETE** (`route.ts:20-23`). +- `src/server/features/cancel-by-collection/handler.ts` — `:17-42`. + +### Findings that contradict the issues + +1. **Supersession does not exist.** `PayloadJobsTaskRunner.enqueue` never cancels or deletes; + `planEnqueue` picks a live job to extend and `extendJob` unions target locales into its `input` + column. A regression test is named for it: *"never cancels or deletes a job when enqueuing"* + (`payload-jobs-runner/__tests__/PayloadJobsTaskRunner.test.ts:234-246`). **#106's second path is + gone.** Even the removed version superseded per *document* and only for `pending` jobs — narrower + than #106's "per (document, target locale)" wording. + +2. **A job is not a (document, locale) pair.** One row covers (document, source locale, strategy, + publish flag, requester) and holds a **list** of target locales (`StoredWorkflowInput.target_lngs`, + `payload-jobs-runner/types.ts:77-91`). `normalizeJobLocales` spreads one `base` — including its + `id` — across every target (`normalizeJob.ts:67-78`). **#107's suggested + `{ fr: id, de: id, it: id }` would repeat one id three times**, and breaks outright for a + `select_all` enqueue spanning many documents. + +3. **The cancel route is `DELETE`, not `POST`** (`cancel/route.ts:20-23`), and its handler holds + **only job ids** (`cancel/handler.ts:14-25`) — no document, no locale, no requester. The row is + hard-deleted (`PayloadJobsTaskRunner.ts:172-175`). `cancel-by-collection` *does* hold `Task[]` + with documents and locales (`:30`) and discards them at the `cancel(ids)` boundary (`:39`). + +4. **The job id already exists at enqueue and is thrown away.** Payload's `jobs.queue` returns the + created `Job`; the plugin's local type alias declares the call `Promise` + (`PayloadJobsTaskRunner.ts:17-23`) and `queueWorkflow` discards it (`:155`). + +5. **There is no opaque channel into `enqueue`.** `TaskInput` has no free-form field + (`types.ts:22-36`). The second parameter is `RequestScope` — a closed shape of `transactionID` + + branded `Requester` (`RequestScope.shapes.ts:9-12`) — and the requester **is part of job identity**: + `pickHost` refuses to host a request from a different requester (`planEnqueue.ts:68-69`), so it + cannot be repurposed without changing which jobs merge. + +6. **Payload's own opaque field is unavailable.** `jobs.queue({ meta })` exists, but the `meta` column + is only added when the host sets `jobs.stats` + (`node_modules/payload/dist/queues/config/collection.js:266-272`). The plugin neither sets it nor + may — the host owns that schema, per `2026-09-04-job-history-pruning.task.md` D1. + +7. **Retries are on by default: 3 attempts, exponential backoff** + (`PayloadJobsRunnerProvider.ts:28-41`, README:372). The attempt number is never read — the task + handler receives only `{ req, input }` (`:149-162`) and never sees the job row's `totalTried`. + **#110's retry claim holds**, and a host that configured nothing gets three `onFailed` calls. + +8. **`withQueuedNotification` is applied only when `onQueued` is present** + (`wireTranslateRunner.ts:87`: `if (!lifecycle.onQueued) return taskRunner;`). A host that supplies + only `onCompleted` gets no decoration at all. Anything delivered through that wrapper would not + reach such a host. + +9. **`TaskRunner` is the hidden implemented contract.** It is *not* exported from `src/index.ts` + (verified against the full 58-symbol export list), yet a third-party `TaskRunnerProvider.create` + must structurally return one (`TaskRunnerProvider.interface.ts:67`). Changing `enqueue`'s return + type is therefore breaking for an implementor who cannot even name the type they must satisfy — + the same gap the `RequestScope` export note describes (`src/index.ts:60-66`). + +10. **Both response shapes are pinned by exact-match tests.** + `enqueue-translation/__tests__/handler.test.ts:159` asserts + `toEqual({ success: true, queued: 2 })`; `get-collection-status/handler.test.ts:136-140` asserts + `toEqual([{id,status},…])`. Adding a field fails both. (`get-document-status` uses + `toMatchObject`, so it is additive-safe.) + +11. **The ordering rationale is untested end to end.** The only ordering test pins `queued` before + `runner.enqueue` (`withQueuedNotification.test.ts:41-59`), not `queued` before `completed`. No + test anywhere registers `onQueued` and `onCompleted` together, and the sync-runner suites contain + no lifecycle references — so the exact scenario the docblock cites as the reason for the ordering + has no coverage. + +12. **The append path is the sharpest constraint on any run id.** `extendJob` unions a new request's + locales into an existing job's `input`. One row can therefore carry locale `fr` from run 1 and + locale `de` from run 2. **A single run id per job row cannot serve both.** Any run identity that + must survive to `onCompleted` has to be recorded *per target locale*, not per job. + +## Size check + +**The four issues are one task, and #123 is not part of it.** + +#107, #110, #106 and #103 are four symptoms of one absence: a translation run has no identity that +outlives the call that started it. They share `TaskRunner`, the lifecycle module and two handlers. + +#123 is a query-shape problem in the same two status handlers. It shares files with #103 but nothing +else — no concept, no contract. Including it would mix a contract change with a performance change in +one review. **Recommendation: leave #123 out, and sequence it immediately after**, so +`get-collection-status` is opened twice deliberately rather than by accident. + +## Problem statement + +A host that layers its own editorial process on top of the plugin cannot connect its records to the +plugin's work. When it asks for a translation it gets back only a count; when the plugin calls back it +gets only `(collection, id, targetLng, strategy)`; when a run is cancelled it is told nothing; and the +collection-wide status list identifies rows by job id alone. The host is forced to join on +`(collection, document, locale)` plus "is a run open", which obliges it to forbid concurrent runs on +the same pair and to disable retries so each run produces exactly one callback. The underlying gap is +that **a run has no name**: nothing the plugin hands out identifies the enqueue that caused the work, +and the one identifier that does exist internally — the job row id — is both discarded at creation and +too coarse, because a job row covers many locales and can be extended by a later, unrelated run. + +## Proposed scope + +**In scope** + +1. A run identity minted or surfaced by the plugin, carried from enqueue through `onQueued`, + `onCompleted` and `onFailed` for the locale it belongs to. +2. Returning that identity from the enqueue endpoint, keyed so it is unambiguous for a multi-locale + and multi-document request. +3. A cancellation signal to the host, for the one path that still exists (explicit cancel). +4. `get-collection-status` carrying document id and target locale. +5. Whatever `TaskRunner` / `TaskRunnerProvider` contract change the above requires, with an explicit + decision on third-party implementors. + +**Out of scope** + +- #123 (query bounds on the status endpoints) — sequenced after, see *Size check*. +- An `onCancelled` for a supersede path: that path does not exist (finding 1). +- Attempt numbering in callbacks: raised by #110, but it is a separable addition to + `TranslationTask`; include only if the design makes it free. +- Any change to what the plugin stores in the host's `payload-jobs` schema beyond the `input` JSON + column the plugin already owns. +- Building the consumer's editorial workflow — out of scope per epic #52. + +**Non-functional scan** + +- **Performance** — relevant. `get-collection-status` already loads a collection's whole job history + unfiltered (`handler.ts:35`). This task must not make that worse; #123 fixes it separately. +- **Security / access control** — relevant. The cancel handler takes ids from the request body and + currently reads no identity; adding a cancellation callback must not leak which jobs exist. Both + status handlers already filter by `visibleIds`. +- **Accessibility** — N/A. No admin UI change is required by any of the four. +- **i18n / localization** — N/A for the plugin's own surface; the data is locale-keyed but no + user-facing strings are added. +- **Observability** — relevant. `LifecycleNotifier.failed` logs when `onFailed` is absent + (`LifecycleNotifier.ts:34-40`); a new callback should follow the same rule or deliberately not. + +## Acceptance criteria (draft) + +1. **Given** an enqueue for one document and three target locales, **when** the host reads the + response, **then** it can tell, per target locale, which run was created — with no ambiguity when + one job row serves all three. *Check: integration test on all three adapters.* +2. **Given** a run identified in (1), **when** `onQueued`, `onCompleted` or `onFailed` fires for one of + those locales, **then** the callback carries the same identity as the response gave for that + locale. *Check: integration test with a real enqueue and a real handler run.* +3. **Given** a live job for document D holding locale `fr` from run 1, **when** a second enqueue adds + locale `de`, **then** the callback for `fr` still carries run 1's identity and the callback for + `de` carries run 2's. *Check: integration test exercising `extendJob`. This is finding 12 and is + the case most likely to be got wrong.* +4. **Given** the default retry configuration (3 attempts), **when** a task fails twice then succeeds, + **then** the host can distinguish the three callbacks from one another. *Check: unit test on the + handler path. Drop this criterion if attempt numbering is ruled out of scope.* +5. **Given** an explicit cancel of a job, **when** the host has registered a cancellation callback, + **then** it is told which document and target locale stopped, and the callback fires before the + row is deleted. *Check: unit test asserting call order against the delete.* +6. **Given** a cancel for an id that does not exist or is not this plugin's job, **then** no + cancellation callback fires and no error surfaces to the caller. *Check: unit test — negative + path; `ownJobs()` already narrows the delete (`PayloadJobsTaskRunner.ts:172-175`).* +7. **Given** a collection with jobs for several documents and locales, **when** + `get-collection-status` is called, **then** each row names its document and its target locale. + *Check: unit test; the exact-match assertion at `handler.test.ts:136-140` must be updated + deliberately.* +8. **Given** a host that supplies only `onCompleted` and no `onQueued`, **when** a translation runs, + **then** `onCompleted` still carries the run identity. *Check: unit test — this is finding 8, and + the current conditional decoration would silently fail it.* +9. **Given** a third-party `TaskRunnerProvider` compiled against 0.15.0, **when** the package is + upgraded, **then** either it still compiles, or the break is deliberate, recorded in + `docs/DEPRECATIONS.md` and shipped per that ledger's policy. *Check: review against finding 9.* +10. **Given** a host that registers no new callback and reads no new field, **when** it upgrades, + **then** its behaviour is unchanged. *Check: the existing suites, unmodified except where a + criterion above says otherwise.* + +## Open questions + +**Coverage scan** + +- *Failure & error states* — covered by AC6; plus: if the host's cancellation callback throws, the + existing `safe` wrapper swallows and logs (`LifecycleNotifier.ts:46-53`), and the new one should + match. No open question. +- *Empty / loading / zero states* — an enqueue that produces no tasks returns 400 before reaching the + runner (`enqueue-translation/handler.ts:72-76`). No open question. +- *Boundaries & limits* — a `select_all` enqueue fans out over every visible document + (`handler.ts:81-83`). Whether the response may carry one identity per document × locale at that + scale is **Q4** below. +- *Permissions & roles* — the cancel handler reads no identity at all today; see **Q3**. +- *Concurrency, idempotency, migration* — the append path is **Q1**; no schema migration is implied + as long as identity rides in the `input` JSON column the plugin already owns. + +**Restate check** + +- "Run" — I read this as *one call to enqueue*, which may cover many documents and many locales. The + consumer's wording in #110 ("the enqueue that produced the job") agrees. **Confirmed from input.** +- "Job" — in current code this is *one row covering one document and a list of locales*, not a + (document, locale) pair. The issues use it in the older sense. **Flagged; this is finding 2.** +- "Cancelled" — now means only *explicit cancel*, since supersession is gone. **Flagged; finding 1.** + +**Questions** + +1. **[blocking] Where does run identity live so it survives to `onCompleted`?** `extendJob` merges a + later run's locales into an earlier run's row (finding 12), so a single id per row is wrong. The + plugin owns the job's `input` JSON column, so a per-locale map needs no schema change — but it + changes `StoredWorkflowInput` and the normalizer. Matters because: a per-row id looks correct in + every single-run test and silently mislabels every appended locale in production. +2. **[blocking] Does `enqueue` start returning something, or does identity flow only through + callbacks and the response body?** `TaskRunner.enqueue` returns `Promise` and is the hidden + implemented contract (finding 9). Changing it breaks third-party providers who cannot name the + type. Matters because: this is the difference between an additive release and one that needs a + deprecation entry. +3. **[blocking] What may a cancellation callback say, given the handler holds only job ids?** + Reporting document and locale requires reading the row before deleting it — an extra query on a + path that currently does none. The alternative is a callback that reports only the id, which the + host cannot join to anything unless Q1/Q2 gave it that id at enqueue. Matters because: it decides + whether #106 is cheap or carries a read. +4. **[non-blocking] For a `select_all` enqueue, is a per-document × locale identity list in the + response acceptable, or should the response stay a count and identity be read elsewhere?** Matters + because: a bulk enqueue over a large collection would return a large body. +5. **[non-blocking] Is attempt numbering in scope?** #110 raises it; it is a separate field on + `TranslationTask` (safe to add, finding: read-only type) but it requires the task handler to see + the job row, which it currently does not (`PayloadJobsRunnerProvider.ts:149-162`). +6. **[non-blocking] Should `withQueuedNotification` stop being conditional on `onQueued`?** Finding 8 + means a host with only `onCompleted` is undecorated. Matters because: AC8 fails without a decision + here. +7. **[non-blocking] Does the sync runner participate?** It mints in-memory ids + (`SyncTaskRunner.ts:29`), creates no rows, and its `cancel` is a no-op (`:72-74`). Whether run + identity must be meaningful there, or may be best-effort, affects how much of the design is + runner-agnostic. + +## Risks & constraints + +- **`TaskRunner` is implemented by third parties but not exported.** Any change to it is a breaking + change to people who cannot name the type. `docs/DEPRECATIONS.md` sets the policy: deprecations ship + as `feat:`, removals batch into one major. +- **The plugin must not touch the host's `payload-jobs` schema.** Settled in + `2026-09-04-job-history-pruning.task.md` D1 and `2026-09-08-one-live-job-per-document.task.md` D5/D6. + The `input` JSON column is the plugin's; new columns are not. +- **`extendJob` writes the `input` column narrowly through the database adapter**, deliberately, to + avoid the read-modify-write clobbering measured in `2026-09-08-one-live-job-per-document.task.md` D2 + (3/120 on Postgres, 1/120 on MongoDB). Any new field in that column inherits that constraint. +- **Two exact-match tests pin the response shapes** (finding 10); changing them is intended, but it + must be deliberate rather than a surprise during implementation. +- **The ordering guarantee has no end-to-end coverage** (finding 11). If the design touches the + ordering, the missing test must be written *first*, or there is nothing to break. +- **Three adapters.** Anything stored in the job row needs SQLite, PostgreSQL and MongoDB evidence, + per the repository's standing practice. + +## Consistency self-check + +- Every in-scope item has at least one acceptance criterion: (1)→AC1-3, (2)→AC1, (3)→AC5-6, + (4)→AC7, (5)→AC9. +- No criterion contradicts an out-of-scope line. AC4 is explicitly conditional on Q5. +- Every criterion names its check. +- Coverage-scan items are either answered inline or carried into Questions 1-7. +- Traceability: AC1-3 trace to #107/#110; AC4 to #110's retry paragraph; AC5-6 to #106 as narrowed by + finding 1; AC7 to #103; AC8 to finding 8 — **`[inferred]`**, no issue asks for it, it rests on + `wireTranslateRunner.ts:87` and would otherwise make AC2 false for a real host configuration; + AC9-10 to findings 9 and 10 — **`[inferred]`**, standard compatibility obligations. + +## Readiness + +- Unresolved **[blocking]** questions: **3** (Q1, Q2, Q3). +- In-scope items with no acceptance criterion: **0**. + +**Needs answers first.** What is still soft: all three blocking questions are about *where identity +lives and who is allowed to see it*, not about effort. They are design decisions with a published +contract on one side and a storage constraint on the other. + +## Suggested next step + +**Ready for `/sp-architect`.** + +This is not a task-level placement decision. It introduces a concept the plugin does not have, threads +it through an interface third parties implement, changes what is persisted per job, and must survive a +merge path that unions two runs into one row. + +Likely architecture vectors: + +| Vector | Why | +|---|---| +| `contract` | `TaskRunner` / `TaskRunnerProvider` are implemented outside this package; two response shapes change | +| `data-model` | identity must be persisted in the job `input` column and survive `extendJob` | +| `evolution` | a breaking change to an implemented contract that cannot currently be named by implementors | +| `concurrency` | the append/merge path, retries, and the queued-before-completed ordering | + +--- + +# Clarifications (2026-10-05) — these supersede the fork above + +The owner resolved the design in conversation. **The "run identity" framing in the sections above is +withdrawn.** It merged two unrelated needs into one invented concept. What follows is the settled +shape; where it contradicts anything earlier in this document, this section wins. + +## C1 — No new identity concept. The key is `(job id, target locale)` + +Rejected: a plugin-minted run name. Rejected: an opaque host-supplied context stored in the job row. + +The second was rejected on the owner's objection, and the objection is the right one: carrying a +host's value would make the plugin a store for data it never reads, visible to anyone with access to +the jobs table, size-unbounded, and empty on the auto-translate path where no host supplied anything. + +The job row already has an id. Surfacing it costs no foreign storage and no new vocabulary. + +`(job id, target locale)` was checked against the three cases that break weaker keys: + +| case | result | +|---|---| +| a locale appended to a live job by a later request (`extendJob`) | **holds** — the appended locale appears once in that row, and the host was handed that row's id when it asked | +| two different requesters on the same document and locale | **holds** — `pickHost` refuses a host from a different requester (`planEnqueue.ts:68-69`), so they get separate rows and separate ids | +| a retried attempt | **does not hold** — same id, same locale. Attempts are distinguished by attempt number, a separate datum (see C4), not by this key | + +## C2 — Per-locale storage was accepted, and is now mostly unnecessary + +The owner accepted the per-locale principle (Q1, option A). Under C1 it mostly dissolves: the key is +built from the job's own id plus the locale the callback already carries, so no new per-locale map +needs persisting for correlation. Finding 12 still stands as the reason a per-*row* value would have +been wrong, and must stay true of anything per-row added later. + +## C3 — `TaskRunner.enqueue` changes, backwards-compatibly, with a deprecation + +The owner's decision: **do not look for a workaround.** Change the contract, keep current +implementors compiling, deprecate the old form, and let the removal batch into the next major per +`docs/DEPRECATIONS.md`. + +Mechanism: + +- Widen the return rather than replace it: `Promise → Promise`. An + existing third-party implementation returning `Promise` still satisfies the union and still + compiles. +- Annotate the `void` form `@deprecated`, with a ledger anchor, matching the precedent already on this + same interface — `findByCollection(slug, documentIds: Array)` carries + `@deprecated … Removed in the next major` (`TaskRunner.interface.ts:34-42`). +- Add the ledger entry. Policy is fixed: deprecations ship `feat:` (minor), removals batch into one + `feat!:` major (`docs/DEPRECATIONS.md:21-22`). +- **Degrade, do not fail.** When a runner returns nothing, the plugin has no ids: the enqueue response + omits them and callbacks carry none. A host that changed nothing observes no change. + +**Export `TaskRunner` from `src/index.ts`.** It is implemented by third parties and currently not +exported (finding 9), so a deprecation on it is unreadable by the people it is addressed to. Additive, +breaks nobody, and closes the same gap the `RequestScope` export closed in 0.14.0 +(`src/index.ts:60-66`). + +## C4 — The handler asks Payload for the job row + +To reach the callbacks, the job id must be available at execution time. It is not today: the +registered task handler takes only `{ req, input }` (`PayloadJobsRunnerProvider.ts:149-162`). Payload +passes the job when asked. This is entirely internal — no external contract moves. + +The same change makes the row's `totalTried` reachable, which is the only way to distinguish retry +attempts. #110's retry complaint is therefore closed by the same edit rather than by a separate +mechanism, and **attempt number moves from "out of scope unless free" to in scope**, because it is now +free. + +Adding fields to `TranslationTask` is safe: hosts only read it (it appears solely in callback +parameter position). Note the speed bump — `taskMapping.test.ts:21-27` and `:41-47` assert exact +shape with `toEqual` and must be updated deliberately. + +## C5 — The sync runner + +It creates no rows but already mints an in-memory id per task (`SyncTaskRunner.ts:29`) which it also +discards. It should return those ids so behaviour is consistent across runners; its `cancel` remains a +documented no-op. + +## C6 — Cancellation: assumption, not yet confirmed + +The one question the owner has not ruled on. **Proceeding on the recommended default:** read the rows +before deleting them, so the callback can name the document and target locale. One added query on a +path that currently performs none (`cancel/handler.ts:14-25` holds only ids; the delete is at +`PayloadJobsTaskRunner.ts:172-175`). The callback fires before the delete. + +Rejected default: reporting only the job id, which would make the signal useless to a host that had +not stored that id. Flag to the owner if this assumption is wrong — it is the only place in this +contract where implementation proceeds without an explicit decision. + +## Revised scope + +In scope, in dependency order: + +1. Export `TaskRunner` (additive). +2. Widen `enqueue`'s return with the deprecation and ledger entry (C3). +3. Both runners return the ids they already have (C5). +4. The enqueue endpoint returns them — **#107**. +5. The task handler receives the job row; job id and attempt number reach `TranslationTask` (C4) — + **#110**. +6. `onCancelled` on the explicit-cancel path, firing before the delete (C6) — **#106**, narrowed: + there is no supersede path to report (finding 1). +7. `get-collection-status` carries document id and target locale — **#103**, pure response mapping. + +Out of scope, unchanged: #123; any host-supplied context; any new column in `payload-jobs`; the +consumer's editorial workflow. + +## Revised acceptance criteria + +Supersedes the draft list above. + +1. An enqueue for one document and three locales returns, per locale, the job id that will run it. + *Integration test, three adapters.* +2. `onQueued`, `onCompleted` and `onFailed` for a locale carry the same job id that the enqueue + response gave for that locale. *Integration test.* +3. A locale appended by a second enqueue into a live job reports that job's id in both the response + and its callbacks, and the first run's locale is unaffected. *Integration test exercising + `extendJob` — the case most likely to be got wrong.* +4. With the default three attempts, a task that fails twice then succeeds produces three callbacks + the host can tell apart by attempt number. *Unit test.* +5. An explicit cancel fires `onCancelled` naming document and target locale, before the row is + deleted. *Unit test asserting order against the delete.* +6. A cancel for an unknown id, or one that is not this plugin's job, fires no callback and surfaces no + error. *Unit test — negative path.* +7. `get-collection-status` names each row's document and target locale. *Unit test; the exact-match + assertion at `get-collection-status/handler.test.ts:136-140` is updated deliberately.* +8. A host supplying only `onCompleted` and no `onQueued` still receives the job id. *Unit test — today + `wireTranslateRunner.ts:87` skips the decoration entirely in that configuration.* +9. A third-party runner that returns `Promise` still compiles, and the plugin degrades to + omitting ids rather than failing. *Type-level check plus a unit test with a void-returning runner.* +10. A host that registers nothing new and reads nothing new sees no behaviour change. *Existing + suites, unmodified except where a criterion above says otherwise.* + +## Revised readiness + +- Unresolved blocking questions: **0** (C6 proceeds on a stated assumption). +- In-scope items with no acceptance criterion: **0**. + +**Ready to implement.** The architecture question that routed this to a design pass is answered by C1 +and C3: no new concept, and the contract change is a deprecation of a shape that already has a +precedent on the same interface. diff --git a/packages/payload-plugin-translator/src/server/features/enqueue-translation/__tests__/handler.test.ts b/packages/payload-plugin-translator/src/server/features/enqueue-translation/__tests__/handler.test.ts index 0fb3ba560..21322f5df 100644 --- a/packages/payload-plugin-translator/src/server/features/enqueue-translation/__tests__/handler.test.ts +++ b/packages/payload-plugin-translator/src/server/features/enqueue-translation/__tests__/handler.test.ts @@ -156,7 +156,35 @@ describe("EnqueueTranslationHandler", () => { expect(response.status).toBe(200); const body = await response.json(); - expect(body.data).toEqual({ success: true, queued: 2 }); + expect( + body.data, + "a runner that answers with nothing leaves the response exactly as it was" + ).toEqual({ success: true, queued: 2 }); + }); + + it("names the job behind each queued locale when the runner says which", async () => { + (mockTaskRunner.enqueue as ReturnType).mockResolvedValue([ + { collectionSlug: "posts", collectionId: "doc-1", targetLng: "fr", jobId: "job-7" }, + { collectionSlug: "posts", collectionId: "doc-2", targetLng: "fr", jobId: "job-7" }, + ]); + const req = createLocalizedRequest({ + source_lng: "en", + target_lng: "fr", + collection_slug: "posts", + collection_id: ["doc-1", "doc-2"], + strategy: "overwrite", + }); + + const body = await (await handler.handle(req)).json(); + + expect(body.data).toEqual({ + success: true, + queued: 2, + jobs: [ + { collection_slug: "posts", collection_id: "doc-1", target_lng: "fr", job_id: "job-7" }, + { collection_slug: "posts", collection_id: "doc-2", target_lng: "fr", job_id: "job-7" }, + ], + }); }); it("calls runner.enqueue with correct tasks", async () => { diff --git a/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts b/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts index 26d4368f7..2765a1035 100644 --- a/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts +++ b/packages/payload-plugin-translator/src/server/features/enqueue-translation/handler.ts @@ -9,7 +9,7 @@ import { isCollectionAvailable, getAllCollectionIds } from "../_lib/collection-u import { Locales } from "../../../core/domain/locales/index.js"; -import { EnqueueInputSchema } from "./model.js"; +import { EnqueueInputSchema, toQueuedJobs } from "./model.js"; import type { EnqueueConfig } from "./model.js"; export class EnqueueTranslationHandler { @@ -90,11 +90,15 @@ export class EnqueueTranslationHandler { })) ); - await runner.enqueue( + const queued = await runner.enqueue( tasks, identityOf(req, authCollectionsOf(req.payload), req.payload.logger) ); - return ServerResponse.success({ success: true, queued: tasks.length }); + return ServerResponse.success({ + success: true, + queued: tasks.length, + ...(queued ? { jobs: toQueuedJobs(queued) } : {}), + }); } } diff --git a/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts b/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts index d1b0718ee..795bb0b16 100644 --- a/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts +++ b/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts @@ -1,6 +1,8 @@ import { z } from "zod"; import type { CollectionSlug } from "payload"; +import type { EnqueueResult } from "../../modules/task-runner/index.js"; + /** * Input validation schema */ @@ -22,3 +24,26 @@ export const EnqueueInputSchema = z.object({ export type EnqueueConfig = { availableCollections: Set; }; + +/** + * One queued locale, in the snake_case the other translation endpoints answer in. + * + * Absent when the runner predates returning ids (docs/DEPRECATIONS.md#enqueue-void-return); empty + * means the runner answered and could name nothing. `queued` counts what was accepted either way. + * + * @since 0.16.0 + */ +export type QueuedJob = { + collection_slug: string; + collection_id: string; + target_lng: string; + job_id: string; +}; + +export const toQueuedJobs = (queued: EnqueueResult): QueuedJob[] => + queued.map((entry) => ({ + collection_slug: entry.collectionSlug, + collection_id: entry.collectionId, + target_lng: entry.targetLng, + job_id: entry.jobId, + })); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunner.interface.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunner.interface.ts index 5e5a32600..5c5ec9081 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunner.interface.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunner.interface.ts @@ -1,6 +1,6 @@ import type { CollectionSlug } from "payload"; -import type { Task, TaskInput, RunResult } from "./types.js"; +import type { ID, Task, TaskInput, RunResult } from "./types.js"; import type { RequestScope } from "../../shared/payload/RequestScope.shapes.js"; /** @@ -14,18 +14,30 @@ import type { RequestScope } from "../../shared/payload/RequestScope.shapes.js"; export interface TaskRunner { /** * Queue translation tasks for execution. - * Implementation handles cancellation of existing tasks for the same documents. * * `scope` joins the reads and writes this makes to the caller's transaction; omit it outside one — * an HTTP route — and each operation opens its own. + * + * Answer with an {@link EnqueueResult}; the `void` form still compiles but is deprecated — see + * {@link LegacyVoidEnqueue}. + * + * @since 0.16.0 the return type widened. */ - enqueue(tasks: TaskInput[], scope?: RequestScope): Promise; + enqueue(tasks: TaskInput[], scope?: RequestScope): Promise; /** * Cancel tasks by IDs. */ cancel(taskIds: string[]): Promise; + /** + * Resolve job ids to the tasks they stand for — the cancel route holds nothing else. Optional: a + * runner that omits it works unchanged, and `onCancelled` never fires for it. + * + * @since 0.16.0 + */ + findByIds?(taskIds: string[]): Promise; + /** * Execute a task immediately. * Returns error status if task not found, already running, or completed. @@ -44,6 +56,29 @@ export interface TaskRunner { findByCollection(collectionSlug: CollectionSlug, filter?: TaskFilter): Promise; } +/** + * What `enqueue` answered before 0.16.0: nothing. Callers degrade to reporting no ids — the enqueue + * response omits them and the lifecycle callbacks carry none. + * + * @since 0.16.0 + * @deprecated Return an {@link EnqueueResult}. Removed in the next major. + * See docs/DEPRECATIONS.md#enqueue-void-return + */ +export type LegacyVoidEnqueue = void; + +/** + * What an enqueue created, one entry per requested target locale. Two locales of the same document + * carry the same `jobId` — one job row covers a document's whole locale list. + * + * @since 0.16.0 + */ +export type EnqueueResult = Array<{ + collectionSlug: CollectionSlug; + collectionId: ID; + targetLng: string; + jobId: string; +}>; + /** * How a {@link TaskRunner["findByCollection"]} call is narrowed. Each field says whether it reaches the * database or is applied in memory over everything the database returned. diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/__tests__/enqueueResult.conformance.types.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/__tests__/enqueueResult.conformance.types.ts new file mode 100644 index 000000000..b49df200f --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/__tests__/enqueueResult.conformance.types.ts @@ -0,0 +1,29 @@ +import type { EnqueueResult, TaskRunner } from "../TaskRunner.interface.js"; +import type { Task, TaskInput } from "../types.js"; + +/** + * A compile-only fixture, enforced by `check-types` rather than by a running test: a runner that + * answers and one that stays silent must both satisfy {@link TaskRunner}, or a third-party + * implementation breaks on upgrade. Narrowing `enqueue`'s return makes this file fail to compile. + */ + +const reads = { + cancel: async (): Promise => undefined, + run: async () => ({ success: false, error: "not_found" }) as const, + findByCollection: async (): Promise => [], +}; + +/** Pre-0.16.0: answers with nothing, and omits the optional by-id read entirely. */ +export const silent: TaskRunner = { ...reads, enqueue: async () => undefined }; + +/** Post-0.16.0: one entry per requested locale, several locales sharing one job row. */ +export const answering: TaskRunner = { + ...reads, + enqueue: async (tasks: TaskInput[]): Promise => + tasks.map((task) => ({ + collectionSlug: task.collectionSlug, + collectionId: task.collectionId, + targetLng: task.targetLng, + jobId: "job-77", + })), +}; diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/index.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/index.ts index 7863bc2cf..034455abc 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/index.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/index.ts @@ -7,5 +7,5 @@ export type { Task, TaskStatus } from "./types.js"; export { createPayloadJobsRunner } from "./payload-jobs-runner/index.js"; export type { PayloadJobsRunnerOptions } from "./payload-jobs-runner/index.js"; export { createSyncRunner } from "./sync-runner/index.js"; -export type { TaskFilter, TaskRunner } from "./TaskRunner.interface.js"; +export type { EnqueueResult, TaskFilter, TaskRunner } from "./TaskRunner.interface.js"; export { toTaskFilter } from "./toTaskFilter.js"; diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts index 25ca60a21..ffe441b76 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsTaskRunner.ts @@ -1,10 +1,10 @@ import type { Payload, Where, CollectionSlug } from "payload"; -import type { TaskFilter, TaskRunner } from "../TaskRunner.interface.js"; +import type { EnqueueResult, TaskFilter, TaskRunner } from "../TaskRunner.interface.js"; import { toTaskFilter } from "../toTaskFilter.js"; import type { Task, TaskInput, RunResult } from "../types.js"; import type { PayloadJobsRunnerConfig, PayloadJob, StoredWorkflowInput } from "./types.js"; -import { normalizeJobLocales } from "./normalizeJob.js"; +import { canStillRun, normalizeJobLocales } from "./normalizeJob.js"; import { planEnqueue } from "./planEnqueue.js"; import type { RequestShape } from "./planEnqueue.js"; import { readCollectionRef } from "./readCollectionRef.js"; @@ -14,13 +14,32 @@ import { freshReq } from "../../../shared/payload/RequestScope.shapes.js"; const APPEND_ATTEMPTS = 2; const ENQUEUE_CONCURRENCY = 10; +/** + * `hostGone`: the row stopped being runnable while we were writing to it — finished, gave up, was + * cancelled or vanished — so none of the request's locales may be claimed for it. + */ +type Extension = { undelivered: string[]; hostGone: boolean }; + +const NO_EXTENSION: Extension = { undelivered: [], hostGone: false }; + type QueueWorkflow = (args: { workflow: string; queue: string; waitUntil?: Date; input: StoredWorkflowInput; req?: { transactionID?: string | number }; -}) => Promise; +}) => Promise<{ id?: unknown } | undefined>; + +const jobIdOf = (queued: { id?: unknown } | undefined): string | null => + typeof queued?.id === "string" || typeof queued?.id === "number" ? String(queued.id) : null; + +const claim = (request: RequestShape, targetLngs: string[], jobId: string): EnqueueResult => + targetLngs.map((targetLng) => ({ + collectionSlug: request.collectionSlug as CollectionSlug, + collectionId: request.collectionId, + targetLng, + jobId, + })); function requestShape(task: TaskInput, scope: RequestScope): RequestShape { return { @@ -48,7 +67,7 @@ export class PayloadJobsTaskRunner implements TaskRunner { private readonly config: PayloadJobsRunnerConfig ) {} - async enqueue(tasks: TaskInput[], scope: RequestScope = {}): Promise { + async enqueue(tasks: TaskInput[], scope: RequestScope = {}): Promise { const byRequest = new Map(); for (const task of tasks) { const key = requestKey(task, scope); @@ -57,7 +76,10 @@ export class PayloadJobsTaskRunner implements TaskRunner { byRequest.set(key, group); } - const live = await this.findRawJobs({ completedAt: { exists: false } }, scope); + const live = await this.findRawJobs( + { and: [{ completedAt: { exists: false } }, { hasError: { not_equals: true } }] }, + scope + ); const liveByDocument = new Map(); for (const job of live) { const { collectionSlug, collectionId } = readCollectionRef(job.input); @@ -67,13 +89,16 @@ export class PayloadJobsTaskRunner implements TaskRunner { const exclusiveQueue = Boolean(this.payload.config.jobs?.enableConcurrencyControl); const groups = [...byRequest.values()]; + const claimed: EnqueueResult = []; for (let i = 0; i < groups.length; i += ENQUEUE_CONCURRENCY) { - await Promise.all( + const served = await Promise.all( groups .slice(i, i + ENQUEUE_CONCURRENCY) .map((group) => this.serve(group, liveByDocument, exclusiveQueue, scope)) ); + for (const entries of served) claimed.push(...entries); } + return claimed; } private async serve( @@ -81,7 +106,7 @@ export class PayloadJobsTaskRunner implements TaskRunner { liveByDocument: Map, exclusiveQueue: boolean, scope: RequestScope - ): Promise { + ): Promise { const [first] = group; const request = requestShape(first, scope); const plan = planEnqueue({ @@ -91,11 +116,25 @@ export class PayloadJobsTaskRunner implements TaskRunner { exclusiveQueue, }); - const undelivered = plan.host - ? await this.extendJob(plan.host, plan.append, first.waitUntil, scope) - : []; - const queue = [...plan.queue, ...undelivered]; - if (queue.length > 0) await this.queueWorkflow(request, queue, first.waitUntil, scope); + const extension = + plan.host && plan.extend + ? await this.extendJob(plan.host, plan.append, first.waitUntil, scope) + : NO_EXTENSION; + const requested = [...new Set(group.map((task) => task.targetLng))]; + // Host gone: nothing of the plan survives it, so every requested locale is queued afresh — + // including any the dead host had already translated. + const queue = extension.hostGone ? requested : [...plan.queue, ...extension.undelivered]; + const goingToANewJob = new Set(queue); + const alreadyOnHost = requested.filter((locale) => !goingToANewJob.has(locale)); + const claimedFromHost = + plan.host && !extension.hostGone ? claim(request, alreadyOnHost, String(plan.host.id)) : []; + + if (queue.length === 0) return claimedFromHost; + + const queuedId = await this.queueWorkflow(request, queue, first.waitUntil, scope); + return queuedId === null + ? claimedFromHost + : [...claimedFromHost, ...claim(request, queue, queuedId)]; } private async extendJob( @@ -103,14 +142,14 @@ export class PayloadJobsTaskRunner implements TaskRunner { locales: string[], waitUntil: Date | undefined, scope: RequestScope - ): Promise { + ): Promise { let current = job; let undelivered = locales; for (let attempt = 0; attempt < APPEND_ATTEMPTS; attempt++) { const listed = current.input?.target_lngs ?? []; const missing = locales.filter((locale) => !listed.includes(locale)); const debounce = waitUntil && !current.processing ? waitUntil.toISOString() : undefined; - if (missing.length === 0 && !debounce) return []; + if (missing.length === 0 && !debounce) return NO_EXTENSION; await this.payload.db.updateOne({ req: freshReq(scope), @@ -123,15 +162,15 @@ export class PayloadJobsTaskRunner implements TaskRunner { returning: false, }); - const reread = await this.findJobById(job.id, scope); - if (!reread || reread.completedAt) return locales; + const reread = await this.findJobById(String(job.id), scope); + if (!reread || !canStillRun(reread)) return { undelivered: [], hostGone: true }; current = reread; const stored = new Set(current.input?.target_lngs); undelivered = locales.filter((locale) => !stored.has(locale)); - if (undelivered.length === 0) return []; + if (undelivered.length === 0) return NO_EXTENSION; } - return undelivered; + return { undelivered, hostGone: false }; } private async queueWorkflow( @@ -139,7 +178,7 @@ export class PayloadJobsTaskRunner implements TaskRunner { targetLngs: string[], waitUntil: Date | undefined, scope: RequestScope - ): Promise { + ): Promise { const input: StoredWorkflowInput = { collection_slug: request.collectionSlug, collection_id: request.collectionId, @@ -152,13 +191,15 @@ export class PayloadJobsTaskRunner implements TaskRunner { }; const queueJob = this.payload.jobs.queue as unknown as QueueWorkflow; - await queueJob({ - workflow: this.config.workflowName, - queue: this.config.queueName, - waitUntil, - input, - req: freshReq(scope), - }); + return jobIdOf( + await queueJob({ + workflow: this.config.workflowName, + queue: this.config.queueName, + waitUntil, + input, + req: freshReq(scope), + }) + ); } async cancel(taskIds: string[]): Promise { @@ -263,6 +304,12 @@ export class PayloadJobsTaskRunner implements TaskRunner { .flatMap(normalizeJobLocales); } + async findByIds(taskIds: string[]): Promise { + if (taskIds.length === 0) return []; + const jobs = await this.findRawJobs({ id: { in: taskIds } }); + return jobs.flatMap(normalizeJobLocales); + } + private ownJobs(): Where { return { or: [ diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.enqueueResult.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.enqueueResult.test.ts new file mode 100644 index 000000000..b379ff508 --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.enqueueResult.test.ts @@ -0,0 +1,215 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import type { Payload, CollectionSlug } from "payload"; + +import { PayloadJobsTaskRunner } from "../PayloadJobsTaskRunner.js"; +import type { PayloadJobsRunnerConfig, PayloadJob } from "../types.js"; +import type { TaskInput } from "../../types.js"; + +describe("PayloadJobsTaskRunner.enqueue answers with the jobs it created or extended", () => { + let mockPayload: { + find: ReturnType; + delete: ReturnType; + update: ReturnType; + db: { updateOne: ReturnType }; + config: { jobs?: { enableConcurrencyControl?: boolean } }; + jobs: { queue: ReturnType; cancel: ReturnType }; + }; + let runner: PayloadJobsTaskRunner; + + const config: PayloadJobsRunnerConfig = { + taskName: "translate_document", + workflowName: "translate_document_locales", + queueName: "translations", + jobsCollection: "payload-jobs", + autoRun: { cron: "* * * * *", limit: 50 }, + staleJobTimeoutMs: 300_000, + }; + + const input = (overrides: Partial = {}): TaskInput => ({ + collectionSlug: "posts" as CollectionSlug, + collectionId: "doc-123", + sourceLng: "en", + targetLng: "de", + strategy: "overwrite", + publishOnTranslation: false, + ...overrides, + }); + + const planSeesThenRereadSees = (atPlan: PayloadJob, atReread: PayloadJob) => + mockPayload.find + .mockResolvedValueOnce({ docs: [atPlan] }) + .mockResolvedValue({ docs: [atReread] }); + + const liveJob = (targetLngs: string[]) => { + const stored = { + collection_slug: "posts", + collection_id: "doc-123", + source_lng: "en", + target_lngs: targetLngs, + strategy: "overwrite", + publish_on_translation: false, + requester_id: null, + requester_collection: null, + }; + const job: PayloadJob = { + id: "job-live", + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + processing: false, + input: stored, + }; + return { job, absorb: (locale: string) => stored.target_lngs.push(locale) }; + }; + + beforeEach(() => { + mockPayload = { + find: vi.fn().mockResolvedValue({ docs: [] }), + delete: vi.fn().mockResolvedValue(undefined), + update: vi.fn().mockResolvedValue({ docs: [] }), + db: { updateOne: vi.fn().mockResolvedValue(undefined) }, + config: { jobs: {} }, + jobs: { + queue: vi.fn().mockResolvedValue({ id: "job-new" }), + cancel: vi.fn().mockResolvedValue(undefined), + }, + }; + runner = new PayloadJobsTaskRunner(mockPayload as unknown as Payload, config); + }); + + it("names the newly queued job for every locale it covers", async () => { + const result = await runner.enqueue([input({ targetLng: "de" }), input({ targetLng: "fr" })]); + + expect(result).toEqual([ + { collectionSlug: "posts", collectionId: "doc-123", targetLng: "de", jobId: "job-new" }, + { collectionSlug: "posts", collectionId: "doc-123", targetLng: "fr", jobId: "job-new" }, + ]); + }); + + it("names the LIVE job for a locale that joined it, not a new one", async () => { + const { job, absorb } = liveJob(["de"]); + mockPayload.find.mockResolvedValue({ docs: [job] }); + mockPayload.db.updateOne.mockImplementation(async () => absorb("fr")); + + const result = await runner.enqueue([input({ targetLng: "fr" })]); + + expect( + result, + "the caller asked for fr and it went into the live job — reporting a new id would be a lie" + ).toEqual([ + { collectionSlug: "posts", collectionId: "doc-123", targetLng: "fr", jobId: "job-live" }, + ]); + expect(mockPayload.jobs.queue, "nothing new was queued").not.toHaveBeenCalled(); + }); + + it("names the live job for a locale it ALREADY covers, since that job is what will run it", async () => { + const { job } = liveJob(["de"]); + mockPayload.find.mockResolvedValue({ docs: [job] }); + + const result = await runner.enqueue([input({ targetLng: "de" })]); + + expect( + result, + "nothing was appended and nothing was queued, but de is still going to be translated — by job-live" + ).toEqual([ + { collectionSlug: "posts", collectionId: "doc-123", targetLng: "de", jobId: "job-live" }, + ]); + expect(mockPayload.jobs.queue).not.toHaveBeenCalled(); + expect(mockPayload.db.updateOne, "there is nothing to append").not.toHaveBeenCalled(); + }); + + it("names the running job under an exclusive queue, where it cannot be extended but still runs the locale", async () => { + const { job } = liveJob(["de"]); + job.processing = true; + mockPayload.config.jobs = { enableConcurrencyControl: true }; + mockPayload.find.mockResolvedValue({ docs: [job] }); + + const result = await runner.enqueue([input({ targetLng: "de" })]); + + expect( + result, + "an exclusive queue stops us extending the job, not the job from translating de" + ).toEqual([ + { collectionSlug: "posts", collectionId: "doc-123", targetLng: "de", jobId: "job-live" }, + ]); + expect( + mockPayload.db.updateOne, + "and nothing may be written to a running job" + ).not.toHaveBeenCalled(); + }); + + it("answers once per locale even if the caller asked for one twice", async () => { + const { job } = liveJob(["de"]); + mockPayload.find.mockResolvedValue({ docs: [job] }); + + const result = await runner.enqueue([input({ targetLng: "de" }), input({ targetLng: "de" })]); + + expect(result, "planEnqueue dedupes; the answer must agree with it").toHaveLength(1); + }); + + it("reports a numeric row id as a string, like the queued path does", async () => { + const { job, absorb } = liveJob(["de"]); + job.id = 1; + mockPayload.find.mockResolvedValue({ docs: [job] }); + mockPayload.db.updateOne.mockImplementation(async () => absorb("fr")); + + const result = await runner.enqueue([input({ targetLng: "fr" })]); + + expect( + result?.[0]?.jobId, + "an autoincrement id must not leak as a number where the type promises a string" + ).toBe("1"); + }); + + it("splits the answer when some locales join the live job and the rest need a new one", async () => { + const { job, absorb } = liveJob(["de"]); + mockPayload.find.mockResolvedValue({ docs: [job] }); + mockPayload.db.updateOne.mockImplementation(async () => absorb("fr")); + + const result = await runner.enqueue([ + input({ targetLng: "fr" }), + input({ targetLng: "it", strategy: "skip_existing" }), + ]); + + const byLocale = Object.fromEntries(result?.map((e) => [e.targetLng, e.jobId]) ?? []); + expect(byLocale.fr).toBe("job-live"); + expect(byLocale.it, "a different request shape cannot share the live job").toBe("job-new"); + }); + + it("queues everything afresh when the host finishes between the plan and the write", async () => { + const { job, absorb } = liveJob(["de"]); + const finished = { ...job, completedAt: "2026-01-01T00:01:00Z" }; + planSeesThenRereadSees(job, finished); + mockPayload.db.updateOne.mockImplementation(async () => absorb("fr")); + + const result = await runner.enqueue([input({ targetLng: "de" }), input({ targetLng: "fr" })]); + + const byLocale = Object.fromEntries(result?.map((e) => [e.targetLng, e.jobId]) ?? []); + expect( + byLocale, + "the host is finished and will run neither locale again — naming it for de would promise a re-translation nothing performs" + ).toEqual({ de: "job-new", fr: "job-new" }); + }); + + it("queues everything afresh when the host gives up between the plan and the write", async () => { + const { job, absorb } = liveJob(["de"]); + const exhausted = { ...job, hasError: true, input: job.input }; + planSeesThenRereadSees(job, exhausted); + mockPayload.db.updateOne.mockImplementation(async () => absorb("fr")); + + const result = await runner.enqueue([input({ targetLng: "de" }), input({ targetLng: "fr" })]); + + const byLocale = Object.fromEntries(result?.map((e) => [e.targetLng, e.jobId]) ?? []); + expect( + byLocale, + "a job that ran out of retries is as unreachable as a finished one — Payload's picker takes neither" + ).toEqual({ de: "job-new", fr: "job-new" }); + }); + + it("answers with nothing rather than a wrong id when the queue call reveals none", async () => { + mockPayload.jobs.queue.mockResolvedValue(undefined); + + const result = await runner.enqueue([input({ targetLng: "de" })]); + + expect(result, "a missing id must not be invented").toEqual([]); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.test.ts index 7e3439d09..e2e24b2af 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.test.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.test.ts @@ -92,11 +92,14 @@ describe("PayloadJobsTaskRunner", () => { }); describe("enqueue", () => { - it("reads only unfinished jobs when planning an enqueue", async () => { + it("reads only jobs that can still run when planning an enqueue", async () => { await runner.enqueue([createInput()]); const whereArg = mockPayload.find.mock.calls[0][0].where; - expect(whereArg).toEqual({ + expect( + whereArg, + "a row that ran out of retries is unfinished forever, so asking for it loads rows no plan can use" + ).toEqual({ and: [ { or: [ @@ -104,7 +107,9 @@ describe("PayloadJobsTaskRunner", () => { { taskSlug: { equals: "translate_document" } }, ], }, - { completedAt: { exists: false } }, + { + and: [{ completedAt: { exists: false } }, { hasError: { not_equals: true } }], + }, ], }); }); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/SyncTaskRunner.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/SyncTaskRunner.ts index 3844835fa..4f6a3e0fc 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/SyncTaskRunner.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/SyncTaskRunner.ts @@ -1,6 +1,6 @@ import type { Payload, CollectionSlug } from "payload"; -import type { TaskFilter, TaskRunner } from "../TaskRunner.interface.js"; +import type { EnqueueResult, TaskFilter, TaskRunner } from "../TaskRunner.interface.js"; import { toTaskFilter } from "../toTaskFilter.js"; import type { TaskHandler } from "../TaskRunnerProvider.interface.js"; import type { Task, TaskInput, RunResult, ID } from "../types.js"; @@ -20,7 +20,9 @@ export class SyncTaskRunner implements TaskRunner { this.tasks = tasks; } - async enqueue(inputs: TaskInput[], scope: RequestScope = {}): Promise { + async enqueue(inputs: TaskInput[], scope: RequestScope = {}): Promise { + const queued: EnqueueResult = []; + for (const input of inputs) { const key = this.getKey(input.collectionSlug, input.collectionId, input.targetLng); const now = new Date().toISOString(); @@ -35,6 +37,12 @@ export class SyncTaskRunner implements TaskRunner { }; this.tasks.set(key, task); + queued.push({ + collectionSlug: input.collectionSlug, + collectionId: input.collectionId, + targetLng: input.targetLng, + jobId: task.id, + }); const markEvictable = (status: "completed" | "failed", error?: Task["error"]) => { const at = new Date().toISOString(); @@ -56,6 +64,7 @@ export class SyncTaskRunner implements TaskRunner { targetLng: input.targetLng, strategy: input.strategy, publishOnTranslation: input.publishOnTranslation, + jobId: task.id, }, scope ); @@ -67,10 +76,13 @@ export class SyncTaskRunner implements TaskRunner { }) ); } + + return queued; } + /** No-op: a synchronous task has already run by the time anyone could cancel it. */ async cancel(_taskIds: string[]): Promise { - // No-op: synchronous tasks execute immediately and cannot be cancelled + return undefined; } async run(_taskId: string): Promise { diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.enqueueResult.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.enqueueResult.test.ts new file mode 100644 index 000000000..fc5103fdc --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.enqueueResult.test.ts @@ -0,0 +1,95 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Payload } from "payload"; + +import { SyncTaskRunner } from "../SyncTaskRunner.js"; +import type { TaskInput, Task } from "../../types.js"; +import type { TaskHandlerInput } from "../../TaskRunnerProvider.interface.js"; +import { LazyMap } from "../../../../shared/utils/index.js"; + +const makeRunner = () => { + const tasks = new LazyMap({ + maxSize: 10, + ttlMs: 60_000, + isRemovable: () => false, + getTimestamp: () => Date.now(), + }); + const seen: TaskHandlerInput[] = []; + const handler = vi.fn(async (_payload: Payload, handlerInput: TaskHandlerInput) => { + seen.push(handlerInput); + }); + return { + runner: new SyncTaskRunner({} as Payload, handler, tasks), + seen, + }; +}; + +const input = (collectionId: string, targetLng: string): TaskInput => ({ + collectionSlug: "posts", + collectionId, + sourceLng: "en", + targetLng, + strategy: "overwrite", + publishOnTranslation: false, +}); + +describe("SyncTaskRunner.enqueue answers with what it ran", () => { + it("names one entry per task", async () => { + const { runner } = makeRunner(); + + const result = await runner.enqueue([input("doc-1", "de"), input("doc-1", "fr")]); + + expect(result).toHaveLength(2); + expect(result?.map((entry) => entry.targetLng)).toEqual(["de", "fr"]); + expect(result?.every((entry) => entry.jobId.length > 0)).toBe(true); + }); + + it("names the same ids its own status read reports, or a caller could not join the two", async () => { + const { runner } = makeRunner(); + + const result = await runner.enqueue([input("doc-1", "de"), input("doc-2", "fr")]); + const found = await runner.findByCollection("posts"); + + expect(new Set(result?.map((entry) => entry.jobId))).toEqual( + new Set(found.map((task) => task.id)) + ); + }); + + it("hands the handler the same id it answered with, so the two never disagree", async () => { + const { runner, seen } = makeRunner(); + + const result = await runner.enqueue([input("doc-1", "de")]); + + expect(seen[0]?.jobId).toBe(result?.[0]?.jobId); + expect(seen[0]?.jobId).toBeTruthy(); + }); + + it("carries the document each entry belongs to", async () => { + const { runner } = makeRunner(); + + const result = await runner.enqueue([input("doc-1", "de"), input("doc-2", "de")]); + + expect(result?.map((entry) => entry.collectionId)).toEqual(["doc-1", "doc-2"]); + expect(result?.every((entry) => entry.collectionSlug === "posts")).toBe(true); + }); + + it("still answers for a task whose handler threw, since the work was queued either way", async () => { + const tasks = new LazyMap({ + maxSize: 10, + ttlMs: 60_000, + isRemovable: () => false, + getTimestamp: () => Date.now(), + }); + const runner = new SyncTaskRunner( + {} as Payload, + vi.fn(async () => { + throw new Error("provider down"); + }), + tasks + ); + + const result = await runner.enqueue([input("doc-1", "de")]); + + expect(result).toHaveLength(1); + expect(result?.[0]?.jobId.length).toBeGreaterThan(0); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.test.ts index 0eab22585..4ad5d9f76 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.test.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/sync-runner/__tests__/SyncTaskRunner.test.ts @@ -36,7 +36,7 @@ describe("SyncTaskRunner", () => { describe("enqueue", () => { it("executes handler immediately", async () => { const input = createInput(); - await runner.enqueue([input]); + const result = await runner.enqueue([input]); expect(mockHandler).toHaveBeenCalledWith( mockPayload, @@ -47,6 +47,7 @@ describe("SyncTaskRunner", () => { targetLng: "de", strategy: "overwrite", publishOnTranslation: false, + jobId: result?.[0]?.jobId, }, {} ); @@ -113,11 +114,15 @@ describe("SyncTaskRunner", () => { mockHandler = vi.fn().mockRejectedValueOnce(fromPayload).mockResolvedValueOnce(undefined); runner = new SyncTaskRunner(mockPayload, mockHandler, tasks); - await expect( - runner.enqueue([createInput({ targetLng: "de" }), createInput({ targetLng: "fr" })]), - "nothing of this caller's was rolled back, so one refused locale must not cancel the rest" - ).resolves.toBeUndefined(); + const queued = await runner.enqueue([ + createInput({ targetLng: "de" }), + createInput({ targetLng: "fr" }), + ]); + expect( + queued?.map((entry) => entry.targetLng), + "nothing of this caller's was rolled back, so one refused locale must not cancel the rest" + ).toEqual(["de", "fr"]); expect(mockHandler).toHaveBeenCalledTimes(2); expect(tasks.get("posts:doc-123:de")?.status).toBe("failed"); expect(tasks.get("posts:doc-123:fr")?.status).toBe("completed"); @@ -130,7 +135,7 @@ describe("SyncTaskRunner", () => { await expect( runner.enqueue([createInput()], { transactionID: "tx-1" }), "a provider failure ran no Payload operation, so the editor's save is still intact" - ).resolves.toBeUndefined(); + ).resolves.toBeDefined(); expect(tasks.get("posts:doc-123:de")?.status).toBe("failed"); }); From 6e8cf08ffab0a3d9b691fc27bba84b0f5822038a Mon Sep 17 00:00:00 2001 From: Siarhei Date: Tue, 6 Oct 2026 14:25:41 +0200 Subject: [PATCH 3/6] feat(translator): carry the job id and retry attempt into the lifecycle callbacks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #110. The task handler now asks Payload for the job row, so a callback can name the job that caused it and which attempt it is on. Together with the target locale the callback already carried, that is what lets a host join a callback to the request behind it. The attempt is counted per locale, not per job: one row carries every locale of a document, so the row's own counter would report a locale's first run as the job's third. Payload writes each counter only after a run finishes, hence the `+ 1`. `onQueued` carries no job id — the decorator fires it before delegating, which is what keeps a synchronous runner's `completed` from arriving first, and at that moment no job exists. --- .../translator/lifecycle-job-id.int.test.ts | 88 +++++++++++++++++ .../translator/retry-attempts.int.test.ts | 98 +++++++++++++++++++ .../runner-predating-the-fields.int.test.ts | 68 +++++++++++++ .../payload-plugin-translator/src/index.ts | 8 +- .../payload-plugin-translator/src/plugin.ts | 6 +- .../translate-document/wireTranslateRunner.ts | 5 +- .../src/server/modules/lifecycle/index.ts | 2 +- .../modules/lifecycle/taskMapping.test.ts | 45 +++++++++ .../server/modules/lifecycle/taskMapping.ts | 13 ++- .../src/server/modules/lifecycle/types.ts | 25 +++++ .../TaskRunnerProvider.interface.ts | 14 +++ .../PayloadJobsRunnerProvider.ts | 48 ++++++++- .../__tests__/attemptOf.test.ts | 45 +++++++++ 13 files changed, 456 insertions(+), 9 deletions(-) create mode 100644 apps/dev/src/integration/translator/lifecycle-job-id.int.test.ts create mode 100644 apps/dev/src/integration/translator/retry-attempts.int.test.ts create mode 100644 apps/dev/src/integration/translator/runner-predating-the-fields.int.test.ts create mode 100644 packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/attemptOf.test.ts diff --git a/apps/dev/src/integration/translator/lifecycle-job-id.int.test.ts b/apps/dev/src/integration/translator/lifecycle-job-id.int.test.ts new file mode 100644 index 000000000..6f914e5b8 --- /dev/null +++ b/apps/dev/src/integration/translator/lifecycle-job-id.int.test.ts @@ -0,0 +1,88 @@ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +type Seen = { targetLng: string; jobId?: string; attempt?: number }; + +const onQueued = vi.fn<(task: Seen) => void>(); +const onCompleted = vi.fn<(task: Seen) => void>(); +const onFailed = vi.fn<(task: Seen) => void>(); + +let ctx: TestPayload; + +const create = async (title: string): Promise => { + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title, note: "A note" }, + }); + return String(made.id); +}; + +const enqueue = (id: string, targetLng: string | string[]) => + callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: targetLng, + collection_slug: "docs", + collection_id: [id], + strategy: "overwrite", + publish_on_translation: true, + }, + }); + +const reportedJobs = (res: { data: unknown }) => + (res.data as { data: { jobs: { target_lng: string; job_id: string }[] } }).data.jobs; + +describe("a lifecycle callback names the job that ran it", () => { + beforeAll(async () => { + ctx = await bootTestPayload({ + runner: createPayloadJobsRunner({ autoRun: false }), + lifecycle: { onQueued, onCompleted, onFailed }, + }); + }); + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("gives onCompleted the same job id the enqueue response reported for that locale", async () => { + onCompleted.mockClear(); + const id = await create("Matched"); + + const reported = reportedJobs(await enqueue(id, ["de", "fr"])); + await ctx.payload.jobs.run({ queue: "translations" }); + + const byLocale = Object.fromEntries( + onCompleted.mock.calls.map(([task]) => [task.targetLng, task.jobId]) + ); + for (const job of reported) { + expect( + byLocale[job.target_lng], + `the callback for ${job.target_lng} must name the job the response promised` + ).toBe(job.job_id); + } + }); + + it("gives onCompleted the attempt it ran on", async () => { + onCompleted.mockClear(); + const id = await create("Attempted"); + + await enqueue(id, "de"); + await ctx.payload.jobs.run({ queue: "translations" }); + + expect(onCompleted.mock.calls[0]?.[0].attempt, "a first run is attempt 1, not 0").toBe(1); + }); + + it("does not pretend onQueued knows a job id — the job does not exist yet", async () => { + onQueued.mockClear(); + const id = await create("Too early"); + + await enqueue(id, "de"); + + expect(onQueued).toHaveBeenCalled(); + expect(onQueued.mock.calls[0]?.[0].jobId).toBeUndefined(); + }); +}); diff --git a/apps/dev/src/integration/translator/retry-attempts.int.test.ts b/apps/dev/src/integration/translator/retry-attempts.int.test.ts new file mode 100644 index 000000000..f20c754d9 --- /dev/null +++ b/apps/dev/src/integration/translator/retry-attempts.int.test.ts @@ -0,0 +1,98 @@ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +const ATTEMPTS = 3; + +type Seen = { targetLng: string; jobId?: string; attempt?: number }; + +const onCompleted = vi.fn<(task: Seen) => void>(); +const onFailed = vi.fn<(task: Seen) => void>(); + +const attemptsOf = (fn: typeof onCompleted, targetLng: string) => + fn.mock.calls.filter(([task]) => task.targetLng === targetLng).map(([task]) => task.attempt); + +let ctx: TestPayload; +let deCalls = 0; +let reportedJobIds: string[] = []; + +describe("one job whose first locale fails twice and whose second locale never fails", () => { + beforeAll(async () => { + ctx = await bootTestPayload({ + runner: createPayloadJobsRunner({ + autoRun: false, + // The shipped default backs off exponentially from five seconds — a minute of waiting for a + // spec about attempt counts. + retries: { attempts: ATTEMPTS, backoff: { type: "fixed", delay: 0 } }, + }), + lifecycle: { onCompleted, onFailed }, + onTranslate: (targetLng) => { + if (targetLng !== "de") return; + deCalls += 1; + if (deCalls <= 2) throw new Error("provider down"); + }, + }); + + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title: "Flaky", note: "A note" }, + }); + + const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: ["de", "fr"], + collection_slug: "docs", + collection_id: [String(made.id)], + strategy: "overwrite", + publish_on_translation: true, + }, + }); + reportedJobIds = (res.data as { data: { jobs: { job_id: string }[] } }).data.jobs.map( + (job) => job.job_id + ); + + for (let run = 0; run < ATTEMPTS; run++) { + await ctx.payload.jobs.run({ queue: "translations" }); + } + }); + + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("reports attempts 1 and 2 on the failures", () => { + expect( + attemptsOf(onFailed, "de"), + "without the attempt number these two callbacks are indistinguishable" + ).toEqual([1, 2]); + }); + + it("reports attempt 3 on the run that finally worked", () => { + expect(attemptsOf(onCompleted, "de").at(-1)).toBe(3); + }); + + it("reports the second locale's first run as attempt 1, though the job is on its third pass", () => { + expect( + attemptsOf(onCompleted, "fr"), + "fr waited behind de and then succeeded outright — the job's two failed passes were not its" + ).toEqual([1]); + }); + + it("names the same job on every attempt, so the host joins all of them to one request", () => { + const ids = new Set([ + ...onFailed.mock.calls.map(([task]) => task.jobId), + ...onCompleted.mock.calls.map(([task]) => task.jobId), + ]); + + expect(ids.size, "one request, one job, four callbacks").toBe(1); + expect( + [...ids], + "and it is the job the enqueue response named — bound directly, not inferred from the other callbacks" + ).toEqual([...new Set(reportedJobIds)]); + }); +}); diff --git a/apps/dev/src/integration/translator/runner-predating-the-fields.int.test.ts b/apps/dev/src/integration/translator/runner-predating-the-fields.int.test.ts new file mode 100644 index 000000000..6bd3ff585 --- /dev/null +++ b/apps/dev/src/integration/translator/runner-predating-the-fields.int.test.ts @@ -0,0 +1,68 @@ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +type Seen = { targetLng: string; jobId?: string; attempt?: number }; + +const onQueued = vi.fn<(task: Seen) => void>(); +const onCompleted = vi.fn<(task: Seen) => void>(); + +let ctx: TestPayload; +let docId: string; + +describe("a runner written before the new handler fields existed", () => { + beforeAll(async () => { + ctx = await bootTestPayload({ lifecycle: { onQueued, onCompleted } }); + + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title: "No ids here", note: "A note" }, + }); + docId = String(made.id); + + await callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: "de", + collection_slug: "docs", + collection_id: [docId], + strategy: "overwrite", + publish_on_translation: true, + }, + }); + }); + + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("still runs the translation to completion", async () => { + expect(onCompleted).toHaveBeenCalled(); + + const de = (await ctx.payload.findByID({ + collection: "docs", + id: docId, + locale: "de", + })) as Record; + + expect(de.title, "the work happened, ids or no ids").toBe("de:No ids here"); + }); + + it("reports the absent attempt as absent rather than empty — it has no retries to count", () => { + expect(onCompleted.mock.calls[0]?.[0].attempt).toBeUndefined(); + }); + + it("names the id it does have, so the callback matches the enqueue response", () => { + expect( + onCompleted.mock.calls[0]?.[0].jobId, + "this runner mints an id and reports it; carrying it in the response but not the callback would be a split personality" + ).toBeTruthy(); + }); + + it("still fires the queued callback", () => { + expect(onQueued).toHaveBeenCalled(); + }); +}); diff --git a/packages/payload-plugin-translator/src/index.ts b/packages/payload-plugin-translator/src/index.ts index d44b9ad05..9fc4df20a 100644 --- a/packages/payload-plugin-translator/src/index.ts +++ b/packages/payload-plugin-translator/src/index.ts @@ -3,7 +3,6 @@ export { translatorPlugin } from "./plugin.js"; export type { TranslatorPluginConfig } from "./plugin.js"; export type { TargetSelectionMode } from "./types/TargetSelection.js"; -// Lifecycle callbacks — the task descriptor passed to onQueued/onCompleted/onFailed export type { TranslationTask, TranslationLifecycleCallbacks, @@ -57,6 +56,13 @@ export type { PayloadJobsRunnerOptions, TaskFilter, } from "./server/modules/task-runner/index.js"; +/** + * The contract a third-party `TaskRunnerProvider.create` must return, and what its `enqueue` may + * answer with. + * + * @since 0.16.0 + */ +export type { EnqueueResult, TaskRunner } from "./server/modules/task-runner/index.js"; /** * A third-party `TaskRunnerProvider` has always had to accept a `RequestScope` — it is the second * parameter of `TaskRunner.enqueue` and of the `TaskHandler` it is handed — but could not name the diff --git a/packages/payload-plugin-translator/src/plugin.ts b/packages/payload-plugin-translator/src/plugin.ts index 286e88ec1..58d634e56 100644 --- a/packages/payload-plugin-translator/src/plugin.ts +++ b/packages/payload-plugin-translator/src/plugin.ts @@ -81,9 +81,9 @@ export type TranslatorPluginConfig = { slug?: string; }; /** - * Optional server-side lifecycle callbacks fired around each translation task - * (`onQueued` / `onCompleted` / `onFailed`). Always available — no schema, no migration, not gated - * by `provenance`. A throwing callback is caught and logged, never failing the translation. + * Optional server-side lifecycle callbacks fired around each translation task. Always available — + * no schema, no migration, not gated by `provenance`. A throwing callback is caught and logged, + * never failing the translation. * @since 0.7.0 */ lifecycle?: TranslationLifecycleCallbacks; diff --git a/packages/payload-plugin-translator/src/server/features/translate-document/wireTranslateRunner.ts b/packages/payload-plugin-translator/src/server/features/translate-document/wireTranslateRunner.ts index 34c2aa1ce..5049ab4d1 100644 --- a/packages/payload-plugin-translator/src/server/features/translate-document/wireTranslateRunner.ts +++ b/packages/payload-plugin-translator/src/server/features/translate-document/wireTranslateRunner.ts @@ -7,6 +7,7 @@ import type { ProvenanceServiceFactory } from "../../modules/provenance/index.js import { LifecycleNotifier, taskFromHandlerInput, + needsDecoration, withQueuedNotification, } from "../../modules/lifecycle/index.js"; import type { TranslationLifecycleCallbacks } from "../../modules/lifecycle/index.js"; @@ -31,7 +32,7 @@ type WireTranslateRunnerParams = { /** * Assemble the document-translation task pipeline: the {@link TranslateDocumentHandler}, the runner * context that wraps each task with lifecycle notifications, the runner's config modifier, and the - * per-request {@link TaskRunnerFactory} (decorated with `onQueued` notification when configured). + * per-request {@link TaskRunnerFactory}. * * Extracted from the plugin's `init()` so the composition root stays a flat list — `plugin.ts` calls * this once and registers the returned `configModifier` through the shared builder. @@ -84,7 +85,7 @@ export function wireTranslateRunner({ const taskRunnerFactory: TaskRunnerFactory = { create: (payload) => { const taskRunner = runner.create(payload, runnerContext.handler); - if (!lifecycle.onQueued) return taskRunner; + if (!needsDecoration(lifecycle)) return taskRunner; return withQueuedNotification(taskRunner, new LifecycleNotifier(lifecycle, payload.logger)); }, }; diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/index.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/index.ts index 25494586b..59e6bf593 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/index.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/index.ts @@ -1,6 +1,6 @@ // Translation lifecycle callbacks — always-on host hooks fired around the runner (not gated by the // provenance opt-in, no schema/migration). export { LifecycleNotifier } from "./LifecycleNotifier.js"; -export { withQueuedNotification } from "./withQueuedNotification.js"; +export { needsDecoration, withQueuedNotification } from "./withQueuedNotification.js"; export { taskFromHandlerInput, taskFromInput } from "./taskMapping.js"; export type { TranslationLifecycleCallbacks, TranslationTask } from "./types.js"; diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.test.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.test.ts index 1affeec2c..954c7de11 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.test.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.test.ts @@ -38,12 +38,57 @@ describe("taskFromHandlerInput", () => { strategy: "overwrite", publishOnTranslation: false, }; + expect( + taskFromHandlerInput(input), + "a runner that cannot name a job leaves the descriptor exactly as it was" + ).toEqual({ + collection: "pages", + id: "doc-2", + sourceLng: "en", + targetLng: "fr", + strategy: "overwrite", + }); + }); + + it("carries the job and the attempt when the runner knows them", () => { + const input: TaskHandlerInput = { + collection: "pages", + collectionId: "doc-2", + sourceLng: "en", + targetLng: "fr", + strategy: "overwrite", + publishOnTranslation: false, + jobId: "job-7", + attempt: 2, + }; + expect(taskFromHandlerInput(input)).toEqual({ + collection: "pages", + id: "doc-2", + sourceLng: "en", + targetLng: "fr", + strategy: "overwrite", + jobId: "job-7", + attempt: 2, + }); + }); + + it("omits each field independently, so a runner with ids but no retries still reports the job", () => { + const input: TaskHandlerInput = { + collection: "pages", + collectionId: "doc-2", + sourceLng: "en", + targetLng: "fr", + strategy: "overwrite", + publishOnTranslation: false, + jobId: "job-7", + }; expect(taskFromHandlerInput(input)).toEqual({ collection: "pages", id: "doc-2", sourceLng: "en", targetLng: "fr", strategy: "overwrite", + jobId: "job-7", }); }); }); diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.ts index 91990fda0..99e48f11f 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/taskMapping.ts @@ -1,4 +1,4 @@ -import type { TaskInput } from "../task-runner/types.js"; +import type { Task, TaskInput } from "../task-runner/types.js"; import type { TaskHandlerInput } from "../task-runner/TaskRunnerProvider.interface.js"; import type { TranslationTask } from "./types.js"; @@ -11,6 +11,15 @@ export const taskFromInput = (task: TaskInput): TranslationTask => ({ strategy: task.strategy, }); +export const taskFromStored = (task: Task): TranslationTask => ({ + collection: task.input.collectionSlug, + id: task.input.collectionId, + sourceLng: task.input.sourceLng, + targetLng: task.input.targetLng, + strategy: task.input.strategy, + jobId: task.id, +}); + /** Map an execution-side {@link TaskHandlerInput} to the public {@link TranslationTask}. */ export const taskFromHandlerInput = (input: TaskHandlerInput): TranslationTask => ({ collection: input.collection, @@ -18,4 +27,6 @@ export const taskFromHandlerInput = (input: TaskHandlerInput): TranslationTask = sourceLng: input.sourceLng, targetLng: input.targetLng, strategy: input.strategy, + ...(input.jobId === undefined ? {} : { jobId: input.jobId }), + ...(input.attempt === undefined ? {} : { attempt: input.attempt }), }); diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/types.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/types.ts index 6c839823b..08cceba47 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/types.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/types.ts @@ -17,6 +17,23 @@ export type TranslationTask = { * public type; host callbacks should treat unknown values gracefully. */ strategy: string; + /** + * The job that will run this locale. Absent when the runner answers `enqueue` with nothing (see + * docs/DEPRECATIONS.md#enqueue-void-return), and always absent on `onQueued` — the job does not + * exist yet when that fires. + * + * @since 0.16.0 + */ + jobId?: string; + /** + * Which attempt at this locale the job is on, counting from 1 — the only field that differs + * between the callbacks of a retried task. It counts the job's attempts at the locale, not the + * caller's: a request that joins a job which already failed this locale sees the job's next + * number, not 1. Absent when the runner does not retry or cannot say. + * + * @since 0.16.0 + */ + attempt?: number; }; /** @@ -43,4 +60,12 @@ export type TranslationLifecycleCallbacks = { onCompleted?: (task: TranslationTask) => void | Promise; /** Fired when a task's translation throws, with the error. @since 0.7.0 */ onFailed?: (task: TranslationTask, error: unknown) => void | Promise; + /** + * A queued translation was cancelled: once per target locale the cancelled job covered, fired + * before the row is deleted. A job that finishes between the read and the delete still reports as + * cancelled. Never fires for a runner whose `TaskRunner` cannot resolve job ids by id. + * + * @since 0.16.0 + */ + onCancelled?: (task: TranslationTask) => void | Promise; }; diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunnerProvider.interface.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunnerProvider.interface.ts index a8c0541ee..61d66c100 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunnerProvider.interface.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/TaskRunnerProvider.interface.ts @@ -14,6 +14,20 @@ export type TaskHandlerInput = { targetLng: string; strategy: TranslationStrategyName; publishOnTranslation: boolean; + /** + * The job this run belongs to, as `enqueue` reported it. Absent when the runner answers `enqueue` + * with nothing (docs/DEPRECATIONS.md#enqueue-void-return); both bundled runners name one. + * + * @since 0.16.0 + */ + jobId?: string; + /** + * Which attempt at this locale the job is on, counting from 1 — the same counting as the + * `attempt` a lifecycle callback receives. Absent when the runner does not retry or cannot say. + * + * @since 0.16.0 + */ + attempt?: number; }; /** diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts index 8cd5fbd69..e2e0f37fa 100644 --- a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/PayloadJobsRunnerProvider.ts @@ -10,7 +10,11 @@ import type { } from "./types.js"; import { PayloadJobsTaskRunner } from "./PayloadJobsTaskRunner.js"; import { readCollectionRef } from "./readCollectionRef.js"; -import type { TaskRunnerContext, TaskRunnerProvider } from "../TaskRunnerProvider.interface.js"; +import type { + TaskHandlerInput, + TaskRunnerContext, + TaskRunnerProvider, +} from "../TaskRunnerProvider.interface.js"; import type { Requester } from "../../../shared/payload/RequestScope.shapes.js"; import { asRequester } from "../../../shared/payload/RequestScope.shapes.js"; import type { TranslationStrategyName } from "../../../../core/translation-pipeline/strategies/index.js"; @@ -51,6 +55,46 @@ export function requesterOf(input: { return asRequester(input.requester_id, input.requester_collection); } +type RunningJob = { + id?: unknown; + workflowSlug?: unknown; + taskStatus?: Record | undefined>; +}; + +/** Payload runs a job that is the task itself under this id, where a workflow passes the locale. */ +const SOLE_TASK_ID = "1"; + +/** + * Which attempt of this locale the handler is running right now. + * + * One job covers every locale of a document, so the row's own `totalTried` counts the job's passes, + * not the locale's; and Payload writes each counter only after a run finishes, hence the `+ 1`. + */ +export function attemptOf( + job: RunningJob | undefined, + taskSlug: string, + targetLng: string +): number | undefined { + if (!job) return undefined; + const taskId = job.workflowSlug ? targetLng : SOLE_TASK_ID; + const tried = job.taskStatus?.[taskSlug]?.[taskId]?.totalTried; + if (tried === undefined) return 1; + return typeof tried === "number" ? tried + 1 : undefined; +} + +function runIdentity( + job: RunningJob | undefined, + taskSlug: string, + targetLng: string +): Partial> { + const attempt = attemptOf(job, taskSlug, targetLng); + const id = job?.id; + return { + ...(typeof id === "string" || typeof id === "number" ? { jobId: String(id) } : {}), + ...(attempt === undefined ? {} : { attempt }), + }; +} + export class PayloadJobsRunnerProvider implements TaskRunnerProvider { private readonly config: PayloadJobsRunnerConfig; @@ -148,6 +192,7 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider { retries, handler: async (args: { req: { payload: Payload }; + job?: RunningJob; input: { collection_slug?: string; collection_id?: string; @@ -170,6 +215,7 @@ export class PayloadJobsRunnerProvider implements TaskRunnerProvider { targetLng: args.input.target_lng, strategy: args.input.strategy, publishOnTranslation: args.input.publish_on_translation ?? false, + ...runIdentity(args.job, taskName, args.input.target_lng), }, { requester: requesterOf(args.input) } ); diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/attemptOf.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/attemptOf.test.ts new file mode 100644 index 000000000..6d73a48bb --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/attemptOf.test.ts @@ -0,0 +1,45 @@ +import { describe, it, expect } from "vitest"; + +import { attemptOf } from "../PayloadJobsRunnerProvider.js"; + +const SLUG = "translate_document"; + +const job = (status: Record) => ({ + workflowSlug: "translate_document_locales", + taskStatus: { [SLUG]: status }, +}); + +describe("attemptOf — which attempt of this locale the handler is running", () => { + it.each([ + ["tried once", 1, 2], + ["tried twice", 2, 3], + ])("reads a locale %s as attempt %i + 1", (_label, totalTried, expected) => { + expect(attemptOf(job({ de: { totalTried } }), SLUG, "de")).toBe(expected); + }); + + it("reads a locale with no status entry as its first attempt", () => { + expect(attemptOf(job({}), SLUG, "de")).toBe(1); + }); + + it("counts the locale's own attempts, not the job's", () => { + expect( + attemptOf(job({ de: { totalTried: 2 } }), SLUG, "fr"), + "one job carries every locale, and fr has not run yet — the job's two passes were de's" + ).toBe(1); + }); + + it("reads a job that is the task itself under the id Payload runs it with", () => { + expect( + attemptOf({ taskStatus: { [SLUG]: { "1": { totalTried: 1 } } } }, SLUG, "de"), + "a pre-workflow row has no workflow and no locale key" + ).toBe(2); + }); + + it("says nothing when there is no job to read", () => { + expect(attemptOf(undefined, SLUG, "de")).toBeUndefined(); + }); + + it("says nothing rather than guessing when the count is not a number", () => { + expect(attemptOf(job({ de: { totalTried: "2" } }), SLUG, "de")).toBeUndefined(); + }); +}); From d6adb227767ed1c4f56165a776df683573eac24b Mon Sep 17 00:00:00 2001 From: Siarhei Date: Tue, 6 Oct 2026 14:26:11 +0200 Subject: [PATCH 4/6] feat(translator): tell the host what a cancellation is stopping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #106. Announced once per target locale the cancelled job covered, before the row is deleted — after the delete there is nothing left to say it from. The cancel route holds only ids, so naming the document and locale needs a read: a new optional `findByIds` on `TaskRunner`. A runner that omits it keeps working and simply never fires the callback. A failed read is logged and never costs the caller their cancellation — otherwise the silence would be indistinguishable from an id that was not ours. The decorator is now applied for any callback it fires, not just `onQueued`; a host that registered only `onCancelled` used to be handed an undecorated runner and heard nothing. --- .../cancelled-only-host.int.test.ts | 59 ++++++ .../modules/lifecycle/LifecycleNotifier.ts | 16 ++ .../lifecycle/cancelNotification.test.ts | 171 ++++++++++++++++++ .../modules/lifecycle/decorationGate.test.ts | 24 +++ .../lifecycle/withQueuedNotification.test.ts | 13 ++ .../lifecycle/withQueuedNotification.ts | 52 +++++- .../PayloadJobsTaskRunner.findByIds.test.ts | 77 ++++++++ 7 files changed, 405 insertions(+), 7 deletions(-) create mode 100644 apps/dev/src/integration/translator/cancelled-only-host.int.test.ts create mode 100644 packages/payload-plugin-translator/src/server/modules/lifecycle/cancelNotification.test.ts create mode 100644 packages/payload-plugin-translator/src/server/modules/lifecycle/decorationGate.test.ts create mode 100644 packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.findByIds.test.ts diff --git a/apps/dev/src/integration/translator/cancelled-only-host.int.test.ts b/apps/dev/src/integration/translator/cancelled-only-host.int.test.ts new file mode 100644 index 000000000..4bfdf4678 --- /dev/null +++ b/apps/dev/src/integration/translator/cancelled-only-host.int.test.ts @@ -0,0 +1,59 @@ +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator"; + +import { bootTestPayload } from "./bootTestPayload"; +import type { TestPayload } from "./bootTestPayload"; +import { callEndpoint } from "./callEndpoint"; + +type Seen = { collection: string; id: string; targetLng: string; jobId?: string }; + +const onCancelled = vi.fn<(task: Seen) => void>(); + +let ctx: TestPayload; + +describe("a host that registers only onCancelled", () => { + beforeAll(async () => { + ctx = await bootTestPayload({ + runner: createPayloadJobsRunner({ autoRun: false }), + lifecycle: { onCancelled }, + }); + }); + afterAll(async () => { + await ctx?.cleanup(); + }); + + it("is told what stopped, once per locale, with the document named", async () => { + const made = await ctx.payload.create({ + collection: "docs", + locale: "en", + data: { _status: "published", title: "To be cancelled", note: "A note" }, + }); + const id = String(made.id); + + const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", { + body: { + source_lng: "en", + target_lng: ["de", "fr"], + collection_slug: "docs", + collection_id: [id], + strategy: "overwrite", + publish_on_translation: true, + }, + }); + const jobs = (res.data as { data: { jobs: { job_id: string }[] } }).data.jobs; + const jobId = jobs[0]?.job_id; + + onCancelled.mockClear(); + await callEndpoint(ctx.payload, "delete", "/translate/cancel", { body: { ids: [jobId] } }); + + expect( + onCancelled.mock.calls.map(([task]) => task.targetLng).sort(), + "both locales of the cancelled job were translations the host was tracking" + ).toEqual(["de", "fr"]); + expect(onCancelled.mock.calls[0]?.[0]).toMatchObject({ + collection: "docs", + id, + jobId, + }); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts index 3c2bd960d..58e5eb423 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts @@ -28,6 +28,11 @@ export class LifecycleNotifier { return this.safe("lifecycle.onCompleted", callback && (() => callback(task))); } + cancelled(task: TranslationTask): Promise { + const callback = this.callbacks.onCancelled; + return this.safe("lifecycle.onCancelled", callback && (() => callback(task))); + } + async failed(task: TranslationTask, error: unknown): Promise { const callback = this.callbacks.onFailed; if (!callback) { @@ -43,6 +48,17 @@ export class LifecycleNotifier { await this.safe("lifecycle.onFailed", () => callback(task, error)); } + /** + * The cancel happened; only the announcement failed. This log is the sole trace that a host's + * `onCancelled` was owed and not fired. + */ + announcementFailed(error: unknown): void { + this.logger.error({ + err: error, + msg: "translator: could not read what a cancellation is stopping", + }); + } + private async safe(name: string, thunk?: () => void | Promise): Promise { if (!thunk) return; try { diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/cancelNotification.test.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/cancelNotification.test.ts new file mode 100644 index 000000000..559cc4848 --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/cancelNotification.test.ts @@ -0,0 +1,171 @@ +import { describe, it, expect, vi } from "vitest"; +import type { CollectionSlug } from "payload"; + +import { withQueuedNotification } from "./withQueuedNotification.js"; +import { LifecycleNotifier } from "./LifecycleNotifier.js"; +import type { TaskRunner } from "../task-runner/TaskRunner.interface.js"; +import type { Task } from "../task-runner/types.js"; + +const task = (targetLng: string, id = "job-77"): Task => ({ + id, + status: "pending", + input: { + collectionSlug: "posts" as CollectionSlug, + collectionId: "doc-1", + sourceLng: "en", + targetLng, + strategy: "overwrite", + publishOnTranslation: false, + }, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + cancelled: false, +}); + +const makeRunner = (overrides: Partial = {}): TaskRunner => ({ + enqueue: vi.fn().mockResolvedValue(undefined), + cancel: vi.fn().mockResolvedValue(undefined), + run: vi.fn().mockResolvedValue({ success: true }), + findByCollection: vi.fn().mockResolvedValue([]), + ...overrides, +}); + +describe("cancelling tells the host what stopped", () => { + it("fires once per target locale the cancelled job covered", async () => { + const onCancelled = vi.fn(); + const runner = makeRunner({ + findByIds: vi.fn().mockResolvedValue([task("de"), task("fr"), task("it")]), + }); + + await withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled }, { error: vi.fn() }) + ).cancel(["job-77"]); + + expect( + onCancelled.mock.calls.map(([t]) => t.targetLng), + "one row covers a list of locales, and each is a translation the host was tracking" + ).toEqual(["de", "fr", "it"]); + expect(onCancelled.mock.calls[0][0]).toMatchObject({ + collection: "posts", + id: "doc-1", + jobId: "job-77", + }); + }); + + it("fires before the job row is deleted, so a host can still read it", async () => { + const order: string[] = []; + const onCancelled = vi.fn(() => { + order.push("callback"); + }); + const runner = makeRunner({ + findByIds: vi.fn().mockResolvedValue([task("de")]), + cancel: vi.fn(async () => { + order.push("cancel"); + }), + }); + + await withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled }, { error: vi.fn() }) + ).cancel(["job-77"]); + + expect(order).toEqual(["callback", "cancel"]); + }); + + it("stays quiet about a locale the job already finished", async () => { + const onCancelled = vi.fn(); + const halfDone = { ...task("de"), status: "completed" as const }; + const runner = makeRunner({ + findByIds: vi.fn().mockResolvedValue([halfDone, task("fr")]), + }); + + await withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled }, { error: vi.fn() }) + ).cancel(["job-77"]); + + expect( + onCancelled.mock.calls.map(([t]) => t.targetLng), + "de is not stopping — the host already had its completed callback for it" + ).toEqual(["fr"]); + }); + + it("says nothing for an id that resolves to no job of ours", async () => { + const onCancelled = vi.fn(); + const runner = makeRunner({ findByIds: vi.fn().mockResolvedValue([]) }); + + await withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled }, { error: vi.fn() }) + ).cancel(["not-ours"]); + + expect(onCancelled).not.toHaveBeenCalled(); + expect(runner.cancel, "the cancel itself still happens").toHaveBeenCalledWith(["not-ours"]); + }); + + it("still cancels, silently, for a runner that cannot resolve ids", async () => { + const onCancelled = vi.fn(); + const runner = makeRunner(); + + await withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled }, { error: vi.fn() }) + ).cancel(["job-77"]); + + expect(onCancelled).not.toHaveBeenCalled(); + expect(runner.cancel).toHaveBeenCalledWith(["job-77"]); + }); + + it("keeps the read available to anyone downstream, like every other method it wraps", () => { + const findByIds = vi.fn().mockResolvedValue([]); + const decorated = withQueuedNotification( + makeRunner({ findByIds }), + new LifecycleNotifier({}, { error: vi.fn() }) + ); + + expect(decorated.findByIds, "decorating must not quietly remove a capability").toBeDefined(); + }); + + it("stays absent when the runner it wraps has none, so the optionality survives", () => { + const decorated = withQueuedNotification( + makeRunner(), + new LifecycleNotifier({}, { error: vi.fn() }) + ); + + expect(decorated.findByIds).toBeUndefined(); + }); + + it("does not let a failed read stop the cancellation", async () => { + const onCancelled = vi.fn(); + const runner = makeRunner({ + findByIds: vi.fn().mockRejectedValue(new Error("sidecar down")), + }); + + await expect( + withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled }, { error: vi.fn() }) + ).cancel(["job-77"]) + ).resolves.toBeUndefined(); + + expect(runner.cancel).toHaveBeenCalledWith(["job-77"]); + }); + + it("logs the failed read, so silence is not mistaken for nothing to announce", async () => { + const error = vi.fn(); + const runner = makeRunner({ + findByIds: vi.fn().mockRejectedValue(new Error("sidecar down")), + }); + + await withQueuedNotification( + runner, + new LifecycleNotifier({ onCancelled: vi.fn() }, { error }) + ).cancel(["job-77"]); + + expect( + error, + "a host that registered onCancelled and heard nothing cannot tell a broken read from an id that was not ours" + ).toHaveBeenCalledWith(expect.objectContaining({ err: expect.any(Error) })); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/decorationGate.test.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/decorationGate.test.ts new file mode 100644 index 000000000..586b777cf --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/decorationGate.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; + +import { needsDecoration } from "./withQueuedNotification.js"; +import type { TranslationLifecycleCallbacks } from "./types.js"; + +const noop = (): void => undefined; + +describe("needsDecoration — whether a host's callbacks require wrapping the runner", () => { + it.each<[string, TranslationLifecycleCallbacks]>([ + ["onQueued alone", { onQueued: noop }], + ["onCancelled alone", { onCancelled: noop }], + ["both", { onQueued: noop, onCancelled: noop }], + ])("wraps for %s", (_label, callbacks) => { + expect(needsDecoration(callbacks)).toBe(true); + }); + + it.each<[string, TranslationLifecycleCallbacks]>([ + ["no callbacks at all", {}], + ["only callbacks the handler fires itself", { onCompleted: noop }], + ["only onFailed", { onFailed: noop }], + ])("leaves the runner alone for %s", (_label, callbacks) => { + expect(needsDecoration(callbacks)).toBe(false); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.test.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.test.ts index 4a88bc7a9..ce581e0f8 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.test.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.test.ts @@ -38,6 +38,19 @@ describe("withQueuedNotification", () => { expect(runner.enqueue).toHaveBeenCalledWith([input("1"), input("2")], undefined); }); + it("hands back whatever the runner answered, so decorating does not cost the caller the ids", async () => { + const answered = [ + { collectionSlug: "posts", collectionId: "1", targetLng: "de", jobId: "job-7" }, + ]; + const runner = makeRunner(); + runner.enqueue = vi.fn().mockResolvedValue(answered); + const notifier = new LifecycleNotifier({ onQueued: vi.fn() }, { error: vi.fn() }); + + const decorated = withQueuedNotification(runner, notifier); + + expect(await decorated.enqueue([input("1")])).toBe(answered); + }); + it("fires queued BEFORE delegating to the runner (ordering vs a synchronous runner)", async () => { const calls: string[] = []; const runner = makeRunner(); diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts index a52457866..1284bd080 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts @@ -1,13 +1,47 @@ import type { TaskRunner } from "../task-runner/TaskRunner.interface.js"; import { toTaskFilter } from "../task-runner/toTaskFilter.js"; import type { LifecycleNotifier } from "./LifecycleNotifier.js"; -import { taskFromInput } from "./taskMapping.js"; +import { taskFromInput, taskFromStored } from "./taskMapping.js"; +import type { TranslationLifecycleCallbacks } from "./types.js"; /** - * Decorate a {@link TaskRunner} so `enqueue` fires the `queued` lifecycle callback for each task. - * Fired BEFORE delegating: a synchronous runner (the sync runner) may execute a task inline during - * `enqueue` and fire `completed`, so emitting `queued` first preserves the queued → completed order. - * All other methods delegate unchanged. + * Whether a host's callbacks need the runner wrapped at all — only the two the decorator itself + * fires count; `onCompleted` and `onFailed` come from the task handler, which needs no decoration. + */ +export const needsDecoration = (callbacks: TranslationLifecycleCallbacks): boolean => + Boolean(callbacks.onQueued ?? callbacks.onCancelled); + +/** + * Say what is about to stop, while the rows still exist to say it from: the cancel route is handed + * ids and nothing else. A failed read is logged rather than thrown — silence would otherwise be + * indistinguishable from an id that was not ours. A locale whose last attempt failed is announced + * too: from the row alone, "failed with retries left" and "gave up" read the same. + */ +async function announceCancellation( + runner: TaskRunner, + notifier: LifecycleNotifier, + taskIds: string[] +): Promise { + if (!runner.findByIds || taskIds.length === 0) return; + try { + const stopping = await runner.findByIds(taskIds); + for (const task of stopping) { + if (task.status === "completed") continue; + await notifier.cancelled(taskFromStored(task)); + } + } catch (error) { + notifier.announcementFailed(error); + } +} + +/** + * Decorate a {@link TaskRunner} so `enqueue` fires `queued` and `cancel` announces what stops. + * + * `queued` fires BEFORE delegating: a synchronous runner may execute a task inline during `enqueue` + * and fire `completed`, so this preserves queued → completed — and is why `queued` carries no job id. + * + * Every method is listed by hand below: an OPTIONAL one added to {@link TaskRunner} and not listed + * vanishes from the wrapper silently. `findByIds` already went missing that way once. */ export function withQueuedNotification( runner: TaskRunner, @@ -16,13 +50,17 @@ export function withQueuedNotification( return { async enqueue(tasks, scope) { await Promise.all(tasks.map((task) => notifier.queued(taskFromInput(task)))); - await runner.enqueue(tasks, scope); + return runner.enqueue(tasks, scope); + }, + async cancel(taskIds) { + await announceCancellation(runner, notifier, taskIds); + return runner.cancel(taskIds); }, - cancel: (taskIds) => runner.cancel(taskIds), run: (taskId) => runner.run(taskId), // Normalized rather than forwarded as-is: the wrapper's own signature comes from the overload // pair, so it cannot pass the deprecated array form straight through. findByCollection: (collectionSlug, filter) => runner.findByCollection(collectionSlug, toTaskFilter(filter)), + ...(runner.findByIds ? { findByIds: (ids: string[]) => runner.findByIds!(ids) } : {}), }; } diff --git a/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.findByIds.test.ts b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.findByIds.test.ts new file mode 100644 index 000000000..d5327a4dd --- /dev/null +++ b/packages/payload-plugin-translator/src/server/modules/task-runner/payload-jobs-runner/__tests__/PayloadJobsTaskRunner.findByIds.test.ts @@ -0,0 +1,77 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import type { Payload, Where } from "payload"; + +import { PayloadJobsTaskRunner } from "../PayloadJobsTaskRunner.js"; +import type { PayloadJobsRunnerConfig, PayloadJob } from "../types.js"; + +const config: PayloadJobsRunnerConfig = { + taskName: "translate_document", + workflowName: "translate_document_locales", + queueName: "translations", + jobsCollection: "payload-jobs", + autoRun: { cron: "* * * * *", limit: 50 }, + staleJobTimeoutMs: 300_000, +}; + +const job = (id: string, targetLngs: string[]): PayloadJob => ({ + id, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + processing: false, + input: { + collection_slug: "posts", + collection_id: "doc-1", + source_lng: "en", + target_lngs: targetLngs, + strategy: "overwrite", + publish_on_translation: false, + requester_id: null, + requester_collection: null, + }, +}); + +describe("PayloadJobsTaskRunner.findByIds — resolving ids back to what they stand for", () => { + let mockPayload: { find: ReturnType }; + let runner: PayloadJobsTaskRunner; + + beforeEach(() => { + mockPayload = { find: vi.fn().mockResolvedValue({ docs: [] }) }; + runner = new PayloadJobsTaskRunner(mockPayload as unknown as Payload, config); + }); + + it("expands one row into one task per target locale", async () => { + mockPayload.find.mockResolvedValue({ docs: [job("job-77", ["de", "fr", "it"])] }); + + const tasks = await runner.findByIds(["job-77"]); + + expect( + tasks.map((task) => task.input.targetLng), + "a row holds a locale LIST, and a cancellation stops every one of them" + ).toEqual(["de", "fr", "it"]); + expect(new Set(tasks.map((task) => task.id)), "all three ride the one row").toEqual( + new Set(["job-77"]) + ); + expect(tasks[0]?.input.collectionId).toBe("doc-1"); + }); + + it("asks only for this plugin's own jobs, so a foreign id resolves to nothing", async () => { + await runner.findByIds(["someone-elses-job"]); + + const where = (mockPayload.find.mock.calls[0][0] as { where: { and: Where[] } }).where; + const narrowing = JSON.stringify(where); + + expect(narrowing, "the plugin's own workflow slug must be part of the query").toContain( + "translate_document_locales" + ); + expect(narrowing).toContain("someone-elses-job"); + }); + + it("reads nothing and asks nothing for an empty list", async () => { + expect(await runner.findByIds([])).toEqual([]); + expect(mockPayload.find).not.toHaveBeenCalled(); + }); + + it("answers with nothing when the ids match no row", async () => { + expect(await runner.findByIds(["gone"])).toEqual([]); + }); +}); From 6ee222c129d0a5a047ffeee49f3d7704bce755ce Mon Sep 17 00:00:00 2001 From: Siarhei Date: Tue, 6 Oct 2026 14:26:34 +0200 Subject: [PATCH 5/6] feat(translator): name the document and locale in the collection status Closes #103. The response carried a job id and a status, and a caller could not resolve either back to what was being translated. Each entry now says which document and which target locale it is for. The job id stays, and it repeats: one job covers a document's whole locale list, so `id` cannot identify an entry on its own. --- .../collectionStatusItem.test.ts | 39 +++++++++++++++++++ .../get-collection-status/handler.test.ts | 22 +++++++++-- .../features/get-collection-status/handler.ts | 4 +- .../features/get-collection-status/model.ts | 16 +++++++- 4 files changed, 74 insertions(+), 7 deletions(-) create mode 100644 packages/payload-plugin-translator/src/server/features/get-collection-status/collectionStatusItem.test.ts diff --git a/packages/payload-plugin-translator/src/server/features/get-collection-status/collectionStatusItem.test.ts b/packages/payload-plugin-translator/src/server/features/get-collection-status/collectionStatusItem.test.ts new file mode 100644 index 000000000..0a82dfe65 --- /dev/null +++ b/packages/payload-plugin-translator/src/server/features/get-collection-status/collectionStatusItem.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from "vitest"; +import type { CollectionSlug } from "payload"; + +import { toCollectionStatusItem } from "./model.js"; +import type { Task } from "../../modules/task-runner/index.js"; + +const task = (targetLng: string, collectionId: string): Task => ({ + id: "job-77", + status: "pending", + input: { + collectionSlug: "posts" as CollectionSlug, + collectionId, + sourceLng: "en", + targetLng, + strategy: "overwrite", + publishOnTranslation: false, + }, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + cancelled: false, +}); + +describe("a collection-status entry says which translation it is", () => { + it("names the document and the target locale beside the job", () => { + expect(toCollectionStatusItem(task("de", "doc-1"))).toEqual({ + id: "job-77", + status: "pending", + collection_id: "doc-1", + target_lng: "de", + }); + }); + + it("tells two locales of one job apart, which the job id alone cannot", () => { + const [de, fr] = [task("de", "doc-1"), task("fr", "doc-1")].map(toCollectionStatusItem); + + expect(de.id, "they really do share a job row").toBe(fr.id); + expect([de.target_lng, fr.target_lng]).toEqual(["de", "fr"]); + }); +}); diff --git a/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.test.ts b/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.test.ts index a3c2e9fd5..4d07a60dc 100644 --- a/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.test.ts +++ b/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.test.ts @@ -134,12 +134,28 @@ describe("GetCollectionStatusHandler", () => { expect(response.status).toBe(200); const body = await response.json(); expect(body.data.docs).toEqual([ - { id: "task-1", status: "completed" }, - { id: "task-2", status: "pending" }, - { id: "task-3", status: "running" }, + { id: "task-1", status: "completed", collection_id: "doc-123", target_lng: "de" }, + { id: "task-2", status: "pending", collection_id: "doc-123", target_lng: "de" }, + { id: "task-3", status: "running", collection_id: "doc-123", target_lng: "de" }, ]); }); + it("tells two locales of the same job apart", async () => { + (mockTaskRunner.findByCollection as ReturnType).mockResolvedValue([ + createMockTask({ id: "job-77", input: { ...createMockTask().input, targetLng: "de" } }), + createMockTask({ id: "job-77", input: { ...createMockTask().input, targetLng: "fr" } }), + ]); + + const body = await ( + await handler.handle(createMockRequest({ collection_slug: "posts" })) + ).json(); + + expect( + body.data.docs.map((doc: { target_lng: string }) => doc.target_lng), + "the job id repeats, so without the locale these two rows are indistinguishable" + ).toEqual(["de", "fr"]); + }); + it("calls findByCollection with correct collection slug", async () => { const req = createMockRequest({ collection_slug: "pages" }); diff --git a/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.ts b/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.ts index f46e9a8a3..9baa798e6 100644 --- a/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.ts +++ b/packages/payload-plugin-translator/src/server/features/get-collection-status/handler.ts @@ -4,7 +4,7 @@ import { ServerResponse } from "../../shared/index.js"; import type { TaskRunnerFactory } from "../../modules/task-runner/index.js"; import { isCollectionAvailable, visibleIds } from "../_lib/collection-utils.js"; -import { GetCollectionStatusInputSchema } from "./model.js"; +import { GetCollectionStatusInputSchema, toCollectionStatusItem } from "./model.js"; import type { GetCollectionStatusConfig } from "./model.js"; /** @@ -39,7 +39,7 @@ export class GetCollectionStatusHandler { return ServerResponse.success({ docs: tasks .filter((task) => visible.has(String(task.input.collectionId))) - .map((task) => ({ id: task.id, status: task.status })), + .map(toCollectionStatusItem), }); } } diff --git a/packages/payload-plugin-translator/src/server/features/get-collection-status/model.ts b/packages/payload-plugin-translator/src/server/features/get-collection-status/model.ts index 7288245d0..4e36268f5 100644 --- a/packages/payload-plugin-translator/src/server/features/get-collection-status/model.ts +++ b/packages/payload-plugin-translator/src/server/features/get-collection-status/model.ts @@ -1,7 +1,7 @@ import { z } from "zod"; import type { CollectionSlug } from "payload"; -import type { TaskStatus } from "../../modules/task-runner/index.js"; +import type { Task, TaskStatus } from "../../modules/task-runner/index.js"; /** * Input validation schema @@ -13,13 +13,25 @@ export const GetCollectionStatusInputSchema = z.object({ export type GetCollectionStatusInput = z.infer; /** - * Summary item for a single document + * One translation in flight: a job, and which document and locale it is for. + * + * `id` is the job, and a job covers a document's whole locale list — so it repeats across entries and + * cannot identify a row on its own. */ export type CollectionStatusItem = { id: string; status: TaskStatus; + collection_id: string; + target_lng: string; }; +export const toCollectionStatusItem = (task: Task): CollectionStatusItem => ({ + id: task.id, + status: task.status, + collection_id: String(task.input.collectionId), + target_lng: task.input.targetLng, +}); + /** * Handler output */ From fe027e5b34eed74199526a413146a38ecb08a2f9 Mon Sep 17 00:00:00 2001 From: Siarhei Date: Tue, 6 Oct 2026 16:36:03 +0200 Subject: [PATCH 6/6] refactor(translator): let the notifier own delivering a cancellation or logging why it could not MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `announceCancellation` mixed three jobs — read the rows, decide whether to announce, log a failed read — and the third forced a log-only method onto `LifecycleNotifier`, whose public surface otherwise means "fire a host callback". Split: `stopping()` answers only what a cancellation is about to stop, and `LifecycleNotifier.cancelling()` owns the try/catch and the log. `cancelled()` is private now, since nothing outside the class fires a single one. Also drops `@since` from `QueuedJob`: it is not re-exported from `src/index.ts`, so the package rule does not reach it, and its neighbour `CollectionStatusItem` carries none. --- .../features/enqueue-translation/model.ts | 2 -- .../modules/lifecycle/LifecycleNotifier.ts | 20 ++++++++----- .../lifecycle/withQueuedNotification.ts | 30 ++++++------------- 3 files changed, 21 insertions(+), 31 deletions(-) diff --git a/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts b/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts index 795bb0b16..57e6e974e 100644 --- a/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts +++ b/packages/payload-plugin-translator/src/server/features/enqueue-translation/model.ts @@ -30,8 +30,6 @@ export type EnqueueConfig = { * * Absent when the runner predates returning ids (docs/DEPRECATIONS.md#enqueue-void-return); empty * means the runner answered and could name nothing. `queued` counts what was accepted either way. - * - * @since 0.16.0 */ export type QueuedJob = { collection_slug: string; diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts index 58e5eb423..57989aa51 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/LifecycleNotifier.ts @@ -28,7 +28,7 @@ export class LifecycleNotifier { return this.safe("lifecycle.onCompleted", callback && (() => callback(task))); } - cancelled(task: TranslationTask): Promise { + private cancelled(task: TranslationTask): Promise { const callback = this.callbacks.onCancelled; return this.safe("lifecycle.onCancelled", callback && (() => callback(task))); } @@ -49,14 +49,18 @@ export class LifecycleNotifier { } /** - * The cancel happened; only the announcement failed. This log is the sole trace that a host's - * `onCancelled` was owed and not fired. + * Announce each translation a cancellation stops. The read can fail where the other callbacks + * cannot, and a host that hears nothing cannot tell that from ids that were never ours. */ - announcementFailed(error: unknown): void { - this.logger.error({ - err: error, - msg: "translator: could not read what a cancellation is stopping", - }); + async cancelling(read: () => Promise): Promise { + try { + for (const task of await read()) await this.cancelled(task); + } catch (error) { + this.logger.error({ + err: error, + msg: "translator: could not read what a cancellation is stopping", + }); + } } private async safe(name: string, thunk?: () => void | Promise): Promise { diff --git a/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts b/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts index 1284bd080..c425ddd42 100644 --- a/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts +++ b/packages/payload-plugin-translator/src/server/modules/lifecycle/withQueuedNotification.ts @@ -2,7 +2,7 @@ import type { TaskRunner } from "../task-runner/TaskRunner.interface.js"; import { toTaskFilter } from "../task-runner/toTaskFilter.js"; import type { LifecycleNotifier } from "./LifecycleNotifier.js"; import { taskFromInput, taskFromStored } from "./taskMapping.js"; -import type { TranslationLifecycleCallbacks } from "./types.js"; +import type { TranslationLifecycleCallbacks, TranslationTask } from "./types.js"; /** * Whether a host's callbacks need the runner wrapped at all — only the two the decorator itself @@ -12,26 +12,14 @@ export const needsDecoration = (callbacks: TranslationLifecycleCallbacks): boole Boolean(callbacks.onQueued ?? callbacks.onCancelled); /** - * Say what is about to stop, while the rows still exist to say it from: the cancel route is handed - * ids and nothing else. A failed read is logged rather than thrown — silence would otherwise be - * indistinguishable from an id that was not ours. A locale whose last attempt failed is announced - * too: from the row alone, "failed with retries left" and "gave up" read the same. + * What cancelling these ids is about to stop — the cancel route is handed ids and nothing else, so + * it can only come from a read. A locale whose last attempt failed is included: from the row alone, + * "failed with retries left" and "gave up" read the same. */ -async function announceCancellation( - runner: TaskRunner, - notifier: LifecycleNotifier, - taskIds: string[] -): Promise { - if (!runner.findByIds || taskIds.length === 0) return; - try { - const stopping = await runner.findByIds(taskIds); - for (const task of stopping) { - if (task.status === "completed") continue; - await notifier.cancelled(taskFromStored(task)); - } - } catch (error) { - notifier.announcementFailed(error); - } +async function stopping(runner: TaskRunner, taskIds: string[]): Promise { + if (!runner.findByIds || taskIds.length === 0) return []; + const rows = await runner.findByIds(taskIds); + return rows.filter((task) => task.status !== "completed").map(taskFromStored); } /** @@ -53,7 +41,7 @@ export function withQueuedNotification( return runner.enqueue(tasks, scope); }, async cancel(taskIds) { - await announceCancellation(runner, notifier, taskIds); + await notifier.cancelling(() => stopping(runner, taskIds)); return runner.cancel(taskIds); }, run: (taskId) => runner.run(taskId),