[workflows] add individual and batch instance deletion

.changeset/workflows-instances-delete.md

@@ -0,0 +1,10 @@
+---
+"wrangler": minor
+"miniflare": minor
+---
+
+Add individual and batch Workflow instance deletion to the runtime and SDK.
+
+- `env.MY_WORKFLOW.get(id).delete()` deletes one instance. Self-deletion stops the current execution.
+- `env.MY_WORKFLOW.deleteBatch(instanceIds)` deletes up to 100 instances and returns `{ deleted, errors }` per input position.
+- `wrangler workflows instances delete <name> [id..]` deletes instances remotely or with `--local`; IDs can also come from a JSON array passed with `--filename`, with a combined limit of 100.

packages/miniflare/scripts/openapi-filter-config.ts

@@ -985,6 +985,107 @@ const config = {
 					tags: ["Workflows"],
 				},
 			},
+			"/workflows/{workflow_name}/instances/batch/delete": {
+				post: {
+					description: "Deletes multiple workflow instances.",
+					operationId: "workflows-batch-delete-instances",
+					parameters: [
+						{
+							in: "path",
+							name: "workflow_name",
+							required: true,
+							schema: {
+								$ref: "#/components/schemas/workflows_workflow-name",
+							},
+						},
+					],
+					requestBody: {
+						required: true,
+						content: {
+							"application/json": {
+								schema: {
+									type: "object",
+									properties: {
+										instances: {
+											type: "array",
+											minItems: 1,
+											maxItems: 100,
+											items: {
+												type: "string",
+												minLength: 1,
+												maxLength: 100,
+												pattern: "^[a-zA-Z0-9_][a-zA-Z0-9-_]*$",
+											},
+										},
+									},
+									required: ["instances"],
+								},
+							},
+						},
+					},
+					responses: {
+						"200": {
+							content: {
+								"application/json": {
+									schema: {
+										allOf: [
+											{
+												$ref: "#/components/schemas/workers_api-response-common",
+											},
+											{
+												type: "object",
+												properties: {
+													result: {
+														type: "object",
+														properties: {
+															deleted: {
+																type: "array",
+																items: {
+																	type: "object",
+																	properties: {
+																		id: { type: "string" },
+																	},
+																	required: ["id"],
+																},
+															},
+															errors: {
+																type: "array",
+																items: {
+																	type: "object",
+																	properties: {
+																		id: { type: "string" },
+																		code: { type: "number" },
+																		message: { type: "string" },
+																	},
+																	required: ["id", "code", "message"],
+																},
+															},
+														},
+														required: ["deleted", "errors"],
+													},
+												},
+											},
+										],
+									},
+								},
+							},
+							description: "Batch delete Workflow Instances response.",
+						},
+						"4XX": {
+							content: {
+								"application/json": {
+									schema: {
+										$ref: "#/components/schemas/workers_api-response-common-failure",
+									},
+								},
+							},
+							description: "Batch delete Workflow Instances response failure.",
+						},
+					},
+					summary: "Batch Delete Workflow Instances",
+					tags: ["Workflows"],
+				},
+			},
 			"/workflows/{workflow_name}/instances/{instance_id}": {
 				get: {
 					description: "Returns the status details of a workflow instance.",

packages/miniflare/src/index.ts

@@ -7,6 +7,7 @@ import net from "node:net";
 import os from "node:os";
 import path from "node:path";
 import { ReadableStream } from "node:stream/web";
+import { setTimeout as wait } from "node:timers/promises";
 import util from "node:util";
 import zlib from "node:zlib";
 import { checkMacOSVersion } from "@cloudflare/cli-shared-helpers";
@@ -928,6 +929,15 @@ export function _initialiseInstanceRegistry() {
 	return (maybeInstanceRegistry = new Map());
 }
 
+type PendingWorkflowStorageDelete = {
+	promise: Promise<void>;
+	failed: boolean;
+	deleted: boolean;
+};
+
+const WORKFLOW_STORAGE_EXTENSIONS = [".sqlite", ".sqlite-shm", ".sqlite-wal"];
+const WORKFLOW_STORAGE_DELETE_ATTEMPTS = 41;
+
 export class Miniflare {
 	#previousSharedOpts?: PluginSharedOptions;
 	#previousWorkerOpts?: PluginWorkerOptions[];
@@ -946,6 +956,10 @@ export class Miniflare {
 		string,
 		{ browserProcess: Process; wsEndpoint: string }
 	> = new Map();
+	#pendingWorkflowStorageDeletes = new Map<
+		string,
+		PendingWorkflowStorageDelete
+	>();
 
 	readonly #runtime?: Runtime;
 	readonly #removeExitHook?: () => void;
@@ -1385,6 +1399,130 @@ export class Miniflare {
 		}
 	}
 
+	async #deleteWorkflowStorageFiles(
+		instancePath: string,
+		pendingDelete: PendingWorkflowStorageDelete
+	): Promise<void> {
+		let firstError: unknown;
+		let failed = false;
+		for (const ext of WORKFLOW_STORAGE_EXTENSIONS) {
+			const filePath = `${instancePath}${ext}`;
+			for (
+				let attempt = 0;
+				attempt < WORKFLOW_STORAGE_DELETE_ATTEMPTS;
+				attempt++
+			) {
+				try {
+					await fs.promises.unlink(filePath);
+					if (ext === ".sqlite") {
+						pendingDelete.deleted = true;
+					}
+					break;
+				} catch (error) {
+					if (isFileNotFoundError(error)) {
+						break;
+					}
+					const code =
+						typeof error === "object" && error !== null && "code" in error
+							? error.code
+							: undefined;
+					if (
+						(code !== "EBUSY" && code !== "EPERM") ||
+						attempt === WORKFLOW_STORAGE_DELETE_ATTEMPTS - 1
+					) {
+						if (!failed) {
+							firstError = error;
+							failed = true;
+						}
+						break;
+					}
+					await wait(50);
+				}
+			}
+		}
+		if (failed) {
+			throw firstError;
+		}
+	}
+
+	async #runWorkflowStorageDelete(
+		instancePath: string,
+		defer: boolean,
+		pendingDelete: PendingWorkflowStorageDelete,
+		previousDelete?: Promise<void>
+	): Promise<void> {
+		await previousDelete;
+		if (defer) {
+			await wait(100);
+		}
+		try {
+			await this.#deleteWorkflowStorageFiles(instancePath, pendingDelete);
+		} catch (error) {
+			pendingDelete.failed = true;
+			this.#log.error(
+				error instanceof Error ? error : new Error(String(error))
+			);
+		}
+		if (
+			!pendingDelete.failed &&
+			this.#pendingWorkflowStorageDeletes.get(instancePath) === pendingDelete
+		) {
+			this.#pendingWorkflowStorageDeletes.delete(instancePath);
+		}
+	}
+
+	#queueWorkflowStorageDelete(
+		instancePath: string,
+		defer: boolean
+	): PendingWorkflowStorageDelete {
+		const previousDelete =
+			this.#pendingWorkflowStorageDeletes.get(instancePath);
+		const pendingDelete: PendingWorkflowStorageDelete = {
+			deleted: previousDelete?.deleted ?? false,
+			failed: false,
+			promise: Promise.resolve(),
+		};
+		this.#pendingWorkflowStorageDeletes.set(instancePath, pendingDelete);
+		pendingDelete.promise = this.#runWorkflowStorageDelete(
+			instancePath,
+			defer,
+			pendingDelete,
+			previousDelete?.promise
+		);
+		return pendingDelete;
+	}
+
+	async #waitForWorkflowStorageDelete(
+		instancePath: string,
+		retried = false
+	): Promise<Response> {
+		const pendingDelete = this.#pendingWorkflowStorageDeletes.get(instancePath);
+		if (pendingDelete === undefined) {
+			return new Response(null, { status: 204 });
+		}
+		await pendingDelete.promise;
+
+		const latestDelete = this.#pendingWorkflowStorageDeletes.get(instancePath);
+		if (latestDelete === undefined) {
+			return new Response(null, { status: 204 });
+		}
+		if (latestDelete !== pendingDelete) {
+			return this.#waitForWorkflowStorageDelete(instancePath, retried);
+		}
+		if (!pendingDelete.failed) {
+			this.#pendingWorkflowStorageDeletes.delete(instancePath);
+			return new Response(null, { status: 204 });
+		}
+		if (retried || this.#disposeController.signal.aborted) {
+			return new Response("Failed to delete workflow instance", {
+				status: 500,
+			});
+		}
+
+		this.#queueWorkflowStorageDelete(instancePath, false);
+		return this.#waitForWorkflowStorageDelete(instancePath, true);
+	}
+
 	/**
 	 * Deletes a Workflow Engine DO instance by removing its .sqlite file
 	 * (and any associated -shm/-wal files) from the persistence directory.
@@ -1405,9 +1543,14 @@ export class Miniflare {
 		const hexId =
 			slashIndex === -1
 				? null
-				: decodeURIComponent(pathAfterPrefix.slice(slashIndex + 1));
+				: decodeURIComponent(
+						pathAfterPrefix.slice(slashIndex + 1)
+					).toLowerCase();
 
 		assert(workflowName, "Workflow name is required");
+		if (url.searchParams.has("waitForPendingDelete") && !hexId) {
+			return new Response("Instance ID is required", { status: 400 });
+		}
 
 		const coreSharedOpts = this.#sharedOpts.core;
 		const workflowsPersistPath = getPersistPath(
@@ -1426,28 +1569,30 @@ export class Miniflare {
 			return new Response("Invalid workflow name", { status: 400 });
 		}
 
-		const extensions = [".sqlite", ".sqlite-shm", ".sqlite-wal"];
-
 		if (hexId) {
-			// Delete a single instance
-			let deleted = false;
-			for (const ext of extensions) {
-				const filePath = path.join(namespacePath, `${hexId}${ext}`);
-				if (!filePath.startsWith(namespacePath + path.sep)) {
-					return new Response("Invalid instance ID", { status: 400 });
-				}
-				try {
-					await fs.promises.unlink(filePath);
-					if (ext === ".sqlite") {
-						deleted = true;
-					}
-				} catch (e) {
-					if (!isFileNotFoundError(e)) {
-						throw e;
-					}
-				}
+			const instancePath = path.join(namespacePath, hexId);
+			if (!instancePath.startsWith(namespacePath + path.sep)) {
+				return new Response("Invalid instance ID", { status: 400 });
 			}
-			if (!deleted) {
+
+			if (url.searchParams.has("waitForPendingDelete")) {
+				return this.#waitForWorkflowStorageDelete(instancePath);
+			}
+
+			const pendingDelete = this.#queueWorkflowStorageDelete(
+				instancePath,
+				url.searchParams.has("defer")
+			);
+			if (url.searchParams.has("defer")) {
+				return new Response("Accepted", { status: 202 });
+			}
+			await pendingDelete.promise;
+			if (pendingDelete.failed) {
+				return new Response("Failed to delete workflow instance", {
+					status: 500,
+				});
+			}
+			if (!pendingDelete.deleted) {
 				return new Response("Not Found", { status: 404 });
 			}
 		} else {
@@ -1456,7 +1601,9 @@ export class Miniflare {
 				const dirEntries = await fs.promises.readdir(namespacePath);
 				await Promise.all(
 					dirEntries
-						.filter((name) => extensions.some((ext) => name.endsWith(ext)))
+						.filter((name) =>
+							WORKFLOW_STORAGE_EXTENSIONS.some((ext) => name.endsWith(ext))
+						)
 						.map((name) =>
 							fs.promises.unlink(path.join(namespacePath, name)).catch(() => {})
 						)
@@ -1639,7 +1786,10 @@ export class Miniflare {
 			} else if (url.pathname.startsWith("/core/do-storage/")) {
 				response = await this.#handleLoopbackDOStorageRequest(url);
 			} else if (url.pathname.startsWith("/core/workflow-storage/")) {
-				if (request.method === "DELETE") {
+				if (
+					request.method === "DELETE" ||
+					url.searchParams.has("waitForPendingDelete")
+				) {
 					response =
 						await this.#handleLoopbackWorkflowStorageDeleteRequest(url);
 				} else {
@@ -3348,6 +3498,11 @@ export class Miniflare {
 			// Cleanup as much as possible even if `#init()` threw
 			await this.#proxyClient?.dispose();
 			await this.#runtime?.dispose();
+			await Promise.all(
+				[...this.#pendingWorkflowStorageDeletes.values()].map(
+					({ promise }) => promise
+				)
+			);
 			// Close the undici Pool used for dispatching fetch requests to the
 			// runtime. This must happen after the runtime is disposed, so that
 			// in-flight connections are broken and close immediately. Without this,