Help us shape the future of Go! Take the 2026 Go Developer Survey

flow

package module
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 5, 2026 License: MIT Imports: 8 Imported by: 0

README

go-flow

CI Go Reference Go Version Release Coverage

A high-performance, minimalist Go structured concurrency and workflow orchestration engine designed for modern Go (1.24+). Built with generics, standard iterators, and zero external dependencies.

Inspired by the structured execution model of C++26 std::execution (NVIDIA/stdexec), go-flow brings declarative, composable execution graphs and execution combinators to idiomatic Go. It replaces unstructured goroutine lifecycles and manual channel boilerplate with strictly scoped, cancellable, and resilient execution pipelines.

Features

  • Generic Step Abstraction: Step[T context.Context] operates natively on standard contexts and custom context types.
  • Fluent Receiver Methods: Chain, compose, and harden steps with .Then(), .Go(), .GoN(), .Race(), .Once(), .Timeout(), .Retry(), .Fallback(), .Catch(), .Recover(), .When(), .Unless(), .Branch().
  • Sequential Execution (Seq): Executes steps in strict order, aborting immediately on the first error.
  • Conditional Branching (Branch): Clean if/else execution paths evaluated dynamically at runtime.
  • Typed Functional Piping (Pipe, PipeSeq, Pipe2, Pipe3): Type-safe functional value transformations and iterator streaming.
  • Unbounded Concurrency (Go): Executes steps concurrently across goroutines, joining all errors.
  • Bounded Concurrency (GoN): Throttles concurrent step execution with an exact worker pool / concurrency limit.
  • Speculative Racing (Race): Races steps concurrently, returning immediately on first success while canceling losing branches.
  • Iterator Streaming (Each, Each2): Streams Go iter.Seq and iter.Seq2 sequences directly through step pipelines.
  • Idempotent Step (Once): Guarantees a step executes at most once across workflows and branches.
  • Topological DAG Engine: Declarative directed acyclic graph execution with cycle validation, maximal branch concurrency, pure function edges (DAGEdges, From.To, Edge), named nodes (DAG, Node.After), and bounded concurrency (DAGN, DAGEdgesN).
  • Zero Dependencies: 100% standard library.

Installation

go get github.com/ntkwan/go-flow

Mental Model

graph LR
    subgraph Sequential [1. Linear Seq]
        A1[Step 1] --> A2[Step 2] --> A3[Step 3]
    end
    subgraph Parallel [2. Concurrent Go / GoN]
        B1[Worker 1]
        B2[Worker 2]
        B3[Worker 3]
    end
    subgraph Speculative [3. Speculative Race]
        R1[Server 1]
        R2[Server 2]
        R3[Server 3]
    end
    subgraph DAG [4. Dependency DAG]
        D1[Fetch User] --> D3[Process Payment]
        D2[Fetch Cart] --> D3
        D3 --> D4[Send Receipt]
    end

Quick Start

[!TIP] Fully runnable, side-by-side comparative examples with and without go-flow are available in the examples/ directory.

1. Fluent Method Chaining

Every Step[T] provides fluent receiver methods for composition, conditional execution, and resilience:

validateInput := flow.Step[context.Context](validateStep)
fetchUserData := flow.Step[context.Context](fetchUserStep)
updateCache   := flow.Step[context.Context](updateCacheStep)

pipeline := validateInput.
	Then(fetchUserData.Retry(3, 100*time.Millisecond).Timeout(2*time.Second)).
	Then(updateCache.Once()).
	Recover()

if err := pipeline.Exec(ctx); err != nil {
	log.Println("Pipeline failed:", err)
}

Examples: Retry | Timeout | Fallback | Middleware Wrap

2. Sequential Execution (Seq)

See runnable example: examples/seq/with_flow vs examples/seq/without_flow

package main

import (
	"context"
	"fmt"

	"github.com/ntkwan/go-flow"
)

func main() {
	workflow := flow.Seq(
		func(ctx context.Context) error {
			fmt.Println("Step 1: Validate input")
			return nil
		},
		func(ctx context.Context) error {
			fmt.Println("Step 2: Save to database")
			return nil
		},
	)

	if err := workflow(context.Background()); err != nil {
		panic(err)
	}
}
3. Parallel Execution (Go)

