async-nodes

async-nodes is a type-safe, asynchronous node engine implementing the Bulk Synchronous Parallel (BSP) computing model in Rust - a control-flow engine with first-class loops, declarative edges, failure recovery, suspension, and serializable graph definitions.

The engine core is control-flow with BSP superstep parallelism: nodes are re-executable units over your own state type S, edges are data carrying declarative conditions, and token-based activation makes loops native topology. The classic dataflow-once pipelines this crate has always run are a profile on that core - a dataflow DAG is just a control-flow graph in which each node fires exactly once, when all its inputs are ready - not a second engine with its own semantics.

Why it's different

Most graph crates schedule DAGs. async-nodes runs workflows:

• Loops are topology, not driver code. A back-edge with an iteration budget is the loop - max_iterations on the edge, max_steps on the engine, iteration counters in the run state. Runaway graphs end in a distinct EngineResult::StepLimitReached, never a silent partial run.
• Suspension is a first-class outcome. A node can freeze the whole run mid-graph (mid-loop, even) with a payload and a list of pins it's waiting on; the host resumes it later by injecting values. Human approvals, webhooks, and long polls become one-line primitives.
• Recovery is policy, not try/catch spaghetti. Per node: fail-fast, retry with jittered exponential backoff (honoring Retry-After-style hints), or route the failure onto a visible OnError edge. Error paths become drawable topology.
• Graphs are data. Every graph round-trips through versioned JSON (GraphDef): stable string ids, declarative conditions by name, per-node config blobs, GUI layout hints. A node-graph GUI can load, edit, save, and instantiate workflows - the palette, the canvas, and the save format are the same data.
• Everything emits events. Supersteps, edges taken, retries, loop iterations, suspensions - one BspInterceptor trait, with a ChannelInterceptor broadcasting over tokio::sync::broadcast for live UIs.
• Parallelism with guardrails. BSP supersteps with barrier synchronization, with_max_concurrency throttling, branch isolation with deterministic, declaration-order merges.

Quickstart

#[node] turns an async fn into a node with typed, named ports. Wiring is compile-time named and validated before execution:

use async_nodes::{BspEngine, BspGraph, BspOutputMapExt, EngineResult, node};

#[node("Source Text")]
async fn source_text() -> String {
    "the quick brown fox jumps over the lazy dog".to_string()
}

#[node("Word Count")]
async fn word_count(text: String) -> i64 {
    text.split_whitespace().count() as i64
}

#[node("Report")]
async fn report(words: i64) -> String {
    format!("{words} words")
}

#[tokio::main]
async fn main() {
    let mut graph = BspGraph::new();
    let source = graph.add_node(SourceTextNode::new());
    let words = graph.add_node(WordCountNode::new());
    let report = graph.add_node(ReportNode::new());

    graph.add_link(source.output(), words.text());   // named after the fn parameter
    graph.add_link(words.output(), report.words());

    let engine = BspEngine::new(graph);
    match engine.run(()).await {
        Ok(EngineResult::Complete(state)) => {
            let text: String = state.get_typed(report.output()).unwrap();
            println!("✅ {text}");
        }
        Ok(EngineResult::Suspended(_)) => println!("graph asked for external input"),
        Ok(EngineResult::StepLimitReached(_)) => println!("hit the step budget"),
        Err(e) => eprintln!("graph failed: {e}"),
    }
}

✅ 9 words

() is the graph's state: a pure dataflow pipeline doesn't need one. Control-flow graphs carry your own S - the engine moves it, clones it across branches, and never inspects it.

Show me something I can't do in a DAG scheduler

All three of these are complete, runnable programs in examples/ - heavily commented at three different audience levels.

A loop that converges - as graph topology

examples/newton_sqrt.rs runs Newton's method as a node with a back-edge guarded by a named condition and an iteration budget:

graph
    .add_edge(improve.id, improve.id)                 // the loop
    .when(Condition::Named("not_converged".into()))
    .with_max_iterations(1_000);                      // the circuit breaker

graph
    .add_edge(improve.id, report.id)                  // the exit
    .when(Condition::Named("converged".into()));

  iteration  1: guess = 154.000000000000
  iteration  2: guess = 78.987012987013
  iteration  5: guess = 25.017148688503
  iteration  8: guess = 24.738633753706
✅ √612 ≈ 24.738633753706
converged in 8 iterations (9 supersteps)

A run that pauses for a human - then resumes mid-superstep

examples/agentic_deploy.rs is a deployment workflow with a retrying preflight check, an approval gate that suspends the entire run, and an always-failing notifier whose error rides a visible OnError edge:

── superstep 1 ─ frontier [2]
   ↻ node 2 retry #1 in 9.335ms (… simulated flake (attempt 1))
   ↻ node 2 retry #2 in 19ms   (… simulated flake (attempt 2))
    [state] preflight green
   ⤷ edge 2 → 3  [always]
── superstep 2 ─ frontier [3]
   ⏸ suspended at node 3: {"plan":[…],"question":"Approve deployment?"}
   ⤷ edge 3 → 4  [named:approved]          ← after engine.resume(approval)

The same example then serializes the graph to JSON and re-runs it from the document - taking the other approval branch:

{
  "version": 1,
  "nodes": [{ "id": "2", "type_name": "preflight", "recovery": "fail_fast" }, …],
  "edges": [{ "from": "3", "to": "4", "condition": { "named": "approved" } }, …]
}

let def = GraphDef::from_json(&json)?;
let graph = BspGraph::from_def(&def, &registry, &ctx)?;  // registry rebinds closures

