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 *State, nodeID string) (T, error)
- func OutputKey(nodeID string) string
- func WithLayerCallback(ctx context.Context, fn LayerCallback) context.Context
- type Builder
- type DAG
- type Edge
- type EdgeMeta
- type ExecutionError
- type LayerCallback
- type Node
- type NodeContext
- type State
- 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 has paused execution (e.g. awaiting human approval). Returning this error preserves partial state so it can be saved to persistent storage.
var GlobalTopology = &TopologyRegistry{}
Functions ¶
func GetNodeOutput ¶
GetNodeOutput reads a node result using the allocation-free direct type path.
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
}
type DAG ¶
type DAG struct {
// contains filtered or unexported fields
}
func (*DAG) Execute ¶
Execute runs DAG layers sequentially, executing nodes in each layer concurrently. Supports HIL resumption: nodes whose output key is already present in initialState are skipped.
A LayerCallback may be attached to ctx with WithLayerCallback; when present it is invoked after each layer with the state that already includes that layer's outputs. A callback error aborts execution and is returned wrapped in *ExecutionError, unless the callback returned ErrSuspended, in which case the sentinel is preserved.
On any error the returned state is non-nil and owns every node output produced so far. The caller must Release it.
The callback is consumed by the outermost DAG that sees it, so nested DAGs never fire it twice.
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 NodeContext ¶
type NodeContext struct {
// Input is the read-only, frozen state from preceding layers.
//
// ⚠️ WARNING: Because DAG state utilizes sync.Pool for memory efficiency,
// you MUST NOT read from Input after your node's handler function returns.
// Spawning detached goroutines that access Input beyond the lifecycle of
// this function will result in use-after-free panics or silent data corruption
// when the state is recycled and reused by subsequent graph executions.
Input *State
Key string // Designated output key for this node
}
NodeContext provides execution context to a DAG node.
type State ¶
type State struct {
// contains filtered or unexported fields
}
State is an immutable-by-convention view of graph state.
func AcquireState ¶
func AcquireState() *State
AcquireState retrieves a State instance from the memory pool.
type TopologyRegistry ¶
type TopologyRegistry struct {
// contains filtered or unexported fields
}
func (*TopologyRegistry) Register ¶
func (t *TopologyRegistry) Register(from, to, channel, protocol string)
func (*TopologyRegistry) ToMermaid ¶
func (t *TopologyRegistry) ToMermaid() string
Source Files