feat: add configurable step limits for Workflows (#12622)

Co-authored-by: emily-shen <69125074+emily-shen@users.noreply.github.com>

Author: LuisDuarte1

.changeset/workflow-step-limits.md

@@ -0,0 +1,27 @@
+---
+"wrangler": minor
+"miniflare": minor
+"@cloudflare/workflows-shared": minor
+---
+
+Add configurable step limits for Workflows
+
+You can now set a maximum number of steps for a Workflow instance via the `limits.steps` configuration in your Wrangler config. When a Workflow instance exceeds this limit, it will fail with an error indicating the limit was reached.
+
+```jsonc
+// wrangler.jsonc
+{
+	"workflows": [
+		{
+			"binding": "MY_WORKFLOW",
+			"name": "my-workflow",
+			"class_name": "MyWorkflow",
+			"limits": {
+				"steps": 5000,
+			},
+		},
+	],
+}
+```
+
+The `steps` value must be an integer between 1 and 25,000. If not specified, the default limit of 10,000 steps is used. Step limits are also enforced in local development via `wrangler dev`.

packages/miniflare/src/plugins/workflows/index.ts

@@ -23,6 +23,7 @@ export const WorkflowsOptionsSchema = z.object({
 				remoteProxyConnectionString: z
 					.custom<RemoteProxyConnectionString>()
 					.optional(),
+				stepLimit: z.number().int().min(1).optional(),
 			})
 		)
 		.optional(),
@@ -151,6 +152,14 @@ export const WORKFLOWS_PLUGIN: Plugin<
 								name: "BINDING_NAME",
 								json: JSON.stringify(bindingName),
 							},
+							...(workflow.stepLimit !== undefined
+								? [
+										{
+											name: "STEP_LIMIT",
+											json: JSON.stringify(workflow.stepLimit),
+										},
+									]
+								: []),
 						],
 					},
 				};

packages/workers-utils/src/config/environment.ts