See runnable example: examples/go/with_flow vs examples/go/without_flow

workflow := flow.Go(
	func(ctx context.Context) error { return fetchUserProfile(ctx) },
	func(ctx context.Context) error { return fetchUserOrders(ctx) },
	func(ctx context.Context) error { return fetchUserPreferences(ctx) },
)

if err := workflow(ctx); err != nil {
	// Returns errors joined via errors.Join
	log.Println("Workflow errors:", err)
}
4. Bounded Concurrency (GoN)

See runnable example: examples/gon/with_flow vs examples/gon/without_flow

var steps []flow.Step[context.Context]
for _, item := range batchItems {
	steps = append(steps, func(ctx context.Context) error {
		return processItem(ctx, item)
	})
}

// Concurrently process items with at most 5 active workers
pipeline := flow.GoN(5, steps...)
if err := pipeline(ctx); err != nil {
	log.Println("Processing failed:", err)
}
5. Speculative Racing (Race)

See runnable example: examples/race/with_flow vs examples/race/without_flow

// Query multiple replicas or fallbacks; returns on first success and cancels the others
fastest := flow.Race(
	func(ctx context.Context) error { return queryPrimaryServer(ctx) },
	func(ctx context.Context) error { return queryMirrorServer(ctx) },
	func(ctx context.Context) error { return queryCachedReplica(ctx) },
)

if err := fastest(ctx); err != nil {
	log.Println("All endpoints failed:", err)
}
6. Iterator Streaming (Each, Each2)

See runnable example: examples/each/with_flow vs examples/each/without_flow

items := slices.Values([]string{"order-1", "order-2", "order-3"})

stream := flow.Each(items, func(ctx context.Context, item string) error {
	return processOrder(ctx, item)
})

if err := stream(ctx); err != nil {
	log.Println("Stream interrupted:", err)
}
7. Idempotent Execution (Once)

See runnable example: examples/once/with_flow vs examples/once/without_flow

// Ensure expensive initialization or connection warmup runs only once
initDB := flow.Once(func(ctx context.Context) error {
	return connectDatabase(ctx)
})

stepA := flow.Seq(initDB, queryUsers)
stepB := flow.Seq(initDB, queryOrders)
8. Directed Acyclic Graph (DAG)

See runnable example: examples/dag/with_flow vs examples/dag/without_flow

go-flow provides multiple ways to define and execute dependency graphs:

A. Pure Function Edges (From.To / Edge)

Wire functions directly into dependency graphs without string keys:

// Using From(...).To(...) fluent syntax
pipeline := flow.DAGEdges(
	flow.From(fetchUser).To(processPayment),
	flow.From(fetchCart).To(processPayment),
	flow.From(processPayment).To(sendReceipt),
)

// Or using Edge(...) helper
pipeline := flow.DAGEdges(
	flow.Edge(fetchUser, processPayment),
	flow.Edge(fetchCart, processPayment),
	flow.Edge(processPayment, sendReceipt),
)

if err := pipeline(ctx); err != nil {
	log.Println("DAG execution failed:", err)
}
B. Named Nodes (Node + .After)

Define explicit node identifiers for self-documenting workflows and logging:

userNode := flow.Node("fetch-user", fetchUser)
cartNode := flow.Node("fetch-cart", fetchCart)

paymentNode := flow.Node("process-payment", processPayment).
	After("fetch-user", "fetch-cart")

receiptNode := flow.Node("send-receipt", sendReceipt).
	After("process-payment")

// Independent branches ("fetch-user" and "fetch-cart") run in parallel.
// "process-payment" runs only after both dependencies succeed.
pipeline := flow.DAG(userNode, cartNode, paymentNode, receiptNode)
if err := pipeline(ctx); err != nil {
	log.Println("DAG execution failed:", err)
}
C. Bounded Concurrency DAGs (DAGN / DAGEdgesN)

Cap the maximum number of concurrent in-flight nodes across the graph:

// Throttle DAG execution to at most 2 concurrent node workers
pipeline := flow.DAGEdgesN(2,
	flow.From(fetchUser).To(processPayment),
	flow.From(fetchCart).To(processPayment),
	flow.From(processPayment).To(sendReceipt),
)

