From ada4b611115c0ce2d4b2b21bf8cc30ed52d6ce3a Mon Sep 17 00:00:00 2001 From: Tobias Nilsson Date: Wed, 29 Jul 2026 20:20:24 +0200 Subject: [PATCH 1/6] fix: Re-run run-scoped results observed from another Run's attempt A Run which attaches to another Run's in-flight attempt cannot consume its run-scoped result, so it re-enters to compute its own. --- src/rust/graph/src/entry.rs | 38 ++++++++++++++++++++++++------------- src/rust/graph/src/lib.rs | 4 ++-- 2 files changed, 27 insertions(+), 15 deletions(-) diff --git a/src/rust/graph/src/entry.rs b/src/rust/graph/src/entry.rs index b1fae65935c..a08c04042fc 100644 --- a/src/rust/graph/src/entry.rs +++ b/src/rust/graph/src/entry.rs @@ -105,10 +105,12 @@ impl EntryResult { } } - fn has_uncacheable_deps(&self) -> bool { + /// The Run which may consume this result, if it is confined to one. + fn uncacheable_run_id(&self) -> Option { match self { - EntryResult::Uncacheable(_, _) | EntryResult::UncacheableDependencies(_, _) => true, - EntryResult::Clean(..) | EntryResult::Dirty(..) => false, + EntryResult::Uncacheable(_, run_id) + | EntryResult::UncacheableDependencies(_, run_id) => Some(*run_id), + EntryResult::Clean(..) | EntryResult::Dirty(..) => None, } } @@ -165,10 +167,11 @@ impl AsRef for EntryResult { } } +/// A Node result, its Generation, and the Run it is confined to, if any. pub type NodeResult = ( Result<::Item, ::Error>, Generation, - bool, + Option, ); #[derive(Debug)] @@ -550,10 +553,19 @@ impl Entry { .. } => { if let Some(receiver) = pending_value.receiver() { + let entry = self.clone(); + let context = context.clone(); return async move { - receiver.recv().await.unwrap_or_else(|| { - (Err(N::Error::invalidated()), generation.next(), true) - }) + let result = receiver.recv().await.unwrap_or_else(|| { + (Err(N::Error::invalidated()), generation.next(), None) + }); + match result { + (Ok(_), _, Some(run_id)) if run_id != context.run_id() => { + // We attached to another Run's attempt: re-enter for our own value. + entry.get_node_result(&context, entry_id).await + } + result => result, + } } .boxed(); } @@ -567,7 +579,7 @@ impl Entry { return future::ready(( Ok(result.as_ref().clone()), generation, - result.has_uncacheable_deps(), + result.uncacheable_run_id(), )) .boxed(); } @@ -644,7 +656,7 @@ impl Entry { receiver .recv() .await - .unwrap_or_else(|| (Err(N::Error::invalidated()), generation.next(), true)) + .unwrap_or_else(|| (Err(N::Error::invalidated()), generation.next(), None)) } .boxed() } @@ -732,7 +744,7 @@ impl Entry { previous_result.dirty(); } generation = generation.next(); - sender.send((Err(e), generation, true)); + sender.send((Err(e), generation, Some(context.run_id()))); EntryState::NotStarted { run_token, generation, @@ -751,7 +763,7 @@ impl Entry { sender.send(( Ok(next_result.as_ref().clone()), generation, - next_result.has_uncacheable_deps(), + next_result.uncacheable_run_id(), )); EntryState::Completed { result: next_result, @@ -774,7 +786,7 @@ impl Entry { sender.send(( Ok(result.as_ref().clone()), generation, - result.has_uncacheable_deps(), + result.uncacheable_run_id(), )); EntryState::Completed { result, @@ -937,7 +949,7 @@ impl Entry { let _ = pending_value.try_interrupt(NodeInterrupt::Aborted(( Err(err.clone()), generation.next(), - true, + None, ))); }; } diff --git a/src/rust/graph/src/lib.rs b/src/rust/graph/src/lib.rs index e9d4d46ccf2..0a0c2014241 100644 --- a/src/rust/graph/src/lib.rs +++ b/src/rust/graph/src/lib.rs @@ -547,7 +547,7 @@ impl Graph { }; if src_id.is_some() - && let Err(e) = context.dep_record(entry_id, generation, uncacheable) + && let Err(e) = context.dep_record(entry_id, generation, uncacheable.is_some()) { return (Err(e), generation); } @@ -653,7 +653,7 @@ impl Graph { // Cleaning succeeded. // // Return true if any dep was uncacheable. - Ok(uncacheable_deps.into_iter().any(|u| u)) + Ok(uncacheable_deps.into_iter().any(|u| u.is_some())) } Err(()) => { // Cleaning failed. From 10705eeb0cfcbbb720be57f2172519f1fad1e4fc Mon Sep 17 00:00:00 2001 From: Tobias Nilsson Date: Wed, 29 Jul 2026 20:20:59 +0200 Subject: [PATCH 2/6] test: Cover cross-run observation of in-flight uncacheable nodes --- src/rust/graph/src/tests.rs | 87 +++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) diff --git a/src/rust/graph/src/tests.rs b/src/rust/graph/src/tests.rs index d2256184be5..4f26930028b 100644 --- a/src/rust/graph/src/tests.rs +++ b/src/rust/graph/src/tests.rs @@ -12,6 +12,7 @@ use futures::future; use parking_lot::Mutex; use rand::RngExt; use task_executor::Executor; +use tokio::sync::watch; use tokio::time::{error::Elapsed, sleep, timeout}; use crate::context::Context; @@ -464,6 +465,54 @@ async fn uncacheable_deps_is_cleaned_for_the_session() { assert_no_change_within_session(&context); } +#[tokio::test] +async fn uncacheable_node_reruns_for_concurrent_run() { + let _logger = env_logger::try_init(); + let graph = empty_graph(); + + let mut uncacheable = HashSet::new(); + uncacheable.insert(TNode::new(1)); + + // Run A blocks in the uncacheable node so run B observes its attempt in flight. + let (gate, mut entered, release) = TGate::new(); + let context_a = { + let mut gates = HashMap::new(); + gates.insert(TNode::new(1), gate); + graph.context( + TContext::new() + .with_uncacheable(uncacheable.clone()) + .with_gates(gates), + ) + }; + let context_b = graph.context(TContext::new().with_uncacheable(uncacheable)); + + let graph_a = graph.clone(); + let context_a2 = context_a.clone(); + let fut_a = tokio::spawn(async move { graph_a.create(TNode::new(2), &context_a2).await }); + entered.wait_for(|entered| *entered).await.unwrap(); + + let graph_b = graph.clone(); + let context_b2 = context_b.clone(); + let fut_b = tokio::spawn(async move { graph_b.create(TNode::new(2), &context_b2).await }); + for _ in 0..10 { + tokio::task::yield_now().await; + } + // Yields are not synchronization: assert B is still pending, i.e. attached to A. + assert!(!fut_b.is_finished()); + release.send(true).unwrap(); + + let expected = vec![T(0, 0), T(1, 0), T(2, 0)]; + assert_eq!(fut_a.await.unwrap(), Ok(expected.clone())); + assert_eq!(fut_b.await.unwrap(), Ok(expected)); + + // B attached to A's attempt, but re-ran the uncacheable node rather than consuming it. + assert_eq!( + context_a.runs(), + vec![TNode::new(2), TNode::new(1), TNode::new(0)] + ); + assert_eq!(context_b.runs(), vec![TNode::new(1)]); +} + #[tokio::test] async fn live_runs_released_when_last_holder_drops() { let _logger = env_logger::try_init(); @@ -944,6 +993,7 @@ struct TContext { edges: Arc>>, delays_pre: Arc>, delays_post: Arc>, + gates: Arc>, // Nodes which should error when they run. errors: Arc>, non_restartable: Arc>, @@ -952,6 +1002,31 @@ struct TContext { runs: Arc>>, } +/// +/// A gate for a TNode: when the node runs, it signals `entered` and then blocks until +/// `release` becomes true. +/// +struct TGate { + entered: watch::Sender, + release: watch::Receiver, +} + +impl TGate { + /// Returns a TGate, a handle to await its entry, and a handle to release it. + fn new() -> (TGate, watch::Receiver, watch::Sender) { + let (entered_tx, entered_rx) = watch::channel(false); + let (release_tx, release_rx) = watch::channel(false); + ( + TGate { + entered: entered_tx, + release: release_rx, + }, + entered_rx, + release_tx, + ) + } +} + impl TContext { fn new() -> TContext { TContext { @@ -959,6 +1034,7 @@ impl TContext { edges: Arc::default(), delays_pre: Arc::default(), delays_post: Arc::default(), + gates: Arc::default(), errors: Arc::default(), non_restartable: Arc::default(), uncacheable: Arc::default(), @@ -984,6 +1060,12 @@ impl TContext { self } + /// Gates awaited before a node requests its dependencies. + fn with_gates(mut self, gates: HashMap) -> TContext { + self.gates = Arc::new(gates); + self + } + fn with_errors(mut self, errors: HashSet) -> TContext { self.errors = Arc::new(errors); self @@ -1030,6 +1112,11 @@ impl TContext { } async fn maybe_delay_pre(&self, node: &TNode) { + if let Some(gate) = self.gates.get(node) { + let _ = gate.entered.send(true); + let mut release = gate.release.clone(); + release.wait_for(|released| *released).await.unwrap(); + } if let Some(delay) = self.delays_pre.get(node) { sleep(*delay).await; } From a05469714dfbcaa239aa584406f6c1cb1f69dc7e Mon Sep 17 00:00:00 2001 From: Tobias Nilsson Date: Sun, 1 Nov 2026 18:14:41 +0100 Subject: [PATCH 3/6] refactor: Make NodeResult a struct Names the result, generation and run_scope fields. --- src/rust/graph/src/entry.rs | 85 +++++++++++++++++++++---------------- src/rust/graph/src/lib.rs | 27 +++++++----- 2 files changed, 65 insertions(+), 47 deletions(-) diff --git a/src/rust/graph/src/entry.rs b/src/rust/graph/src/entry.rs index a08c04042fc..e1087b43ff2 100644 --- a/src/rust/graph/src/entry.rs +++ b/src/rust/graph/src/entry.rs @@ -167,12 +167,31 @@ impl AsRef for EntryResult { } } -/// A Node result, its Generation, and the Run it is confined to, if any. -pub type NodeResult = ( - Result<::Item, ::Error>, - Generation, - Option, -); +#[derive(Clone, Debug)] +pub(crate) struct NodeResult { + pub(crate) result: Result, + pub(crate) generation: Generation, + /// The Run which produced this result, if it may only be consumed within that Run. + pub(crate) run_scope: Option, +} + +impl NodeResult { + fn completed(result: &EntryResult, generation: Generation) -> Self { + NodeResult { + result: Ok(result.as_ref().clone()), + generation, + run_scope: result.uncacheable_run_id(), + } + } + + fn invalidated(generation: Generation) -> Self { + NodeResult { + result: Err(N::Error::invalidated()), + generation: generation.next(), + run_scope: None, + } + } +} #[derive(Debug)] pub(crate) enum EntryState { @@ -471,7 +490,7 @@ impl Entry { match interrupt_item { Some(NodeInterrupt::Aborted(res)) => { // We were aborted via terminate: complete with the given res. - break (Some(res.0), context2.complete(entry2.node())) + break (Some(res.result), context2.complete(entry2.node())) } Some(NodeInterrupt::Dirtied) => { // Attempt to clean the Node, and cancel it if we fail. @@ -556,15 +575,18 @@ impl Entry { let entry = self.clone(); let context = context.clone(); return async move { - let result = receiver.recv().await.unwrap_or_else(|| { - (Err(N::Error::invalidated()), generation.next(), None) - }); - match result { - (Ok(_), _, Some(run_id)) if run_id != context.run_id() => { + let node_result = receiver + .recv() + .await + .unwrap_or_else(|| NodeResult::invalidated(generation)); + match node_result.run_scope { + Some(run_id) + if node_result.result.is_ok() && run_id != context.run_id() => + { // We attached to another Run's attempt: re-enter for our own value. entry.get_node_result(&context, entry_id).await } - result => result, + _ => node_result, } } .boxed(); @@ -576,12 +598,7 @@ impl Entry { generation, .. } if result.is_clean(context) => { - return future::ready(( - Ok(result.as_ref().clone()), - generation, - result.uncacheable_run_id(), - )) - .boxed(); + return future::ready(NodeResult::completed(result, generation)).boxed(); } _ => (), }; @@ -656,7 +673,7 @@ impl Entry { receiver .recv() .await - .unwrap_or_else(|| (Err(N::Error::invalidated()), generation.next(), None)) + .unwrap_or_else(|| NodeResult::invalidated(generation)) } .boxed() } @@ -744,7 +761,11 @@ impl Entry { previous_result.dirty(); } generation = generation.next(); - sender.send((Err(e), generation, Some(context.run_id()))); + sender.send(NodeResult { + result: Err(e), + generation, + run_scope: Some(context.run_id()), + }); EntryState::NotStarted { run_token, generation, @@ -760,11 +781,7 @@ impl Entry { // Node was re-executed (ie not cleaned) and had a different result value. generation = generation.next() }; - sender.send(( - Ok(next_result.as_ref().clone()), - generation, - next_result.uncacheable_run_id(), - )); + sender.send(NodeResult::completed(&next_result, generation)); EntryState::Completed { result: next_result, pollers: Vec::new(), @@ -783,11 +800,7 @@ impl Entry { self.cacheable_with_output(Some(result.as_ref())), has_uncacheable_deps, ); - sender.send(( - Ok(result.as_ref().clone()), - generation, - result.uncacheable_run_id(), - )); + sender.send(NodeResult::completed(&result, generation)); EntryState::Completed { result, pollers: Vec::new(), @@ -946,11 +959,11 @@ impl Entry { .. } = state { - let _ = pending_value.try_interrupt(NodeInterrupt::Aborted(( - Err(err.clone()), - generation.next(), - None, - ))); + let _ = pending_value.try_interrupt(NodeInterrupt::Aborted(NodeResult { + result: Err(err.clone()), + generation: generation.next(), + run_scope: None, + })); }; } } diff --git a/src/rust/graph/src/lib.rs b/src/rust/graph/src/lib.rs index 0a0c2014241..fc1d89e3097 100644 --- a/src/rust/graph/src/lib.rs +++ b/src/rust/graph/src/lib.rs @@ -5,7 +5,7 @@ mod context; mod entry; mod node; -use crate::entry::{Entry, Generation, RunToken}; +use crate::entry::{Entry, Generation, NodeResult, RunToken}; use std::collections::VecDeque; use std::fs::File; @@ -528,9 +528,11 @@ impl Graph { // Return the state of the destination, retrying the dst to handle Node invalidation. let context = context.clone(); - let (result, generation, uncacheable) = loop { + let node_result = loop { match entry.get_node_result(&context, entry_id).await { - (Err(err), _, _) if err == N::Error::invalidated() => { + NodeResult { + result: Err(err), .. + } if err == N::Error::invalidated() => { let node = { let inner = self.inner.lock(); inner.unsafe_entry_for_id(entry_id).node().clone() @@ -547,12 +549,16 @@ impl Graph { }; if src_id.is_some() - && let Err(e) = context.dep_record(entry_id, generation, uncacheable.is_some()) + && let Err(e) = context.dep_record( + entry_id, + node_result.generation, + node_result.run_scope.is_some(), + ) { - return (Err(e), generation); + return (Err(e), node_result.generation); } - (result, generation) + (node_result.result, node_result.generation) } /// @@ -625,11 +631,10 @@ impl Graph { .clone(); async move { - let (_, generation, uncacheable) = - dep_entry.get_node_result(context, dep_id).await; - if generation == previous_dep_generation { + let node_result = dep_entry.get_node_result(context, dep_id).await; + if node_result.generation == previous_dep_generation { // Matched. - Ok(uncacheable) + Ok(node_result.run_scope.is_some()) } else { // Did not match. We error here to trigger fail-fast in `try_join_all`. log::debug!( @@ -653,7 +658,7 @@ impl Graph { // Cleaning succeeded. // // Return true if any dep was uncacheable. - Ok(uncacheable_deps.into_iter().any(|u| u.is_some())) + Ok(uncacheable_deps.into_iter().any(|u| u)) } Err(()) => { // Cleaning failed. From 83833e67fea19847bffae131505b16dafd71de55 Mon Sep 17 00:00:00 2001 From: Tobias Nilsson Date: Sun, 1 Nov 2026 18:14:57 +0100 Subject: [PATCH 4/6] refactor: Pass only the error in NodeInterrupt::Aborted complete() sets the generation and run scope itself. --- src/rust/graph/src/entry.rs | 21 ++++++--------------- 1 file changed, 6 insertions(+), 15 deletions(-) diff --git a/src/rust/graph/src/entry.rs b/src/rust/graph/src/entry.rs index e1087b43ff2..97ef7815f3d 100644 --- a/src/rust/graph/src/entry.rs +++ b/src/rust/graph/src/entry.rs @@ -58,7 +58,7 @@ impl Generation { #[derive(Debug)] pub(crate) enum NodeInterrupt { Dirtied, - Aborted(NodeResult), + Aborted(N::Error), } /// @@ -488,9 +488,9 @@ impl Entry { tokio::select! { interrupt_item = sender.interrupted() => { match interrupt_item { - Some(NodeInterrupt::Aborted(res)) => { - // We were aborted via terminate: complete with the given res. - break (Some(res.result), context2.complete(entry2.node())) + Some(NodeInterrupt::Aborted(err)) => { + // We were aborted via terminate: complete with the given error. + break (Some(Err(err)), context2.complete(entry2.node())) } Some(NodeInterrupt::Dirtied) => { // Attempt to clean the Node, and cancel it if we fail. @@ -953,17 +953,8 @@ impl Entry { let inner = &mut *self.state.lock(); test_trace_log!("Terminating node {:?} with {:?}", self.node, err); for state in inner.states_mut() { - if let EntryState::Running { - pending_value, - generation, - .. - } = state - { - let _ = pending_value.try_interrupt(NodeInterrupt::Aborted(NodeResult { - result: Err(err.clone()), - generation: generation.next(), - run_scope: None, - })); + if let EntryState::Running { pending_value, .. } = state { + let _ = pending_value.try_interrupt(NodeInterrupt::Aborted(err.clone())); }; } } From d19d3f8ed887dbade76b44066cb7b6ade1e4aeab Mon Sep 17 00:00:00 2001 From: Tobias Nilsson Date: Sun, 1 Nov 2026 18:15:07 +0100 Subject: [PATCH 5/6] refactor: Rename uncacheable_run_id to run_scope --- src/rust/graph/src/entry.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/rust/graph/src/entry.rs b/src/rust/graph/src/entry.rs index 97ef7815f3d..397b99fcac5 100644 --- a/src/rust/graph/src/entry.rs +++ b/src/rust/graph/src/entry.rs @@ -105,8 +105,8 @@ impl EntryResult { } } - /// The Run which may consume this result, if it is confined to one. - fn uncacheable_run_id(&self) -> Option { + /// The Run which produced this result, if it may only be consumed within that Run. + fn run_scope(&self) -> Option { match self { EntryResult::Uncacheable(_, run_id) | EntryResult::UncacheableDependencies(_, run_id) => Some(*run_id), @@ -180,7 +180,7 @@ impl NodeResult { NodeResult { result: Ok(result.as_ref().clone()), generation, - run_scope: result.uncacheable_run_id(), + run_scope: result.run_scope(), } } From 4fe1ae7a6a05f30b4d81fad092c5bc595da8af10 Mon Sep 17 00:00:00 2001 From: Tobias Nilsson Date: Sun, 1 Nov 2026 18:22:53 +0100 Subject: [PATCH 6/6] test: Gate TNodes with oneshot channels --- src/rust/graph/src/tests.rs | 40 ++++++++++++++++++++++--------------- 1 file changed, 24 insertions(+), 16 deletions(-) diff --git a/src/rust/graph/src/tests.rs b/src/rust/graph/src/tests.rs index 4f26930028b..bcaeefea2b5 100644 --- a/src/rust/graph/src/tests.rs +++ b/src/rust/graph/src/tests.rs @@ -12,7 +12,7 @@ use futures::future; use parking_lot::Mutex; use rand::RngExt; use task_executor::Executor; -use tokio::sync::watch; +use tokio::sync::oneshot; use tokio::time::{error::Elapsed, sleep, timeout}; use crate::context::Context; @@ -474,7 +474,7 @@ async fn uncacheable_node_reruns_for_concurrent_run() { uncacheable.insert(TNode::new(1)); // Run A blocks in the uncacheable node so run B observes its attempt in flight. - let (gate, mut entered, release) = TGate::new(); + let (gate, entered, release) = TGate::new(); let context_a = { let mut gates = HashMap::new(); gates.insert(TNode::new(1), gate); @@ -489,7 +489,7 @@ async fn uncacheable_node_reruns_for_concurrent_run() { let graph_a = graph.clone(); let context_a2 = context_a.clone(); let fut_a = tokio::spawn(async move { graph_a.create(TNode::new(2), &context_a2).await }); - entered.wait_for(|entered| *entered).await.unwrap(); + entered.await.unwrap(); let graph_b = graph.clone(); let context_b2 = context_b.clone(); @@ -499,7 +499,7 @@ async fn uncacheable_node_reruns_for_concurrent_run() { } // Yields are not synchronization: assert B is still pending, i.e. attached to A. assert!(!fut_b.is_finished()); - release.send(true).unwrap(); + release.send(()).unwrap(); let expected = vec![T(0, 0), T(1, 0), T(2, 0)]; assert_eq!(fut_a.await.unwrap(), Ok(expected.clone())); @@ -1003,28 +1003,38 @@ struct TContext { } /// -/// A gate for a TNode: when the node runs, it signals `entered` and then blocks until -/// `release` becomes true. +/// A gate for a TNode: the first time the node runs, it signals `entered` and then blocks until +/// `release` fires. /// struct TGate { - entered: watch::Sender, - release: watch::Receiver, + entered: Mutex>>, + release: Mutex>>, } impl TGate { /// Returns a TGate, a handle to await its entry, and a handle to release it. - fn new() -> (TGate, watch::Receiver, watch::Sender) { - let (entered_tx, entered_rx) = watch::channel(false); - let (release_tx, release_rx) = watch::channel(false); + fn new() -> (TGate, oneshot::Receiver<()>, oneshot::Sender<()>) { + let (entered_tx, entered_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); ( TGate { - entered: entered_tx, - release: release_rx, + entered: Mutex::new(Some(entered_tx)), + release: Mutex::new(Some(release_rx)), }, entered_rx, release_tx, ) } + + async fn pass(&self) { + if let Some(entered) = self.entered.lock().take() { + let _ = entered.send(()); + } + let release = self.release.lock().take(); + if let Some(release) = release { + release.await.unwrap(); + } + } } impl TContext { @@ -1113,9 +1123,7 @@ impl TContext { async fn maybe_delay_pre(&self, node: &TNode) { if let Some(gate) = self.gates.get(node) { - let _ = gate.entered.send(true); - let mut release = gate.release.clone(); - release.wait_for(|released| *released).await.unwrap(); + gate.pass().await; } if let Some(delay) = self.delays_pre.get(node) { sleep(*delay).await;