@@ -675,6 +675,11 @@ export type WorkflowBinding = {
 	script_name?: string;
 	/** Whether the Workflow should be remote or not in local development */
 	remote?: boolean;
+	/** Optional limits for the Workflow */
+	limits?: {
+		/** Maximum number of steps a Workflow instance can execute */
+		steps?: number;
+	};
 };
 
 /**

packages/workers-utils/src/config/validation.ts

@@ -2584,12 +2584,54 @@ const validateWorkflowBinding: ValidatorFn = (diagnostics, field, value) => {
 		isValid = false;
 	}
 
+	if (hasProperty(value, "limits") && value.limits !== undefined) {
+		if (
+			typeof value.limits !== "object" ||
+			value.limits === null ||
+			Array.isArray(value.limits)
+		) {
+			diagnostics.errors.push(
+				`"${field}" bindings should, optionally, have an object "limits" field but got ${JSON.stringify(
+					value
+				)}.`
+			);
+			isValid = false;
+		} else {
+			const limits = value.limits as Record<string, unknown>;
+			if (limits.steps !== undefined) {
+				if (
+					typeof limits.steps !== "number" ||
+					!Number.isInteger(limits.steps) ||
+					limits.steps < 1
+				) {
+					diagnostics.errors.push(
+						`"${field}" bindings "limits.steps" field must be a positive integer but got ${JSON.stringify(
+							limits.steps
+						)}.`
+					);
+					isValid = false;
+				} else if (limits.steps > 25_000) {
+					diagnostics.warnings.push(
+						`"${field}" has a step limit of ${limits.steps}, which exceeds the production maximum of 25,000. This configuration may not work when deployed.`
+					);
+				}
+			}
+			validateAdditionalProperties(
+				diagnostics,
+				`${field}.limits`,
+				Object.keys(limits),
+				["steps"]
+			);
+		}
+	}
+
 	validateAdditionalProperties(diagnostics, field, Object.keys(value), [
 		"binding",
 		"name",
 		"class_name",
 		"script_name",
 		"remote",
+		"limits",
 	]);
 
 	return isValid;

packages/workers-utils/src/worker.ts

@@ -181,6 +181,9 @@ export interface CfWorkflow {
 	script_name?: string;
 	remote?: boolean;
 	raw?: boolean;
+	limits?: {
+		steps?: number;
+	};
 }
 
 export interface CfQueue {

packages/workflows-shared/src/context.ts

@@ -47,6 +47,7 @@ export class Context extends RpcTarget {
 	#state: DurableObjectState;
 
 	#counters: Map<string, number> = new Map();
+	#lifetimeStepCounter: number = 0;
 
 	constructor(engine: Engine, state: DurableObjectState) {
 		super();
@@ -85,6 +86,15 @@ export class Context extends RpcTarget {
 			stepConfig = {};
 		}
 
+		this.#lifetimeStepCounter++;
+
+		const stepLimit = this.#engine.stepLimit;
+		if (this.#lifetimeStepCounter > stepLimit) {
+			throw new WorkflowFatalError(
+				`The limit of ${stepLimit} steps has been reached. This limit can be changed in your worker configuration.`
+			);
+		}
+
 		if (!isValidStepName(name)) {
 			// NOTE(lduarte): marking errors as user error allows the observability layer to avoid leaking
 			// user errors to sentry while making everything more observable. `isUserError` is not serialized

packages/workflows-shared/src/engine.ts

@@ -23,6 +23,7 @@ import type { WorkflowEntrypoint, WorkflowEvent } from "cloudflare:workers";
 
 interface Env {
 	USER_WORKFLOW: WorkflowEntrypoint;
+	STEP_LIMIT?: string; // JSON-encoded number from miniflare binding
 }
 
 export type DatabaseWorkflow = {
@@ -74,6 +75,8 @@ const ENGINE_STATUS_KEY = "ENGINE_STATUS";
 
 const EVENT_MAP_PREFIX = "EVENT_MAP";
 
+export const DEFAULT_STEP_LIMIT = 10_000;
+
 export class Engine extends DurableObject<Env> {
 	logs: Array<unknown> = [];
 
@@ -83,13 +86,19 @@ export class Engine extends DurableObject<Env> {
 	workflowName: string | undefined;
 	timeoutHandler: GracePeriodSemaphore;
 	priorityQueue: TimePriorityQueue | undefined;
+	stepLimit: number;
 
 	waiters: Map<string, Array<(event: Event | PromiseLike<Event>) => void>> =
 		new Map();
 	eventMap: Map<string, Array<Event>> = new Map();
 
 	constructor(state: DurableObjectState, env: Env) {
 		super(state, env);
+
+		this.stepLimit = env.STEP_LIMIT
+			? JSON.parse(env.STEP_LIMIT)
+			: DEFAULT_STEP_LIMIT;
+
 		void this.ctx.blockConcurrencyWhile(async () => {
 			this.ctx.storage.transactionSync(() => {
 				try {

packages/workflows-shared/tests/engine.test.ts

@@ -5,7 +5,7 @@ import {
 } from "cloudflare:test";
 import { NonRetryableError } from "cloudflare:workflows";
 import { describe, it, vi } from "vitest";
-import { InstanceEvent, InstanceStatus } from "../src";
+import { DEFAULT_STEP_LIMIT, InstanceEvent, InstanceStatus } from "../src";
 import type {
 	DatabaseInstance,
 	DatabaseVersion,
@@ -271,4 +271,118 @@ describe("Engine", () => {
 			logs.logs.some((log) => log.event === InstanceEvent.WORKFLOW_START)
 		).toBe(true);
 	});
+
+	describe("step limits", () => {
+		it("should enforce step limit when exceeded", async ({ expect }) => {
+			const stepLimit = 3;
+
+			const engineStub = await runWorkflow(
+				"STEP-LIMIT-EXCEEDED",
+				async (_event, step) => {
+					// Try to run more steps than the limit
+					for (let i = 0; i < stepLimit + 1; i++) {
+						await step.do(`step-${i}`, async () => `result-${i}`);
+					}
+				}
+			);
+
+			// Set the step limit on the engine
+			await runInDurableObject(engineStub, (engine) => {
+				engine.stepLimit = stepLimit;
+			});
+
+			// Re-init to run with the new limit
+			await setWorkflowEntrypoint(engineStub, async (_event, step) => {
+				for (let i = 0; i < stepLimit + 1; i++) {
+					await step.do(`step-${i}`, async () => `result-${i}`);
+				}
+			});
+
+			const engineId = env.ENGINE.idFromName("STEP-LIMIT-EXCEEDED-2");
+			const freshStub = env.ENGINE.get(engineId);
+
+			await runInDurableObject(freshStub, (engine) => {
+				engine.stepLimit = stepLimit;
+			});
+
+			await setWorkflowEntrypoint(freshStub, async (_event, step) => {
+				for (let i = 0; i < stepLimit + 1; i++) {
+					await step.do(`step-${i}`, async () => `result-${i}`);
+				}
+			});
+
+			await freshStub.init(
+				12346,
+				{} as DatabaseWorkflow,
+				{} as DatabaseVersion,
+				{ id: "STEP-LIMIT-EXCEEDED-2" } as DatabaseInstance,
+				{
+					payload: {},
+					timestamp: new Date(),
+					instanceId: "STEP-LIMIT-EXCEEDED-2",
+				}
+			);
+
+			const logs = (await freshStub.readLogs()) as EngineLogs;
+
+			expect(
+				logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE)
+			).toBe(true);
+		});
+
+		it("should succeed when steps are exactly at the limit", async ({
+			expect,
+		}) => {
+			const stepLimit = 3;
+
+			const engineId = env.ENGINE.idFromName("STEP-LIMIT-AT-LIMIT");
+			const freshStub = env.ENGINE.get(engineId);
+
+			await runInDurableObject(freshStub, (engine) => {
+				engine.stepLimit = stepLimit;
+			});
+
+			await setWorkflowEntrypoint(freshStub, async (_event, step) => {
+				for (let i = 0; i < stepLimit; i++) {
+					await step.do(`step-${i}`, async () => `result-${i}`);
+				}
+				return "done";
+			});
+
+			await freshStub.init(
+				12346,
+				{} as DatabaseWorkflow,
+				{} as DatabaseVersion,
+				{ id: "STEP-LIMIT-AT-LIMIT" } as DatabaseInstance,
+				{
+					payload: {},
+					timestamp: new Date(),
+					instanceId: "STEP-LIMIT-AT-LIMIT",
+				}
+			);
+
+			const logs = (await freshStub.readLogs()) as EngineLogs;
+
+			expect(
+				logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_SUCCESS)
+			).toBe(true);
+			expect(
+				logs.logs.some((val) => val.event === InstanceEvent.WORKFLOW_FAILURE)
+			).toBe(false);
+		});
+
+		it("should use DEFAULT_STEP_LIMIT when no limit is configured", async ({
+			expect,
+		}) => {
+			const engineId = env.ENGINE.idFromName("STEP-LIMIT-DEFAULT");
+			const freshStub = env.ENGINE.get(engineId);
+
+			const stepLimit = await runInDurableObject(
+				freshStub,
+				(engine) => engine.stepLimit
+			);
+
+			expect(stepLimit).toBe(DEFAULT_STEP_LIMIT);
+		});
+	});
 });