// Or with named nodes
boundedDAG := flow.DAGN(2, userNode, cartNode, paymentNode, receiptNode)
D. Visual Graph Export (Mermaid & Graphviz DOT)

See runnable example: examples/order_checkout/with_flow vs examples/order_checkout/without_flow

Generate Mermaid markdown diagrams or Graphviz DOT graphs directly from Go code for documentation, live telemetry, and debugging:

plan := flow.NewDAG(
	flow.Node("validate-order", validateOrder),
	flow.Node("fetch-user", fetchUser).After("validate-order"),
	flow.Node("fetch-inventory", fetchInventory).After("validate-order"),
	flow.Node("calculate-discounts", calculateDiscounts).After("fetch-user"),
	flow.Node("process-payment", processPayment).After("calculate-discounts", "fetch-inventory"),
	flow.Node("update-inventory", updateInventory).After("process-payment"),
	flow.Node("generate-invoice", generateInvoice).After("process-payment"),
	flow.Node("notify-customer", notifyCustomer).After("generate-invoice"),
	flow.Node("dispatch-warehouse", dispatchWarehouse).After("update-inventory"),
)

// Export to Mermaid markdown syntax
mermaid, err := plan.ToMermaid()
fmt.Println(mermaid)

// Export to Graphviz DOT syntax
dot, err := plan.ToDOT()

// Execute as Step[T] or bounded Step[T]
pipeline := plan.Step()

Generated Mermaid Diagram:

graph TD
    validate_order["validate-order"] --> fetch_user["fetch-user"]
    validate_order["validate-order"] --> fetch_inventory["fetch-inventory"]
    fetch_user["fetch-user"] --> calculate_discounts["calculate-discounts"]
    calculate_discounts["calculate-discounts"] --> process_payment["process-payment"]
    fetch_inventory["fetch-inventory"] --> process_payment["process-payment"]
    process_payment["process-payment"] --> update_inventory["update-inventory"]
    process_payment["process-payment"] --> generate_invoice["generate-invoice"]
    generate_invoice["generate-invoice"] --> notify_customer["notify-customer"]
    update_inventory["update-inventory"] --> dispatch_warehouse["dispatch-warehouse"]

Standalone export functions (flow.DAGToMermaid, flow.DAGToDOT, flow.DAGEdgesToMermaid, flow.DAGEdgesToDOT) are also available.

9. Custom Domain Contexts (Step[T])

Step[T] is generic over any type satisfying context.Context, providing compile-time type safety for domain workflows without type assertions:

type OrderContext struct {
	context.Context
	OrderID string
	UserID  string
	Total   int64
	Paid    bool
}

func validateOrder(ctx *OrderContext) error {
	if ctx.Total <= 0 {
		return errors.New("invalid order total")
	}
	return nil
}

func chargePayment(ctx *OrderContext) error {
	ctx.Paid = true
	return nil
}

// Entire pipeline is strongly typed to *OrderContext
orderPipeline := flow.Seq(
	flow.Step[*OrderContext](validateOrder),
	flow.Step[*OrderContext](chargePayment),
)

orderCtx := &OrderContext{
	Context: context.Background(),
	OrderID: "ord-883",
	UserID:  "usr-102",
	Total:   15000,
}

if err := orderPipeline(orderCtx); err != nil {
	log.Fatal(err)
}
10. Conditional Branching (Branch)

See runnable example: examples/branch/with_flow vs examples/branch/without_flow

Execute different branches dynamically based on context or state:

// Standalone Branch combinator
evalRisk := flow.Branch(
	func(ctx *OrderContext) bool { return ctx.Total > 100000 },
	manualReviewStep, // Executed when true
	autoApproveStep,  // Executed when false
)

// Or chained with fluent .Branch()
pipeline := validateOrder.
	Branch(
		func(ctx *OrderContext) bool { return ctx.IsVIP },
		applyDiscountStep,
		standardPricingStep,
	).
	Then(chargePayment)
11. Typed Functional Piping & Streaming (Pipe, PipeSeq)

See runnable example: examples/pipe/with_flow vs examples/pipe/without_flow

Bridge pure functions and typed value transformations cleanly into workflows and stream processing:

