diff --git a/.changeset/fix-workflows-duplicate-ids.md b/.changeset/fix-workflows-duplicate-ids.md new file mode 100644 index 00000000000..16f5267fc97 --- /dev/null +++ b/.changeset/fix-workflows-duplicate-ids.md @@ -0,0 +1,8 @@ +--- +"miniflare": patch +"wrangler": patch +--- + +Emulate the deterministic-ID uniqueness contract in the local Workflows binding + +The local Workflows binding now matches the documented production behavior for deterministic instance IDs: `create({ id })` with an ID that already exists throws and retains the existing instance, and `createBatch()` skips IDs that already exist — or repeat within the batch — excluding them from the result instead of creating duplicate executions. Previously both paths silently created duplicates, so code relying on deterministic IDs for idempotency (for example a Queue consumer creating one workflow per message) appeared to work locally while double-executing workflow bodies. diff --git a/packages/miniflare/test/dev-registry.spec.ts b/packages/miniflare/test/dev-registry.spec.ts index 76d11c41c7f..447be3ccc2e 100644 --- a/packages/miniflare/test/dev-registry.spec.ts +++ b/packages/miniflare/test/dev-registry.spec.ts @@ -884,7 +884,17 @@ describe.sequential("DevRegistry", () => { script: ` export default { async fetch(request, env, ctx) { - const instance = await env.MY_WORKFLOW.create({ id: "cross-worker-instance" }); + // Deterministic ids are unique: create() throws once the + // instance exists, so later attempts read it instead. + let instance; + try { + instance = await env.MY_WORKFLOW.create({ id: "cross-worker-instance" }); + } catch (e) { + if (!String(e).includes("instance.already_exists")) { + throw e; + } + instance = await env.MY_WORKFLOW.get("cross-worker-instance"); + } return Response.json({ id: instance.id }); } } diff --git a/packages/miniflare/test/plugins/workflows/index.spec.ts b/packages/miniflare/test/plugins/workflows/index.spec.ts index 9ab62d25b6d..f921d48a083 100644 --- a/packages/miniflare/test/plugins/workflows/index.spec.ts +++ b/packages/miniflare/test/plugins/workflows/index.spec.ts @@ -17,7 +17,17 @@ export class MyWorkflow extends WorkflowEntrypoint { } export default { async fetch(request, env, ctx) { - const workflow = await env.MY_WORKFLOW.create({id: "an-id"}) + // Deterministic ids are unique: create() throws once the instance + // exists, so later requests read the existing instance instead. + let workflow; + try { + workflow = await env.MY_WORKFLOW.create({id: "an-id"}) + } catch (e) { + if (!String(e).includes("instance.already_exists")) { + throw e; + } + workflow = await env.MY_WORKFLOW.get("an-id") + } return new Response(JSON.stringify(await workflow.status())) }, diff --git a/packages/vite-plugin-cloudflare/playground/external-workflows/worker-a/index.ts b/packages/vite-plugin-cloudflare/playground/external-workflows/worker-a/index.ts index c86cda82414..663155339ab 100644 --- a/packages/vite-plugin-cloudflare/playground/external-workflows/worker-a/index.ts +++ b/packages/vite-plugin-cloudflare/playground/external-workflows/worker-a/index.ts @@ -8,9 +8,24 @@ export default { const id = url.searchParams.get("id"); if (url.pathname === "/create") { - const instance = await env.MY_WORKFLOW.create( - id === null ? undefined : { id } - ); + let instance: WorkflowInstance; + try { + instance = await env.MY_WORKFLOW.create( + id === null ? undefined : { id } + ); + } catch (e) { + // Deterministic ids are unique: create() throws once the instance + // exists, so read the existing instance instead. Any other + // failure is real and should surface. + const isDuplicate = + id !== null && + e instanceof Error && + e.message.includes("instance.already_exists"); + if (!isDuplicate) { + throw e; + } + instance = await env.MY_WORKFLOW.get(id); + } return Response.json({ id: instance.id, diff --git a/packages/vite-plugin-cloudflare/playground/workflows/src/index.ts b/packages/vite-plugin-cloudflare/playground/workflows/src/index.ts index b586ce0ea89..931001e76f5 100644 --- a/packages/vite-plugin-cloudflare/playground/workflows/src/index.ts +++ b/packages/vite-plugin-cloudflare/playground/workflows/src/index.ts @@ -31,9 +31,24 @@ export default { const id = url.searchParams.get("id"); if (url.pathname === "/create") { - const instance = await env.MY_WORKFLOW.create( - id === null ? undefined : { id } - ); + let instance: WorkflowInstance; + try { + instance = await env.MY_WORKFLOW.create( + id === null ? undefined : { id } + ); + } catch (e) { + // Deterministic ids are unique: create() throws once the instance + // exists, so read the existing instance instead. Any other + // failure is real and should surface. + const isDuplicate = + id !== null && + e instanceof Error && + e.message.includes("instance.already_exists"); + if (!isDuplicate) { + throw e; + } + instance = await env.MY_WORKFLOW.get(id); + } return Response.json({ id: instance.id, diff --git a/packages/workflows-shared/src/binding.ts b/packages/workflows-shared/src/binding.ts index 9574eeb4c48..8cebada033e 100644 --- a/packages/workflows-shared/src/binding.ts +++ b/packages/workflows-shared/src/binding.ts @@ -1,6 +1,7 @@ import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; import { InstanceEvent, instanceStatusName } from "./instance"; import { + duplicateInstanceError, isUserTriggeredPause, isUserTriggeredRestart, isUserTriggeredTerminate, @@ -141,16 +142,39 @@ export class WorkflowBinding extends WorkerEntrypoint { super(ctx, env); } - public async create({ - id = crypto.randomUUID(), - params = {}, - }: WorkflowInstanceCreateOptions = {}): Promise<{ + async #instanceExists(id: string): Promise { + const stub = this.env.ENGINE.get(this.env.ENGINE.idFromName(id)); + return await stub.hasInstance(); + } + + public async create(options: WorkflowInstanceCreateOptions = {}): Promise<{ id: string; }> { + // Destructuring defaults apply only to absent fields: an explicit null + // id must reach the validation below rather than becoming a generated + // id. + const { id = crypto.randomUUID() } = options; if (!isValidWorkflowInstanceId(id)) { throw new WorkflowError("Workflow instance has invalid id"); } + // Deterministic (caller-provided) ids carry a documented uniqueness + // contract: creating an instance with an id that already exists throws + // and the existing instance is retained. + if (options.id !== undefined && (await this.#instanceExists(id))) { + throw duplicateInstanceError(id); + } + + return this.#createUnchecked(id, options); + } + + // Creation body shared by create() and createBatch(), which perform their + // own validation and existence checks before calling this. + async #createUnchecked( + id: string, + options: WorkflowInstanceCreateOptions + ): Promise<{ id: string }> { + const { params = {} } = options; const stubId = this.env.ENGINE.idFromName(id); const stub = this.env.ENGINE.get(stubId); const introspectionSession = workflowIntrospectionSessions.get( @@ -232,12 +256,50 @@ export class WorkflowBinding extends WorkerEntrypoint { ); } - return await Promise.all( - batch.map(async (val) => { - const res = await this.create(val); - return res; + // Reject malformed ids before anything is probed or created: probing + // an id constructs its engine Durable Object, which persists storage, + // and a bad batch must not be partially applied. + for (const options of batch) { + if (options.id !== undefined && !isValidWorkflowInstanceId(options.id)) { + throw new WorkflowError("Workflow instance has invalid id"); + } + } + + // Probe each distinct caller-provided id once, concurrently, instead of + // sequentially per entry (and a second time inside create()). + const providedIds = [ + ...new Set( + batch + .map((options) => options.id) + .filter((id): id is string => id !== undefined) + ), + ]; + const existing = new Set(); + await Promise.all( + providedIds.map(async (id) => { + if (await this.#instanceExists(id)) { + existing.add(id); + } }) ); + + // The documented batch contract is idempotent creation: ids that already + // exist, or that repeat within the batch, are skipped and excluded from + // the result rather than throwing, and instances are created in batch + // order. + const results: { id: string }[] = []; + const seenIds = new Set(); + for (const options of batch) { + if (options.id !== undefined) { + if (seenIds.has(options.id) || existing.has(options.id)) { + continue; + } + seenIds.add(options.id); + } + const { id = crypto.randomUUID() } = options; + results.push(await this.#createUnchecked(id, options)); + } + return results; } public async unsafeGetBindingName(): Promise { diff --git a/packages/workflows-shared/src/engine.ts b/packages/workflows-shared/src/engine.ts index a6e96855e1f..75ee8910e27 100644 --- a/packages/workflows-shared/src/engine.ts +++ b/packages/workflows-shared/src/engine.ts @@ -1186,6 +1186,12 @@ export class Engine extends DurableObject { void this.init(accountId, workflow, version, instance, event); } + async hasInstance(): Promise { + // INSTANCE_METADATA is written exactly once, by the first init() for this + // id, so its presence is the durable marker that the instance exists. + return (await this.ctx.storage.get(INSTANCE_METADATA)) !== undefined; + } + async init( accountId: number, workflow: DatabaseWorkflow, diff --git a/packages/workflows-shared/src/lib/errors.ts b/packages/workflows-shared/src/lib/errors.ts index 25614b3cd46..91623d9c3a8 100644 --- a/packages/workflows-shared/src/lib/errors.ts +++ b/packages/workflows-shared/src/lib/errors.ts @@ -125,3 +125,10 @@ export function stepNotFoundError(name: string): WorkflowError { "instance.cannot_restart" ); } + +export function duplicateInstanceError(id: string): WorkflowError { + return createWorkflowError( + `Workflow instance with id "${id}" already exists`, + "instance.already_exists" + ); +} diff --git a/packages/workflows-shared/tests/binding.test.ts b/packages/workflows-shared/tests/binding.test.ts index 03417019660..d7944dcb641 100644 --- a/packages/workflows-shared/tests/binding.test.ts +++ b/packages/workflows-shared/tests/binding.test.ts @@ -108,6 +108,15 @@ describe("WorkflowBinding", () => { "Workflow instance has invalid id" ); }); + + it("should reject a null id instead of generating one", async ({ + expect, + }) => { + const binding = createBinding(); + await expect( + binding.create({ id: null as unknown as string }) + ).rejects.toThrow("Workflow instance has invalid id"); + }); }); describe("get()", () => { @@ -181,6 +190,112 @@ describe("WorkflowBinding", () => { "WorkflowError: batchCreate should have at least 1 instance" ); }); + + it("should skip and exclude ids that already exist", async ({ expect }) => { + const existing = uniqueId("dedup-existing"); + const fresh = uniqueId("dedup-fresh"); + const binding = createBinding(); + const engineStub = env.ENGINE.get(env.ENGINE.idFromName(existing)); + setTestWorkflowCallback(async () => "done"); + + const first = await binding.createBatch([{ id: existing }]); + expect(first.map((r) => r.id)).toEqual([existing]); + await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS); + + const second = await binding.createBatch([{ id: existing }]); + expect(second).toEqual([]); + + const mixed = await binding.createBatch([ + { id: existing }, + { id: fresh }, + ]); + expect(mixed.map((r) => r.id)).toEqual([fresh]); + + const freshStub = env.ENGINE.get(env.ENGINE.idFromName(fresh)); + await waitUntilLogEvent(freshStub, InstanceEvent.WORKFLOW_SUCCESS); + }); + + it("should collapse duplicate ids within a single batch", async ({ + expect, + }) => { + const id = uniqueId("dedup-in-batch"); + const binding = createBinding(); + const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id)); + setTestWorkflowCallback(async () => "done"); + + const results = await binding.createBatch([{ id }, { id }, { id }]); + expect(results.map((r) => r.id)).toEqual([id]); + + await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS); + }); + + it("should reject the whole batch before creating anything when an id is invalid", async ({ + expect, + }) => { + const good = uniqueId("batch-invalid-good"); + const binding = createBinding(); + setTestWorkflowCallback(async () => "done"); + + await expect( + binding.createBatch([{ id: good }, { id: "#invalid!" }]) + ).rejects.toThrow("Workflow instance has invalid id"); + + // The valid entry listed before the invalid one must not have been + // created. + await expect(binding.get(good)).rejects.toThrow("instance.not_found"); + }); + + it("should create batch entries without ids under generated ids", async ({ + expect, + }) => { + const binding = createBinding(); + setTestWorkflowCallback(async () => "done"); + + const results = await binding.createBatch([{}, {}]); + expect(results).toHaveLength(2); + expect(results[0].id).not.toBe(results[1].id); + + // Wait for both workflows to complete so the fire-and-forget + // init() RPCs settle before teardown. + for (const { id } of results) { + const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id)); + await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS); + } + }); + }); + + describe("deterministic id uniqueness", () => { + it("should throw when creating an instance with an existing id", async ({ + expect, + }) => { + const id = uniqueId("dup-create"); + const binding = createBinding(); + const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id)); + setTestWorkflowCallback(async () => "done"); + + await binding.create({ id }); + await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS); + + await expect(binding.create({ id })).rejects.toThrow( + `(instance.already_exists) Workflow instance with id "${id}" already exists` + ); + }); + + it("should not throw for auto-generated ids", async ({ expect }) => { + const binding = createBinding(); + setTestWorkflowCallback(async () => "done"); + + const first = await binding.create(); + const second = await binding.create(); + expect(first.id).not.toBe(second.id); + + // Wait for both workflows to complete so the fire-and-forget + // init() RPCs settle before teardown. + for (const { id } of [first, second]) { + const engineStub = env.ENGINE.get(env.ENGINE.idFromName(id)); + await waitUntilLogEvent(engineStub, InstanceEvent.WORKFLOW_SUCCESS); + } + }); }); });