[workflows-shared] Emulate deterministic-ID uniqueness in the local Workflows binding

.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.

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 });
 					}
 				}

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()))
 	},

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,

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,

packages/workflows-shared/src/binding.ts

@@ -1,6 +1,8 @@
 import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers";
 import { InstanceEvent, instanceStatusName } from "./instance";
 import {
+	duplicateInstanceError,
+	isDuplicateInstanceError,
 	isUserTriggeredPause,
 	isUserTriggeredRestart,
 	isUserTriggeredTerminate,
@@ -141,16 +143,29 @@ export class WorkflowBinding extends WorkerEntrypoint<Env> {
 		super(ctx, env);
 	}
 
-	public async create({
-		id = crypto.randomUUID(),
-		params = {},
-	}: WorkflowInstanceCreateOptions = {}): Promise<{
+	async #instanceExists(id: string): Promise<boolean> {
+		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(), params = {} } = 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);
+		}
+
 		const stubId = this.env.ENGINE.idFromName(id);
 		const stub = this.env.ENGINE.get(stubId);
 		const introspectionSession = workflowIntrospectionSessions.get(
@@ -232,12 +247,42 @@ export class WorkflowBinding extends WorkerEntrypoint<Env> {
 			);
 		}
 
-		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");
+			}
+		}
+
+		// 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<string>();
+		for (const options of batch) {
+			const id = options.id;
+			if (id !== undefined) {
+				if (seenIds.has(id) || (await this.#instanceExists(id))) {
+					continue;
+				}
+				seenIds.add(id);
+			}
+			try {
+				results.push(await this.create(options));
+			} catch (e) {
+				// A concurrent create can claim the id between the existence
+				// check above and create(); the batch contract skips such ids
+				// rather than failing the batch.
+				if (id !== undefined && isDuplicateInstanceError(e)) {
+					continue;
+				}
+				throw e;
+			}
+		}
+		return results;
 	}
 
 	public async unsafeGetBindingName(): Promise<string> {

packages/workflows-shared/src/engine.ts

@@ -1186,6 +1186,12 @@ export class Engine extends DurableObject<Env> {
 		void this.init(accountId, workflow, version, instance, event);
 	}
 
+	async hasInstance(): Promise<boolean> {
+		// 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,

packages/workflows-shared/src/lib/errors.ts

@@ -125,3 +125,17 @@ 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"
+	);
+}
+
+export function isDuplicateInstanceError(error: unknown): boolean {
+	return (
+		error instanceof WorkflowError &&
+		error.message.includes("(instance.already_exists)")
+	);
+}

packages/workflows-shared/tests/binding.test.ts

@@ -3,6 +3,7 @@ import { env } from "cloudflare:workers";
 import { describe, it, vi } from "vitest";
 import { InstanceEvent, InstanceStatus } from "../src";
 import { WorkflowBinding } from "../src/binding";
+import { duplicateInstanceError } from "../src/lib/errors";
 import { setTestWorkflowCallback } from "./test-entry";
 import type { WorkflowHandle } from "../src/binding";
 import type { Engine, EngineLogs } from "../src/engine";
@@ -108,6 +109,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 +191,119 @@ 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 skip an id claimed by a concurrent create instead of failing the batch", async ({
+			expect,
+		}) => {
+			const raced = uniqueId("batch-race");
+			const fresh = uniqueId("batch-race-fresh");
+			const binding = createBinding();
+			const freshStub = env.ENGINE.get(env.ENGINE.idFromName(fresh));
+			setTestWorkflowCallback(async () => "done");
+
+			// Simulate another caller winning the id between createBatch's
+			// existence check and create().
+			const realCreate = binding.create.bind(binding);
+			vi.spyOn(binding, "create").mockImplementation(async (options) => {
+				if (options?.id === raced) {
+					throw duplicateInstanceError(raced);
+				}
+				return realCreate(options);
+			});
+
+			const results = await binding.createBatch([{ id: raced }, { id: fresh }]);
+			expect(results.map((r) => r.id)).toEqual([fresh]);
+
+			await waitUntilLogEvent(freshStub, 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);
+			}
+		});
 	});
 });