// Transform typed values and write output back into context
calculateTax := flow.Pipe(
	func(ctx *OrderContext, subtotal int64) (int64, error) {
		return int64(float64(subtotal) * 0.08), nil
	},
	func(ctx *OrderContext) int64 { return ctx.Subtotal },
	func(ctx *OrderContext, tax int64) { ctx.Tax = tax },
)

// Stream standard Go iterators through a transform step into a sink
streamPipeline := flow.PipeSeq(
	slices.Values([]int{1, 2, 3, 4, 5}),
	func(ctx context.Context, item int) (string, error) {
		return fmt.Sprintf("item-%d", item*2), nil
	},
	func(ctx context.Context, item string) error {
		return saveItem(ctx, item)
	},
)

License

MIT License

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func DAGEdgesToDOT added in v0.4.0 

func DAGEdgesToDOT[T context.Context](connections ...DAGConnection[T]) (string, error)

func DAGEdgesToMermaid added in v0.4.0 

func DAGEdgesToMermaid[T context.Context](connections ...DAGConnection[T]) (string, error)

func DAGToDOT added in v0.4.0 

func DAGToDOT[T context.Context](nodes ...*DAGNode[T]) (string, error)

func DAGToMermaid added in v0.4.0 

func DAGToMermaid[T context.Context](nodes ...*DAGNode[T]) (string, error)

func Pipe2 added in v0.3.0 

func Pipe2[T context.Context, A, B, C any](
	f1 func(ctx T, a A) (B, error),
	f2 func(ctx T, b B) (C, error),
) func(ctx T, a A) (C, error)

func Pipe3 added in v0.3.0 

func Pipe3[T context.Context, A, B, C, D any](
	f1 func(ctx T, a A) (B, error),
	f2 func(ctx T, b B) (C, error),
	f3 func(ctx T, c C) (D, error),
) func(ctx T, a A) (D, error)

Types

type DAGConnection 

type DAGConnection[T context.Context] struct {
	// contains filtered or unexported fields
}

func Edge 

func Edge[T context.Context](from Step[T], to ...Step[T]) DAGConnection[T]

type DAGEdgeBuilder 

type DAGEdgeBuilder[T context.Context] struct {
	// contains filtered or unexported fields
}

func From 

func From[T context.Context](from Step[T]) DAGEdgeBuilder[T]

func (DAGEdgeBuilder[T]) To 

func (b DAGEdgeBuilder[T]) To(to ...Step[T]) DAGConnection[T]

type DAGNode 

type DAGNode[T context.Context] struct {
	// contains filtered or unexported fields
}

func Node 

func Node[T context.Context](name string, step Step[T]) *DAGNode[T]

func (*DAGNode[T]) After 

func (n *DAGNode[T]) After(deps ...string) *DAGNode[T]

type DAGPlan added in v0.4.0 

type DAGPlan[T context.Context] struct {
	// contains filtered or unexported fields
}

func NewDAG added in v0.4.0 

func NewDAG[T context.Context](nodes ...*DAGNode[T]) *DAGPlan[T]

func NewDAGEdges added in v0.4.0 

func NewDAGEdges[T context.Context](connections ...DAGConnection[T]) (*DAGPlan[T], error)

func (*DAGPlan[T]) Step added in v0.4.0 

func (p *DAGPlan[T]) Step() Step[T]

func (*DAGPlan[T]) StepN added in v0.4.0 

func (p *DAGPlan[T]) StepN(limit int) Step[T]

func (*DAGPlan[T]) ToDOT added in v0.4.0 

func (p *DAGPlan[T]) ToDOT() (string, error)

func (*DAGPlan[T]) ToMermaid added in v0.4.0 

func (p *DAGPlan[T]) ToMermaid() (string, error)

type Middleware 

type Middleware[T context.Context] func(next Step[T]) Step[T]

type Step 

type Step[T context.Context] func(ctx T) error

func Branch added in v0.3.0 

func Branch[T context.Context](condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]

func Chunk 

func Chunk[T context.Context, V any](seq iter.Seq[V], size int, step func(ctx T, batch []V) error) Step[T]

func Chunk2 

func Chunk2[T context.Context, K, V any](seq iter.Seq2[K, V], size int, step func(ctx T, keys []K, vals []V) error) Step[T]

