[workflows] add individual and batch instance deletion

.changeset/workflows-instances-delete.md

@@ -0,0 +1,9 @@
+---
+"wrangler": minor
+"miniflare": minor
+---
+
+Add individual and batch workflow instance deletion support.
+
+- `wrangler workflows instances delete <name> <id..>` deletes one or up to 100 workflow instances (works remotely and in local dev via `--local`).
+- The Workflows binding now supports both `env.MY_WORKFLOW.get(id).delete()` and `env.MY_WORKFLOW.deleteBatch(instanceIds)` in local development.

packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts

@@ -3,6 +3,7 @@ import type {
 	WorkflowInstanceRestartOptions,
 	WorkflowInstanceTerminateOptions,
 } from "@cloudflare/workflows-shared/src/binding";
+import type { WorkflowBatchDeleteResult } from "@cloudflare/workflows-shared/src/types";
 import type { WorkflowIntrospectionOperation } from "@cloudflare/workflows-shared/src/types";
 
 class WorkflowImpl implements Workflow {
@@ -34,6 +35,10 @@ class WorkflowImpl implements Workflow {
 		});
 	}
 
+	async deleteBatch(instanceIds: string[]): Promise<WorkflowBatchDeleteResult> {
+		return this.binding.deleteBatch({ instances: instanceIds });
+	}
+
 	async unsafeGetBindingName(): Promise<string> {
 		return this.binding.unsafeGetBindingName();
 	}
@@ -124,6 +129,12 @@ class InstanceImpl implements WorkflowInstance {
 		await instance.restart(options);
 	}
 
