From f6d156e691c19c308d8d6e56783b3fd04ff2b0b8 Mon Sep 17 00:00:00 2001 From: vaish Date: Thu, 23 Jul 2026 15:33:10 +0530 Subject: [PATCH] feat: support individual and batch workflow instance deletion --- .changeset/workflows-instances-delete.md | 9 + .../scripts/openapi-filter-config.ts | 102 +++++++++ .../miniflare/src/plugins/workflows/index.ts | 10 + .../workers/local-explorer/explorer.worker.ts | 13 ++ .../workers/local-explorer/generated/index.ts | 5 + .../local-explorer/generated/types.gen.ts | 42 ++++ .../local-explorer/generated/zod.gen.ts | 46 ++++ .../workers/local-explorer/openapi.local.json | 109 +++++++++ .../local-explorer/resources/workflows.ts | 58 ++++- .../src/workers/local-explorer/route-names.ts | 4 + .../workflows/wrapped-binding.worker.ts | 61 ++++- .../test/plugins/workflows/index.spec.ts | 76 +++++++ packages/workflows-shared/src/binding.ts | 87 ++++++++ packages/workflows-shared/src/engine.ts | 12 + packages/workflows-shared/src/lib/errors.ts | 5 + .../workflows-shared/src/lib/validators.ts | 6 +- packages/workflows-shared/src/types.ts | 9 + .../workflows-shared/tests/binding.test.ts | 164 ++++++++++++++ .../workflows-shared/tests/validators.test.ts | 3 + .../wrangler/src/__tests__/workflows.test.ts | 210 ++++++++++++++++++ packages/wrangler/src/index.ts | 5 + .../workflows/commands/instances/delete.ts | 104 +++++++++ packages/wrangler/src/workflows/local.ts | 17 ++ packages/wrangler/src/workflows/utils.ts | 20 ++ 24 files changed, 1167 insertions(+), 10 deletions(-) create mode 100644 .changeset/workflows-instances-delete.md create mode 100644 packages/wrangler/src/workflows/commands/instances/delete.ts diff --git a/.changeset/workflows-instances-delete.md b/.changeset/workflows-instances-delete.md new file mode 100644 index 0000000000..45a42b8114 --- /dev/null +++ b/.changeset/workflows-instances-delete.md @@ -0,0 +1,9 @@ +--- +"wrangler": minor +"miniflare": minor +--- + +Add a workflow instance deletion command and local development support. + +- `wrangler workflows instances delete ` deletes one or up to 100 workflow instances remotely or in local development with `--local`. +- Local development now supports the existing `env.MY_WORKFLOW.get(id).delete()` and `env.MY_WORKFLOW.deleteBatch(instanceIds)` runtime APIs. diff --git a/packages/miniflare/scripts/openapi-filter-config.ts b/packages/miniflare/scripts/openapi-filter-config.ts index d0968087c3..552fdb7c5f 100644 --- a/packages/miniflare/scripts/openapi-filter-config.ts +++ b/packages/miniflare/scripts/openapi-filter-config.ts @@ -985,6 +985,108 @@ 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: + "^(?!batch$|terminate$|terminateAll$)[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.", diff --git a/packages/miniflare/src/plugins/workflows/index.ts b/packages/miniflare/src/plugins/workflows/index.ts index a89b521912..f32a9b27b7 100644 --- a/packages/miniflare/src/plugins/workflows/index.ts +++ b/packages/miniflare/src/plugins/workflows/index.ts @@ -9,6 +9,7 @@ import { PersistenceSchema, ProxyNodeBinding, SERVICE_DEV_REGISTRY_PROXY, + WORKER_BINDING_SERVICE_LOOPBACK, } from "../shared"; import type { Service } from "../../runtime"; import type { Plugin, RemoteProxyConnectionString } from "../shared"; @@ -68,6 +69,15 @@ export const WORKFLOWS_PLUGIN: Plugin< entrypoint: "WorkflowBinding", }, }, + ...(workflow.remoteProxyConnectionString === undefined + ? [ + WORKER_BINDING_SERVICE_LOOPBACK, + { + name: "MINIFLARE_WORKFLOW_NAME", + json: JSON.stringify(workflow.name), + }, + ] + : []), ], }, }) diff --git a/packages/miniflare/src/workers/local-explorer/explorer.worker.ts b/packages/miniflare/src/workers/local-explorer/explorer.worker.ts index e9b4b12786..3258ee6f6c 100644 --- a/packages/miniflare/src/workers/local-explorer/explorer.worker.ts +++ b/packages/miniflare/src/workers/local-explorer/explorer.worker.ts @@ -18,6 +18,7 @@ import { zWorkersKvNamespaceListANamespaceSKeysData, zWorkersKvNamespaceListNamespacesData, zObservabilityQueryData, + zWorkflowsBatchDeleteInstancesData, zWorkflowsChangeInstanceStatusData, zWorkflowsListInstancesData, } from "./generated/zod.gen"; @@ -45,6 +46,7 @@ import { createWorkflowInstance, deleteWorkflow, deleteWorkflowInstance, + deleteWorkflowInstances, getWorkflowDetails, getWorkflowInstanceDetails, listWorkflowInstances, @@ -315,6 +317,17 @@ app.post("/api/workflows/:workflow_name/instances", (c) => createWorkflowInstance(c, c.req.param("workflow_name")) ); +app.post( + "/api/workflows/:workflow_name/instances/batch/delete", + validateRequestBody(zWorkflowsBatchDeleteInstancesData.shape.body), + (c) => + deleteWorkflowInstances( + c, + c.req.param("workflow_name"), + c.req.valid("json") + ) +); + app.get("/api/workflows/:workflow_name/instances/:instance_id", (c) => getWorkflowInstanceDetails( c, diff --git a/packages/miniflare/src/workers/local-explorer/generated/index.ts b/packages/miniflare/src/workers/local-explorer/generated/index.ts index 4533d80886..92472cf8f4 100644 --- a/packages/miniflare/src/workers/local-explorer/generated/index.ts +++ b/packages/miniflare/src/workers/local-explorer/generated/index.ts @@ -174,6 +174,11 @@ export type { WorkersNamespaceWritable, WorkersObject, WorkersSchemasId, + WorkflowsBatchDeleteInstancesData, + WorkflowsBatchDeleteInstancesError, + WorkflowsBatchDeleteInstancesErrors, + WorkflowsBatchDeleteInstancesResponse, + WorkflowsBatchDeleteInstancesResponses, WorkflowsChangeInstanceStatusData, WorkflowsChangeInstanceStatusError, WorkflowsChangeInstanceStatusErrors, diff --git a/packages/miniflare/src/workers/local-explorer/generated/types.gen.ts b/packages/miniflare/src/workers/local-explorer/generated/types.gen.ts index bbe24d8029..e67a637b92 100644 --- a/packages/miniflare/src/workers/local-explorer/generated/types.gen.ts +++ b/packages/miniflare/src/workers/local-explorer/generated/types.gen.ts @@ -1666,6 +1666,48 @@ export type WorkflowsCreateInstanceResponses = { export type WorkflowsCreateInstanceResponse = WorkflowsCreateInstanceResponses[keyof WorkflowsCreateInstanceResponses]; +export type WorkflowsBatchDeleteInstancesData = { + body: { + instances: Array; + }; + path: { + workflow_name: WorkflowsWorkflowName; + }; + query?: never; + url: "/workflows/{workflow_name}/instances/batch/delete"; +}; + +export type WorkflowsBatchDeleteInstancesErrors = { + /** + * Batch delete Workflow Instances response failure. + */ + "4XX": WorkersApiResponseCommonFailure; +}; + +export type WorkflowsBatchDeleteInstancesError = + WorkflowsBatchDeleteInstancesErrors[keyof WorkflowsBatchDeleteInstancesErrors]; + +export type WorkflowsBatchDeleteInstancesResponses = { + /** + * Batch delete Workflow Instances response. + */ + 200: WorkersApiResponseCommon & { + result?: { + deleted: Array<{ + id: string; + }>; + errors: Array<{ + id: string; + code: number; + message: string; + }>; + }; + }; +}; + +export type WorkflowsBatchDeleteInstancesResponse = + WorkflowsBatchDeleteInstancesResponses[keyof WorkflowsBatchDeleteInstancesResponses]; + export type WorkflowsDeleteInstanceData = { body?: never; path: { diff --git a/packages/miniflare/src/workers/local-explorer/generated/zod.gen.ts b/packages/miniflare/src/workers/local-explorer/generated/zod.gen.ts index 29cf7fd6b1..f038e97bc1 100644 --- a/packages/miniflare/src/workers/local-explorer/generated/zod.gen.ts +++ b/packages/miniflare/src/workers/local-explorer/generated/zod.gen.ts @@ -1047,6 +1047,52 @@ export const zWorkflowsCreateInstanceResponse = zWorkersApiResponseCommon.and( }) ); +export const zWorkflowsBatchDeleteInstancesData = z.object({ + body: z.object({ + instances: z + .array( + z + .string() + .min(1) + .max(100) + .regex( + /^(?!batch$|terminate$|terminateAll$)[a-zA-Z0-9_][a-zA-Z0-9-_]*$/ + ) + ) + .min(1) + .max(100), + }), + path: z.object({ + workflow_name: zWorkflowsWorkflowName, + }), + query: z.never().optional(), +}); + +/** + * Batch delete Workflow Instances response. + */ +export const zWorkflowsBatchDeleteInstancesResponse = + zWorkersApiResponseCommon.and( + z.object({ + result: z + .object({ + deleted: z.array( + z.object({ + id: z.string(), + }) + ), + errors: z.array( + z.object({ + id: z.string(), + code: z.number(), + message: z.string(), + }) + ), + }) + .optional(), + }) + ); + export const zWorkflowsDeleteInstanceData = z.object({ body: z.never().optional(), path: z.object({ diff --git a/packages/miniflare/src/workers/local-explorer/openapi.local.json b/packages/miniflare/src/workers/local-explorer/openapi.local.json index 700e8af3e6..c0eb711f09 100644 --- a/packages/miniflare/src/workers/local-explorer/openapi.local.json +++ b/packages/miniflare/src/workers/local-explorer/openapi.local.json @@ -1632,6 +1632,115 @@ "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": "^(?!batch$|terminate$|terminateAll$)[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.", diff --git a/packages/miniflare/src/workers/local-explorer/resources/workflows.ts b/packages/miniflare/src/workers/local-explorer/resources/workflows.ts index f626a48d5a..568d85dd99 100644 --- a/packages/miniflare/src/workers/local-explorer/resources/workflows.ts +++ b/packages/miniflare/src/workers/local-explorer/resources/workflows.ts @@ -7,6 +7,7 @@ import { errorResponse, wrapResponse } from "../common"; import type { AppContext } from "../common"; import type { Env } from "../explorer.worker"; import type { + WorkflowsBatchDeleteInstancesData, WorkflowsChangeInstanceStatusData, WorkflowsWorkflow, } from "../generated"; @@ -15,6 +16,7 @@ import type { RestartFromStep, WorkflowInstanceTerminateOptions, } from "@cloudflare/workflows-shared/src/binding"; +import type { WorkflowBatchDeleteResult } from "@cloudflare/workflows-shared/src/types"; import type { z } from "zod"; // ============================================================================ @@ -33,6 +35,10 @@ interface DirectoryEntry { birthtimeMs: number; } +interface WorkflowWithBatchDelete extends Workflow { + deleteBatch(instanceIds: string[]): Promise; +} + /** Methods on a WorkflowInstance handle (from workflow.get()). */ interface WorkflowHandle { pause(): Promise; @@ -176,12 +182,15 @@ function getEngineNamespace( * Get the Workflow proxy binding for a workflow. * Used for operations that go through the standard Workflow interface (create). */ -function getWorkflowBinding(env: Env, workflowName: string): Workflow | null { +function getWorkflowBinding( + env: Env, + workflowName: string +): WorkflowWithBatchDelete | null { const info = env.LOCAL_EXPLORER_BINDING_MAP.workflows[workflowName]; if (!info) { return null; } - return env[info.binding] as Workflow; + return env[info.binding] as WorkflowWithBatchDelete; } function getLocalWorkflows(env: Env): WorkflowsWorkflow[] { @@ -945,6 +954,51 @@ export async function changeWorkflowInstanceStatus( } } +/** + * Delete multiple workflow instances through the local Workflow binding. + */ +export async function deleteWorkflowInstances( + c: AppContext, + workflowName: string, + body: WorkflowsBatchDeleteInstancesData["body"] +): Promise { + const workflow = getWorkflowBinding(c.env, workflowName); + + if (!workflow) { + const ownerMiniflare = await findWorkflowOwner(c, workflowName); + if (ownerMiniflare) { + const response = await fetchFromPeer( + ownerMiniflare, + `/workflows/${encodeURIComponent(workflowName)}/instances/batch/delete`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + } + ); + if (response) { + return response; + } + } + + return errorResponse( + 404, + WORKFLOW_ERROR_NOT_FOUND, + `Workflow '${workflowName}' not found.` + ); + } + + try { + const result = await workflow.deleteBatch(body.instances); + statusCountsCache.delete(workflowName); + return c.json(wrapResponse(result)); + } catch (error) { + const message = + error instanceof Error ? error.message : "Failed to delete instances"; + return errorResponse(500, 10001, message); + } +} + /** * Delete a workflow instance by removing its .sqlite persistence files. * diff --git a/packages/miniflare/src/workers/local-explorer/route-names.ts b/packages/miniflare/src/workers/local-explorer/route-names.ts index e5ed4b3390..6c5b49bf22 100644 --- a/packages/miniflare/src/workers/local-explorer/route-names.ts +++ b/packages/miniflare/src/workers/local-explorer/route-names.ts @@ -16,6 +16,10 @@ const ROUTE_PATTERNS: [RegExp, string][] = [ [/^\/r2\/buckets\/[^/]+\/objects$/, "r2.objects"], [/^\/r2\/buckets\/[^/]+$/, "r2.bucket"], [/^\/r2\/buckets$/, "r2.buckets"], + [ + /^\/workflows\/[^/]+\/instances\/batch\/delete$/, + "workflows.instances.batch_delete", + ], [ /^\/workflows\/[^/]+\/instances\/[^/]+\/events\/[^/]+$/, "workflows.instance.event", diff --git a/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts b/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts index fdaa952106..cde6f5c068 100644 --- a/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts +++ b/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts @@ -3,13 +3,42 @@ 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"; +type Env = { + binding: WorkflowBinding; + MINIFLARE_LOOPBACK?: Fetcher; + MINIFLARE_WORKFLOW_NAME?: string; +}; + +async function deletePersistedInstance(env: Env, id: string): Promise { + if ( + env.MINIFLARE_LOOPBACK === undefined || + env.MINIFLARE_WORKFLOW_NAME === undefined + ) { + return; + } + + const hexId = await env.binding.unsafeGetInstanceStorageId(id); + const response = await env.MINIFLARE_LOOPBACK.fetch( + `http://localhost/core/workflow-storage/${encodeURIComponent(env.MINIFLARE_WORKFLOW_NAME)}/${hexId}`, + { method: "DELETE" } + ); + if (!response.ok && response.status !== 404) { + throw new Error(`Failed to delete persisted workflow instance '${id}'`); + } +} + class WorkflowImpl implements Workflow { - constructor(private binding: WorkflowBinding) {} + private binding: WorkflowBinding; + + constructor(private env: Env) { + this.binding = env.binding; + } async get(id: string): Promise { - const instanceHandle = new InstanceImpl(id, this.binding); + const instanceHandle = new InstanceImpl(id, this.binding, this.env); // throws instance.not_found if instance doesn't exist // this is needed for backwards compat await instanceHandle.status(); @@ -22,7 +51,7 @@ class WorkflowImpl implements Workflow { using result = (await this.binding.create(options)) as WorkflowInstance & Disposable; - return new InstanceImpl(result.id, this.binding); + return new InstanceImpl(result.id, this.binding, this.env); } async createBatch( @@ -30,10 +59,20 @@ class WorkflowImpl implements Workflow { ): Promise { const result = await this.binding.createBatch(options); return result.map((res) => { - return new InstanceImpl(res.id, this.binding); + return new InstanceImpl(res.id, this.binding, this.env); }); } + async deleteBatch(instanceIds: string[]): Promise { + const result = await this.binding.deleteBatch({ instances: instanceIds }); + await Promise.allSettled( + [...new Set(result.deleted.map(({ id }) => id))].map((id) => + deletePersistedInstance(this.env, id) + ) + ); + return result; + } + async unsafeGetBindingName(): Promise { return this.binding.unsafeGetBindingName(); } @@ -88,7 +127,8 @@ class WorkflowImpl implements Workflow { class InstanceImpl implements WorkflowInstance { constructor( public id: string, - private binding: WorkflowBinding + private binding: WorkflowBinding, + private env: Env ) {} private async getInstance(): Promise { @@ -124,6 +164,13 @@ class InstanceImpl implements WorkflowInstance { await instance.restart(options); } + public async delete(): Promise { + using instance = await this.getInstance(); + // TODO(vaish): remove cast once @cloudflare/workers-types ships instance delete + await (instance as unknown as { delete(): Promise }).delete(); + await deletePersistedInstance(this.env, this.id); + } + public async status(): Promise { using instance = await this.getInstance(); using res = (await instance.status()) as InstanceStatus & Disposable; @@ -139,8 +186,8 @@ class InstanceImpl implements WorkflowInstance { } } -export function makeBinding(env: { binding: WorkflowBinding }): Workflow { - return new WorkflowImpl(env.binding); +export function makeBinding(env: Env): Workflow { + return new WorkflowImpl(env); } export default makeBinding; diff --git a/packages/miniflare/test/plugins/workflows/index.spec.ts b/packages/miniflare/test/plugins/workflows/index.spec.ts index 56061c1fb7..c03cc3aa58 100644 --- a/packages/miniflare/test/plugins/workflows/index.spec.ts +++ b/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 } }); @@ -195,6 +207,15 @@ function lifecycleMiniflareOpts(tmp: string): MiniflareOptions { }; } +async function getPersistedInstanceFiles(tmp: string): Promise { + const files = await fs.readdir( + `${tmp}/miniflare-workflows-LIFECYCLE_WORKFLOW` + ); + return files.filter( + (file) => file.endsWith(".sqlite") && file !== "metadata.sqlite" + ); +} + async function waitForStatus( mf: Miniflare, id: string, @@ -305,6 +326,61 @@ 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(); + + expect(await getPersistedInstanceFiles(tmp)).toHaveLength(1); + const deleteResponse = await mf.dispatchFetch( + "http://localhost/delete?id=delete-one" + ); + expect(await deleteResponse.json()).toEqual({ ok: true }); + expect(await getPersistedInstanceFiles(tmp)).toHaveLength(0); + + const statusResponse = await mf.dispatchFetch( + "http://localhost/status?id=delete-one" + ); + expect(statusResponse.status).toBe(500); + expect(await statusResponse.text()).toContain("instance.not_found"); + }); + + 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(); + } + + expect(await getPersistedInstanceFiles(tmp)).toHaveLength(2); + const response = await mf.dispatchFetch( + "http://localhost/deleteBatch?id=delete-1&id=delete-2&id=delete-1" + ); + expect(await response.json()).toEqual({ + deleted: [{ id: "delete-1" }, { id: "delete-2" }, { id: "delete-1" }], + errors: [], + }); + expect(await getPersistedInstanceFiles(tmp)).toHaveLength(0); + + for (const id of ["delete-1", "delete-2"]) { + const statusResponse = await mf.dispatchFetch( + `http://localhost/status?id=${id}` + ); + expect(statusResponse.status).toBe(500); + await statusResponse.text(); + } + }); + test("restart a running workflow", async ({ expect }) => { const tmp = await useTmp(); const mf = new Miniflare(lifecycleMiniflareOpts(tmp)); diff --git a/packages/workflows-shared/src/binding.ts b/packages/workflows-shared/src/binding.ts index 9574eeb4c4..fc583de72f 100644 --- a/packages/workflows-shared/src/binding.ts +++ b/packages/workflows-shared/src/binding.ts @@ -1,8 +1,10 @@ import { RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; import { InstanceEvent, instanceStatusName } from "./instance"; import { + isUserTriggeredDelete, isUserTriggeredPause, isUserTriggeredRestart, + createWorkflowError, isUserTriggeredTerminate, WorkflowError, } from "./lib/errors"; @@ -16,11 +18,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; BINDING_NAME: string; @@ -240,6 +245,77 @@ export class WorkflowBinding extends WorkerEntrypoint { ); } + public async deleteBatch(options: { + instances: string[]; + }): Promise { + const instanceIds = options?.instances; + if (!Array.isArray(instanceIds)) { + throw createWorkflowError("Provided argument is invalid", "body"); + } + if (instanceIds.length > 100) { + throw createWorkflowError( + "batchDeleteInstances only supports 100 instances at a time", + "body" + ); + } + if (instanceIds.length === 0) { + throw createWorkflowError( + "batchDeleteInstances should have at least 1 instance", + "body" + ); + } + if (!instanceIds.every(isValidWorkflowInstanceId)) { + throw createWorkflowError( + "Instance ID is invalid", + "instance.invalid_id" + ); + } + + const uniqueIds = [...new Set(instanceIds)]; + const deletions: Promise[] = []; + 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 unsafeGetInstanceStorageId(id: string): Promise { + return this.env.ENGINE.idFromName(id).toString(); + } + public async unsafeGetBindingName(): Promise { // async because of rpc return this.env.BINDING_NAME; @@ -377,6 +453,17 @@ export class WorkflowHandle extends RpcTarget implements WorkflowInstance { } } + public async delete(): Promise { + try { + await this.stub.deleteInstance(); + } catch (e) { + // delete aborts the instance + if (!isUserTriggeredDelete(e)) { + throw e; + } + } + } + public async restart( options?: WorkflowInstanceRestartOptions ): Promise { diff --git a/packages/workflows-shared/src/engine.ts b/packages/workflows-shared/src/engine.ts index a6e96855e1..9d5550d5dc 100644 --- a/packages/workflows-shared/src/engine.ts +++ b/packages/workflows-shared/src/engine.ts @@ -1037,6 +1037,18 @@ export class Engine extends DurableObject { await this.abort(ABORT_REASONS.USER_TERMINATE); } + async deleteInstance(): Promise { + 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(); diff --git a/packages/workflows-shared/src/lib/errors.ts b/packages/workflows-shared/src/lib/errors.ts index 25614b3cd4..af9632da56 100644 --- a/packages/workflows-shared/src/lib/errors.ts +++ b/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; diff --git a/packages/workflows-shared/src/lib/validators.ts b/packages/workflows-shared/src/lib/validators.ts index 6818581abb..12d9b38a9c 100644 --- a/packages/workflows-shared/src/lib/validators.ts +++ b/packages/workflows-shared/src/lib/validators.ts @@ -14,6 +14,7 @@ const ALLOWED_WORKFLOW_INSTANCE_ID_REGEX = new RegExp( ALLOWED_STRING_ID_PATTERN ); const ALLOWED_WORKFLOW_NAME_REGEX = ALLOWED_WORKFLOW_INSTANCE_ID_REGEX; +const RESERVED_WORKFLOW_INSTANCE_IDS = ["batch", "terminate", "terminateAll"]; // eslint-disable-next-line no-control-regex -- intentional use of control character range to detect invalid characters in workflow names const CONTROL_CHAR_REGEX = new RegExp("[\x00-\x1F]"); @@ -34,7 +35,10 @@ export function isValidWorkflowInstanceId(id: string): boolean { return false; } - if (id.length > MAX_WORKFLOW_INSTANCE_ID_LENGTH) { + if ( + RESERVED_WORKFLOW_INSTANCE_IDS.includes(id) || + id.length > MAX_WORKFLOW_INSTANCE_ID_LENGTH + ) { return false; } diff --git a/packages/workflows-shared/src/types.ts b/packages/workflows-shared/src/types.ts index ae8358ac35..d5ae9b869e 100644 --- a/packages/workflows-shared/src/types.ts +++ b/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; diff --git a/packages/workflows-shared/tests/binding.test.ts b/packages/workflows-shared/tests/binding.test.ts index 0341701966..1f468485c0 100644 --- a/packages/workflows-shared/tests/binding.test.ts +++ b/packages/workflows-shared/tests/binding.test.ts @@ -130,6 +130,7 @@ describe("WorkflowBinding", () => { resume: expect.any(Function), terminate: expect.any(Function), restart: expect.any(Function), + delete: expect.any(Function), }); // Wait for the workflow to complete before the test ends so @@ -144,6 +145,169 @@ describe("WorkflowBinding", () => { }); }); + describe("delete()", () => { + it("should delete an instance and wipe its stored state", async ({ + expect, + }) => { + const id = uniqueId(); + const binding = createBinding(); + + setTestWorkflowCallback(async () => "done"); + await binding.create({ id }); + + const instance = await binding.get(id); + await vi.waitUntil( + async () => { + const status = await instance.status(); + return status.status === "complete"; + }, + { timeout: 5000 } + ); + + await expect( + (instance as unknown as { delete(): Promise }).delete() + ).resolves.toBeUndefined(); + await expect(binding.get(id)).rejects.toThrow("instance.not_found"); + }); + + it("should let a running instance delete itself and stop execution", async ({ + expect, + }) => { + const id = uniqueId(); + const binding = createBinding(); + let deleteStarted = false; + let continuedAfterDelete = false; + + setTestWorkflowCallback(async () => { + deleteStarted = true; + const instance = await binding.get(id); + await (instance as unknown as { delete(): Promise }).delete(); + continuedAfterDelete = true; + }); + await binding.create({ id }); + await vi.waitUntil(() => deleteStarted, { timeout: 5000 }); + + await vi.waitUntil( + async () => { + try { + await binding.get(id); + return false; + } catch { + return true; + } + }, + { timeout: 5000 } + ); + + await scheduler.wait(50); + expect(continuedAfterDelete).toBe(false); + await expect(binding.get(id)).rejects.toThrow("instance.not_found"); + }); + }); + + describe("deleteBatch()", () => { + it("should delete instances and wipe their stored state", async ({ + expect, + }) => { + const ids = [uniqueId(), uniqueId()]; + const binding = createBinding(); + + setTestWorkflowCallback(async () => "done"); + await binding.createBatch(ids.map((id) => ({ id }))); + + for (const id of ids) { + const instance = await binding.get(id); + await vi.waitUntil( + async () => { + const status = await instance.status(); + return status.status === "complete"; + }, + { timeout: 5000 } + ); + } + + await expect(binding.deleteBatch({ instances: ids })).resolves.toEqual({ + deleted: ids.map((id) => ({ id })), + errors: [], + }); + for (const id of ids) { + await expect(binding.get(id)).rejects.toThrow("instance.not_found"); + } + }); + + it("should report duplicate missing instances once per request entry", async ({ + expect, + }) => { + const binding = createBinding(); + await expect( + binding.deleteBatch({ + instances: ["missing-instance", "missing-instance"], + }) + ).resolves.toEqual({ + deleted: [], + errors: [ + { + id: "missing-instance", + code: 10400, + message: "workflows.api.error.instance.not_found", + }, + { + id: "missing-instance", + code: 10400, + message: "workflows.api.error.instance.not_found", + }, + ], + }); + }); + + it("should normalize unexpected deletion errors", async ({ expect }) => { + const deleteInstance = vi + .fn() + .mockRejectedValue(new Error("sensitive failure")); + const binding = new WorkflowBinding(createExecutionContext(), { + ENGINE: { + idFromName: (id: string) => id, + get: () => ({ deleteInstance }), + } as unknown as DurableObjectNamespace, + BINDING_NAME: "TEST_WORKFLOW", + WORKFLOW_NAME: "test-workflow", + }); + + await expect( + binding.deleteBatch({ instances: ["broken-instance"] }) + ).resolves.toEqual({ + deleted: [], + errors: [ + { + id: "broken-instance", + code: 10001, + message: "workflows.api.error.internal_server", + }, + ], + }); + expect(deleteInstance).toHaveBeenCalledOnce(); + }); + + it("should reject invalid batches", async ({ expect }) => { + const binding = createBinding(); + await expect(binding.deleteBatch({ instances: [] })).rejects.toThrow( + "(body) batchDeleteInstances should have at least 1 instance" + ); + await expect( + binding.deleteBatch({ + instances: Array.from({ length: 101 }, (_, i) => `instance-${i}`), + }) + ).rejects.toThrow( + "(body) batchDeleteInstances only supports 100 instances at a time" + ); + for (const id of ["#invalid!", "batch"]) { + await expect(binding.deleteBatch({ instances: [id] })).rejects.toThrow( + "(instance.invalid_id) Instance ID is invalid" + ); + } + }); + }); + describe("createBatch()", () => { it("should create multiple instances in a batch", async ({ expect }) => { const binding = createBinding(); diff --git a/packages/workflows-shared/tests/validators.test.ts b/packages/workflows-shared/tests/validators.test.ts index deb6ac736c..5f46884ffd 100644 --- a/packages/workflows-shared/tests/validators.test.ts +++ b/packages/workflows-shared/tests/validators.test.ts @@ -42,6 +42,9 @@ describe("Workflow instance ID validation", () => { "\n\nhello", "w".repeat(MAX_WORKFLOW_INSTANCE_ID_LENGTH + 1), "#1231231!!!!", + "batch", + "terminate", + "terminateAll", ])("should reject invalid IDs", (value, { expect }) => { expect(isValidWorkflowInstanceId(value as string)).toBe(false); }); diff --git a/packages/wrangler/src/__tests__/workflows.test.ts b/packages/wrangler/src/__tests__/workflows.test.ts index 148dbc75c5..afac1aeba2 100644 --- a/packages/wrangler/src/__tests__/workflows.test.ts +++ b/packages/wrangler/src/__tests__/workflows.test.ts @@ -19,6 +19,7 @@ import { runWrangler } from "./helpers/run-wrangler"; import { writeWorkerSource } from "./helpers/write-worker-source"; import type { Instance, Workflow } from "../workflows/types"; import type { RawConfig } from "@cloudflare/workers-utils"; +import type { WorkflowBatchDeleteResult } from "@cloudflare/workflows-shared/src/types"; import type { ExpectStatic } from "vitest"; describe("wrangler workflows", () => { @@ -199,6 +200,7 @@ describe("wrangler workflows", () => { wrangler workflows instances restart Restart a workflow instance wrangler workflows instances pause Pause a workflow instance wrangler workflows instances resume Resume a workflow instance + wrangler workflows instances delete [id..] Delete workflow instances GLOBAL FLAGS -c, --config Path to Wrangler configuration file [string] @@ -687,6 +689,97 @@ describe("wrangler workflows", () => { }); }); + describe("instances delete", () => { + const mockDeleteInstances = ( + expect: ExpectStatic, + expectedIds: string[], + result: WorkflowBatchDeleteResult = { + deleted: expectedIds.map((id) => ({ id })), + errors: [], + } + ) => { + msw.use( + http.post( + `*/accounts/:accountId/workflows/:workflowName/instances/batch/delete`, + async ({ request }) => { + expect(await request.json()).toEqual({ instances: expectedIds }); + return HttpResponse.json({ + success: true, + errors: [], + messages: [], + result, + }); + }, + { once: true } + ) + ); + }; + + it("should delete multiple instances", async ({ expect }) => { + writeWranglerConfig(); + mockDeleteInstances(expect, ["foo", "bar"]); + + await runWrangler(`workflows instances delete some-workflow foo bar`); + expect(std.info).toMatchInlineSnapshot( + `"🗑️ Deleted workflow instances from "some-workflow": "foo", "bar""` + ); + }); + + it("should report per-instance errors after logging deletions", async ({ + expect, + }) => { + writeWranglerConfig(); + mockDeleteInstances(expect, ["foo", "bar"], { + deleted: [{ id: "foo" }], + errors: [{ id: "bar", code: 500, message: "delete failed" }], + }); + + await expect( + runWrangler(`workflows instances delete some-workflow foo bar`) + ).rejects.toThrow( + "Failed to delete 1 workflow instance(s):\n - bar: delete failed" + ); + expect(std.info).toContain('"foo"'); + }); + + it("should read instance IDs from a file", async ({ expect }) => { + writeWranglerConfig(); + fs.writeFileSync("instance-ids.txt", "foo\nbar\n"); + mockDeleteInstances(expect, ["foo", "bar"]); + + await runWrangler( + "workflows instances delete some-workflow --ids-file instance-ids.txt" + ); + expect(std.info).toContain('"foo", "bar"'); + }); + + it("should require at least one instance ID", async ({ expect }) => { + writeWranglerConfig(); + await expect( + runWrangler("workflows instances delete some-workflow") + ).rejects.toThrow("Provide at least one workflow instance ID"); + }); + + it("should report an unreadable IDs file", async ({ expect }) => { + writeWranglerConfig(); + await expect( + runWrangler( + "workflows instances delete some-workflow --ids-file missing.txt" + ) + ).rejects.toThrow('Could not read IDs file "missing.txt"'); + }); + + it("should reject more than 100 instances", async ({ expect }) => { + writeWranglerConfig(); + const ids = Array.from({ length: 101 }, (_, i) => `instance-${i}`); + await expect( + runWrangler(`workflows instances delete some-workflow ${ids.join(" ")}`) + ).rejects.toThrow( + "You can delete at most 100 workflow instances at a time" + ); + }); + }); + describe("instances restart", () => { const mockInstances: Instance[] = [ { @@ -1808,6 +1901,123 @@ describe("wrangler workflows", () => { }); }); + describe("workflows instances delete --local", () => { + it("should delete multiple instances in local dev session", async ({ + expect, + }) => { + writeWranglerConfig(); + const ids = ["instance-123", "instance-456", "instance-123"]; + + msw.use( + http.post( + `${LOCAL_BASE}/workflows/:workflowName/instances/batch/delete`, + async ({ params, request }) => { + expect(params.workflowName).toEqual("my-workflow"); + expect(await request.json()).toEqual({ instances: ids }); + return HttpResponse.json({ + success: true, + errors: [], + messages: [], + result: { + deleted: ids.map((id) => ({ id })), + errors: [], + }, + }); + } + ) + ); + + await runWrangler( + "workflows instances delete my-workflow instance-123 instance-456 instance-123 --local" + ); + expect(std.info).toMatchInlineSnapshot( + `"🗑️ Deleted workflow instances from "my-workflow": "instance-123", "instance-456", "instance-123""` + ); + }); + + it("should report local per-instance errors", async ({ expect }) => { + writeWranglerConfig(); + + msw.use( + http.post( + `${LOCAL_BASE}/workflows/:workflowName/instances/batch/delete`, + () => + HttpResponse.json({ + success: true, + errors: [], + messages: [], + result: { + deleted: [], + errors: [ + { + id: "missing-instance", + code: 10400, + message: "workflows.api.error.instance.not_found", + }, + { + id: "broken-instance", + code: 10001, + message: "workflows.api.error.internal_server", + }, + ], + }, + }) + ) + ); + + await expect( + runWrangler( + "workflows instances delete my-workflow missing-instance broken-instance --local" + ) + ).rejects.toThrow( + "Failed to delete 2 workflow instance(s):\n" + + " - missing-instance: workflows.api.error.instance.not_found\n" + + " - broken-instance: workflows.api.error.internal_server" + ); + }); + + it("should resolve latest before local deletion", async ({ expect }) => { + writeWranglerConfig(); + msw.use( + http.get(`${LOCAL_BASE}/workflows/:workflowName/instances`, () => + HttpResponse.json({ + success: true, + errors: [], + messages: [], + result: [ + { + id: "newest-instance", + created_on: "2024-06-01T00:00:00Z", + }, + ], + }) + ), + http.post( + `${LOCAL_BASE}/workflows/:workflowName/instances/batch/delete`, + async ({ request }) => { + expect(await request.json()).toEqual({ + instances: ["newest-instance"], + }); + return HttpResponse.json({ + success: true, + errors: [], + messages: [], + result: { + deleted: [{ id: "newest-instance" }], + errors: [], + }, + }); + } + ) + ); + + await runWrangler( + "workflows instances delete my-workflow latest --local" + ); + expect(std.info).toContain('"newest-instance"'); + }); + }); + describe("workflows instances restart --local", () => { it("should restart an instance in local dev session", async ({ expect, diff --git a/packages/wrangler/src/index.ts b/packages/wrangler/src/index.ts index 2501578b7e..8199c2f65d 100644 --- a/packages/wrangler/src/index.ts +++ b/packages/wrangler/src/index.ts @@ -536,6 +536,7 @@ import { websearchSearchCommand } from "./websearch/search"; import { workflowsInstanceNamespace, workflowsNamespace } from "./workflows"; import { workflowsDeleteCommand } from "./workflows/commands/delete"; import { workflowsDescribeCommand } from "./workflows/commands/describe"; +import { workflowsInstancesDeleteCommand } from "./workflows/commands/instances/delete"; import { workflowsInstancesDescribeCommand } from "./workflows/commands/instances/describe"; import { workflowsInstancesListCommand } from "./workflows/commands/instances/list"; import { workflowsInstancesPauseCommand } from "./workflows/commands/instances/pause"; @@ -2269,6 +2270,10 @@ export function createCLIParser(argv: string[]) { command: "wrangler workflows instances resume", definition: workflowsInstancesResumeCommand, }, + { + command: "wrangler workflows instances delete", + definition: workflowsInstancesDeleteCommand, + }, ]); registry.registerNamespace("workflows"); diff --git a/packages/wrangler/src/workflows/commands/instances/delete.ts b/packages/wrangler/src/workflows/commands/instances/delete.ts new file mode 100644 index 0000000000..592300a03c --- /dev/null +++ b/packages/wrangler/src/workflows/commands/instances/delete.ts @@ -0,0 +1,104 @@ +import { readFileSync } from "node:fs"; +import { UserError } from "@cloudflare/workers-utils"; +import { createCommand } from "../../../core/create-command"; +import { logger } from "../../../logger"; +import { requireAuth } from "../../../user"; +import { + deleteLocalInstances, + getLocalInstanceIdFromArgs, + localWorkflowArgs, +} from "../../local"; +import { deleteInstances, getInstanceIdFromArgs } from "../../utils"; +import type { WorkflowBatchDeleteResult } from "@cloudflare/workflows-shared/src/types"; + +export const workflowsInstancesDeleteCommand = createCommand({ + metadata: { + description: "Delete workflow instances", + owner: "Product: Workflows", + status: "stable", + }, + positionalArgs: ["name", "id"], + args: { + ...localWorkflowArgs, + name: { + describe: "Name of the workflow", + type: "string", + demandOption: true, + }, + id: { + describe: + "IDs of the instances - you can type 'latest' to get the latest instance and delete it", + type: "string", + array: true, + }, + "ids-file": { + describe: "Path to a file containing one instance ID per line", + type: "string", + }, + }, + + async handler(args, { config }) { + let fileIds: string[] = []; + if (args.idsFile) { + try { + fileIds = readFileSync(args.idsFile, "utf8") + .split(/\r?\n/) + .map((id) => id.trim()) + .filter(Boolean); + } catch (error) { + throw new UserError(`Could not read IDs file "${args.idsFile}"`, { + telemetryMessage: "workflows batch delete ids file unreadable", + cause: error, + }); + } + } + + const requestedIds = [...(args.id ?? []), ...fileIds]; + if (requestedIds.length === 0) { + throw new UserError("Provide at least one workflow instance ID", { + telemetryMessage: "workflows batch delete no ids", + }); + } + if (requestedIds.length > 100) { + throw new UserError( + "You can delete at most 100 workflow instances at a time", + { + telemetryMessage: "workflows batch delete too large", + } + ); + } + + let ids: string[]; + let result: WorkflowBatchDeleteResult; + + if (args.local) { + ids = await Promise.all( + requestedIds.map((id) => + getLocalInstanceIdFromArgs(args.port, { id, name: args.name }) + ) + ); + result = await deleteLocalInstances(args.port, args.name, ids); + } else { + const accountId = await requireAuth(config); + ids = await Promise.all( + requestedIds.map((id) => + getInstanceIdFromArgs(accountId, { id, name: args.name }, config) + ) + ); + result = await deleteInstances(config, accountId, args.name, ids); + } + + if (result.deleted.length > 0) { + logger.info( + `🗑️ Deleted workflow instances from "${args.name}": ${result.deleted.map(({ id }) => `"${id}"`).join(", ")}` + ); + } + + if (result.errors.length > 0) { + throw new UserError( + `Failed to delete ${result.errors.length} workflow instance(s):\n${result.errors.map(({ id, message }) => ` - ${id}: ${message}`).join("\n")}`, + { telemetryMessage: "workflows batch delete partial failure" } + ); + } + }, +}); diff --git a/packages/wrangler/src/workflows/local.ts b/packages/wrangler/src/workflows/local.ts index 505872aedd..b88ff851ec 100644 --- a/packages/wrangler/src/workflows/local.ts +++ b/packages/wrangler/src/workflows/local.ts @@ -5,6 +5,7 @@ import type { InstanceStatusAndLogs, WorkflowInstanceRestartFrom, } from "./types"; +import type { WorkflowBatchDeleteResult } from "@cloudflare/workflows-shared/src/types"; const LOCAL_EXPLORER_BASE_PATH = "/cdn-cgi/explorer/api"; const DEFAULT_LOCAL_PORT = 8787; @@ -163,6 +164,22 @@ export async function updateLocalInstanceStatus( ); } +export async function deleteLocalInstances( + port: number, + workflowName: string, + instanceIds: string[] +): Promise { + return fetchLocalResult( + port, + `/workflows/${encodeURIComponent(workflowName)}/instances/batch/delete`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ instances: instanceIds }), + } + ); +} + // ============================================================================ // Local response types (differ slightly from remote API types) // ============================================================================ diff --git a/packages/wrangler/src/workflows/utils.ts b/packages/wrangler/src/workflows/utils.ts index da821d5e71..542fbf94be 100644 --- a/packages/wrangler/src/workflows/utils.ts +++ b/packages/wrangler/src/workflows/utils.ts @@ -7,6 +7,7 @@ import type { WorkflowInstanceRestartFrom, } from "./types"; import type { Config } from "@cloudflare/workers-utils"; +import type { WorkflowBatchDeleteResult } from "@cloudflare/workflows-shared/src/types"; export const emojifyInstanceStatus = (status: InstanceStatus) => { switch (status) { @@ -143,3 +144,22 @@ export async function updateInstanceStatus( } ); } + +export async function deleteInstances( + config: Config, + accountId: string, + workflowName: string, + instanceIds: string[] +): Promise { + return fetchResult( + config, + `/accounts/${accountId}/workflows/${workflowName}/instances/batch/delete`, + { + method: "POST", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify({ instances: instanceIds }), + } + ); +}