diff --git a/.changeset/workflows-instances-delete.md b/.changeset/workflows-instances-delete.md new file mode 100644 index 0000000000..5197581e37 --- /dev/null +++ b/.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 ` 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. diff --git a/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts b/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts index fdaa952106..941d462d6e 100644 --- a/packages/miniflare/src/workers/workflows/wrapped-binding.worker.ts +++ b/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 { + return this.binding.deleteBatch({ instances: instanceIds }); + } + async unsafeGetBindingName(): Promise { return this.binding.unsafeGetBindingName(); } @@ -124,6 +129,12 @@ 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(); + } + public async status(): Promise { using instance = await this.getInstance(); using res = (await instance.status()) as InstanceStatus & Disposable; diff --git a/packages/miniflare/test/plugins/workflows/index.spec.ts b/packages/miniflare/test/plugins/workflows/index.spec.ts index 56061c1fb7..6484147eef 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 } }); @@ -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)); diff --git a/packages/workflows-shared/src/binding.ts b/packages/workflows-shared/src/binding.ts index 9574eeb4c4..23f3dfa0dc 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 { + 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; BINDING_NAME: string; @@ -240,6 +244,64 @@ export class WorkflowBinding extends WorkerEntrypoint { ); } + public async deleteBatch(options: { + instances: string[]; + }): Promise { + 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[] = []; + 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 { // async because of rpc return this.env.BINDING_NAME; @@ -377,6 +439,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/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..c11772f400 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,165 @@ 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(); + for (const instances of [ + [], + Array.from({ length: 101 }, (_, i) => `instance-${i}`), + ]) { + await expect(binding.deleteBatch({ instances })).rejects.toThrow( + "Batch must contain between 1 and 100 instance IDs" + ); + } + await expect( + binding.deleteBatch({ instances: ["#invalid!"] }) + ).rejects.toThrow("Workflow instance has invalid id"); + }); + }); + describe("createBatch()", () => { it("should create multiple instances in a batch", async ({ expect }) => { const binding = createBinding(); diff --git a/packages/wrangler/src/__tests__/workflows.test.ts b/packages/wrangler/src/__tests__/workflows.test.ts index 148dbc75c5..fb2997db17 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 Delete workflow instances GLOBAL FLAGS -c, --config Path to Wrangler configuration file [string] @@ -687,6 +689,65 @@ 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", 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)"); + }); + + 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 +1869,78 @@ describe("wrangler workflows", () => { }); }); + describe("workflows instances delete --local", () => { + it("should delete multiple instances in local dev session", async ({ + expect, + }) => { + writeWranglerConfig(); + const deleted: string[] = []; + + msw.use( + http.delete( + `${LOCAL_BASE}/workflows/:workflowName/instances/:instanceId`, + ({ params }) => { + expect(params.workflowName).toEqual("my-workflow"); + deleted.push(params.instanceId as string); + return HttpResponse.json({ + success: true, + errors: [], + messages: [], + result: { success: true }, + }); + } + ) + ); + + await runWrangler( + "workflows instances delete my-workflow instance-123 instance-456 instance-123 --local" + ); + expect(deleted).toEqual(["instance-123", "instance-456"]); + expect(std.info).toMatchInlineSnapshot( + `"🗑️ Deleted workflow instances from "my-workflow": "instance-123", "instance-456", "instance-123""` + ); + }); + + it("should normalize local deletion errors", async ({ expect }) => { + writeWranglerConfig(); + + msw.use( + http.delete( + `${LOCAL_BASE}/workflows/:workflowName/instances/:instanceId`, + ({ params }) => { + const notFound = params.instanceId === "missing-instance"; + return HttpResponse.json( + { + success: false, + errors: [ + { + code: notFound ? 10501 : 10001, + message: notFound + ? "Instance not found" + : "sensitive failure", + }, + ], + messages: [], + result: null, + }, + { status: notFound ? 404 : 500 } + ); + } + ) + ); + + 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" + ); + }); + }); + 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..668d252f1c --- /dev/null +++ b/packages/wrangler/src/workflows/commands/instances/delete.ts @@ -0,0 +1,79 @@ +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, + demandOption: true, + }, + }, + + async handler(args, { config }) { + if (args.id.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( + args.id.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( + args.id.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..1f03bdbfbf 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; @@ -78,20 +79,27 @@ export async function fetchLocalResult( const json = (await response .json() .catch(() => null)) as LocalApiResponse | null; - const errorMessage = - json?.errors?.[0]?.message ?? `HTTP ${response.status}`; - throw new UserError(`Local API error: ${errorMessage}`, { - telemetryMessage: "workflows local api error response", - }); + const apiError = json?.errors?.[0]; + throw new UserError( + `Local API error: ${apiError?.message ?? `HTTP ${response.status}`}`, + { + telemetryMessage: "workflows local api error response", + cause: apiError, + } + ); } const json = (await response.json()) as LocalApiResponse; if (!json.success) { - const errorMessage = json.errors?.[0]?.message ?? "Unknown local API error"; - throw new UserError(`Local API error: ${errorMessage}`, { - telemetryMessage: "workflows local api unsuccessful response", - }); + const apiError = json.errors?.[0]; + throw new UserError( + `Local API error: ${apiError?.message ?? "Unknown local API error"}`, + { + telemetryMessage: "workflows local api unsuccessful response", + cause: apiError, + } + ); } return json.result; @@ -163,6 +171,64 @@ export async function updateLocalInstanceStatus( ); } +async function deleteLocalInstance( + port: number, + workflowName: string, + instanceId: string +): Promise { + await fetchLocalResult<{ success: boolean }>( + port, + `/workflows/${encodeURIComponent(workflowName)}/instances/${encodeURIComponent(instanceId)}`, + { + method: "DELETE", + } + ); +} + +export async function deleteLocalInstances( + port: number, + workflowName: string, + instanceIds: string[] +): Promise { + const uniqueIds = [...new Set(instanceIds)]; + const deletions: Promise[] = []; + for (const id of uniqueIds) { + deletions.push(deleteLocalInstance(port, workflowName, id)); + } + + 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") { + result.deleted.push({ id }); + continue; + } + + const cause = + deletion.reason instanceof Error ? deletion.reason.cause : undefined; + const apiCode = + typeof cause === "object" && cause !== null && "code" in cause + ? cause.code + : undefined; + const isNotFound = apiCode === 10501; + result.errors.push({ + id, + code: isNotFound ? 10400 : 10001, + message: isNotFound + ? "workflows.api.error.instance.not_found" + : "workflows.api.error.internal_server", + }); + } + return result; +} + // ============================================================================ // 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 }), + } + ); +}