Chapter 1 · Workflow state machines

Flow Like Water

A TypeScript workflow engine that turns fragile scripts into explicit states, transitions, and recoverable execution.

Narrative

Flow Like Water grew out of a Kubernetes provisioning project where a dozen shell commands had to run in strict order, with guardrails and retries. The official instructions were clear, but brittle: miss a step, and you re-run everything from scratch.

The library reframes that situation as a state machine. Each command becomes a task that knows when it can run and where it should transition next. Tasks can be grouped, replayed, and observed with lifecycle events, which turns procedural scripts into a workflow you can reason about.

Core idea

Treat every step as a state. Transitions are explicit. Retries are built into the model rather than bolted on afterwards.

Example: provisioning flow

This example shows the typical pattern: declare tasks, define conditions, group them, and let the controller run the workflow.

workflow.ts   
    import { Task, TaskGroup, FlowControl } from "flow-like-water"; 
  const createNamespace = new Task({ 
    id: "create-namespace", 
    execute: async () => await kubectl.createNamespace("app"), 
    checkCondition: async () => { 
      const namespaces = await kubectl.listNamespaces(); 
      return !namespaces.includes("app"); 
    retries: 3, 
    waitTime: 1500, 
  const deployApp = new Task({ 
    id: "deploy-app", 
    execute: async () => await kubectl.applyManifest("app.yaml"), 
    retries: 2, 
  const group = new TaskGroup("bootstrap"); 
  group.addChild(createNamespace); 
  group.addChild(deployApp); 
  const flow = new FlowControl(); 
  flow.addGroup(group); 
  await flow.run(); 
  

More examples

1. Progress events

Subscribe to lifecycle events to drive logs, dashboards, or notifications.

events.ts   
    flow.on("taskStarted", (task) => { 
    console.log(`Starting ${task.id}`); 
  flow.on("taskCompleted", (task) => { 
    console.log(`Finished ${task.id}`); 
  

2. State snapshots

Persist workflow state and resume where you left off after an interruption.

resume.ts   
    const state = flow.getSerializedState(); 
  await fs.writeFile("workflow.json", JSON.stringify(state)); 
  const saved = JSON.parse(await fs.readFile("workflow.json")); 
  const next = Object.entries(saved).find(([_, data]) => data.state !== "completed"); 
  if (next) { 
    await flow.runTask(next[0]); 
  

3. Retry with backoff

Encapsulate retry logic inside the task definition instead of writing wrapper loops.

retry.ts   
    const deployPod = new Task({ 
    id: "deploy-pod", 
    execute: async () => await kubectl.createPod(podSpec), 
    retries: 5, 
    waitTime: (attempt) => Math.min(1000 * 2 ** attempt, 30000), 
    checkCondition: async () => { 
      const status = await kubectl.getPodStatus("my-pod"); 
      return status === "Running"; 
  

1	import { Task, TaskGroup, FlowControl } from "flow-like-water";
2
3	const createNamespace = new Task({
4	id: "create-namespace",
5	execute: async () => await kubectl.createNamespace("app"),
6	checkCondition: async () => {
7	const namespaces = await kubectl.listNamespaces();
8	return !namespaces.includes("app");
9	},
10	retries: 3,
11	waitTime: 1500,
12	});
13
14	const deployApp = new Task({
15	id: "deploy-app",
16	execute: async () => await kubectl.applyManifest("app.yaml"),
17	retries: 2,
18	});
19
20	const group = new TaskGroup("bootstrap");
21	group.addChild(createNamespace);
22	group.addChild(deployApp);
23
24	const flow = new FlowControl();
25	flow.addGroup(group);
26
27	await flow.run();

1	flow.on("taskStarted", (task) => {
2	console.log(`Starting ${task.id}`);
3	});
4
5	flow.on("taskCompleted", (task) => {
6	console.log(`Finished ${task.id}`);
7	});

1	const state = flow.getSerializedState();
2	await fs.writeFile("workflow.json", JSON.stringify(state));
3
4	const saved = JSON.parse(await fs.readFile("workflow.json"));
5	const next = Object.entries(saved).find(([_, data]) => data.state !== "completed");
6	if (next) {
7	await flow.runTask(next[0]);
8	}

1	const deployPod = new Task({
2	id: "deploy-pod",
3	execute: async () => await kubectl.createPod(podSpec),
4	retries: 5,
5	waitTime: (attempt) => Math.min(1000 * 2 ** attempt, 30000),
6	checkCondition: async () => {
7	const status = await kubectl.getPodStatus("my-pod");
8	return status === "Running";
9	},
10	});