Documentation
¶
Overview ¶
Package dag provides a small action-oriented directed acyclic graph.
Nodes are Nexss actions. Compile validates dependencies and produces deterministic execution layers. Execute runs each layer with cancellation and merges results only at layer boundaries.
Index ¶
- Variables
- func GetNodeOutput[T any](state ReadState, nodeID string) (T, error)
- func InnerStateKey(nodeID string) string
- func OutputKey(nodeID string) string
- func Suspend(reason string, payload any) error
- func WithLayerCallback(ctx context.Context, fn LayerCallback) context.Context
- type Builder
- func (b *Builder) AddConditionalEdge(from string, router RouterFunc) *Builder
- func (b *Builder) AddEdge(from, to string) *Builder
- func (b *Builder) AddNamedConditionalEdge(from, label string, router RouterFunc) *Builder
- func (b *Builder) AddNode(id, _ string, act action.AnyAction) *Builder
- func (b *Builder) Compile() (*DAG, error)
- func (b *Builder) WithEntry(nodeID string) *Builder
- func (b *Builder) WithMaxSteps(n int) *Builder
- type DAG
- type Edge
- type EdgeMeta
- type ExecutionError
- type LayerCallback
- type Node
- type NodeContext
- type ReadState
- type RouterFunc
- type State
- type SuspendError
- type TopologyRegistry
- type Value
Constants ¶
This section is empty.
Variables ¶
var ErrSuspended = errors.New("graph: execution suspended for human intervention")
ErrSuspended signals that a node paused execution (e.g. awaiting human approval). Returning it preserves partial state for later resume.
var GlobalTopology = &TopologyRegistry{}
Functions ¶
func GetNodeOutput ¶
GetNodeOutput reads a node's typed output from a read-only state view.
Hot path: single type assertion, zero allocations. Slow path (checkpoint resume): falls back to action.Coerce for JSON-decoded map values.
func InnerStateKey ¶ added in v0.14.0
InnerStateKey returns the state key that holds a nested DAG's partially-completed inner state. On nested failure the parent DAG stashes the inner *State under this key so a resume run can continue the inner graph from where it stopped.
func Suspend ¶ added in v0.11.1
Suspend pauses DAG execution gracefully. The returned error matches ErrSuspended.
func WithLayerCallback ¶ added in v0.11.0
func WithLayerCallback(ctx context.Context, fn LayerCallback) context.Context
WithLayerCallback attaches a callback to ctx.
Use it to checkpoint the DAG after every layer. A caller that keeps the last successful state, checks that the flow definition has not changed, and re-executes with that state gets resume-from-failure for free: DAG.Execute already skips any node whose output key is present in the initial state, so only the nodes that did not run are re-run.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder accumulates nodes and edges, then compiles them into an immutable DAG. The first builder error short-circuits further calls.
func (*Builder) AddConditionalEdge ¶ added in v0.27.0
func (b *Builder) AddConditionalEdge(from string, router RouterFunc) *Builder
AddConditionalEdge makes the graph dynamic: execution walks node by node from the entry point, following RouterFunc decisions instead of running topological layers.
Only one router per source node is allowed. The label defaults to the source node ID; see AddNamedConditionalEdge for a distinct one.
func (*Builder) AddEdge ¶
AddEdge declares that `to` depends on `from`. Duplicate edges are rejected at build time.
func (*Builder) AddNamedConditionalEdge ¶ added in v0.27.0
func (b *Builder) AddNamedConditionalEdge(from, label string, router RouterFunc) *Builder
AddNamedConditionalEdge is AddConditionalEdge with an explicit label that appears in diagnostics when the router points at an unknown node.
func (*Builder) AddNode ¶
AddNode registers a node. id must be unique and non-empty; act must implement action.Executable. The second argument (a display name) is accepted for API compatibility and ignored — the node's output key is always derived from id.
func (*Builder) Compile ¶
Compile validates the graph and produces the immutable DAG. Errors accumulated by AddNode / AddEdge / AddConditionalEdge surface here.
func (*Builder) WithEntry ¶ added in v0.27.0
WithEntry designates the node a dynamic DAG starts from. Only meaningful when the DAG has at least one conditional edge; ignored otherwise.
The default entry is the first node added via AddNode. Set this explicitly when the natural start is not the first AddNode call — e.g. when a common setup node is registered before the actual branch point.
func (*Builder) WithMaxSteps ¶ added in v0.27.0
WithMaxSteps bounds the number of dynamic-execution steps when the DAG contains conditional edges. Without a bound, a router that returns a cycle would loop forever. The default (32) is applied at execute time when the value is <= 0.
Ignored for DAGs without conditional edges — layer execution is always bounded by the layer count.
type DAG ¶
type DAG struct {
// contains filtered or unexported fields
}
DAG is an immutable, compiled directed acyclic graph.
A DAG is either static (only AddEdge calls, executed layer by layer) or dynamic (at least one AddConditionalEdge, executed node by node starting from entryID and following RouterFunc decisions). The two modes are mutually exclusive; see Execute.
func (*DAG) AsAction ¶
AsAction wraps the DAG as a standard Nexss action. The wrapped action accepts a ReadState and returns a freshly owned *State that the caller must Release.
func (*DAG) Execute ¶
Execute runs the DAG.
Two execution modes, selected by the compiled graph's shape:
Layer mode (default). Nodes are grouped into topological layers by the compiler. Each layer runs concurrently with errgroup; layers run sequentially. Resume-from-failure skips any node whose output key is already present in initialState.
Dynamic mode (when the DAG has at least one conditional edge). The engine walks from entryID, one node per step, following RouterFunc decisions until a router returns "" or "end", or until maxSteps is exceeded. Layers are ignored; resume state is honored the same way (nodes whose output key exists are not re-executed, so a resumed dynamic run continues from where it stopped).
Layer callback: attach with WithLayerCallback. In layer mode it fires once per completed layer. In dynamic mode it is not fired — there are no layers to checkpoint. Callers that need checkpointing for dynamic DAGs must use a Journal (see action.Durable) instead. The callback is consumed by the outermost DAG that sees it, so nested DAGs never fire it twice.
On any error the returned *State is non-nil and owns every node output produced so far. The caller must Release it.
type ExecutionError ¶ added in v0.11.0
type ExecutionError struct {
Layer int
FailedNode string
Cause error
State *State
Completed []string
}
ExecutionError reports that a layer stopped before every node in it succeeded. State carries the partial result: every node whose output is present already ran, so a caller that persists State and later re-executes with it will re-run only the missing nodes.
The caller owns State and must call Release on it, exactly as with a successful return.
func (*ExecutionError) Error ¶ added in v0.11.0
func (e *ExecutionError) Error() string
func (*ExecutionError) Unwrap ¶ added in v0.11.0
func (e *ExecutionError) Unwrap() error
type LayerCallback ¶ added in v0.11.0
LayerCallback is invoked by DAG.Execute after each layer completes.
The state passed to the callback already includes every node output from the completed layer. A callback error aborts execution and is returned to the caller; the state at the point of failure is preserved, not released, so the caller can inspect or persist it.
Nested DAGs do not fire the callback. The outermost DAG that finds a callback in its context consumes it, and every nested DAG in the same execution tree sees an empty slot. This makes the callback a checkpoint hook for one top-level execution, not for every graph in the tree.
func LayerCallbackFromCtx ¶ added in v0.11.0
func LayerCallbackFromCtx(ctx context.Context) LayerCallback
LayerCallbackFromCtx returns the callback attached by WithLayerCallback. It returns nil when no callback is set, and also when the callback has already been consumed by an outer DAG.Execute.
type Node ¶
type Node struct {
ID string
OutputKey string
Action action.Executable
}
Node is a single unit of DAG work.
type NodeContext ¶
NodeContext is the execution context handed to every node.
Input is read-only by construction: the compiler enforces that nodes cannot mutate the parent layer's state. There is no Set method on ReadState, and the engine never exposes a *State through this struct.
A node may retain Input past its own return (e.g. spawn a goroutine that reads it later) — the underlying map is guaranteed alive as long as any ReadState view of it exists.
type ReadState ¶ added in v0.14.0
type ReadState struct {
// contains filtered or unexported fields
}
ReadState is the read-only view of a State handed to nodes. It shares the backing map with the origin State; copying a ReadState is free (one map-header copy) and keeps the map alive.
func (ReadState) Clone ¶ added in v0.14.0
Clone returns a fresh *State with a shallow copy of the read view's backing map. Execute uses it to derive the layer's mutable state from the caller's initial snapshot.
type RouterFunc ¶ added in v0.27.0
RouterFunc decides the next node to visit after its owning edge's source node. Returning "" or "end" terminates the dynamic walk. An error aborts the walk and is returned to the caller unchanged.
type State ¶
type State struct {
// contains filtered or unexported fields
}
State is the mutable graph state. Only the DAG engine writes to it; nodes receive a read-only view via NodeContext.Input.
Ownership: whoever receives a *State from Execute owns it and calls Release when finished. Release drops this State's reference to its backing map; any outstanding ReadState view keeps the map alive. The GC reclaims the map once every reference is gone.
State is deliberately NOT pooled. Nodes receive NodeContext.Input and may legally retain that reference inside their own scope; recycling States across executions would create use-after-free hazards the compiler cannot catch.
func AcquireState ¶
func AcquireState() *State
AcquireState returns a fresh, empty State. The caller owns it and should call Release when finished.
func (*State) AsRead ¶ added in v0.14.0
AsRead returns the read-only view of this state. Zero cost: it copies the map header. Safe on a nil receiver.
func (*State) Clone ¶
Clone returns a fresh State with a shallow copy of the map. Values are shared, not deep-copied; nodes must not return mutable values they intend to modify after returning.
func (*State) Get ¶
Get on *State delegates to the read-only view so callers that hold the writable handle can read without an explicit AsRead hop. Nil-safe.
func (*State) Release ¶
func (s *State) Release()
Release drops this State's reference to its backing map. It is a hint, not a lifetime operation: State is not pooled, so a forgotten Release is not a leak — the GC reclaims the map once every *State and ReadState view goes out of scope. Outstanding ReadState views (a node that captured Input into a goroutine) keep the map readable regardless of how many times Release has been called.
Safe on a nil receiver and safe to call multiple times.
type SuspendError ¶ added in v0.11.1
SuspendError carries the reason and payload for a graceful pause. Reason is human-readable; Payload is what the approver needs to see.
func (*SuspendError) Error ¶ added in v0.11.1
func (e *SuspendError) Error() string
func (*SuspendError) Is ¶ added in v0.11.1
func (e *SuspendError) Is(target error) bool
type TopologyRegistry ¶
type TopologyRegistry struct {
// contains filtered or unexported fields
}
func (*TopologyRegistry) Register ¶
func (t *TopologyRegistry) Register(from, to, channel, protocol string)
func (*TopologyRegistry) Reset ¶ added in v0.14.0
func (t *TopologyRegistry) Reset()
func (*TopologyRegistry) ToMermaid ¶
func (t *TopologyRegistry) ToMermaid() string
Source Files