# async-nodes
[](https://crates.io/crates/async-nodes)
[](https://docs.rs/async-nodes)
[](LICENSE)
[](https://github.com/sponsors/xangelix)
`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:
```rust
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}"),
}
}
```
```text
✅ 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/`](examples) - heavily
commented at three different audience levels.
### A loop that converges - as graph topology
[`examples/newton_sqrt.rs`](examples/newton_sqrt.rs) runs Newton's method as a node with a
**back-edge guarded by a named condition and an iteration budget**:
```rust
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()));
```
```text
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`](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:
```text
── 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:
```json
{
"version": 1,
"nodes": [{ "id": "2", "type_name": "preflight", "recovery": "fail_fast" }, …],
"edges": [{ "from": "3", "to": "4", "condition": { "named": "approved" } }, …]
}
```
```rust
let def = GraphDef::from_json(&json)?;
let graph = BspGraph::from_def(&def, ®istry, &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`](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**:
```text
+---------------------------------------+
| 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.
```rust
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
}
```
```rust
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:
```rust
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:
```rust
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
| [`quickstart.rs`](examples/quickstart.rs) | Beginner | `#[node]` macros, typed named ports, fan-out + barrier join, typed outputs |
| [`newton_sqrt.rs`](examples/newton_sqrt.rs) | Advanced | custom `BspNode<S>`, named conditions, back-edge + `max_iterations`, entries, counters |
| [`agentic_deploy.rs`](examples/agentic_deploy.rs) | Mastery | retry policy, suspend/resume HITL, `OnError` edges, live event trace, `GraphDef` JSON round-trip through a registry |
Run any of them:
```bash
cargo run --example quickstart
cargo run --example newton_sqrt
cargo run --example agentic_deploy
```
## License
MIT