bilaldemir.dev

Building a DAG Workflow Engine in Go

A workflow engine sounds big, but the core idea is small. A workflow is a directed acyclic graph: nodes are steps (call an LLM, post to Slack, upload a file), edges say "this runs after that", and data flows along the edges. The engine's job is to run every node once, as early as possible, with the outputs of the nodes it depends on.

I built this twice. First as a prototype, go-dag. Then for real, as the execution engine of BlockNext, an open-source AI workflow platform. The prototype is useful mostly as a list of mistakes, so let's start there.

The prototype

go-dag takes a workflow as JSON: a list of nodes with their inputs, and a list of edges. Inputs can refer to another node's output with $node.key.

{
  "nodes": [
    { "id": "openai",       "inputs": { "prompt": "Create a futuristic city illustration" } },
    { "id": "dall-e",       "inputs": { "prompt": "$openai.response" } },
    { "id": "google-drive", "inputs": { "file": "$dall-e.image" } },
    { "id": "slack",        "inputs": { "message": "Image created: $google-drive.result" } },
    { "id": "telegram",     "inputs": { "message": "Image created: $google-drive.result" } }
  ],
  "edges": [
    { "source": "openai",       "target": "dall-e" },
    { "source": "dall-e",       "target": "google-drive" },
    { "source": "google-drive", "target": "slack" },
    { "source": "google-drive", "target": "telegram" }
  ]
}

The engine sorts the nodes topologically with a depth-first search, which also detects cycles. Then it resolves each node's $references from the outputs collected so far, and runs the node. Each node type is a small struct that registers itself in a global registry from an init() function. Adding an integration means adding a package and one blank import.

That part held up. The rest is where it went wrong.

What the prototype got wrong

It looked parallel, but it wasn't. The graph type had a sync.WaitGroup, every node had a mutex, and outputs lived in a sync.Map. All the parts of a concurrent engine were there. But the main loop called each node synchronously and then waited for it:

for _, nodeConfig := range sortedNodes {
	node, _ := g.nodeManager.GetNode(nodeConfig.ID)
	inputs := resolveInputs(nodeConfig.Inputs)

	g.wg.Add(1)
	g.ExecuteNode(node, inputs) // synchronous call
	g.wg.Wait()                 // ...and wait for it
}

In the example above, Slack and Telegram don't depend on each other and could run at the same time. In the prototype they ran one after the other. All the synchronization was paying for concurrency the engine never had.

One bad workflow could take down the server. The HTTP handler called log.Fatalf when loading a workflow failed. In Go that exits the process. Any client could crash the service with a workflow that had a cycle.

Node settings never reached the node. Settings from the JSON, like API keys, were stored on a per-run node struct, but Process was a method on the registered processor, which is a shared singleton. So the processor read its own empty settings, not the ones from the workflow. And if it had read them, every concurrent run would have shared and overwritten the same settings.

No context, no timeout, no errors. Process took a map and returned a map. A slow API call couldn't be cancelled, and a failed one had no way to say so, other than putting something in the output map and hoping the next node noticed.

None of this is unusual for a prototype. But each one is a design decision, and it's much cheaper to fix in a few hundred lines than in a production system.

Ordering: Kahn's algorithm with priorities

BlockNext's dag package builds forward and reverse adjacency lists, finds start nodes (no incoming edges, starter type), and drops nodes that can't be reached from any start node, with a warning instead of an error. Ordering uses Kahn's algorithm over a min-heap:

for pq.Len() > 0 {
	item := heap.Pop(pq).(*Item)
	nodeID := item.ID
	result = append(result, nodeID)

	for _, neighbor := range d.adjacencyList[nodeID] {
		if reachable[neighbor] {
			inDegree[neighbor]--
			if inDegree[neighbor] == 0 {
				heap.Push(pq, &Item{
					ID:       neighbor,
					Priority: d.getNodePriority(neighbor),
				})
			}
		}
	}
}

if len(result) != len(reachable) {
	return ErrCycleDetected
}

The heap gives a stable order: starters first, then conditions, then actions, with the node ID as tie-breaker. The same workflow always sorts the same way, which makes logs and tests deterministic. If some reachable node never reaches in-degree zero, there is a cycle, and that is a validation error, not a crash.

The important change from the prototype: the sorted list is used to validate the graph, not to run it.

Execution: driven by dependencies, not by a list

The coordinator starts every start node in its own goroutine. When a node finishes, it looks at the node's children. A child is ready when it is on a branch that was actually taken and all of its parents are done. All ready children start at once:

var wg sync.WaitGroup
errChan := make(chan error, len(readyChildren))

for _, childNode := range readyChildren {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	wg.Add(1)
	go func(node *dag.Node) {
		defer wg.Done()
		if err := c.executeNode(ctx, task, node); err != nil {
			errChan <- err
		}
	}(childNode)
}

wg.Wait()
close(errChan)

That one change turns the Slack/Telegram example into what it should be: both run at the same time as soon as the Drive upload finishes. Checking ctx.Done() before starting each child means a cancelled run stops spreading instead of finishing the graph in the background.

Joins: run once, no matter how many parents

A node with two parents, like a merge, creates a race. Both parents finish around the same time, both see "all parents done", and both try to start the child. The fix is one atomic check:

if _, started := task.StartedNodes.LoadOrStore(node.ID, struct{}{}); started {
	return nil // another parent already started this join node
}

LoadOrStore on a per-task sync.Map guarantees that exactly one goroutine wins. The prototype tried to solve this with a mutex and a Done flag per node. That works only if every path remembers to take the lock in the right order.

Branches

Conditions make the graph dynamic: after a condition runs, only one of its outgoing branches should continue. In BlockNext every node takes an array of items and returns an array, and a condition sends each item down the branch its own comparison chose. An output with no items simply skips whatever hangs off it. The coordinator checks whether a branch was actually reached before treating a child as ready, so nodes on a branch that wasn't taken never start.

Retries, timeouts and partial failure

Every node carries its own execution settings:

type NodeSettings struct {
	MaxRetries      float64 `json:"maxRetries"`
	RetryDelay      float64 `json:"retryDelay"`
	Timeout         float64 `json:"timeout"`
	ContinueOnError bool    `json:"continueOnError"`
	Disabled        bool    `json:"disabled"`
}

Each attempt runs under context.WithTimeout, failed attempts are retried with a fixed delay, and ContinueOnError lets the children run even if this node failed. That is useful for "notify, but don't block the rest" steps. Disabled nodes are marked as skipped and count as finished, so they don't block their children. Rerunning a failed workflow reuses the stored outputs of the nodes that already succeeded, so you don't pay for the same LLM call twice.

What I deliberately left out

There are no loops, no sub-workflows and no expression language. Each of those makes a workflow engine more powerful and much harder for a non-developer to read. The batch model (arrays in, arrays out, references per item) covers most of what loops are used for.

If you're building one

← All posts