fix NonRetryableError thrown with an empty error message not stopping workflow retries locally (#10219)

Author: dario-piotrowicz

.changeset/shy-coats-drum.md

@@ -0,0 +1,6 @@
+---
+"@cloudflare/workflows-shared": patch
+"wrangler": patch
+---
+
+fix `NonRetryableError` thrown with an empty error message not stopping workflow retries locally

fixtures/workflow/src/index.ts

@@ -5,6 +5,7 @@ import {
 	WorkflowEvent,
 	WorkflowStep,
 } from "cloudflare:workers";
+import { NonRetryableError } from "cloudflare:workflows";
 
 type Params = {
 	name: string;
@@ -56,14 +57,47 @@ export class Demo2 extends WorkflowEntrypoint<{}, Params> {
 	}
 }
 
+export class Demo3 extends WorkflowEntrypoint<{}, Params> {
+	async run(
+		event: WorkflowEvent<Params & { doRetry: boolean; errorMessage: string }>,
+		step: WorkflowStep
+	) {
+		let runs = -1;
+		try {
+			await step.do(
+				"First (and only step) step",
+				{
+					retries: {
+						limit: 3,
+						delay: 100,
+					},
+				},
+				async function () {
+					runs++;
+					if (event.payload.doRetry) {
+						throw new Error(event.payload.errorMessage);
+					} else {
+						throw new NonRetryableError(event.payload.errorMessage);
+					}
+				}
+			);
+		} catch {}
+
+		return `The step was retried ${runs} time${runs === 1 ? "" : "s"}`;
+	}
+}
+
 type Env = {
 	WORKFLOW: Workflow;
 	WORKFLOW2: Workflow;
+	WORKFLOW3: Workflow<{ doRetry: boolean; errorMessage: string }>;
 };
 export default class extends WorkerEntrypoint<Env> {
 	async fetch(req: Request) {
 		const url = new URL(req.url);
 		const id = url.searchParams.get("workflowName");
+		const doRetry = url.searchParams.get("doRetry");
+		const errorMessage = url.searchParams.get("errorMessage");
 
 		if (url.pathname === "/favicon.ico") {
 			return new Response(null, { status: 404 });
@@ -89,6 +123,18 @@ export default class extends WorkerEntrypoint<Env> {
 			} else {
 				handle = await this.env.WORKFLOW2.create({ id });
 			}
+		} else if (url.pathname === "/createDemo3") {
+			if (id === null) {
+				handle = await this.env.WORKFLOW3.create();
+			} else {
+				handle = await this.env.WORKFLOW3.create({
+					id,
+					params: {
+						doRetry: doRetry === "false" ? false : true,
+						errorMessage: errorMessage ?? "",
+					},
+				});
+			}
 		} else if (url.pathname === "/sendEvent") {
 			handle = await this.env.WORKFLOW2.get(id);
 
@@ -98,6 +144,8 @@ export default class extends WorkerEntrypoint<Env> {
 			});
 		} else if (url.pathname === "/get2") {
 			handle = await this.env.WORKFLOW2.get(id);
+		} else if (url.pathname === "/get3") {
+			handle = await this.env.WORKFLOW3.get(id);
 		} else {
 			handle = await this.env.WORKFLOW.get(id);
 		}

fixtures/workflow/tests/index.test.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from "crypto";
 import { rm } from "fs/promises";
 import { resolve } from "path";
 import { fetch } from "undici";
@@ -139,6 +140,88 @@ describe("Workflows", () => {
 		]);
 	});
 
+	describe("retrying a step", () => {
+		test("should retry a step if a generic Error (with a generic error message) is thrown", async ({
+			expect,
+		}) => {
+			const name = randomUUID();
+			await fetchJson(
+				`http://${ip}:${port}/createDemo3?workflowName=${name}&doRetry=true&errorMessage=generic_error_message`
+			);
+
+			await vi.waitFor(
+				async () => {
+					const result = await fetchJson(
+						`http://${ip}:${port}/get3?workflowName=${name}`
+					);
+
+					expect(result["output"]).toEqual("The step was retried 3 times");
+				},
+				{ timeout: 1500 }
+			);
+		});
+
+		test("should retry a step if a generic Error (with an empty error message) is thrown", async ({
+			expect,
+		}) => {
+			const name = randomUUID();
+			await fetchJson(
+				`http://${ip}:${port}/createDemo3?workflowName=${name}&doRetry=true&errorMessage=`
+			);
+
+			await vi.waitFor(
+				async () => {
+					const result = await fetchJson(
+						`http://${ip}:${port}/get3?workflowName=${name}`
+					);
+
+					expect(result["output"]).toEqual("The step was retried 3 times");
+				},
+				{ timeout: 1500 }
+			);
+		});
+
+		test("should not retry a step if a NonRetryableError (with a generic error message) is thrown", async ({
+			expect,
+		}) => {
+			const name = randomUUID();
+			await fetchJson(
+				`http://${ip}:${port}/createDemo3?workflowName=${name}&doRetry=false&errorMessage=generic_error_message"`
+			);
+
+			await vi.waitFor(
+				async () => {
+					const result = await fetchJson(
+						`http://${ip}:${port}/get3?workflowName=${name}`
+					);
+
+					expect(result["output"]).toEqual("The step was retried 0 times");
+				},
+				{ timeout: 1500 }
+			);
+		});
+
+		test("should not retry a step if a NonRetryableError (with an empty error message) is thrown", async ({
+			expect,
+		}) => {
+			const name = randomUUID();
+			await fetchJson(
+				`http://${ip}:${port}/createDemo3?workflowName=${name}&doRetry=false&errorMessage=`
+			);
+
+			await vi.waitFor(
+				async () => {
+					const result = await fetchJson(
+						`http://${ip}:${port}/get3?workflowName=${name}`
+					);
+
+					expect(result["output"]).toEqual("The step was retried 0 times");
+				},
+				{ timeout: 1500 }
+			);
+		});
+	});
+
 	test("waitForEvent should work", async ({ expect }) => {
 		await fetchJson(`http://${ip}:${port}/createDemo2?workflowName=something`);
 

fixtures/workflow/wrangler.jsonc

@@ -13,5 +13,10 @@
 			"name": "my-workflow2",
 			"class_name": "Demo2",
 		},
+		{
+			"binding": "WORKFLOW3",
+			"name": "my-workflow3",
+			"class_name": "Demo3",
+		},
 	],
 }

packages/workflows-shared/src/context.ts

@@ -350,7 +350,7 @@ export class Context extends RpcTarget {
 				if (
 					e instanceof Error &&
 					(error.name === "NonRetryableError" ||
-						error.message.startsWith("NonRetryableError:"))
+						error.message.startsWith("NonRetryableError"))
 				) {
 					this.#engine.writeLog(
 						InstanceEvent.ATTEMPT_FAILURE,