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..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 { @@ -59,9 +66,112 @@ 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 *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_indices: Vec = completed + .iter() + .enumerate() + .filter(|(_, activation)| activation.node == relief.source) + .map(|(index, _)| index) + .collect(); + if source_indices.is_empty() { + continue; + } + let branch_taken = source_indices.iter().any(|index| { + resolved[*index].iter().any(|target| { + self.reaches_deterministically( + target.node(), + &relief.relief_node, + &relief.barrier_node, + ) + }) + }); + // 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 { + 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) } + /// 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 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(),