+	public async delete(): Promise<void> {
+		using instance = await this.getInstance();
+		// TODO(vaish): remove cast once @cloudflare/workers-types ships instance delete
+		await (instance as unknown as { delete(): Promise<void> }).delete();
+	}
+
 	public async status(): Promise<InstanceStatus> {
 		using instance = await this.getInstance();
 		using res = (await instance.status()) as InstanceStatus & Disposable;

packages/miniflare/test/plugins/workflows/index.spec.ts

@@ -169,6 +169,18 @@ export default {
 		return Response.json(await instance.status());
 	}
 
+	if (url.pathname === "/delete") {
+		const instance = await env.LIFECYCLE_WORKFLOW.get(id);
+		await instance.delete();
+		return Response.json({ ok: true });
+	}
+
+	if (url.pathname === "/deleteBatch") {
+		return Response.json(
+			await env.LIFECYCLE_WORKFLOW.deleteBatch(url.searchParams.getAll("id"))
+		);
+	}
+
 		if (url.pathname === "/sendEvent") {
 			const instance = await env.LIFECYCLE_WORKFLOW.get(id);
 			await instance.sendEvent({ type: "continue", payload: { sent: true } });
@@ -305,6 +317,43 @@ describe("workflow instance lifecycle methods", () => {
 		await waitForStatus(mf, "terminate-test", "terminated");
 	});
 
+	test("delete a workflow", async ({ expect }) => {
+		const tmp = await useTmp();
+		const mf = new Miniflare(lifecycleMiniflareOpts(tmp));
+		useDispose(mf);
+
+		const createResponse = await mf.dispatchFetch(
+			"http://localhost/create?id=delete-one"
+		);
+		await createResponse.text();
+
+		const deleteResponse = await mf.dispatchFetch(
+			"http://localhost/delete?id=delete-one"
+		);
+		expect(await deleteResponse.json()).toEqual({ ok: true });
+	});
+
+	test("delete multiple workflows", async ({ expect }) => {
+		const tmp = await useTmp();
+		const mf = new Miniflare(lifecycleMiniflareOpts(tmp));
+		useDispose(mf);
+
+		for (const id of ["delete-1", "delete-2"]) {
+			const response = await mf.dispatchFetch(
+				`http://localhost/create?id=${id}`
+			);
+			await response.text();
+		}
+
+		const response = await mf.dispatchFetch(
+			"http://localhost/deleteBatch?id=delete-1&id=delete-2"
+		);
+		expect(await response.json()).toEqual({
+			deleted: [{ id: "delete-1" }, { id: "delete-2" }],
+			errors: [],
+		});
+	});
+
 	test("restart a running workflow", async ({ expect }) => {
 		const tmp = await useTmp();
 		const mf = new Miniflare(lifecycleMiniflareOpts(tmp));

packages/workflows-shared/src/binding.ts

@@ -1,6 +1,7 @@
 import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers";
 import { InstanceEvent, instanceStatusName } from "./instance";
 import {
+	isUserTriggeredDelete,
 	isUserTriggeredPause,
 	isUserTriggeredRestart,
 	isUserTriggeredTerminate,
@@ -16,11 +17,14 @@ import type {
 } from "./engine";
 import type { InstanceStatus as EngineInstanceStatus } from "./instance";
 import type {
+	WorkflowBatchDeleteResult,
 	WorkflowInstanceModifier,
 	WorkflowIntrospectionOperation,
 	WorkflowIntrospectionStreamResult,
 } from "./types";
 
+export type { WorkflowBatchDeleteResult } from "./types";
+
 type Env = {
 	ENGINE: DurableObjectNamespace<Engine>;
 	BINDING_NAME: string;
@@ -240,6 +244,64 @@ export class WorkflowBinding extends WorkerEntrypoint<Env> {
 		);
 	}
 
+	public async deleteBatch(options: {
+		instances: string[];
+	}): Promise<WorkflowBatchDeleteResult> {
+		const instanceIds = options?.instances;
+		if (
+			!Array.isArray(instanceIds) ||
+			instanceIds.length === 0 ||
+			instanceIds.length > 100
+		) {
+			throw new WorkflowError(
+				"Batch must contain between 1 and 100 instance IDs"
+			);
+		}
+		if (!instanceIds.every(isValidWorkflowInstanceId)) {
+			throw new WorkflowError("Workflow instance has invalid id");
+		}
+
+		const uniqueIds = [...new Set(instanceIds)];
+		const deletions: Promise<void>[] = [];
+		for (const id of uniqueIds) {
+			const stub = this.env.ENGINE.get(this.env.ENGINE.idFromName(id));
+			deletions.push(stub.deleteInstance());
+		}
+
+		const settled = await Promise.allSettled(deletions);
+		const resultsById = new Map(
+			uniqueIds.map((id, index) => [id, settled[index]])
+		);
+		const result: WorkflowBatchDeleteResult = { deleted: [], errors: [] };
+		for (const id of instanceIds) {
+			const deletion = resultsById.get(id);
+			if (deletion === undefined) {
+				throw new Error("Missing batch deletion result");
+			}
+			if (
+				deletion.status === "fulfilled" ||
+				isUserTriggeredDelete(deletion.reason)
+			) {
+				result.deleted.push({ id });
+				continue;
+			}
+
+			const message =
+				deletion.reason instanceof Error
+					? deletion.reason.message
+					: String(deletion.reason);
+			const isNotFound = message.includes("(instance.not_found)");
+			result.errors.push({
+				id,
+				code: isNotFound ? 10400 : 10001,
+				message: isNotFound
+					? "workflows.api.error.instance.not_found"
+					: "workflows.api.error.internal_server",
+			});
+		}
+		return result;
+	}
+
 	public async unsafeGetBindingName(): Promise<string> {
 		// async because of rpc
 		return this.env.BINDING_NAME;
@@ -377,6 +439,17 @@ export class WorkflowHandle extends RpcTarget implements WorkflowInstance {
 		}
 	}
 
+	public async delete(): Promise<void> {
+		try {
+			await this.stub.deleteInstance();
+		} catch (e) {
+			// delete aborts the instance
+			if (!isUserTriggeredDelete(e)) {
+				throw e;
+			}
+		}
+	}
+
 	public async restart(
 		options?: WorkflowInstanceRestartOptions
 	): Promise<void> {

packages/workflows-shared/src/engine.ts

@@ -1037,6 +1037,18 @@ export class Engine extends DurableObject<Env> {
 		await this.abort(ABORT_REASONS.USER_TERMINATE);
 	}
 
+	async deleteInstance(): Promise<void> {
+		if ((await this.ctx.storage.get(INSTANCE_METADATA)) === undefined) {
+			throw createWorkflowError(
+				"Instance does not exist",
+				"instance.not_found"
+			);
+		}
+
+		await this.ctx.storage.deleteAll();
+		await this.abort(ABORT_REASONS.USER_DELETE);
+	}
+
 	async userTriggeredPause() {
 		const status = await this.getStatus();
 

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

@@ -70,6 +70,7 @@ export const ABORT_REASONS = {
 	USER_PAUSE: `${ABORT_PREFIX} User called pause`,
 	USER_RESTART: `${ABORT_PREFIX} User called restart`,
 	USER_TERMINATE: `${ABORT_PREFIX} User called terminate`,
+	USER_DELETE: `${ABORT_PREFIX} User called delete`,
 	NON_RETRYABLE_ERROR: `${ABORT_PREFIX} A step threw a NonRetryableError`,
 	NOT_SERIALISABLE: `${ABORT_PREFIX} Value is not serialisable`,
 	STORAGE_LIMIT_EXCEEDED: `${ABORT_PREFIX} Storage limit exceeded`,
@@ -110,6 +111,10 @@ export function isUserTriggeredTerminate(e: unknown): boolean {
 	return getErrorMessage(e) === ABORT_REASONS.USER_TERMINATE;
 }
 
+export function isUserTriggeredDelete(e: unknown): boolean {
+	return getErrorMessage(e) === ABORT_REASONS.USER_DELETE;
+}
+
 function getCompatFlag(name: string): boolean {
 	// eslint-disable-next-line @typescript-eslint/no-explicit-any -- safe globalThis access for environments where cloudflare global may not exist
 	return (globalThis as any).Cloudflare?.compatibilityFlags?.[name] ?? false;

packages/workflows-shared/src/types.ts

@@ -1,3 +1,12 @@
+export interface WorkflowBatchDeleteResult {
+	deleted: { id: string }[];
+	errors: Array<{
+		id: string;
+		code: number;
+		message: string;
+	}>;
+}
+
 export type WorkflowStepSelector = {
 	name: string;
 	index?: number;