Draftish RFC: Cancellable Scopes for Cloudflare Workflows

rfc-part-1.md

This is a more structured draft on a way to cancel Workflow steps that I would like to see after I made an attempt at improving parallel execution of steps within Cloudflare Workflows with cf-forklift.

Some details may need a refresh and more thinking, but I wanted to post this to get thoughts and see if/where it breaks in the real world. Please do comment on how well/poorly this would work for different usecases as I think this can be iterated on and be shipped as a first class runtime API

I came across @wishee0's tweet about wanting feedback for cloudflare workflows, and saw that @ksw_arman from @thecontextco mentioned that wrapping parallel steps in promise.all() and promise.allSettled() isn't a great experience.

I've looked through the docs, fiddled around with the API and shipped a workflow app. And I do think that there's an opportunity here to improve this drastically.

// A deceptively simple example of parallel workflow step execution within Cloudflare Workflows
const [verifyBankStep, screenRiskStep] = await Promise.all([
	step.do("verify bank", () => verifyBank(merchantId)),
	step.do("screen risk", () => screenRisk(merchantId)),
]);

There are two main problems with this approach:

1. Promises are eagerly evaluated during runtime
2. Promises lack built in interruption handling handling

promise.all() fails fast, but it does not cancel sibling promises (and the steps.do() callbacks are promises). It only rejects the parent await.

So if "verify bank" fails first, there is no cancellation/abort signal that's automatically sent into "screen risk" step.

And this is where promise.all() starts to feel wrong for durable workflow operations especially for parallel steps.

If user wrote this and it throws:

await Promise.all([
	step.do("verify bank", () => verifyBank(merchantId)),
	step.do("screen risk", () => screenRisk(merchantId)),
]);

There is no parent scoped step called "verify merchant" that owns those two parallel steps as siblings. They are just separate WorkflowStep operations that user code is waiting on with promise.all().

That matters because the sibling operation may already be inside a step.do callback, sleeping, or waiting for an event.

Cloudflare (workflows visualization in the dashboard) can show that those steps ran in parallel, but it cannot represent the whole thing as one cancellable WorkflowStep scope that can say: "verify merchant failed, cancel unfinished sibling work, and show that in history".

What I tried in userland is shipping a library to help with structured parallel workflowstep operations.

/*
npm i cf-forklift
*/

cf-forklift borrows concepts from Netflix's workflow engine Conductor by providing fork/join operations. It lets users describe named parallel branches, then join them as required or settled branches.

const workflow = withWorkflow(step);

// a fork lets you define named branches of workflow steps
const verifyMerchant = workflow.fork("verify merchant", {
  // each branch is a named function that returns a Workflow step
  profile: ({ step }) => step.do("verify profile", verifyProfile),
  // they're lazily executed
  bank: ({ step }) => step.do("verify bank", () => verifyBank(merchantId)),
  risk: ({ step }) => step.do("screen risk", () => screenRisk(merchantId)),
});

// An alternative to promise.all, it starts every branch, 
// waits for every branch to settle, and returns keyed values 
// only if every branch succeeds
const verification = await workflow.join.required(verifyMerchant);

// example 2

const enrichMerchant = workflow.fork("enrich merchant", {
  website: ({ step }) => 
    step.do("check website reputation", () => checkWebsite(merchantId)),
  marketplace: ({ step }) =>
    step.do("check marketplace footprint", () =>
      checkMarketplace(merchantId)
    ),
  support: ({ step }) =>
    step.do("check support profile", () =>
      checkSupportProfile(merchantId)
    ),
});

// An alternative to promise.allSettled(), it starts every branch, 
// waits for every branch to settle, and returns keyed 
// outcomes without throwing for branch failures
const enrichment = await workflow.join.settled(enrichMerchant);

if (enrichment.website.status === "failure") {
	// Continue with partial enrichment.
}

What allowed me to ship this was that Cloudflare recently shipped step rollbacks. That allows me to roll an Abort Signal style cancellation scope around parallel WorkflowSteps

const fork = workflow.fork("verify merchant", {
	bank: ({ step }) => {
		// let's say that this can throw
		return step.do("verify bank", () => verifyBank(merchantId));
	},
	risk: async ({ step, cancellation }) => {
		await step.do("risk check 1", () => riskCheckOne(merchantId));

		// this can skip future work after a sibling has failed.
		// it cannot interrupt the step that already ran above.
		cancellation.throwIfRequested();

		return step.do("risk check 2", () => riskCheckTwo(merchantId));
	},
});

await workflow.join.required(fork, {
	abortOnFailure: "cooperative",
});

My library covers a lot of gaps for the problems with parallel workflow execution in Cloudflare Workflows.

• parallel operations are lazily executed and not eagerly like promises
• you can name sibiling parallel steps into a "named branch"
• results are typesafe and keyed from branch results
• join.required() drain every branch before throwing
• join.settled() return keyed outcomes
• cooperative abort checkpoints between WorkflowStep operations and pass-through rollbacks on completed step.do work

