Crate async_nodes

Expand description

async-nodes is a generic, type-safe execution engine implementing the Bulk Synchronous Parallel (BSP) standard for control-flow graphs with loops, with the classic dataflow-once pipelines expressed as a profile on the same core.

§One core, two faces

The engine core is control-flow with BSP superstep parallelism: nodes are re-executable units over a consumer-owned state S, edges are data carrying declarative Conditions, and activation is token-based (Kahn-style firing), which makes loops first-class topology — a back-edge simply delivers another token.

A dataflow DAG is a control-flow graph in which each node fires exactly once, when all its inputs are ready. That profile (pins, links, #[node] macros, BspSkip propagation, static inputs) keeps its guarantees — static analyzability, trivial scheduling, no infinite loops by construction — without being a second engine.

§The moving parts

BspGraph — the blueprint: nodes, graph::Edges, static inputs, and the consumer’s named-condition / merger registries. Author graphs with the fluent builder; exchange them as def::GraphDef JSON.
BspEngine — the scheduler: frontier supersteps under buffer_unordered, bounded by max_concurrency; circuit breakers (max_steps, per-edge max_iterations, per-node max_executions, per-edge token caps); recovery policies; suspension.
BspNode — the consumer’s computation unit, generic over the consumer state S.
NodeRegistry — node factories + named conditions + mergers, making graphs instantiable from JSON.
BspInterceptor — the live event stream (supersteps, edges taken, loop iterations, retries, suspensions) a GUI subscribes to.

§A loop, briefly

entry ──▶ Generate ──[has_more]──▶ Tools ──[always]──▶ Generate   (back-edge, max_iterations = 30)
              │
              └──────[done]──────▶ next-node…

Joins inside loops are almost always Activation::Any or merger-joins: an All-join on a cycle starves (one incoming edge fires once ever, the other once per iteration) — see engine for the diagram, and expect a validation warning.

Re-exports§

pub use typemap_rev;

Modules§

condition: Declarative edge conditions.
context
def: Serializable graph definitions — the GUI / n8n-enabler.
engine: The dynamic scheduler: token-based activation over BSP supersteps.
error
events: The live event stream: telemetry observers for runs, and a broadcast bridge.
graph: Graph topology: nodes, pins, and declarative edges.
merge: Fork/join branch merging.
node: The node contract: re-executable units over consumer state S.
recovery: Per-node failure recovery: retry with backoff, or route to a declarative error edge.
registry: The node & condition registry: the engine/definition separation made concrete.
run: Run-scoped state: the value store, the schedulable snapshot RunState, the node-facing RunContext, and suspension packaging.
value

Macros§

register_bsp_type

Structs§

BspContext: Shared execution context supporting dependency injection and state mapping across AST graphs.
BspEngine: The dynamic scheduler that organizes nodes into supersteps and executes them concurrently (bounded), re-executing nodes as loops deliver new activation tokens.
BspGraph: The blueprint definition containing nodes, edges, and initial inputs.
BspSkip: A marker type used to propagate bypassed execution paths in dynamic control-flow topologies.
BspValue: A thread-safe, reference-counted container wrapping arbitrary Rust types.
ChannelInterceptor: A BspInterceptor that forwards every event onto a tokio::sync::broadcast channel — every GUI / CLI / TUI then just subscribes.
ConditionEval: Evaluation context handed to conditions when a source node completes (or fails).
Edge: A declarative connection between two nodes — data, not a closure.
EdgeDef: A serializable edge definition.
GraphDef: A fully serializable graph description.
InputPin: An input pin on a specific node.
NamedConditionId: Registry key for a consumer-registered named condition, e.g. "my-crate.has_tool_calls".
NodeDef: A serializable node definition.
NodeHandle: A legacy non-generic identifier wrapping a designated node with port-link builder logic.
NodeRegistry: A per-S registry of node factories, named conditions, and mergers.
OutputPin: An output pin on a specific node.
PendingSuspension: Single record of a suspended node waiting for external inputs.
RetryPolicy: Declarative backoff policy governing automatic retry of failed node executions.
RunContext: Everything a node may touch, nothing it may not.
RunState: The complete schedulable state of one run: consumer state, value store, and all activation accounting. Suspension, resumption, and step-limit results all package this.
StaticInputDef: A serializable static input: an initial pin value injected before execution.
SuspendedExecution: Package containing suspension parameters to resume execution later.
TypedNodeHandle: A compile-time type-safe identifier wrapping a designated node. This allows compiler-checked, named port accessors generated by macro attributes.
ValidationReport: The result of graph validation: fatal errors plus non-fatal findings.
ValueStore: The engine-owned store of pin values (the dataflow profile’s IO surface).

Enums§

Activation: How a node joins a frontier, i.e. how its pending activation tokens are counted.
ActivationDef: Serializable activation policy: "any" (control-flow default) or "all" (join).
BspError: Standardized error types for the BSP graph compiler and execution runtime.
BspEvent: One owned event from the execution stream, as broadcast by ChannelInterceptor.
Condition: A declarative predicate deciding whether an edge fires.
ConditionDef: The serializable condition subset.
EngineResult: Represents the return status of a BSP execution run.
NodeOutcome: Defines the evaluated outcome of a single node’s execution turn.
Recovery: How a failed node execution should be recovered, if at all.
RecoveryDef: Serializable recovery policy: "fail_fast", "error_edge", or {"retry": {...}} (backoff parameters only — classification never serializes).
RecoveryPolicy: Per-node recovery policy, declared by the node and enforced by the engine.

Constants§

DEFAULT_MAX_PENDING_TOKENS: The default cap on pending (unconsumed) tokens per edge (token-flood guard).
DEFAULT_MAX_STEPS: The default global superstep budget (circuit breaker). The pre-rewrite engine had no cap; loops make one mandatory, so a generous default is baked in — tune it with BspEngine::with_max_steps.
ERROR_PORT: The value-store port a failing node writes its error to when its recovery policy routes to an error edge (dataflow profile; see crate::recovery::RecoveryPolicy::RouteToErrorEdge).
GRAPH_DEF_VERSION: The current GraphDef schema version. Bump on any breaking format change.

Traits§

BspInterceptor: Telemetry observer to audit performance and trace the state transitions of a run.
BspNode: The programmatic contract for writing execution nodes.
BspOutputMapExt
DataflowNode: The dataflow-profile authoring trait: plain-Rust node authoring without macros.
Merger: How a branched state rejoins the trunk.

Type Aliases§

BranchTransform: A transform applied to each branch-state clone at multi-node frontier formation, right after the clone is taken from the trunk (see BspEngine::with_branch_transform). The usize is the branch’s ordinal within the frontier formation (declaration order, 0-based).
EdgeId: Edge identifier: the edge’s index in declaration order.
NodeId: Programmatic node identifier: positional, stable within one graph build.
PortIndex: A port index on a node (dataflow profile).

Attribute Macros§

node: Procedural attribute transforming an asynchronous function into an explicit BspNode.
node_blocking: Procedural attribute transforming a synchronous, blocking function into a BspNode.

Derive Macros§

BspType: Custom derive macro providing serialization support to arbitrary developer types.