From 7ec82a6aef7da8558c7fd69d4cda39cc894c5c15 Mon Sep 17 00:00:00 2001 From: cyrus Date: Thu, 16 Jul 2026 22:15:01 +0530 Subject: [PATCH 1/3] feat(graph): add BarrierRelief primitive for mixed fan-in barriers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A waiting/barrier node (add_waiting_edge) normally activates only once every registered predecessor has arrived. When one predecessor is only reachable via a conditional branch, and the brancher routes elsewhere, that predecessor never arrives and the barrier deadlocks forever. Add GraphBuilder::add_barrier_relief(source, relief_node, barrier_node): when `source` completes a superstep without routing to `relief_node`, route_completed registers a phantom arrival of `relief_node` at `barrier_node`, letting the barrier clear on its remaining predecessors instead of hanging. This fixes the deadlock without weakening the barrier into a plain edge, which would let a *taken* branch's data race past the merge and be silently dropped (a stale-snapshot data loss bug). Threaded through GraphBuilder -> CompiledGraph::from_parts (compiled- graph metadata; no checkpoint/serialization changes needed since barrier_arrivals already persist). Adds barrier_relief_fires_when_source_skips_relief_node covering the fix. Part of the tinyflows Eng §5 mixed fan-in deadlock fix (companion change in tinyflows/src/engine.rs). --- src/graph/builder/mod.rs | 58 +++++++++++++++++++++++++++++++ src/graph/builder/types.rs | 3 ++ src/graph/compiled/mod.rs | 5 ++- src/graph/compiled/routing.rs | 32 +++++++++++++++++ src/graph/compiled/test.rs | 65 +++++++++++++++++++++++++++++++++++ src/graph/compiled/types.rs | 5 ++- 6 files changed, 166 insertions(+), 2 deletions(-) diff --git a/src/graph/builder/mod.rs b/src/graph/builder/mod.rs index d148fae..eaf4e8f 100644 --- a/src/graph/builder/mod.rs +++ b/src/graph/builder/mod.rs @@ -29,6 +29,32 @@ use crate::graph::reducer::{OverwriteStateReducer, StateReducer}; use crate::harness::ids::{GraphId, NodeId}; use crate::{Result, TinyAgentsError}; +/// A relief registration for a mixed fan-in barrier. +/// +/// A waiting/barrier node (see [`GraphBuilder::add_waiting_edge`]) normally +/// activates only once *every* registered predecessor has arrived. When one +/// of those predecessors (`relief_node`) is only reachable via a conditional +/// branch out of `source`, and `source` routes elsewhere instead, that +/// predecessor never runs and the barrier would wait forever. +/// +/// A [`BarrierRelief`] fixes that without weakening the barrier into a plain +/// edge (which would let a *taken* branch's downstream data race the merge +/// and get silently dropped): when `source` completes a superstep without +/// routing to `relief_node`, the executor registers a phantom arrival of +/// `relief_node` at `barrier_node`, so the barrier can still clear on the +/// predecessors that actually ran. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BarrierRelief { + /// The brancher node whose conditional routing determines whether + /// `relief_node` runs. + pub source: NodeId, + /// The conditional-only predecessor of `barrier_node` that `source` may + /// or may not route to. + pub relief_node: NodeId, + /// The mixed fan-in (all-waiting) node gated on `relief_node`'s arrival. + pub barrier_node: NodeId, +} + impl Default for GraphBuilder where State: Clone + Send + Sync + 'static, @@ -55,6 +81,7 @@ where branches: HashMap::new(), command_nodes: HashSet::new(), waiting: HashMap::new(), + barrier_reliefs: Vec::new(), reducer: None, recursion_limit: 50, parallel: false, @@ -201,6 +228,35 @@ where self } + /// Registers a barrier relief for a mixed fan-in barrier. + /// + /// When `source` completes a superstep without routing to `relief_node` + /// — because `relief_node` sits behind a conditional branch `source` did + /// not take this run — this records a phantom arrival of `relief_node` + /// at `barrier_node`'s waiting-edge barrier (registered separately via + /// [`Self::add_waiting_edge`]), so an all-waiting merge downstream of a + /// mixed fan-in (one unconditionally reachable predecessor plus one + /// reachable only via a conditional branch) can still clear on its + /// remaining predecessors instead of deadlocking on a branch that will + /// never run. + /// + /// `relief_node` must be one of `barrier_node`'s registered waiting + /// predecessors; a relief for a barrier with no matching waiting + /// registration is a no-op at execution time. + pub fn add_barrier_relief( + mut self, + source: impl Into, + relief_node: impl Into, + barrier_node: impl Into, + ) -> Self { + self.barrier_reliefs.push(BarrierRelief { + source: source.into(), + relief_node: relief_node.into(), + barrier_node: barrier_node.into(), + }); + self + } + /// Sets the entry node, i.e. `add_edge(START, node)`. pub fn set_entry(self, node: impl Into) -> Self { self.add_edge(START, node) @@ -406,6 +462,7 @@ where max_concurrency, node_timeout, node_meta, + barrier_reliefs, } = self; Ok(CompiledGraph::from_parts( @@ -423,6 +480,7 @@ where max_concurrency, node_timeout, node_meta, + barrier_reliefs, )) } diff --git a/src/graph/builder/types.rs b/src/graph/builder/types.rs index d9f5311..ef5ffa3 100644 --- a/src/graph/builder/types.rs +++ b/src/graph/builder/types.rs @@ -222,6 +222,9 @@ pub struct GraphBuilder { /// Barrier/waiting edges: target node -> set of predecessor nodes that must /// all have completed (across steps) before the target activates. pub(crate) waiting: HashMap>, + /// Mixed fan-in barrier relief registrations; see + /// [`super::BarrierRelief`]. + pub(crate) barrier_reliefs: Vec, pub(crate) reducer: Option>>, pub(crate) recursion_limit: usize, /// When true, active nodes in a superstep run concurrently (fan-out). diff --git a/src/graph/compiled/mod.rs b/src/graph/compiled/mod.rs index 9bfe022..28c5f04 100644 --- a/src/graph/compiled/mod.rs +++ b/src/graph/compiled/mod.rs @@ -78,7 +78,8 @@ use std::sync::Arc; use std::time::{Duration, SystemTime}; use crate::graph::builder::{ - Branch, BuilderNode, END, ForkId, NodeContext, NodeFuture, NodeHandler, NodeMeta, START, + BarrierRelief, Branch, BuilderNode, END, ForkId, NodeContext, NodeFuture, NodeHandler, + NodeMeta, START, }; use crate::graph::checkpoint::{ BarrierArrivals, Checkpoint, CheckpointConfig, CheckpointTuple, Checkpointer, DurabilityMode, @@ -253,6 +254,7 @@ impl CompiledGraph { max_concurrency: Option, node_timeout: Option, node_meta: HashMap, + barrier_reliefs: Vec, ) -> Self { Self { graph_id, @@ -262,6 +264,7 @@ impl CompiledGraph { branches: Arc::new(branches), command_nodes: Arc::new(command_nodes), waiting: Arc::new(waiting), + barrier_reliefs: Arc::new(barrier_reliefs), node_meta: Arc::new(node_meta), entry, reducer, diff --git a/src/graph/compiled/routing.rs b/src/graph/compiled/routing.rs index bf5062b..a15ac70 100644 --- a/src/graph/compiled/routing.rs +++ b/src/graph/compiled/routing.rs @@ -59,6 +59,38 @@ where } } } + + // Mixed fan-in barrier relief: a waiting/barrier node normally + // activates only once every registered predecessor has arrived. When + // one of those predecessors is reachable only via a conditional + // branch, and the brancher routed elsewhere this superstep, that + // predecessor will never arrive on its own — register a phantom + // arrival on its behalf so the barrier can still clear on the + // predecessors that actually ran, instead of deadlocking forever. + for relief in self.barrier_reliefs.iter() { + let source_completed = completed.iter().any(|a| a.node == relief.source); + if !source_completed || next_seen.contains(&relief.relief_node) { + continue; + } + let Some(required) = self.waiting.get(&relief.barrier_node) else { + continue; + }; + let arrived = barrier_arrivals + .entry(relief.barrier_node.clone()) + .or_default(); + arrived.insert(relief.relief_node.clone()); + if !required.is_subset(arrived) { + continue; + } + barrier_arrivals.remove(&relief.barrier_node); + if next_seen.insert(relief.barrier_node.clone()) { + next.push(Activation { + node: relief.barrier_node.clone(), + send_arg: None, + }); + } + } + Ok(next) } diff --git a/src/graph/compiled/test.rs b/src/graph/compiled/test.rs index 705a203..8eb370c 100644 --- a/src/graph/compiled/test.rs +++ b/src/graph/compiled/test.rs @@ -1243,6 +1243,71 @@ async fn barrier_arrivals_survive_interrupt_and_resume() { assert_eq!(done.state.value, 103); } +#[tokio::test] +async fn barrier_relief_fires_when_source_skips_relief_node() { + // Mixed fan-in: `m` waits on both `a` and `c`. `condition` never routes to + // `a` — it always takes the `skip` route to END, simulating an untaken + // conditional branch — so without a barrier relief `m` would deadlock + // forever waiting on a predecessor that never runs. + // `add_barrier_relief("condition", "a", "m")` registers `a`'s phantom + // arrival at `m` whenever `condition` completes without activating `a`, + // so `m` still fires once `c`'s real arrival lands. + let graph = GraphBuilder::, String>::new() + .with_parallel(true) + .set_reducer(ClosureStateReducer::new(|mut s: Vec, u: String| { + s.push(u); + Ok(s) + })) + .add_node("start", |_s, _c: NodeContext| async move { + Ok(NodeResult::Command( + Command::default().with_goto(["condition", "c"]), + )) + }) + .add_node("condition", |_s, _c: NodeContext| async move { + Ok(NodeResult::Update("condition".to_string())) + }) + .add_node("a", |_s, _c: NodeContext| async move { + Ok(NodeResult::Update("a".to_string())) + }) + .add_node("c", |_s, _c: NodeContext| async move { + Ok(NodeResult::Update("c".to_string())) + }) + .add_node("m", |_s, _c: NodeContext| async move { + Ok(NodeResult::Update("m".to_string())) + }) + .set_entry("start") + .mark_command_routing("start") + // `condition` always takes the `skip` route to END — `a` is never + // reached via a real edge. + .add_conditional_edges( + "condition", + |_s: &Vec| "skip".to_string(), + [("skip", END)], + ) + .add_waiting_edge("a", "m") + .add_waiting_edge("c", "m") + .add_barrier_relief("condition", "a", "m") + .set_finish("m") + .compile() + .unwrap(); + + let run = graph.run(Vec::new()).await.unwrap(); + + assert!( + run.visited.iter().any(|n| n.as_str() == "m"), + "m must activate via the barrier relief even though `a` never ran" + ); + assert!( + !run.visited.iter().any(|n| n.as_str() == "a"), + "a must never have run (condition always skipped it)" + ); + assert_eq!( + run.state, + vec!["condition".to_string(), "c".to_string(), "m".to_string()], + "m fires off condition+c's real contributions, with no phantom `a` update" + ); +} + #[tokio::test] async fn reducer_error_at_boundary_transitions_run_to_failed() { // A reducer error raised at the step boundary (after the node ran) must diff --git a/src/graph/compiled/types.rs b/src/graph/compiled/types.rs index e220827..2419bdb 100644 --- a/src/graph/compiled/types.rs +++ b/src/graph/compiled/types.rs @@ -4,7 +4,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; use crate::graph::builder::START; -use crate::graph::builder::{Branch, BuilderNode, NodeMeta}; +use crate::graph::builder::{BarrierRelief, Branch, BuilderNode, NodeMeta}; use crate::graph::checkpoint::{ CheckpointConfig, CheckpointMetadata, Checkpointer, DurabilityMode, }; @@ -32,6 +32,8 @@ pub struct CompiledGraph { /// Barrier/waiting edges: target -> the predecessor set that must all /// complete (across steps) before the target activates. pub(crate) waiting: Arc>>, + /// Mixed fan-in barrier relief registrations; see [`BarrierRelief`]. + pub(crate) barrier_reliefs: Arc>, /// Behavior-free per-node markers/metadata surfaced by the topology export. pub(crate) node_meta: Arc>, pub(crate) entry: NodeId, @@ -106,6 +108,7 @@ impl Clone for CompiledGraph { branches: self.branches.clone(), command_nodes: self.command_nodes.clone(), waiting: self.waiting.clone(), + barrier_reliefs: self.barrier_reliefs.clone(), node_meta: self.node_meta.clone(), entry: self.entry.clone(), reducer: self.reducer.clone(), From f60135b57d30b9ed81cc91ce9ed406eb5da09e70 Mon Sep 17 00:00:00 2001 From: cyrus Date: Thu, 16 Jul 2026 22:26:43 +0530 Subject: [PATCH 2/3] fix(graph): key barrier relief on resolved routing target, not same-superstep scheduling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit barrier_relief_fires_when_source_skips_relief_node covered only the direct (one-hop) case: `source --branch--> relief_node`. The check — "is relief_node freshly scheduled into next this step" — is a same- superstep proxy that breaks for a multi-hop conditional predecessor (`source --branch--> x --main--> relief_node`, x a plain pass-through): relief_node is not in next on the step source itself completes, even on the TAKEN branch, since x hasn't run x yet. The relief fired a premature phantom arrival regardless, letting the barrier clear before the real predecessor's data ever committed — reintroducing the exact data-loss bug BarrierRelief exists to prevent. Fix: resolve source's actual routing target(s) this step (reusing the existing pure self.route()) and walk them forward through the compiled graph's deterministic (single-successor) edges via a new reaches_deterministically helper, stopping at barrier_node. This answers "was the branch leading to relief_node taken" directly, correct for any chain length of plain/waiting-edge pass-throughs — not just one hop. Keeps the same-superstep next_seen check as a defensive fallback. A further conditional/command node between source and relief_node (no deterministic self.edges entry) is a second runtime decision this walk cannot resolve ahead of time and conservatively reports unreachable — documented as a residual limitation in the new helper's doc comment. --- src/graph/compiled/routing.rs | 86 +++++++++++++++++++++++++++++++++-- 1 file changed, 83 insertions(+), 3 deletions(-) diff --git a/src/graph/compiled/routing.rs b/src/graph/compiled/routing.rs index a15ac70..d73e7fd 100644 --- a/src/graph/compiled/routing.rs +++ b/src/graph/compiled/routing.rs @@ -63,13 +63,60 @@ where // Mixed fan-in barrier relief: a waiting/barrier node normally // activates only once every registered predecessor has arrived. When // one of those predecessors is reachable only via a conditional - // branch, and the brancher routed elsewhere this superstep, that + // branch, and the *taken* branch does not lead toward it, that // predecessor will never arrive on its own — register a phantom // arrival on its behalf so the barrier can still clear on the // predecessors that actually ran, instead of deadlocking forever. + // + // The check is keyed on `source`'s *resolved routing target* this + // step (via `reaches_deterministically`), not on whether + // `relief_node` is freshly scheduled into `next` this same + // superstep. A same-superstep check is wrong for a multi-hop + // conditional predecessor (`source --branch--> x --main--> + // relief_node`, `x` a plain pass-through): `relief_node` would not + // yet be in `next` on the step `source` itself completes — even on + // the *taken* branch, where `relief_node` WILL run once `x` does — + // so a same-superstep check would fire a premature phantom arrival + // and let the barrier clear before the real predecessor's data + // commits, reintroducing the exact data-loss bug this primitive + // exists to prevent. Resolving `source`'s actual target and walking + // it forward through deterministic (non-branching) edges is correct + // for both direct and multi-hop cases because it answers "was the + // branch leading to `relief_node` taken", not "did `relief_node` + // happen to run in lockstep with `source`". for relief in self.barrier_reliefs.iter() { - let source_completed = completed.iter().any(|a| a.node == relief.source); - if !source_completed || next_seen.contains(&relief.relief_node) { + let source_indices: Vec = completed + .iter() + .enumerate() + .filter(|(_, activation)| activation.node == relief.source) + .map(|(index, _)| index) + .collect(); + if source_indices.is_empty() { + continue; + } + let mut branch_taken = false; + for index in &source_indices { + let targets = self.route( + &relief.source, + goto_map.get(index).map(Vec::as_slice), + state, + )?; + if targets.iter().any(|target| { + self.reaches_deterministically( + target.node(), + &relief.relief_node, + &relief.barrier_node, + ) + }) { + branch_taken = true; + break; + } + } + // A relief_node freshly scheduled into `next` this step (a + // direct, single-hop predecessor) is also proof the branch was + // taken; kept as a defensive fallback alongside the resolved- + // target check above. + if branch_taken || next_seen.contains(&relief.relief_node) { continue; } let Some(required) = self.waiting.get(&relief.barrier_node) else { @@ -94,6 +141,39 @@ where Ok(next) } + /// Whether `to` is reachable from `from` by following only deterministic + /// static routing — plain/waiting edges (`self.edges`), which resolve to + /// exactly one successor with no runtime decision — without ever + /// expanding through `stop`. + /// + /// This is what makes barrier-relief evaluation correct for a + /// multi-hop conditional predecessor: once a brancher's routing decision + /// for this step is resolved to a concrete target, whether that target + /// eventually leads to a barrier's conditional predecessor is a static + /// property of the compiled topology for any chain of plain pass-through + /// nodes — it does not depend on when each hop happens to run. A further + /// conditional/command node along the way (no `self.edges` entry) is a + /// second runtime decision this walk cannot resolve ahead of time, so it + /// conservatively reports unreachable there (falling back to the + /// same-superstep check). + fn reaches_deterministically(&self, from: &NodeId, to: &NodeId, stop: &NodeId) -> bool { + if from == to { + return true; + } + let mut current = from; + let mut seen: HashSet<&NodeId> = HashSet::new(); + while let Some(next) = self.edges.get(current) { + if next == to { + return true; + } + if next == stop || !seen.insert(next) { + return false; + } + current = next; + } + false + } + /// Resolves the next routing targets for `node_id`. /// /// Command `goto` (which may include [`Send`] packets) wins over static and From a7b3d195decc29f41b3926c891eba0c729e3073e Mon Sep 17 00:00:00 2001 From: cyrus Date: Fri, 17 Jul 2026 12:15:23 +0530 Subject: [PATCH 3/3] fix(graph): reuse first-pass resolved targets in barrier-relief check Addresses @coderabbitai on PR #62 (routing.rs:87-114): the barrier-relief pass called self.route() a second time for the same activation to get its resolved targets, which would invoke a stateful/nondeterministic router closure twice for one activation. Cache each activation's resolved targets (already computed once in the main completed loop) and have the relief pass index into that cache instead of recomputing. --- src/graph/compiled/routing.rs | 24 +++++++++++------------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/src/graph/compiled/routing.rs b/src/graph/compiled/routing.rs index d73e7fd..e555bf7 100644 --- a/src/graph/compiled/routing.rs +++ b/src/graph/compiled/routing.rs @@ -21,9 +21,16 @@ where ) -> Result> { let mut next: Vec = Vec::new(); let mut next_seen: HashSet = HashSet::new(); + // Resolved targets per activation index, captured once here and + // reused by the barrier-relief pass below instead of calling + // `self.route` a second time — a router closure is only guaranteed + // pure/idempotent per the `route`/`add_conditional_edges` contract, + // not safe to invoke twice for the same activation. + let mut resolved: Vec> = Vec::with_capacity(completed.len()); for (index, activation) in completed.iter().enumerate() { let node_id = &activation.node; let targets = self.route(node_id, goto_map.get(&index).map(Vec::as_slice), state)?; + resolved.push(targets.clone()); for target in targets { let tnode = target.node().clone(); if tnode.as_str() == END { @@ -94,24 +101,15 @@ where if source_indices.is_empty() { continue; } - let mut branch_taken = false; - for index in &source_indices { - let targets = self.route( - &relief.source, - goto_map.get(index).map(Vec::as_slice), - state, - )?; - if targets.iter().any(|target| { + let branch_taken = source_indices.iter().any(|index| { + resolved[*index].iter().any(|target| { self.reaches_deterministically( target.node(), &relief.relief_node, &relief.barrier_node, ) - }) { - branch_taken = true; - break; - } - } + }) + }); // A relief_node freshly scheduled into `next` this step (a // direct, single-hop predecessor) is also proof the branch was // taken; kept as a defensive fallback alongside the resolved-