Cloudflare Workflows is a sturdy execution engine that permits you to chain steps, retry on failure, and persist state throughout long-running processes. Builders use Workflows to energy background brokers, handle knowledge pipelines, construct human-in-the-loop approval techniques, and extra.
Final month, we introduced that each workflow deployed to Cloudflare now has a whole visible diagram within the dashboard.
We constructed this as a result of having the ability to visualize your functions is extra essential now than ever earlier than. Coding brokers are writing code that you could be or will not be studying. Nevertheless, the form of what will get constructed nonetheless issues: how the steps join, the place they department, and what’s truly occurring.
In case you’ve seen diagrams from visible workflow builders earlier than, these are often working from one thing declarative: JSON configs, YAML, drag-and-drop. Nevertheless, Cloudflare Workflows are simply code. They will embrace Guarantees, Promise.all, loops, conditionals, and/or be nested in capabilities or courses. This dynamic execution mannequin makes rendering a diagram a bit extra difficult.
We use Summary Syntax Bushes (ASTs) to statically derive the graph, monitoring Promise and await relationships to grasp what runs in parallel, what blocks, and the way the items join.
Hold studying to find out how we constructed these diagrams, or deploy your first workflow and see the diagram for your self.
Right here’s an instance of a diagram generated from Cloudflare Workflows code:
Dynamic workflow execution
Usually, workflow engines can execute based on both dynamic or sequential (static) execution order. Sequential execution would possibly look like the extra intuitive answer: set off workflow → step A → step B → step C, the place step B begins executing instantly after the engine completes Step A, and so forth.
Cloudflare Workflows observe the dynamic execution mannequin. Since workflows are simply code, the steps execute because the runtime encounters them. When the runtime discovers a step, that step will get handed over to the workflow engine, which manages its execution. The steps aren’t inherently sequential except awaited — the engine executes all unawaited steps in parallel. This manner, you’ll be able to write your workflow code as circulation management with out extra wrappers or directives. Right here’s how the handoff works:
An engine, which is a “supervisor” Sturdy Object for that occasion, spins up. The engine is chargeable for the logic of the particular workflow execution.
The engine triggers a person employee by way of dynamic dispatch, passing management over to Employees runtime.
When Runtime encounters a
step.do, it passes the execution again to the engine.The engine executes the step, persists the outcome (or throws an error, if relevant) and triggers the person Employee once more.
With this structure, the engine doesn’t inherently “know” the order of the steps that it’s executing — however for a diagram, the order of steps turns into essential info. The problem right here lies in getting the overwhelming majority of workflows translated precisely right into a diagnostically useful graph; with the diagrams in beta, we’ll proceed to iterate and enhance on these representations.
Fetching the script at deploy time, as a substitute of run time, permits us to parse the workflow in its entirety to statically generate the diagram.
Taking a step again, right here is the lifetime of a workflow deployment:
To create the diagram, we fetch the script after it has been bundled by the interior configuration service which deploys Employees (step 2 underneath Workflow deployment). Then, we use a parser to create an summary syntax tree (AST) representing the workflow, and our inside service generates and traverses an intermediate graph with all WorkflowEntrypoints and calls to workflows steps. We render the diagram primarily based on the ultimate outcome on our API.
When a Employee is deployed, the configuration service bundles (utilizing esbuild by default) and minifies the code except specified in any other case. This presents one other problem — whereas Workflows in TypeScript observe an intuitive sample, their minified Javascript (JS) could be dense and indigestible. There are additionally completely different ways in which code could be minified, relying on the bundler.
Right here’s an instance of Workflow code that exhibits brokers executing in parallel:
const summaryPromise = step.do(
`abstract agent (loop ${loop})`,
async () => {
return runAgentPrompt(
this.env,
SUMMARY_SYSTEM,
buildReviewPrompt(
'Summarize this textual content in 5 bullet factors.',
draft,
enter.context
)
);
}
);
const correctnessPromise = step.do(
`correctness agent (loop ${loop})`,
async () => {
return runAgentPrompt(
this.env,
CORRECTNESS_SYSTEM,
buildReviewPrompt(
'Record correctness points and urged fixes.',
draft,
enter.context
)
);
}
);
const clarityPromise = step.do(
`readability agent (loop ${loop})`,
async () => {
return runAgentPrompt(
this.env,
CLARITY_SYSTEM,
buildReviewPrompt(
'Record readability points and urged fixes.',
draft,
enter.context
)
);
}
);Bundling with rspack, a snippet of the minified code appears to be like like this:
class pe extends e{async run(e,t){de("workflow.run.start",{instanceId:e.instanceId});const r=await t.do("validate payload",async()=>{if(!e.payload.r2Key)throw new Error("r2Key is required");if(!e.payload.telegramChatId)throw new Error("telegramChatId is required");return{r2Key:e.payload.r2Key,telegramChatId:e.payload.telegramChatId,context:e.payload.context?.trim()}}),s=await t.do("load source document from r2",async()=>{const e=await this.env.REVIEW_DOCUMENTS.get(r.r2Key);if(!e)throw new Error(`R2 object not discovered: ${r.r2Key}`);const t=(await e.textual content()).trim();if(!t)throw new Error("R2 object is empty");return t}),n=Quantity(this.env.MAX_REVIEW_LOOPS??"5"),o=this.env.RESPONSE_TIMEOUT??"7 days",a=async(s,i,c)=>{if(s>n)return le("workflow.loop.max_reached",{instanceId:e.instanceId,maxLoops:n}),await t.do("notify max loop reached",async()=>{await se(this.env,r.telegramChatId,`Evaluation stopped after ${n} loops for ${e.instanceId}. Begin once more if you happen to nonetheless want revisions.`)}),{accepted:!1,loops:n,finalText:i};const h=t.do(`abstract agent (loop ${s})`,async()=>te(this.env,"You summarize documents. Keep the output short, concrete, and factual.",ue("Summarize this text in 5 bullet points.",i,r.context)))...Or, bundling with vite, here’s a minified snippet:
class ht extends pe {
async run(e, r) {
b("workflow.run.start", { instanceId: e.instanceId });
const s = await r.do("validate payload", async () => {
if (!e.payload.r2Key)
throw new Error("r2Key is required");
if (!e.payload.telegramChatId)
throw new Error("telegramChatId is required");
return {
r2Key: e.payload.r2Key,
telegramChatId: e.payload.telegramChatId,
context: e.payload.context?.trim()
};
}), n = await r.do(
"load source document from r2",
async () => {
const i = await this.env.REVIEW_DOCUMENTS.get(s.r2Key);
if (!i)
throw new Error(`R2 object not discovered: ${s.r2Key}`);
const c = (await i.textual content()).trim();
if (!c)
throw new Error("R2 object is empty");
return c;
}
), o = Quantity(this.env.MAX_REVIEW_LOOPS ?? "5"), l = this.env.RESPONSE_TIMEOUT ?? "7 days", a = async (i, c, u) => {
if (i > o)
return H("workflow.loop.max_reached", {
instanceId: e.instanceId,
maxLoops: o
}), await r.do("notify max loop reached", async () => {
await J(
this.env,
s.telegramChatId,
`Evaluation stopped after ${o} loops for ${e.instanceId}. Begin once more if you happen to nonetheless want revisions.`
);
}), {
accepted: !1,
loops: o,
finalText: c
};
const h = r.do(
`abstract agent (loop ${i})`,
async () => _(
this.env,
et,
Ok(
"Summarize this text in 5 bullet points.",
c,
s.context
)
)
)...Minified code can get fairly gnarly — and relying on the bundler, it could possibly get gnarly in a bunch of various instructions.
We would have liked a method to parse the assorted types of minified code shortly and exactly. We determined oxc-parser from the JavaScript Oxidation Compiler (OXC) was good for the job. We first examined this concept by having a container working Rust. Each script ID was despatched to a Cloudflare Queue, after which messages have been popped and despatched to the container to course of. As soon as we confirmed this method labored, we moved to a Employee written in Rust. Employees helps working Rust by way of WebAssembly, and the bundle was sufficiently small to make this easy.
The Rust Employee is chargeable for first changing the minified JS into AST node varieties, then changing the AST node varieties into the graphical model of the workflow that’s rendered on the dashboard. To do that, we generate a graph of pre-defined node varieties for every workflow and translate into our graph illustration via a collection of node mappings.
There have been two challenges to rendering a diagram model of the workflow: the way to monitor step and performance relationships appropriately, and the way to outline the workflow node varieties as merely as attainable whereas masking all of the floor space.
To ensure that step and performance relationships are tracked appropriately, we would have liked to gather each the perform and step names. As we mentioned earlier, the engine solely has details about the steps, however a step could also be depending on a perform, or vice versa. For instance, builders would possibly wrap steps in capabilities or outline capabilities as steps. They may additionally name steps inside a perform that come from completely different modules or rename steps.
Though the library passes the preliminary hurdle by giving us the AST, we nonetheless should resolve the way to parse it. Some code patterns require extra creativity. For instance, capabilities — inside a WorkflowEntrypoint, there could be capabilities that decision steps instantly, not directly, or by no means. Think about functionA, which incorporates console.log(await functionB(), await functionC()) the place functionB calls a step.do(). In that case, each functionA and functionB must be included on the workflow diagram; nevertheless, functionC mustn’t. To catch all capabilities which embrace direct and oblique step calls, we create a subgraph for every perform and test whether or not it incorporates a step name itself or whether or not it calls one other perform which could. These subgraphs are represented by a perform node, which incorporates all of its related nodes. If a perform node is a leaf of the graph, that means it has no direct or oblique workflow steps inside it, it’s trimmed from the ultimate output.
We test for different patterns as nicely, together with a listing of static steps from which we will infer the workflow diagram or variables, outlined in as much as ten alternative ways. In case your script incorporates a number of workflows, we observe an identical sample to the subgraphs created for capabilities, abstracted one stage larger.
For each AST node sort, we needed to contemplate each manner they may very well be used within a workflow: loops, branches, guarantees, parallels, awaits, arrow capabilities… the record goes on. Even inside these paths, there are dozens of potentialities. Think about just some of the attainable methods to loop:
// for...of
for (const merchandise of things) {
await step.do(`course of ${merchandise}`, async () => merchandise);
}
// whereas
whereas (shouldContinue) {
await step.do('ballot', async () => getStatus());
}
// map
await Promise.all(
objects.map((merchandise) => step.do(`map ${merchandise}`, async () => merchandise)),
);
// forEach
await objects.forEach(async (merchandise) => {
await step.do(`every ${merchandise}`, async () => merchandise);
});And past looping, the way to deal with branching:
// swap / case
swap (motion.sort) {
case 'create':
await step.do('deal with create', async () => {});
break;
default:
await step.do('deal with unknown', async () => {});
break;
}
// if / else if / else
if (standing === 'pending') {
await step.do('pending path', async () => {});
} else if (standing === 'energetic') {
await step.do('energetic path', async () => {});
} else {
await step.do('fallback path', async () => {});
}
// ternary operator
await (cond
? step.do('ternary true department', async () => {})
: step.do('ternary false department', async () => {}));
// nullish coalescing with step on RHS
const myStepResult =
variableThatCanBeNullUndefined ??
(await step.do('nullish fallback step', async () => 'default'));
// strive/catch with lastly
strive {
await step.do('strive step', async () => {});
} catch (_e) {
await step.do('catch step', async () => {});
} lastly {
await step.do('lastly step', async () => {});
}Our purpose was to create a concise API that communicated what builders must know with out overcomplicating it. However changing a workflow right into a diagram meant accounting for each sample (whether or not it follows finest practices, or not) and edge case attainable. As we mentioned earlier, every step shouldn’t be explicitly sequential, by default, to another step. If a workflow doesn’t make the most of await and Promise.all(), we assume that the steps will execute within the order by which they’re encountered. But when a workflow included await, Promise or Promise.all(), we would have liked a method to monitor these relationships.
We selected monitoring execution order, the place every node has a begins: and resolves: area. The begins and resolves indices inform us when a promise began executing and when it ends relative to the primary promise that began with out a right away, subsequent conclusion. This correlates to vertical positioning within the diagram UI (i.e., all steps with begins:1 will likely be inline). If steps are awaited when they’re declared, then begins and resolves will likely be undefined, and the workflow will execute within the order of the steps’ look to the runtime.
Whereas parsing, after we encounter an unawaited Promise or Promise.all(), that node (or nodes) are marked with an entry quantity, surfaced within the begins area. If we encounter an await on that promise, the entry quantity is incremented by one and saved because the exit quantity (which is the worth in resolves). This enables us to know which guarantees run on the identical time and once they’ll full in relation to one another.
export class ImplicitParallelWorkflow extends WorkflowEntrypoint {
async run(occasion: WorkflowEvent, step: WorkflowStep) {
const branchA = async () => {
const a = step.do("task a", async () => "a"); //begins 1
const b = step.do("task b", async () => "b"); //begins 1
const c = await step.waitForEvent("task c", { sort: "my-event", timeout: "1 hour" }); //begins 1 resolves 2
await step.do("task d", async () => JSON.stringify(c)); //begins 2 resolves 3
return Promise.all([a, b]); //resolves 3
};
const branchB = async () => {
const e = step.do("task e", async () => "e"); //begins 1
const f = step.do("task f", async () => "f"); //begins 1
return Promise.all([e, f]); //resolves 2
};
await Promise.all([branchA(), branchB()]);
await step.sleep("final sleep", 1000);
}
} You possibly can see the steps’ alignment within the diagram:
After accounting for all of these patterns, we settled on the next record of node varieties:
| StepSleep
| StepDo
| StepWaitForEvent
| StepSleepUntil
| LoopNode
| ParallelNode
| TryNode
| BlockNode
| IfNode
| SwitchNode
| StartNode
| FunctionCall
| FunctionDef
| BreakNode;Listed below are a couple of samples of API output for various behaviors:
perform name:
{
"functions": {
"runLoop": {
"name": "runLoop",
"nodes": []
}
}
}if situation branching to step.do:
{
"type": "if",
"branches": [
{
"condition": "loop > maxLoops",
"nodes": [
{
"type": "step_do",
"name": "notify max loop reached",
"config": {
"retries": {
"limit": 5,
"delay": 1000,
"backoff": "exponential"
},
"timeout": 10000
},
"nodes": []
}
]
}
]
}parallel with step.do and waitForEvent:
{
"type": "parallel",
"kind": "all",
"nodes": [
{
"type": "step_do",
"name": "correctness agent (loop ${...})",
"config": {
"retries": {
"limit": 5,
"delay": 1000,
"backoff": "exponential"
},
"timeout": 10000
},
"nodes": [],
"starts": 1
},
...
{
"type": "step_wait_for_event",
"name": "wait for user response (loop ${...})",
"options": {
"event_type": "user-response",
"timeout": "unknown"
},
"starts": 3,
"resolves": 4
}
]
}Finally, the purpose of those Workflow diagrams is to function a full-service debugging device. Which means you’ll have the ability to:
Hint an execution via the graph in actual time
Uncover errors, watch for human-in-the-loop approvals, and skip steps for testing
Entry visualizations in native growth
Take a look at the diagrams in your Workflow overview pages. In case you have any characteristic requests or discover any bugs, share your suggestions instantly with the Cloudflare group by becoming a member of the Cloudflare Builders group on Discord.
