Flow Like Water | DJ Petersen

The story

In 2023 I was provisioning a Kubernetes cluster with Talos. The official guide was a long sequence of terminal commands — some synchronous, some that kicked off work I then had to wait on — and a transient failure midway through meant starting the whole sequence over.

The obvious fix was retry logic. But wrapping each command in try-catch blocks turned a readable script into nested callback hell, and tracking which steps had already succeeded meant scattering state across variables. The script that was supposed to save time was consuming it.

What I actually needed wasn't a script — it was a state machine. Each task moves through a small set of states: not_started →in_progress → completed, failed, or skipped. Once I framed it that way, the rest followed: tasks declare their own retries and conditions, groups coordinate related tasks, event listeners report progress, and the whole run serializes so it can pick up where it stopped.

Flow Like Water is that engine, written in TypeScript and tested with Jest. It's a small library I built because I needed it. The repo is public; I haven't touched it since early 2024, and dormant is the honest label.

How it works

Three ideas carry the library. First, retries are declared on the task instead of wrapped around it — waitTime accepts a function of the attempt number, so exponential backoff is one line:

retry-with-backoff.ts
const deployPod = new Task({
  id: 'deploy-pod',
  execute: async () => {
    await kubectl.createPod(podSpec);
  // Retry up to 5 times with exponential backoff
  retries: 5,
  waitTime: (attempt) => Math.min(1000 * Math.pow(2, attempt), 30000),
  // Ensure pod is actually running before considering task complete
  checkCondition: async () => {
    const status = await kubectl.getPodStatus('my-pod');
    return status === 'Running';

Second, transitions are dynamic. execute returns the id of the next task to run, so a run can branch on its own results. AndcheckCondition acts as an idempotency guard — if the namespace already exists, the task is skipped rather than re-run:

dynamic-transitions.ts
const createNamespace = new Task({
  id: 'create-namespace',
  execute: async () => {
    const result = await kubectl.createNamespace('my-app');
    return result.success ? 'deploy-app' : undefined;
  checkCondition: async () => {
    const namespaces = await kubectl.listNamespaces();
    return !namespaces.includes('my-app');
  retries: 3,
  waitTime: 2000

Third, the whole run serializes. Persist the state anywhere, and after a crash you find the first incomplete task and resume from there:

resume-from-state.ts
// Save the current state
const state = flow.getSerializedState();
await fs.writeFile('workflow-state.json', JSON.stringify(state));
// Later, you can use this state to determine which tasks need to be re-run
const savedState = JSON.parse(await fs.readFile('workflow-state.json'));
const incompleteTask = Object.entries(savedState)
  .find(([_, data]) => data.state !== 'completed');
if (incompleteTask) {
  await flow.runTask(incompleteTask[0]);

Demo

Simulated run — scripted illustration, including the failure

Flow Like WaterSimulated

The run is canned: the transform task fails once, backs off, and the pipeline resumes from where it stopped — which is the library's entire pitch.

Artifacts & further reading

[1]GitHubgithub.com

1	const deployPod = new Task({
2	id: 'deploy-pod',
3	execute: async () => {
4	await kubectl.createPod(podSpec);
5	},
6	// Retry up to 5 times with exponential backoff
7	retries: 5,
8	waitTime: (attempt) => Math.min(1000 * Math.pow(2, attempt), 30000),
9	// Ensure pod is actually running before considering task complete
10	checkCondition: async () => {
11	const status = await kubectl.getPodStatus('my-pod');
12	return status === 'Running';
13	}
14	});

1	const createNamespace = new Task({
2	id: 'create-namespace',
3	execute: async () => {
4	const result = await kubectl.createNamespace('my-app');
5	return result.success ? 'deploy-app' : undefined;
6	},
7	checkCondition: async () => {
8	const namespaces = await kubectl.listNamespaces();
9	return !namespaces.includes('my-app');
10	},
11	retries: 3,
12	waitTime: 2000
13	});

1	// Save the current state
2	const state = flow.getSerializedState();
3	await fs.writeFile('workflow-state.json', JSON.stringify(state));
4
5	// Later, you can use this state to determine which tasks need to be re-run
6	const savedState = JSON.parse(await fs.readFile('workflow-state.json'));
7	const incompleteTask = Object.entries(savedState)
8	.find(([_, data]) => data.state !== 'completed');
9
10	if (incompleteTask) {
11	await flow.runTask(incompleteTask[0]);
12	}