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
- type DAG
- type Edge
- type EdgeMeta
- type ExecutionError
- type LayerCallback
- type Node
- type NodeContext
- type ReadState
- 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) AddEdge ¶
AddEdge declares that `to` depends on `from`. Duplicate edges are rejected at build time.
type DAG ¶
type DAG struct {
// contains filtered or unexported fields
}
DAG is an immutable, compiled directed acyclic graph.
func (*DAG) AsAction ¶
AsAction wraps the DAG as a standard Nexss action. The wrapped action accepts a ReadState (typically the parent layer's view) and returns a freshly owned *State that the caller must Release.
Nesting: the inner DAG clones the parent's state internally, so the parent is never mutated by the inner graph.
func (*DAG) Execute ¶
Execute runs DAG layers sequentially, executing nodes within each layer concurrently.
Resume-from-failure: any node whose output key is already present in initialState is skipped. Re-execute with a persisted state to run only the nodes that did not complete.
Layer callback: attach with WithLayerCallback. Fired after each layer with the accumulated state. A callback error aborts execution and is returned wrapped in *ExecutionError, unless the callback returned ErrSuspended (preserved as the sentinel). 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 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