That is the whole n8n-style story: workflows you can draw, save, and hand to someone else.

The beginner one

examples/quickstart.rs - the pipeline above, with fan-out and a barrier join, commented for a first-day reader.

What is Bulk Synchronous Parallel (BSP)?

The BSP model, originally proposed by Leslie Valiant, structures concurrent computation into global iterations called supersteps:

            +---------------------------------------+
            |        SUPERSTEP EXECUTION LOOP       |
            +---------------------------------------+
            |  Phase 1: Concurrent Computation      |
            |   - Workers execute local logic       |
            +-------------------+-------------------+
                                |
                                v
            +-------------------+-------------------+
            |  Phase 2: Communication               |
            |   - Outputs routed to target inputs   |
            +-------------------+-------------------+
                                |
                                v
            +-------------------+-------------------+
            |  Phase 3: Barrier Synchronization     |
            |   - All tasks complete before step    |
            |     increment                         |
            +-------------------+-------------------+

1. Concurrent Computation: nodes perform computation independently using locally available inputs.
2. Communication: outputs of completed nodes are routed to the inputs of downstream waiting nodes.
3. Barrier Synchronization: all active tasks in the current superstep complete before the step counter increments; the next superstep's frontier is computed from freshly satisfied edges and inputs.

Structuring concurrency into explicit supersteps eliminates data races, ad-hoc locks, and unstructured state tracking - execution stays deterministic, predictable, and debuggable.

Dependency & Context Injection

Nodes often need shared, thread-safe dependencies (connection pools, clients, config) that don't originate from graph data flow. BspContext is a type-keyed DI store; the #[context] attribute extracts and clones dependencies before execution, so no lock guards survive across .await points.

use std::sync::Arc;
use async_nodes::typemap_rev::TypeMapKey;

#[derive(Clone, Debug)]
pub struct DatabaseConnection {
    pub url: String,
}

impl TypeMapKey for DatabaseConnection {
    type Value = Self; // the storage representation of the key
}

use async_nodes::node;

#[node("QueryUser")]
async fn query_user(
    #[context] db: DatabaseConnection, // extracted and cloned from BspContext
    user_id: f64,                      // passed normally from wired graph ports
) -> String {
    format!("User retrieved from {} with ID {}", db.url, user_id)
}

For reference-counted heavy resources, map the key to its Arc form explicitly:

impl TypeMapKey for HeavyConfig {
    type Value = Arc<Self>;
}

#[node("VerifyPermissions")]
async fn verify_permissions(
    #[context(HeavyConfig)] config: Arc<HeavyConfig>,
    user_id: String,
) -> bool {
    config.keys.contains(&user_id)
}

Attach the context at engine construction:

use async_nodes::{BspContext, BspEngine, BspGraph, BspOutputMapExt, EngineResult};

#[tokio::main]
async fn main() {
    let mut graph = BspGraph::new();
    let query_node = graph.add_node(QueryUserNode::new());
    graph.set_static_input(query_node.user_id(), 42.0);

    let ctx = BspContext::new();
    ctx.insert::<DatabaseConnection>(DatabaseConnection {
        url: "postgres://localhost:5432".to_string(),
    });

    let engine = BspEngine::new(graph).with_context(ctx);
    if let Ok(EngineResult::Complete(state)) = engine.run(()).await {
        let result: String = state.get_typed(query_node.output()).unwrap();
        println!("{result}");
    }
}

Critical Execution & Safety Guidelines

1. Cooperative Concurrency

The engine drives supersteps concurrently via cooperative stream buffering. Non-blocking #[node] async methods must remain cooperative - periodically tokio::task::yield_now().await inside long computations, or use #[node_blocking] for CPU-bound work (it routes through tokio::task::spawn_blocking).

2. Re-entrancy & Idempotency on Resumption

When a graph is suspended via NodeOutcome::Suspend and later resumed, the suspended node's execute block runs from the very beginning (with the injected inputs now present). Write suspend-capable nodes idempotently: check whether side effects already happened before repeating them.

3. Skip Propagation Behavior

Dynamic bypass uses BspSkip propagation: if any incoming input to a dataflow node carries a BspSkip marker (even into an Option<T> port), the node's execution is bypassed and the skip propagates downstream. Branch selection in the control-flow layer should use mutually exclusive edge conditions instead.

4. Loops Are Legal - and Bounded

Cycles are native topology: a back-edge re-activates an earlier node, and activation tokens (not an "already executed" set) drive scheduling. Because runaway loops are possible by construction, bounds are part of the core:

• Put max_iterations on every cycle-closing edge (BspError::LoopBoundExceeded when tripped). Validation warns on cycles lacking one.
• The global with_max_steps budget (default 1024 supersteps) ends runaway runs with a distinct EngineResult::StepLimitReached carrying the live state - distinguish it from natural completion instead of treating it as one.
• Joins inside loops are almost always Activation::Any or merger-joins. An All-join on a cycle starves: one incoming edge's token arrives once per iteration while another's arrives once ever. Validation warns on this shape; at runtime it ends in BspError::Deadlock.
• Retry policies re-invoke execute verbatim. Attach RecoveryPolicy::Retry only to read-only nodes, or to nodes that undo partial writes on re-entry via RunContext.execution_no - retrying a side-effecting node replays the side effect.

Examples

Run any of them:

cargo run --example quickstart
cargo run --example newton_sqrt
cargo run --example agentic_deploy

License

MIT