flow

flow is a small, type-safe control-flow library for Go. Compose ordinary functions into reusable nodes, then opt into workflow only when a definition must come from JSON, a database, or a visual editor.

The library is in-process and explicit:

• one protocol: Run(context.Context, input) (output, error);
• compile-time types for Go-defined pipelines;
• runtime DAGs, named ports, and strict JSON Schema validation when needed;
• checkpointed suspension and resumption without a resident scheduler; and
• standard cancellation and error inspection with context, errors.Is, and errors.As.

Install

flow requires Go 1.26 or newer.

go get github.com/Tangerg/flow

Choose the smallest package

Package	Use it for
flow	Typed sequence, selection, iteration, mapping, and races
flowx	Derived conveniences such as fan-out, fallback, and same-type chains
workflow	Named state, runtime definitions, streaming, and resumption
workflow/expr	Optional data-driven branch and loop rules
workflow/diagram	Deterministic ASCII and Mermaid Graph renderings

Start with flow. Use workflow when the definition itself must be assembled at run time.

Quick start

Adapt ordinary functions with NodeFunc, then connect unlike types with Then. The complete runnable version is example/node_test.go:

package main

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/Tangerg/flow"
)

func main() {
	parse := flow.NodeFunc[string, int](
		func(_ context.Context, in string) (int, error) {
			return strconv.Atoi(strings.TrimSpace(in))
		},
	)
	double := flow.NodeFunc[int, int](
		func(_ context.Context, in int) (int, error) {
			return in * 2, nil
		},
	)

	pipeline := flow.Then(parse, double)
	out, err := pipeline.Run(context.Background(), " 21 ")
	if err != nil {
		panic(err)
	}
	fmt.Println(out) // 42
}

pipeline is a Node[string, int]. A composition remains a node, so it can be run, tested, wrapped, or composed again.

Typed composition

The root package revolves around one interface:

type Node[I, O any] interface {
	Run(ctx context.Context, in I) (O, error)
}

API	Meaning
NodeFunc	Adapt a function to Node
Then	Pass one node's output to the next
Switch	Select a node from the current input
Loop	Repeat a node until a condition is met
Map	Run one node for every element and preserve input order
Race	Run several nodes and return the first success

Map and Race own the goroutines they start, propagate cancellation, and wait for started calls to return. Cancellation remains cooperative: nodes must observe the context they receive.

flowx contains only derived composition shapes: Chain, FanOut, Combine, and Fallback. Retry, timeout, tracing, and circuit breaking are policies; implement them as Node[I, O] decorators or use a dedicated package.

See Nodes and Then and Composition and concurrency for the typed path.

Runtime-defined workflows

workflow.Step is an alias for flow.Node[Store, Store]. A leaf reads a named value and publishes its result under its own ID. The complete runnable version is example/workflow_test.go:

clean := workflow.LeafFunc(
	"clean",
	workflow.Output("input"),
	func(_ context.Context, in string) (string, error) {
		return strings.TrimSpace(in), nil
	},
)
greet := workflow.LeafFunc(
	"greet",
	workflow.Output("clean"),
	func(_ context.Context, name string) (string, error) {
		return "hello, " + name, nil
	},
)

out, err := workflow.Sequence(clean, greet).Run(
	ctx,
	workflow.NewStore().WithOutput("input", " Ada "),
)
if err != nil {
	return err
}
message, err := workflow.Get[string](out, workflow.Output("greet"))

The Store is immutable and copy-on-write. Stored values are shared as-is, so treat maps, slices, pointers, and other mutable values as immutable after insertion. Prefer workflow.Get[T] for typed reads, including after a Store has been serialized and restored.

Runtime definitions have two complementary forms:

Form	Best for
Graph	Flat DAGs with named-port edges, bounded concurrency, and conditional routes
Spec	Nested sequence, parallel, branch, loop, iteration, and sealed subgraphs

A code-built or compiled Step can describe itself with workflow.Describe. Descriptions use the same typed workflow.Kind values as Spec—for example, workflow.KindGraph and workflow.KindSubgraph—so tooling does not compare undocumented string literals.

A Registry is the capability boundary between external data and executable Go code. Both definition forms compile to an ordinary Step:

JSON -> strict decode -> JSON Schema -> Registry validation -> Step

if err := workflow.ValidateGraphJSON(data); err != nil {
	return err
}
step, err := registry.CompileGraphJSON(data)

GraphJSONSchema and SpecJSONSchema return self-contained Draft 2020-12 schemas for editors and API endpoints. Graph input ports are also dependency edges: ready nodes start as soon as their own dependencies finish, subject to Graph.Concurrency. Subgraph provides an isolated, reusable region with declared inputs and one projected result.

The tutorials cover Stores and references, Graph compilation, the JSON DSL, and conditional graphs.

Per-run services

Definitions are reusable. Observation, streaming output, and replay state belong to one call:

out, err := workflow.Run(ctx, step, input, workflow.RunConfig{
	Observer: observer,
	Emitter:  emitter,
	Journal:  journal,
})

Service	Purpose
Observer	Low-volume lifecycle events
Emitter	High-volume intermediate application values
Journal	Completed boundaries, decisions, and interrupt responses

StreamFunc remains an ordinary typed Node; Leaf gives its chunks workflow identity. Emission is synchronous and applies backpressure. Chunks describe attempted output, not durable delivery: rerunning an incomplete leaf may repeat a prefix. See Streaming output.

Await, Interrupt, and Suspend stop a run with an error matching ErrSuspended. Persist the returned Store and Journal, record the external response, and run the same definition again. Replay restores completed boundaries instead of serializing a Go call stack. See Suspension and resumption.

Store and Journal JSON are persistence values, not an application run record. Persist the application run ID, active waits, authorization and status, the workflow-definition version, and the flow module version separately. A Journal document carries its own wire-format version and unsupported versions are rejected.

Execution boundary

Package dependencies point toward the typed core:

workflow/diagram ---> workflow ---> flow
workflow/expr ------> workflow
flowx ---------------------------> flow

There is no central orchestrator object or background scheduler. The project does not provide distributed workers, queues, timers, leases, workflow migration, deterministic instruction-level replay, or exactly-once external effects.

Resumption is checkpoint-and-restart at Step boundaries. If a process stops after an external side effect succeeds but before its result is journaled, the effect may run again. Use idempotency keys, a transactional outbox, or domain compensation. Choose a durable workflow service when distributed scheduling and durable timers are requirements.

Documentation

• Tutorials — progressive paths from one node to runtime-defined, resumable workflows.
• Executable examples — public-API examples with asserted output.
• Documentation index — API reference and maintainer documents.
• Roadmap — remaining stabilization work and engine boundaries.
• Changelog — user-visible work for the next release.
• Contributing — development and API-review requirements.

Package comments and examples are the canonical API reference:

go doc github.com/Tangerg/flow
go doc github.com/Tangerg/flow/workflow

Stability

The project follows Semantic Versioning. Before v1, minor releases may refine public APIs and will document migrations in the changelog. After v1, exported behavior and error contracts become compatibility commitments.

Use keyed fields for exported structs and prefer constructors such as Output, At, Item, and ItemIndex where provided.

Development

Run the local gate before submitting a change:

test -z "$(gofmt -l .)"
go mod tidy -diff
go test -race -cover ./...
go vet ./...

See CONTRIBUTING.md for the complete checks and design boundaries.