I would delete this library if Cloudflare had a native API for cancellation scope. Cause the abort signal method isn't true cancellation of tasks and it relies on saga compensation style rollbacks for already executed steps before the throw failure (which also means that it is inherently racey).

Userland cannot cancel an inflight step.do callback, remove a pending sleep from the priority queue, remove a waitForEvent waiter, write a native "cancelled" history event, or preserve scope identity through restart-from-step.

You can just .pause() and .terminate() an overaching Workflow class, but not WorkflowStep. And it's quite naive to provide a step.interrupt() / step.cancel() as well

rfc-part-2.md

/*
The new Runtime API that I propose
*/

const verification = await step.cancellableScope("verify merchant", async ({ step }) => {
	const [bank, risk] = await Promise.all([
		step.do("verify bank", async (ctx) => {
			// ctx.signal is aborted if the "verify merchant" scope is cancelled
			return verifyBank(merchantId, ctx.signal);
		}),

		step.do("screen risk", async (ctx) => {
			// same signal shape as fetch(), so user code can stop cooperatively
			return screenRisk(merchantId, ctx.signal);
		}),
	]);

	return { bank, risk };
});


// The astute of you would recognize this pattern within Effect 

const program = Effect.gen(function* () {
	const [profile, bank, risk] = yield* Effect.all(
		[
			verifyProfile(merchantId),
			verifyBank(merchantId),
			screenRisk(merchantId),
		],
		// Effect owns these child effects and can interrupt siblings
		{ concurrency: "unbounded" }
	);

	return { profile, bank, risk };
});

// sorry, I've been Effect pilled for almost two years now. I blame dillon

The beauty of providing step.cancellableScope is that it keeps the normal promise APIs that users are used to writing, and it doesn't change their code drastically to improve this.

When the code calls step.do, step.sleep, step.sleepUntil, or step.waitForEvent inside a cancellable scope, the runtime registers that operation under that named scope.

// API Surface (to glance through)
// this definitely needs more thought and tests against the engine runtime

interface WorkflowStep {
	cancellableScope<T>(
		name: string,
		callback: (scope: WorkflowCancellationScope) => Promise<T>,
	): Promise<T>;

	cancellableScope<T>(
		name: string,
		options: WorkflowCancellationScopeOptions,
		callback: (scope: WorkflowCancellationScope) => Promise<T>,
	): Promise<T>;
}

type WorkflowCancellationScopeOptions = {
	timeout?: string | number;
};

type WorkflowCancellationSource =
	| "user"
	| "scope-timeout"
	| "parent-scope"
	| "scope-close";

type WorkflowCancellationReason = {
	name: "WorkflowCancellationError";
	message: string;
	source: WorkflowCancellationSource;
	scope?: { name: string; count: number; key: string };
	cause?: { name: string; message: string };
};

class WorkflowCancellationError extends Error {
	name = "WorkflowCancellationError";
}

type WorkflowCancellationScope = {
	name: string;
	signal: AbortSignal;
	step: WorkflowStep;
	cancelRequested: Promise<never>;
	cancel(reason?: unknown): void;
	nonCancellable<T>(callback: (scope: WorkflowCancellationScope) => Promise<T>): Promise<T>;
};

type WorkflowStepContext = {
	step: { name: string; count: number };
	attempt: number;
	config: WorkflowStepConfig;
	signal: AbortSignal;
};

Promise.all() should stay native (and not monkey patched).

The Structured behavior comes from the Workflow operation promises being attached to the current cancellable scope.

If the scope later fails or is cancelled, the runtime has the operation records it needs to cancel unfinished children and wait for terminal outcomes.

Cancellation should be recorded as its own outcome, and not hidden as a generic failure. Cancellation is also not rollback but pairs super well with saga style compensation.

• cancellation stops unfinished work
• rollback compensates completed work if the Workflow fails normally

await step.cancellableScope("upload files", async ({ step, nonCancellable }) => {
	try {
		// both uploads belong to the "upload files" scope
		// if one fails, the scope can cancel the unfinished sibling
		await Promise.all([
			step.do("upload a", async (ctx) => uploadA(ctx.signal)),
			step.do("upload b", async (ctx) => uploadB(ctx.signal)),
		]);
	} catch (error) {
		// the upload scope is already failed/cancelled here
		// run cleanup in a shielded scope so it can actually happen
		await nonCancellable(async ({ step }) => {
			await step.do("cleanup uploads", async () => cleanupUploads());
		});

		// don't swallow the original failure
		// let normal Workflow failure + rollback semantics happen
		throw error;
	}
});

type WorkflowOperationOutcome<T> =
	| { status: "running" }
	| { status: "success"; value: T }
	| { status: "failure"; error: { name: string; message: string } }
	| { status: "cancelled"; reason: WorkflowCancellationReason };