func DAG 

func DAG[T context.Context](nodes ...*DAGNode[T]) Step[T]

func DAGEdges 

func DAGEdges[T context.Context](connections ...DAGConnection[T]) Step[T]

func DAGEdgesN 

func DAGEdgesN[T context.Context](limit int, connections ...DAGConnection[T]) Step[T]

func DAGN 

func DAGN[T context.Context](limit int, nodes ...*DAGNode[T]) Step[T]

func Each 

func Each[T context.Context, V any](seq iter.Seq[V], step func(ctx T, item V) error) Step[T]

func Each2 

func Each2[T context.Context, K, V any](seq iter.Seq2[K, V], step func(ctx T, key K, val V) error) Step[T]

func Go 

func Go[T context.Context](steps ...Step[T]) Step[T]

func GoN 

func GoN[T context.Context](limit int, steps ...Step[T]) Step[T]

func Once 

func Once[T context.Context](step Step[T]) Step[T]

func Pipe added in v0.3.0 

func Pipe[T context.Context, In, Out any](
	fn func(ctx T, in In) (Out, error),
	get func(ctx T) In,
	set func(ctx T, out Out),
) Step[T]

func PipeSeq added in v0.3.0 

func PipeSeq[T context.Context, In, Out any](
	seq iter.Seq[In],
	transform func(ctx T, item In) (Out, error),
	sink func(ctx T, item Out) error,
) Step[T]

func PipeSeq2 added in v0.3.0 

func PipeSeq2[T context.Context, K, V, Out any](
	seq iter.Seq2[K, V],
	transform func(ctx T, key K, val V) (Out, error),
	sink func(ctx T, item Out) error,
) Step[T]

func Race 

func Race[T context.Context](steps ...Step[T]) Step[T]

func Seq 

func Seq[T context.Context](steps ...Step[T]) Step[T]

func (Step[T]) Branch added in v0.3.0 

func (s Step[T]) Branch(condition func(ctx T) bool, ifBranch, elseBranch Step[T]) Step[T]

func (Step[T]) Catch 

func (s Step[T]) Catch(handler func(ctx T, err error) error) Step[T]

func (Step[T]) Exec 

func (s Step[T]) Exec(ctx T) error

func (Step[T]) Fallback 

func (s Step[T]) Fallback(fallback Step[T]) Step[T]

func (Step[T]) Go 

func (s Step[T]) Go(steps ...Step[T]) Step[T]

func (Step[T]) GoN 

func (s Step[T]) GoN(limit int, steps ...Step[T]) Step[T]

func (Step[T]) Once 

func (s Step[T]) Once() Step[T]

func (Step[T]) Race 

func (s Step[T]) Race(steps ...Step[T]) Step[T]

func (Step[T]) Recover 

func (s Step[T]) Recover() Step[T]

func (Step[T]) Retry 

func (s Step[T]) Retry(attempts int, delay time.Duration) Step[T]

func (Step[T]) Then 

func (s Step[T]) Then(steps ...Step[T]) Step[T]

func (Step[T]) Timeout 

func (s Step[T]) Timeout(d time.Duration) Step[T]

func (Step[T]) Unless 

func (s Step[T]) Unless(predicate func(ctx T) bool) Step[T]

func (Step[T]) When 

func (s Step[T]) When(predicate func(ctx T) bool) Step[T]

func (Step[T]) Wrap 

func (s Step[T]) Wrap(middlewares ...Middleware[T]) Step[T]

Source Files

View all Source files

Directories

Path Synopsis
cmd
gen-diagrams command
examples
branch/with_flow command
branch/without_flow command
dag/with_flow command
dag/without_flow command
each/with_flow command
each/without_flow command
fallback/with_flow command
fallback/without_flow command
go/with_flow command
go/without_flow command
gon/with_flow command
gon/without_flow command
once/with_flow command
once/without_flow command
order_checkout/with_flow command
order_checkout/without_flow command
pipe/with_flow command
pipe/without_flow command
race/with_flow command
race/without_flow command
retry/with_flow command
retry/without_flow command
seq/with_flow command
seq/without_flow command
timeout/with_flow command
timeout/without_flow command
wrap/with_flow command
wrap/without_flow command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.