flowgraph

README

flowgraph

[!CAUTION] This library is fully AI-generated. The implementation, tests, examples and documentation were all written by an AI agent.

It is exercised by the test suite in this repository, but its only real-world validation is against production workloads specific to Heru, Inc. Anything those workloads do not exercise — other concurrency shapes, other scales, other cancellation and failure patterns — is unproven, and the concurrency and lifetime semantics are exactly where that gap is most likely to hurt.

Use it with caution: read the code, and treat Semantics fine print as claims to verify against your own workload rather than as guarantees.

Compose a program as a graph of lazy, memoized, typed computations. A node is one async computation producing one value; nothing runs until something demands it, demand propagates backward from whoever needs the answer, and a value computed once is shared by every later consumer. Dependencies flow as Node[T] handles — no central map[string]any registry, no interface{} on any value path. An orchestrator is just a node whose body wires other nodes, so modules nest arbitrarily deep, across package and even Go-module boundaries. The engine is about 120 lines; the combinators layered on top are another 200.

Taste

root := flowgraph.NewScope(ctx, nil)

raw := flowgraph.Compute(root, func(ctx context.Context) ([]byte, error) {
	return os.ReadFile("orders.json") // runs once, for whoever asks first
})
orders := flowgraph.Then(root, raw, parseOrders)   // []Order
rates := flowgraph.Compute(root, fetchFxRates)     // slow HTTP call

// Zip2 prefetches both sides, so the file read and the HTTP call overlap.
totals := flowgraph.Zip2(root, orders, rates, computeTotals)

// Nothing above has run yet. This line is what pulls the graph:
report, err := totals.Get(ctx)

// A second consumer of `orders` costs zero re-parsing — it is memoized.
biggest, err := flowgraph.Then(root, orders, largestOrder).Get(ctx)

Install

go get github.com/heru-opensource/flowgraph

Requires Go 1.26 or newer. The only dependency is golang.org/x/sync.

Concepts

Lazy and memoized. Compute builds a handle and runs nothing. The body runs on the first Get (demand it now) or Prefetch (start it, I'll check later), at most once per node instance. Concurrent demanders join the one in-flight execution; later ones get the stored result immediately. A node nobody demands never runs, so a module that isn't needed under the current inputs costs nothing.

Lifetime belongs to the owner, not the consumer. A node body runs under its owning Scope's context — never a caller's. A consumer that walks away (cancellation, early failure, a timeout) abandons only its own wait: the node keeps running and memoizes for everyone else. To cancel work, cancel the scope that owns it. That one split is what makes shared providers safe to depend on: your failure cannot poison a value someone else still needs.

Scopes share by type, outermost first. Provide[T](scope, node) registers a shared node; Resolve[T](scope) walks from that scope to the root and the outermost provider wins, so the widest set of consumers shares one instance. That inverts normal lexical shadowing on purpose — the goal is sharing, not shadowing. Override[T] restores shadowing for exactly one subtree, and is invisible to ancestors and siblings. Use a distinct named type per shared dependency (type SamplingRate float64, not bare float64) and the keys cannot collide.

Composition is recursion. A node body may build and wire its own child nodes, so "orchestrator" is not a separate concept — it is a node, and the "controller" is just the root node you call Get on. Expand takes this to its conclusion: a node whose value is []Node[B], letting the graph grow at runtime to match data only knowable at runtime.

Combinators

Function	Shape	Semantics
Compute(s, fn)	Node[Out]	the primitive: a lazy, memoized node owned by s
Const(s, v)	Node[T]	a known value as a node (inputs, defaults, tests)
Then(s, a, f)	Node[B]	1-ary derivation; the building block for pipelines
Zip2 / Zip3	Node[C] / Node[D]	join 2 or 3 nodes; both/all sides prefetched, so they overlap
All(s, ns...)	Node[[]T]	join any number of same-typed nodes, in order
Fallback(s, a, b)	Node[T]	force a; only on error force b. Chains into a config cascade
Race(s, ns...)	Node[T]	first success wins; losers are never cancelled. All fail → joined error
Map(s, list, n, f)	Node[[]B]	bounded parallel fan-out, ordered output, fail-fast within the batch
MapCollect(...)	Node[[]Result[B]]	same fan-out, per-item outcomes; one bad item is data, not a batch failure
Expand(s, list, f)	Node[[]Node[B]]	items promoted to nodes: per-item laziness and cross-consumer sharing
Gather(ctx, ns)	([]B, error)	prefetch all, then collect in order — the terminal step after Expand
GetWithTimeout(...)	(T, error)	bound how long this caller waits; the node stays alive
Provide / Override	—	register a shared provider / cap upward resolution for a subtree
Resolve / MustResolve	Node[T]	upward lookup; MustResolve panics, which is what you want at wiring time
Prefetchable	interface{ Prefetch() }	the one type-erased seam: start a set of differently-typed nodes in one loop, then collect each through its own typed handle. Exposes no value

Everything is a free function taking *Scope because Go forbids generic methods — Compute, Provide and Resolve each need their own type parameters. That is a language constraint, not a style choice. Node[T].Get can be a method because Node[T] is a generic type.

Three decision rules

Goroutines vs nodes. Structured concurrency inside a node body is ordinary Go and encouraged when the parallelism is a private implementation detail (that is exactly what Map is). Promote work to its own node only when the result needs node properties: independently demandable, per-item lazy, or memoized across consumers.

Map vs Expand. A batch consumed once by one consumer → Map. Different consumers picking different items, items that may go undemanded, or item results feeding several downstream nodes → Expand.

Provide vs an explicit argument. Pass a Node[T] as a constructor argument when there is exactly one obvious consumer wiring — that is compile-time checked and easiest to follow. Provide into a scope when many consumers at unknown depths must share one instance; that is wire-time checked and outermost-wins.

Examples

Runnable programs, each narrating its own output. They live in their own Go module so protobuf never becomes a dependency of the library:

cd examples && go run ./pipeline    # linear ETL: Then + Zip2, overlap and memoization
cd examples && go run ./router      # heterogeneous outputs behind one interface; scope resolution
cd examples && go run ./fanout      # Map vs MapCollect vs Expand over a runtime list
cd examples && go run ./resilience  # layered config, Race, deadlines, scope teardown
cd examples/multimodule && go run ./app   # integration across Go-module boundaries

• examples/pipeline — the smallest realistic program: derivation, joining, shared upstreams running once, and an undemanded node that never runs.
• examples/router — a router that inspects its input's discriminant, forces exactly one branch, and returns concrete values through a common behavioral interface (proto.Message), with the discriminant moved onto the wire as anypb.Any. Also documents when to use a sealed union + type switch instead.
• examples/fanout — the three fan-out shapes side by side, plus many consumers over one shared mid-value.
• examples/resilience — the cancellation model made visceral: a Race loser finishing on its own, a timed-out caller picking up the memoized value on a second try, and a scope teardown that does cancel.
• examples/multimodule — a standalone module behind a typed seam, New vs NewInScope, and a contract the child module owns but the integrator implements.

Semantics fine print

1. Laziness. A node whose Get/Prefetch is never called never executes its body.
2. Single execution. A node's body runs at most once per node instance, regardless of concurrent or repeated demand. Later Gets return the stored result immediately — including to a caller whose context is already cancelled, since there is no longer any work to abandon.
3. Owner-context execution. The body runs under its owning scope's context, never a caller's. This is the whole cancellation model in one line.
4. Caller cancellation abandons only the wait. A cancelled Get returns the caller's ctx.Err(); the node keeps running and memoizes for everyone else.
5. Scope teardown cancels owned nodes. Cancel a scope's context and in-flight bodies it owns observe ctx.Done(). Lifetime cascades by ownership, never by consumption.
6. Outermost-preference resolution. Resolve[T] walks from the querying scope to the root; the highest plain Provide wins.
7. Override caps the walk, downward only. An Override at scope S makes S and its descendants resolve to it; ancestors and sibling subtrees are unaffected.
8. Type is the key. The provider map holds any internally, but Provide[T] and Resolve[T] are its only accessors, so the assertion cannot fail and users never see any. Providing the same type twice in one scope replaces the earlier registration.
9. Wiring-time failure. MustResolve panics for an unprovided type. Resolve during graph construction so a missing provider fails before anything runs.
10. Combinator promises. Race never cancels losers and joins all errors if every candidate fails. Fallback never runs the secondary when the primary succeeds, and reports both errors when neither works. Map is fail-fast within its batch and isolated outside it. MapCollect never fails the batch for an item error. Expand mints handles without forcing them. GetWithTimeout leaves the node alive.
11. Concurrency. Node and Scope are safe for concurrent use.
12. Panics are not recovered. A node body runs on its own goroutine, so a panic in one crashes the program as any unrecovered goroutine panic does — and the demanding goroutine's stack will not be in the trace. Return errors from node bodies instead.

The test suite pins every one of these; flow_test.go is organised so each test names the rule it protects.

FAQ

Why free functions instead of methods? Go forbids generic methods. Compute, Provide and Resolve each need their own type parameters, so they must be package-level. Not a style choice.

Why does the provider map hold any — isn't that the thing you banned? The ban is on any in user-facing value paths. The map is private, Provide[T] and Resolve[T] are its only accessors, and since Provide[T] is the sole writer of key T, the assertion in Resolve[T] cannot fail. Users only ever write and read Node[T].

Why outermost-wins? Every scoping system I know shadows inner-over-outer. Because the goal is sharing, not lexical shadowing: a higher provider is visible to more consumers, so preferring it maximises reuse of one instance. Override restores shadowing exactly where a subtree genuinely needs its own binding — the inversion is the default, not a prison.

Why not cancel a node when its last waiter leaves? Isolation is deliberate: a computed value may serve future consumers, and one consumer's failure should not poison shared state. Lifetime belongs to the owning scope. If abandoned work becomes a real cost, tear down a child scope when its orchestrator finishes — never refcount individual consumers.

Why is a shared cross-module contract an interface owned by the child module? Import cycles. Resolve[T] must name T at compile time inside the resolving package, so if the concrete type lived in the integrator (which imports the child), the child could not import it back. The child owns the interface and ships a default; the integrator provides the implementation. All edges keep pointing downward. See examples/multimodule.

What about cycles in the graph? Under pure pull with one output per node, a dependency cycle deadlocks: A's Get waits on B's Get, which waits on A's memoization that will never complete. This release does not detect that — see the roadmap. Until then, resolve dependencies at wiring time and keep constructors acyclic; a cycle you can build with MustResolve at construction time is a cycle you can see.

How do I publish a value from the middle of a module? Name the midpoint as its own node. The producer's first half becomes the shared node; its second half forces that and continues to the producer's own final output. No mutable cells, no publish/subscribe. examples/fanout part 1 does exactly this.

Roadmap

Deliberately out of scope for v0.1, and documented rather than half-built:

• Cycle detection — thread the demand path through Get so forcing a node already on the current demand stack errors instead of deadlocking. Complicated by owner-context execution (the body does not receive the caller's chain), so it likely needs an explicit demand token — worth designing carefully before it is built.
• Streaming / multi-shot nodes — Node[T] is deliberately produce-once. A Stream[T] sibling is a separate abstraction, not an overload of Node.
• Observability hooks — optional per-scope OnStart/OnDone callbacks and a ComputeNamed variant, zero-cost when unset.
• Graph export — record construction edges (an upper bound; dynamic demand makes exact edges unknowable statically) and emit DOT.
• Refcounted child scopes for reclaiming abandoned work.
• Panic containment — optionally recover a panicking node body into an error, or re-panic it in the demander with the demand path attached. Today a panic in a body behaves like any other unrecovered goroutine panic.

Releasing

A Go module is published by pushing a semver tag — there is no registry to upload to. The module proxy fetches the tag from this repository on demand, and pkg.go.dev indexes it from there.

# 1. land the release notes FIRST — a `## [0.1.0]` heading in CHANGELOG.md is a
#    hard requirement, and the tag cannot be reused if you forget it. Push that
#    to main.
# 2. then tag the commit CI is green on:
git tag v0.1.0
git push origin v0.1.0

Pushing the tag triggers release.yml, which re-runs every CI gate against the tagged commit, checks the tag is one Go can actually consume and that the version is documented, cuts a GitHub Release with generated notes, and asks proxy.golang.org for the version so pkg.go.dev indexes it promptly instead of waiting for someone's first go get.

Tags are effectively immutable. Once the proxy has fetched a version it caches it permanently; moving or deleting the tag does not un-publish anything, and consumers may already have the old bytes in their go.sum. A bad release is fixed by cutting the next patch version, never by retagging. That is why the workflow's pre-flight checks fail loudly rather than trying to paper over anything:

Check	Why it is fatal
strict semver (vMAJOR.MINOR.PATCH[-pre])	Go silently ignores tags it cannot parse, so a typo is a release that never appears
major version agrees with the module path	v2.0.0 needs the module path to end in /v2, or nobody can import it
no replace directives in go.mod	replace does not apply to consumers, so the module would not resolve for them
## [VERSION] heading in CHANGELOG.md	every released version must be documented; an undocumented release is not a release

Because the changelog check is a hard gate, write the entry before you tag. If a tag does fail this check, no GitHub Release is written — add the entry on the default branch and cut the next patch version.

Two things deliberately stay unpublished: examples/ and examples/multimodule/* are separate modules carrying replace directives that point at this working tree. They exist to be read and run from a clone, and the v* tag filter never matches them.

License

MIT — see LICENSE.

Documentation

Overview

Package flowgraph is a lazy, demand-driven, type-enforced module graph.

Caution: this library is fully AI-generated

The implementation, tests, examples and documentation were all written by an AI agent.

The test suite in the repository exercises every rule described below, but the only real-world validation is against production workloads specific to Heru, Inc. Anything those workloads do not exercise — other concurrency shapes, other scales, other cancellation and failure patterns — is unproven, and the concurrency and lifetime semantics are exactly where that gap is most likely to hurt. Use it with caution: read the code, and treat the rules below as claims to verify against your own workload rather than as guarantees.

Overview

The whole system is one idea: a module is a lazy, memoized, shareable async computation that produces exactly one typed value. Nothing runs until its output is demanded; demand propagates backward from whoever ultimately needs the answer; and a value computed once is shared by every later consumer.

Composition is recursion: a node's body may construct and wire its own child nodes, so an "orchestrator" is just a node, and the "controller" is just the root node you call Get on. There is no central registry and no interface{} on any value path — dependencies flow as typed Node handles. The single type-erased seam, Prefetchable, deliberately exposes no value.

The four rules

Laziness. A node whose Get or Prefetch is never called never runs. A module that is not needed under the current inputs costs nothing.

Memoization. A node's body runs at most once per node instance. Concurrent demanders join the one in-flight execution; later demanders get the stored result immediately.

Owner-context execution. A node body runs under its owning Scope's context — never a caller's. So a consumer that walks away (cancellation, early failure) abandons only its own wait: the node keeps running and memoizes for everyone else. Cancel a scope to cancel the nodes that scope owns. Lifetime cascades by ownership, never by consumption.

Hierarchical sharing with outermost preference. Shared dependencies are registered into a scope chain by their type (Provide) and looked up by walking upward (Resolve). The outermost provider wins, so the widest set of consumers shares one instance. Override caps that walk for one subtree when it genuinely needs its own binding.

Shape of the API

Go forbids generic methods, so Compute, Provide, Resolve and every combinator are free functions taking a *Scope. That is forced by the language, not a style choice; Node[T].Get is a method because Node[T] is a generic type rather than a generic method.

Build nodes with Compute, derive them with Then, Zip2, Zip3 and All, fan out with Map, MapCollect and Expand, and recover with Fallback and Race. The core is deliberately tiny; the combinators exist so applications never need to touch its internals.

Index

• func Gather[B any](ctx context.Context, nodes []Node[B]) ([]B, error)
• func GetWithTimeout[T any](ctx context.Context, n Node[T], d time.Duration) (T, error)
• func Override[T any](s *Scope, n Node[T])
• func Provide[T any](s *Scope, n Node[T])
• type Node
  • func All[T any](s *Scope, nodes ...Node[T]) Node[[]T]
  • func Compute[Out any](s *Scope, run func(ctx context.Context) (Out, error)) Node[Out]
  • func Const[T any](s *Scope, v T) Node[T]
  • func Expand[A, B any](s *Scope, list Node[[]A], f func(ctx context.Context, item A) (B, error)) Node[[]Node[B]]
  • func Fallback[T any](s *Scope, primary, secondary Node[T]) Node[T]
  • func Map[A, B any](s *Scope, list Node[[]A], maxParallel int, ...) Node[[]B]
  • func MapCollect[A, B any](s *Scope, list Node[[]A], maxParallel int, ...) Node[[]Result[B]]
  • func MustResolve[T any](s *Scope) Node[T]
  • func Race[T any](s *Scope, candidates ...Node[T]) Node[T]
  • func Resolve[T any](s *Scope) (Node[T], bool)
  • func Then[A, B any](s *Scope, a Node[A], f func(ctx context.Context, a A) (B, error)) Node[B]
  • func Zip2[A, B, C any](s *Scope, a Node[A], b Node[B], ...) Node[C]
  • func Zip3[A, B, C, D any](s *Scope, a Node[A], b Node[B], c Node[C], ...) Node[D]
• type Prefetchable
• type Result
• type Scope
  • func NewScope(ctx context.Context, parent *Scope) *Scope
  • func (s *Scope) Child(ctx context.Context) *Scope
  • func (s *Scope) Context() context.Context

Examples

• Compute
• Expand
• Fallback
• Map
• MapCollect
• Provide
• Then
• Zip2

Functions

func Gather

func Gather[B any](ctx context.Context, nodes []Node[B]) ([]B, error)

Gather forces every node in a slice concurrently (Prefetch all, then Get each) and collects the results in order. The natural terminal step after Expand when a consumer does want all items.

It returns the first error in slice order, having already started every node; the nodes it stops waiting on keep running under their owners' contexts and stay available to later consumers.

func GetWithTimeout

func GetWithTimeout[T any](ctx context.Context, n Node[T], d time.Duration) (T, error)

GetWithTimeout bounds how long THIS CALLER waits. On timeout the node keeps running (owner lifetime) and its result stays available to later callers — a later Get can pick up the memoized value.

func Override

func Override[T any](s *Scope, n Node[T])

Override registers n as a provider of T that CAPS upward resolution. This scope and everything below it resolve to n regardless of any outer Provide. Use it when a subtree legitimately needs a different T than its ancestors.

An Override is downward-only: ancestors and sibling subtrees never see it.

func Provide

func Provide[T any](s *Scope, n Node[T])

Provide registers n as a shared provider of T. It participates in outermost-preference: when several scopes Provide T, Resolve picks the one closest to the root, so the widest set of consumers shares one instance.

Providing T twice in the same scope replaces the earlier registration.

Example

ExampleProvide shares one node with consumers at unknown depths. Resolution walks upward and the outermost provider wins, so the widest set of consumers shares a single instance; Override caps that walk for one subtree.

package main

import (
	"context"
	"fmt"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()

	// A distinct named type is the sharing key — never a bare builtin.
	type Region string

	root := flowgraph.NewScope(ctx, nil)
	service := root.Child(ctx)
	tenant := service.Child(ctx)

	flowgraph.Provide(root, flowgraph.Const(root, Region("us-east-1")))

	// A plain Provide deeper down does not shadow the outer one: sharing beats
	// lexical scoping by default.
	flowgraph.Provide(tenant, flowgraph.Const(tenant, Region("eu-west-1")))
	got, _ := flowgraph.MustResolve[Region](tenant).Get(ctx)
	fmt.Println("tenant resolves:", got)

	// An Override does shadow it, for this scope and its descendants only.
	flowgraph.Override(tenant, flowgraph.Const(tenant, Region("eu-west-1")))
	got, _ = flowgraph.MustResolve[Region](tenant).Get(ctx)
	fmt.Println("after override:  ", got)

	sibling := service.Child(ctx)
	got, _ = flowgraph.MustResolve[Region](sibling).Get(ctx)
	fmt.Println("sibling still:   ", got)

}

Output:
tenant resolves: us-east-1
after override:   eu-west-1
sibling still:    us-east-1

Types

type Node

type Node[Out any] interface {
	// Get forces the node (if not already started) and blocks until the result
	// is ready or the CALLER's context is cancelled. Crucially, the node body
	// runs under its OWNER's context, not the caller's — so a consumer that
	// walks away (early failure, lost interest) never cancels the node. The
	// node keeps running and memoizes its result for everyone else. That single
	// split is what delivers isolation + cancel-immune shared providers.
	//
	// A cancelled caller gets ctx.Err(). The one exception is a node that has
	// already finished: its stored result is returned even to a cancelled
	// caller, because there is no longer any work to abandon.
	Get(ctx context.Context) (Out, error)

	// Prefetch starts the node without blocking. This is request-mode (a):
	// "I'll need this, start working on it, I'll check later." Get is mode (b):
	// "I need this now." Both are idempotent; the body runs at most once.
	Prefetch()
}

Node is a lazy, memoized, shareable computation producing one Out.

A Node is a handle, not a value: holding one costs nothing and starts nothing. The body runs at most once per Node instance, however many consumers demand it and however often.

func All

func All[T any](s *Scope, nodes ...Node[T]) Node[[]T]

All joins any number of same-typed nodes: prefetch all, gather in order. See Gather for the error semantics.

func Compute

func Compute[Out any](s *Scope, run func(ctx context.Context) (Out, error)) Node[Out]

Compute builds a node owned by scope s. The node's body runs under s's context, so the node lives exactly as long as its owning scope — never as long as a consumer. Tear down a scope to cancel everything it owns.

Building a node is pure bookkeeping: run is not called until something demands the node. A node body may itself call Compute to build child nodes, which is how the graph grows at runtime (see Expand).

A panic inside run is not recovered. Because the body runs on its own goroutine, a panic crashes the program exactly as any unrecovered goroutine panic does, and the demanding goroutine's stack will not appear in the trace. Return errors from node bodies rather than panicking.

Example

ExampleCompute shows the two properties every node has: it does not run until something demands it, and it runs at most once however many consumers ask.

package main

import (
	"context"
	"fmt"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	runs := 0
	price := flowgraph.Compute(root, func(ctx context.Context) (int, error) {
		runs++
		fmt.Println("expensive lookup running")
		return 21, nil
	})
	fmt.Println("graph built, runs so far:", runs)

	a, _ := price.Get(ctx)
	b, _ := price.Get(ctx) // second consumer: memoized, no second run
	fmt.Printf("values: %d %d after %d run\n", a, b, runs)

}

Output:
graph built, runs so far: 0
expensive lookup running
values: 21 21 after 1 run

func Const

func Const[T any](s *Scope, v T) Node[T]

Const wraps an already-known value as a node, so plain values can be handed to anything expecting a Node[T] (inputs, defaults, tests).

func Expand

func Expand[A, B any](
	s *Scope,
	list Node[[]A],
	f func(ctx context.Context, item A) (B, error),
) Node[[]Node[B]]

Expand is the NODE-SHAPED fan-out: dynamic graph expansion. It forces the list node and mints one lazy child Node[B] per item, returning the slice of handles WITHOUT forcing any of them.

Use this when per-item results need node properties: several downstream consumers demand individual items independently (each item computes once, memoized, shared); or some items may never be demanded at all (per-item laziness). The item nodes are owned by scope s, so they inherit its lifetime and are immune to any single consumer's cancellation — same rules as every other node.

Note the type is Node[[]Node[B]]: a node whose VALUE is nodes. That is legal here precisely because composition is recursion; the graph grows at runtime to match data only knowable at runtime. Gather is the usual terminal step when a consumer does end up wanting every item.

Example

ExampleExpand promotes list items to nodes. Minting the handles runs nothing; each item then computes at most once, no matter how many consumers want it, and items nobody demands never run at all.

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	runs := map[string]int{}
	pages := flowgraph.Const(root, []string{"cover", "body", "appendix"})
	rendered := flowgraph.Expand(root, pages, func(ctx context.Context, page string) (string, error) {
		runs[page]++
		return strings.ToUpper(page), nil
	})

	handles, _ := rendered.Get(ctx)
	fmt.Println("handles minted:", len(handles), "renders so far:", len(runs))

	// Two consumers both want the cover; nobody wants the appendix.
	first, _ := handles[0].Get(ctx)
	again, _ := handles[0].Get(ctx)
	body, _ := handles[1].Get(ctx)
	fmt.Println(first, again, body)
	fmt.Println("cover renders:", runs["cover"], "appendix renders:", runs["appendix"])

}

Output:
handles minted: 3 renders so far: 0
COVER COVER BODY
cover renders: 1 appendix renders: 0

func Fallback

func Fallback[T any](s *Scope, primary, secondary Node[T]) Node[T]

Fallback forces primary; on error it forces secondary instead. The primary's error is wrapped into the secondary's error if BOTH fail. Laziness holds: secondary never runs if primary succeeds — which is what makes a chain of Fallbacks the natural spelling of a config cascade.

Example

ExampleFallback is the layered-config cascade: laziness means each tier is only consulted if the tier above it failed.

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	fromEnv := flowgraph.Compute(root, func(ctx context.Context) (string, error) {
		fmt.Println("checking env")
		return "", errors.New("FLOWGRAPH_DSN unset")
	})
	fromFile := flowgraph.Compute(root, func(ctx context.Context) (string, error) {
		fmt.Println("reading config file")
		return "postgres://from-file", nil
	})
	defaults := flowgraph.Compute(root, func(ctx context.Context) (string, error) {
		fmt.Println("falling back to defaults") // never printed: the file tier won
		return "postgres://localhost", nil
	})

	dsn := flowgraph.Fallback(root, fromEnv, flowgraph.Fallback(root, fromFile, defaults))
	v, _ := dsn.Get(ctx)
	fmt.Println("dsn:", v)

}

Output:
checking env
reading config file
dsn: postgres://from-file

func Map

func Map[A, B any](
	s *Scope,
	list Node[[]A],
	maxParallel int,
	f func(ctx context.Context, item A) (B, error),
) Node[[]B]

Map is the SIMPLE fan-out: a node that forces the list node, then transforms every item in parallel (bounded by maxParallel; <=0 means unbounded) and memoizes the whole []B as one value. Output order matches input order.

Use this when the per-item work is a private implementation detail of the consumer: no other node will ever demand a single item's result, so the items don't need to be nodes. Structured concurrency inside a node body is ordinary Go — not a violation of the architecture.

Error semantics are fail-fast WITHIN the batch (the first error cancels the sibling items via the group context, and Map returns that error) but isolated OUTSIDE it: the group context derives from this node's own owner context, so a batch failure never propagates to other nodes, and a consumer walking away never cancels the batch.

Example

ExampleMap fans out over a list whose length is only known at runtime, with a concurrency cap. Output order always matches input order.

package main

import (
	"context"
	"fmt"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	docs := flowgraph.Const(root, []string{"alpha", "bravo", "charlie", "delta"})
	lengths := flowgraph.Map(root, docs, 2, func(ctx context.Context, d string) (int, error) {
		return len(d), nil
	})

	out, err := lengths.Get(ctx)
	fmt.Println(out, err)

}

Output:
[5 5 7 5] <nil>

func MapCollect

func MapCollect[A, B any](
	s *Scope,
	list Node[[]A],
	maxParallel int,
	f func(ctx context.Context, item A) (B, error),
) Node[[]Result[B]]

MapCollect is Map's collect-all-errors sibling: every item is processed (no intra-batch fail-fast), and each item's success or failure is reported in order. Use when partial progress matters more than stopping early.

maxParallel bounds concurrency exactly as in Map (<=0 means unbounded). The node itself fails only if the list node fails — an item error is data, carried in that item's Result.

Example

ExampleMapCollect keeps going after a bad item and reports each outcome, where Map would fail the whole batch.

package main

import (
	"context"
	"fmt"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	ids := flowgraph.Const(root, []int{1, -2, 3})
	results := flowgraph.MapCollect(root, ids, 3, func(ctx context.Context, id int) (string, error) {
		if id < 0 {
			return "", fmt.Errorf("id %d: negative", id)
		}
		return fmt.Sprintf("record-%d", id), nil
	})

	out, err := results.Get(ctx) // the batch itself succeeds
	fmt.Println("batch error:", err)
	for i, r := range out {
		if r.Err != nil {
			fmt.Printf("item %d failed: %v\n", i, r.Err)
			continue
		}
		fmt.Printf("item %d ok: %s\n", i, r.Val)
	}

}

Output:
batch error: <nil>
item 0 ok: record-1
item 1 failed: id -2: negative
item 2 ok: record-3

func MustResolve

func MustResolve[T any](s *Scope) Node[T]

MustResolve is the wiring-phase form: it panics when T has no provider. Do all dependency resolution at graph construction, before any Get, so an unsatisfied dependency fails at build time rather than deep inside a run. This is the wire-time safety floor that stands in for the compile-time checking we trade away for automatic resolution.

func Race

func Race[T any](s *Scope, candidates ...Node[T]) Node[T]

Race forces every candidate concurrently and resolves to the FIRST success. Consistent with the isolation rule, losers are NOT cancelled — they keep running under their owners' contexts and memoize for any other consumer. Race only stops WAITING on them. If all candidates fail, the errors are joined (see errors.Join).

func Resolve

func Resolve[T any](s *Scope) (Node[T], bool)

Resolve implements outermost-with-local-override. Walking from s up to the root: each provider found overwrites the choice, so the highest (outermost) normal provider wins by default. An Override breaks the walk at its scope, so the most-local override caps the search.

The returned node is not forced. Resolve reports whether a provider exists; use MustResolve when a missing provider is a wiring bug.

func Then

func Then[A, B any](s *Scope, a Node[A], f func(ctx context.Context, a A) (B, error)) Node[B]

Then derives a new node from one upstream node: force a, transform. The building block for linear pipelines.

Example

ExampleThen builds a linear pipeline. Each stage forces the one before it, so demanding the last stage pulls the whole chain — and nothing more.

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	raw := flowgraph.Const(root, "  Widget, Gadget , Doohickey ")
	fields := flowgraph.Then(root, raw, func(ctx context.Context, s string) ([]string, error) {
		out := strings.Split(s, ",")
		for i := range out {
			out[i] = strings.TrimSpace(out[i])
		}
		return out, nil
	})
	count := flowgraph.Then(root, fields, func(ctx context.Context, f []string) (int, error) {
		return len(f), nil
	})

	names, _ := fields.Get(ctx)
	n, _ := count.Get(ctx) // reuses the memoized fields value
	fmt.Println(names, n)

}

Output:
[Widget Gadget Doohickey] 3

func Zip2

func Zip2[A, B, C any](s *Scope, a Node[A], b Node[B], f func(ctx context.Context, a A, b B) (C, error)) Node[C]

Zip2 joins two nodes: both are prefetched (so they run concurrently), then combined. If a side fails, its error propagates and f is not called; the other side still completes under its own owner's context and memoizes.

Example

ExampleZip2 joins two independent nodes. Both are prefetched, so they run concurrently rather than one after the other.

package main

import (
	"context"
	"fmt"

	"github.com/heru-opensource/flowgraph"
)

func main() {
	ctx := context.Background()
	root := flowgraph.NewScope(ctx, nil)

	subtotal := flowgraph.Const(root, 4000)
	taxRate := flowgraph.Const(root, 0.07)
	total := flowgraph.Zip2(root, subtotal, taxRate,
		func(ctx context.Context, cents int, rate float64) (int, error) {
			return cents + int(float64(cents)*rate), nil
		})

	v, _ := total.Get(ctx)
	fmt.Println("total cents:", v)

}

Output:
total cents: 4280

func Zip3

func Zip3[A, B, C, D any](s *Scope, a Node[A], b Node[B], c Node[C], f func(ctx context.Context, a A, b B, c C) (D, error)) Node[D]

Zip3 joins three nodes; same semantics as Zip2.

type Prefetchable

type Prefetchable interface{ Prefetch() }

Prefetchable is the non-generic view of a node, used to kick off a set of differently-typed dependencies (e.g. an orchestrator holding []Prefetchable) without ever surfacing their values as `any`. It is the only type-erased seam, and it deliberately exposes no value.

Every Node[T] satisfies it, so a slice of unrelated node types can be started in one loop and then collected individually through their typed handles.

type Result

type Result[T any] struct {
	Val T
	Err error
}

Result carries a per-item outcome for collect-all-errors fan-out.

type Scope

type Scope struct {
	// contains filtered or unexported fields
}

Scope is one orchestrator's level in the dependency-injection chain. A scope sees its own providers and its ancestors' — never its descendants' or a sibling subtree's privates. Visibility runs strictly downward, which is what "information flows downward" means concretely.

A scope also carries the context that owns every node built under it. This is the lifetime boundary: cancelling a scope cancels its nodes; a single consumer's cancellation never reaches them.

A Scope is safe for concurrent use: providers may be registered and resolved from different goroutines. The recommended discipline is still to do all wiring during graph construction, before anything is demanded, so a missing provider fails before any work starts (see MustResolve).

func NewScope

func NewScope(ctx context.Context, parent *Scope) *Scope

NewScope creates a root scope whose context owns every node built under it. Pass nil as parent for a self-rooted scope (nothing resolves above it); pass a parent to chain onto an existing scope, which is what Scope.Child does.

Cancelling ctx is how a graph is torn down: every node owned by this scope (and by its descendants, if they derive their contexts from ctx) observes the cancellation in its body.

func (*Scope) Child

func (s *Scope) Child(ctx context.Context) *Scope

Child opens a sub-scope. Pass a derived context to make the child cancellable independently of its parent (that is how an orchestrator opts into fail-fast over the nodes IT owns, without touching shared siblings). Pass s.Context() to give the child exactly its parent's lifetime.

func (*Scope) Context

func (s *Scope) Context() context.Context

Context returns the scope's context: the one context every node owned by this scope runs under, regardless of which context a consumer passes to Get.