feat(#2772): make a dependency edge a set of accepted outcomes
`DepWhen` was two named cases, so every new combination wanted a new
variant. It is now a set over the terminal outcomes: a `u8` bitset
newtype, no dependency, with `AFTER_OK` / `AFTER_ANY` kept as the two
constants the templates actually use. "Run regardless" is all outcomes,
"anything that isn't a failure" is `{Done, Cancelled}`, a compensation
branch is `{Failed}` — closed under combination, so it never needs
another variant.
`TerminalState` is its own type rather than a subset of `State`, so an
edge cannot name `Pending` / `Running` / `Finishing`. Those are
meaningless in a dependency and are better unrepresentable than
validated against. The empty set is the one thing that can't be typed
away — nothing satisfies it, so `validate` rejects it next to the cycle
check.
Two consequences worth calling out:
- `cascade_cancel` collapses to one rule: a pending node is doomed once
any edge it names can no longer be satisfied. The hardcoded `AfterOk`
special case is gone, and a weak-edged node survives its dependency's
cancellation because of its own edge rather than by exemption.
- The cascade now runs on **any** terminal outcome, `Done` included.
With sets, success rules dependents out just as failure does — a
`{Failed}` branch is unsatisfiable the moment its dependency succeeds,
and leaving it `Pending` would wedge the subtree non-terminal forever.
That is a hang, not a wrong answer, so it is the load-bearing half of
this commit.
Edges are conjunctive, so "any of these N failed" is not directly
sayable. The composition that works is in the tests: the success branch
depends `AFTER_OK` on every root, so it is itself cancelled the moment
one of them doesn't succeed, and the failure branch hangs off *that*
with `{Cancelled}`. Exactly one of the two runs.
Also deletes hive-c0re's duplicate `DepWhen` enum and the
`to_crate_when` translation beside it. The copy bought nothing and had
to be widened in lockstep with the crate's edge model — it is the
in-between layer #2772 exists to remove, and it is what broke the build
when the crate's spelling changed.
All 34 jobq tests pass, including the four new ones covering both
directions of a failure-only branch, weak-edge survival of a cancelled
dependency, and the aggregator composition.
This commit is contained in:
atlas
mara
parent
7110a25cf6
commit
affedecaa5
6 changed files with 303 additions and 82 deletions
|
|
@ -44,12 +44,12 @@ use chrono::{DateTime, Utc};
|
|||
use hive_host_sock::jobs::NodeView;
|
||||
use hive_jobq::resources::ResourceTable;
|
||||
use hive_jobq::scheduler::{Outcome, Scheduler};
|
||||
use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState};
|
||||
use hive_jobq::{Dep, Graph, NodeId, State as JobState};
|
||||
use hive_sh4re::wire_time::now_unix;
|
||||
use tokio::sync::Notify;
|
||||
|
||||
use crate::coordinator::TransientKind;
|
||||
pub use model::{DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State};
|
||||
pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State};
|
||||
use resource::Resource;
|
||||
|
||||
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
|
||||
|
|
@ -190,14 +190,6 @@ impl Default for JobQueue {
|
|||
}
|
||||
}
|
||||
|
||||
/// Map a spec dependency edge kind onto the crate's.
|
||||
fn to_crate_when(when: DepWhen) -> JobDepWhen {
|
||||
match when {
|
||||
DepWhen::AfterOk => JobDepWhen::AfterOk,
|
||||
DepWhen::AfterAny => JobDepWhen::AfterAny,
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a crate node state onto the wire state (`Pending` ↔ `Queued`;
|
||||
/// `Finishing` — own logic done, sub-nodes still running — reads as `Running`).
|
||||
fn to_wire_state(state: JobState) -> State {
|
||||
|
|
@ -241,7 +233,7 @@ fn insert_group(
|
|||
for d in &ns.deps {
|
||||
deps.push(Dep::Node {
|
||||
id: ids[dep_index(d.on)],
|
||||
when: to_crate_when(d.when),
|
||||
when: d.when,
|
||||
});
|
||||
}
|
||||
let parent = match ns.parent {
|
||||
|
|
|
|||
|
|
@ -17,18 +17,11 @@ use serde::Serialize;
|
|||
|
||||
use crate::coordinator::TransientKind;
|
||||
|
||||
/// When a dependency edge is considered satisfied.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum DepWhen {
|
||||
/// Dep must reach `Done`. A `Failed` / `Cancelled` dep cancels this
|
||||
/// node (cancel-downstream).
|
||||
AfterOk,
|
||||
/// Dep must merely reach a terminal state (ok *or* fail). Used only
|
||||
/// by `rebuild`'s tail `Reconcile` so the recovery-start runs even
|
||||
/// when `Swap` failed.
|
||||
AfterAny,
|
||||
}
|
||||
/// When a dependency edge is satisfied — re-exported from [`hive_jobq`] rather
|
||||
/// than mirrored here. It used to be a duplicate enum with a `to_crate_when`
|
||||
/// translation beside it; the copy bought nothing and had to be widened in
|
||||
/// lockstep every time the crate's edge model grew (#2772).
|
||||
pub use hive_jobq::DepWhen;
|
||||
|
||||
/// A dependency edge (intra-DAG only — cross-DAG ordering comes from
|
||||
/// the per-agent lease + dedup, never from edges between DAGs).
|
||||
|
|
@ -230,7 +223,7 @@ pub enum NodeKind {
|
|||
/// Tail node of an approval-carrying DAG (spawn / opaque deploy / config-PR
|
||||
/// merge): resolve the approval row from how the work actually ended.
|
||||
///
|
||||
/// Weak-edged (`DepWhen::AfterAny`) like [`NodeKind::DeployTail`], so it runs on
|
||||
/// Weak-edged (`DepWhen::AFTER_ANY`) like [`NodeKind::DeployTail`], so it runs on
|
||||
/// success, failure **and cancel** alike and decides internally. It reads its
|
||||
/// dependencies' terminal states off its own [`Claim::deps`] rather than
|
||||
/// re-deriving them from the world the way `DeployTail` reads git: a node is
|
||||
|
|
|
|||
|
|
@ -39,7 +39,7 @@ use crate::coordinator::TransientKind;
|
|||
pub(crate) fn after_ok(on: u64) -> Vec<Dep> {
|
||||
vec![Dep {
|
||||
on,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}]
|
||||
}
|
||||
|
||||
|
|
@ -55,7 +55,7 @@ pub(crate) fn after_any_all(ons: &[u64]) -> Vec<Dep> {
|
|||
ons.iter()
|
||||
.map(|&on| Dep {
|
||||
on,
|
||||
when: DepWhen::AfterAny,
|
||||
when: DepWhen::AFTER_ANY,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
|
@ -135,7 +135,7 @@ pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u64) -> Vec<NodeSpe
|
|||
NodeKind::Reconcile { agent: a() },
|
||||
vec![Dep {
|
||||
on: base + 1,
|
||||
when: DepWhen::AfterAny,
|
||||
when: DepWhen::AFTER_ANY,
|
||||
}],
|
||||
),
|
||||
]
|
||||
|
|
@ -174,11 +174,11 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str) -> Vec<NodeSpec> {
|
|||
vec![
|
||||
Dep {
|
||||
on: 1,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
},
|
||||
Dep {
|
||||
on: 5,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
},
|
||||
],
|
||||
));
|
||||
|
|
@ -256,7 +256,7 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec
|
|||
NodeKind::DeployTail { agent: a() },
|
||||
vec![Dep {
|
||||
on: 2,
|
||||
when: DepWhen::AfterAny,
|
||||
when: DepWhen::AFTER_ANY,
|
||||
}],
|
||||
),
|
||||
node(
|
||||
|
|
|
|||
|
|
@ -154,7 +154,7 @@ fn cyclic_dag_is_rejected_at_submit() {
|
|||
},
|
||||
deps: vec![Dep {
|
||||
on: 1,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}],
|
||||
parent: None,
|
||||
},
|
||||
|
|
@ -164,7 +164,7 @@ fn cyclic_dag_is_rejected_at_submit() {
|
|||
},
|
||||
deps: vec![Dep {
|
||||
on: 0,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}],
|
||||
parent: None,
|
||||
},
|
||||
|
|
@ -183,7 +183,7 @@ fn unknown_dep_is_rejected_at_submit() {
|
|||
},
|
||||
deps: vec![Dep {
|
||||
on: 9,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}],
|
||||
parent: None,
|
||||
}];
|
||||
|
|
|
|||
|
|
@ -56,29 +56,105 @@ impl NodeId {
|
|||
}
|
||||
}
|
||||
|
||||
/// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the
|
||||
/// current queue carries as `DepWhen`, load-bearing for failure safety.
|
||||
/// How a node finished. The terminal subset of [`State`], as its own type so an
|
||||
/// edge condition cannot name `Pending` / `Running` / `Finishing` — those are
|
||||
/// meaningless in a dependency and are better unrepresentable than rejected.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||
pub enum DepWhen {
|
||||
/// The dependency must reach [`State::Done`]. This is the default chain
|
||||
/// edge: if the dependency *fails*, the dependent must not run and is
|
||||
/// cancelled ([`State::Cancelled`]) down the chain — e.g. a failed
|
||||
/// `Prebuild` must not let `StopForUpdate` stop a healthy container.
|
||||
AfterOk,
|
||||
/// The dependency need only be terminal — success or failure both satisfy
|
||||
/// it. For steps that must converge regardless, e.g. `Reconcile` running
|
||||
/// even when the preceding `Swap` failed.
|
||||
AfterAny,
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum TerminalState {
|
||||
/// Own logic succeeded and every sub-node did too.
|
||||
Done,
|
||||
/// Own logic failed, or a sub-node did.
|
||||
Failed,
|
||||
/// Never ran — an edge it depended on became unsatisfiable.
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
impl TerminalState {
|
||||
/// Bit for this outcome in a [`DepWhen`] set.
|
||||
const fn bit(self) -> u8 {
|
||||
match self {
|
||||
TerminalState::Done => 1,
|
||||
TerminalState::Failed => 1 << 1,
|
||||
TerminalState::Cancelled => 1 << 2,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl State {
|
||||
/// This state as a [`TerminalState`], or `None` while the node is still
|
||||
/// in flight.
|
||||
#[must_use]
|
||||
pub fn terminal(self) -> Option<TerminalState> {
|
||||
match self {
|
||||
State::Done => Some(TerminalState::Done),
|
||||
State::Failed => Some(TerminalState::Failed),
|
||||
State::Cancelled => Some(TerminalState::Cancelled),
|
||||
State::Pending | State::Running | State::Finishing => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Which outcomes of a dependency satisfy a [`Dep::Node`] edge — **a set**, not
|
||||
/// a fixed set of named cases.
|
||||
///
|
||||
/// Naming the cases (`AfterOk` / `AfterFail` / …) means a new variant every time
|
||||
/// a combination is wanted. A set is closed under combination: "run regardless"
|
||||
/// (systemd's `After=`) is all three; "anything that isn't a failure" is
|
||||
/// `{Done, Cancelled}`; a compensating branch is `{Failed}`. [`AFTER_OK`] and
|
||||
/// [`AFTER_ANY`] stay as named constants because they're the two the templates
|
||||
/// overwhelmingly use.
|
||||
///
|
||||
/// The empty set satisfies nothing, so a node carrying one could never run;
|
||||
/// [`Graph::validate`] rejects it.
|
||||
///
|
||||
/// [`AFTER_OK`]: DepWhen::AFTER_OK
|
||||
/// [`AFTER_ANY`]: DepWhen::AFTER_ANY
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||
pub struct DepWhen(u8);
|
||||
|
||||
impl DepWhen {
|
||||
/// Whether a dependency in `dep_state` satisfies this edge.
|
||||
/// The dependency must reach [`TerminalState::Done`]. The default chain
|
||||
/// edge: if the dependency fails, the dependent must not run and is
|
||||
/// cancelled down the chain — e.g. a failed `Prebuild` must not let
|
||||
/// `StopForUpdate` stop a healthy container.
|
||||
pub const AFTER_OK: Self = Self(TerminalState::Done.bit());
|
||||
/// The dependency need only be terminal — any outcome satisfies. For steps
|
||||
/// that must converge regardless, e.g. `Reconcile` running even when the
|
||||
/// preceding `Swap` failed, or a tail node that reports how the work ended.
|
||||
pub const AFTER_ANY: Self = Self(
|
||||
TerminalState::Done.bit() | TerminalState::Failed.bit() | TerminalState::Cancelled.bit(),
|
||||
);
|
||||
|
||||
/// An edge satisfied by exactly the listed outcomes.
|
||||
#[must_use]
|
||||
pub const fn of(outcomes: &[TerminalState]) -> Self {
|
||||
let mut bits = 0u8;
|
||||
let mut i = 0;
|
||||
while i < outcomes.len() {
|
||||
bits |= outcomes[i].bit();
|
||||
i += 1;
|
||||
}
|
||||
Self(bits)
|
||||
}
|
||||
|
||||
/// Whether `outcome` satisfies this edge.
|
||||
#[must_use]
|
||||
pub const fn accepts(self, outcome: TerminalState) -> bool {
|
||||
self.0 & outcome.bit() != 0
|
||||
}
|
||||
|
||||
/// An edge no outcome can satisfy — rejected at [`Graph::validate`].
|
||||
#[must_use]
|
||||
pub const fn is_empty(self) -> bool {
|
||||
self.0 == 0
|
||||
}
|
||||
|
||||
/// Whether a dependency in `dep_state` satisfies this edge. A non-terminal
|
||||
/// dependency never does.
|
||||
#[must_use]
|
||||
pub fn satisfied_by(self, dep_state: State) -> bool {
|
||||
match self {
|
||||
DepWhen::AfterOk => dep_state == State::Done,
|
||||
DepWhen::AfterAny => dep_state.is_terminal(),
|
||||
}
|
||||
dep_state.terminal().is_some_and(|t| self.accepts(t))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -207,6 +283,16 @@ pub enum GraphError {
|
|||
/// A node's `parent` named an id not present in the graph.
|
||||
#[error("parent references unknown node {0:?}")]
|
||||
UnknownParent(NodeId),
|
||||
/// A [`Dep::Node`] edge carries an empty [`DepWhen`] set, which no outcome
|
||||
/// can satisfy — the node could never run. Rejected at insert rather than
|
||||
/// left to wedge its subtree non-terminal at runtime.
|
||||
#[error("node {node:?} has an unsatisfiable dependency on {dep:?}: empty outcome set")]
|
||||
UnsatisfiableDep {
|
||||
/// The node that could never run.
|
||||
node: NodeId,
|
||||
/// The dependency whose edge accepts nothing.
|
||||
dep: NodeId,
|
||||
},
|
||||
/// A node's [`Dep::Node`] edge points outside its own parent group — the
|
||||
/// target must be a proper descendant of the depender's `parent` (a sibling
|
||||
/// or a sibling's sub-node), never the parent itself or a node in another
|
||||
|
|
@ -427,10 +513,16 @@ impl<N, R> Graph<N, R> {
|
|||
return Err(GraphError::UnknownParent(p));
|
||||
}
|
||||
for dep in &node.deps {
|
||||
if let Dep::Node { id, .. } = dep {
|
||||
if let Dep::Node { id, when } = dep {
|
||||
if self.node(*id).is_none() {
|
||||
return Err(GraphError::UnknownDep(*id));
|
||||
}
|
||||
if when.is_empty() {
|
||||
return Err(GraphError::UnsatisfiableDep {
|
||||
node: node.id,
|
||||
dep: *id,
|
||||
});
|
||||
}
|
||||
if !self.dep_target_in_group(node.parent, *id) {
|
||||
return Err(GraphError::DepOutsideParent {
|
||||
dep: *id,
|
||||
|
|
@ -465,7 +557,7 @@ mod tests {
|
|||
"update",
|
||||
vec![Dep::Node {
|
||||
id: a,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}],
|
||||
None,
|
||||
)
|
||||
|
|
@ -494,19 +586,19 @@ mod tests {
|
|||
fn after_ok_needs_success_after_any_needs_terminal() {
|
||||
// AfterOk: only Done satisfies; a Failed/Cancelled dep does NOT (the
|
||||
// dependent must be cancelled, not run).
|
||||
assert!(DepWhen::AfterOk.satisfied_by(State::Done));
|
||||
assert!(!DepWhen::AfterOk.satisfied_by(State::Failed));
|
||||
assert!(!DepWhen::AfterOk.satisfied_by(State::Cancelled));
|
||||
assert!(!DepWhen::AfterOk.satisfied_by(State::Running));
|
||||
assert!(DepWhen::AFTER_OK.satisfied_by(State::Done));
|
||||
assert!(!DepWhen::AFTER_OK.satisfied_by(State::Failed));
|
||||
assert!(!DepWhen::AFTER_OK.satisfied_by(State::Cancelled));
|
||||
assert!(!DepWhen::AFTER_OK.satisfied_by(State::Running));
|
||||
// AfterAny: any terminal state satisfies.
|
||||
assert!(DepWhen::AfterAny.satisfied_by(State::Done));
|
||||
assert!(DepWhen::AfterAny.satisfied_by(State::Failed));
|
||||
assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled));
|
||||
assert!(!DepWhen::AfterAny.satisfied_by(State::Pending));
|
||||
assert!(DepWhen::AFTER_ANY.satisfied_by(State::Done));
|
||||
assert!(DepWhen::AFTER_ANY.satisfied_by(State::Failed));
|
||||
assert!(DepWhen::AFTER_ANY.satisfied_by(State::Cancelled));
|
||||
assert!(!DepWhen::AFTER_ANY.satisfied_by(State::Pending));
|
||||
// Finishing satisfies neither — a dependent waits until the node rolls
|
||||
// up to a terminal state (all its sub-nodes done).
|
||||
assert!(!DepWhen::AfterOk.satisfied_by(State::Finishing));
|
||||
assert!(!DepWhen::AfterAny.satisfied_by(State::Finishing));
|
||||
assert!(!DepWhen::AFTER_OK.satisfied_by(State::Finishing));
|
||||
assert!(!DepWhen::AFTER_ANY.satisfied_by(State::Finishing));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -515,7 +607,7 @@ mod tests {
|
|||
let bogus = NodeId(42);
|
||||
let deps = vec![Dep::Node {
|
||||
id: bogus,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}];
|
||||
assert_eq!(
|
||||
g.insert("x", deps, None).unwrap_err(),
|
||||
|
|
@ -547,7 +639,7 @@ mod tests {
|
|||
"b".to_owned(),
|
||||
vec![Dep::Node {
|
||||
id: a,
|
||||
when: DepWhen::AfterAny,
|
||||
when: DepWhen::AFTER_ANY,
|
||||
}],
|
||||
None,
|
||||
)
|
||||
|
|
@ -568,7 +660,7 @@ mod tests {
|
|||
// roll-up model — the parent stays `Finishing` awaiting its children).
|
||||
let on_parent = vec![Dep::Node {
|
||||
id: root,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}];
|
||||
assert_eq!(
|
||||
g.insert("child", on_parent, Some(root)).unwrap_err(),
|
||||
|
|
@ -598,7 +690,7 @@ mod tests {
|
|||
fn after_ok_dep(on: NodeId) -> Dep<String> {
|
||||
Dep::Node {
|
||||
id: on,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -613,7 +705,7 @@ mod tests {
|
|||
payload: "x".to_owned(),
|
||||
deps: vec![Dep::Node {
|
||||
id: NodeId(99),
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}],
|
||||
state: State::Pending,
|
||||
started_at: None,
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ use std::collections::HashMap;
|
|||
use std::hash::Hash;
|
||||
|
||||
use crate::resources::ResourceTable;
|
||||
use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State};
|
||||
use crate::{Dep, Graph, GraphError, NodeId, State};
|
||||
|
||||
/// The result of a node's own execution, reported to [`Scheduler::complete`].
|
||||
///
|
||||
|
|
@ -244,6 +244,12 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
|
|||
/// Transition a node whose own logic just *succeeded* to its resulting state:
|
||||
/// [`State::Finishing`] while any child is still non-terminal, else `Failed`
|
||||
/// if a child failed, else `Done`. A node with no children skips `Finishing`.
|
||||
///
|
||||
/// Cascades on **any** terminal outcome, `Done` included. Since an edge names
|
||||
/// the set of outcomes it accepts, success can rule a dependent out just as
|
||||
/// failure can — a `{Failed}` compensation branch is unsatisfiable the moment
|
||||
/// its dependency succeeds, and leaving it `Pending` would wedge the subtree
|
||||
/// non-terminal forever.
|
||||
fn settle_terminal(&mut self, id: NodeId) {
|
||||
let state = if !self.all_children_terminal(id) {
|
||||
State::Finishing
|
||||
|
|
@ -253,7 +259,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
|
|||
State::Done
|
||||
};
|
||||
self.graph.set_state(id, state);
|
||||
if state == State::Failed {
|
||||
if state.is_terminal() {
|
||||
self.cascade_cancel(id);
|
||||
}
|
||||
}
|
||||
|
|
@ -276,14 +282,13 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
|
|||
State::Done
|
||||
};
|
||||
self.graph.set_state(a, state);
|
||||
if state == State::Failed {
|
||||
self.cascade_cancel(a);
|
||||
}
|
||||
// Any terminal outcome can rule a dependent out — see `settle_terminal`.
|
||||
self.cascade_cancel(a);
|
||||
cur = self.graph.node(a).and_then(|n| n.parent);
|
||||
}
|
||||
}
|
||||
|
||||
/// Cancel a still-*pending* node (and cascade to its `AfterOk` dependents):
|
||||
/// Cancel a still-*pending* node (and cascade to the dependents it rules out):
|
||||
/// mark it [`State::Cancelled`] and report whether it was cancellable. A
|
||||
/// node that has already started (`Running`) or finished is left untouched —
|
||||
/// an in-flight node's work is not interruptible. A pending node holds no
|
||||
|
|
@ -316,16 +321,26 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
|
|||
.collect()
|
||||
}
|
||||
|
||||
/// Propagate cancellation out from a just-failed/cancelled `origin`: every
|
||||
/// Propagate cancellation out from a just-terminal `origin`: every
|
||||
/// still-`Pending` node that can no longer run gets marked `Cancelled`,
|
||||
/// transitively. Two edges carry it: (a) an `AfterOk` dep on a cancelled node
|
||||
/// (a strong dependency failed), and (b) being a *child* of one (its parent
|
||||
/// will never reach `Finishing`, so it was gated from ever starting — and
|
||||
/// leaving it pending would wedge the subtree non-terminal). Cancelled nodes
|
||||
/// were `Pending`, so they hold no resources.
|
||||
/// transitively. Cancelled nodes were `Pending`, so they hold no resources.
|
||||
///
|
||||
/// One rule decides it: **a node is doomed once any edge it names can never
|
||||
/// be satisfied** — the dep settled on an outcome that edge does not accept.
|
||||
/// `AFTER_OK` on a `Failed` dep dooms (a strong dependency failed);
|
||||
/// `AFTER_ANY` never dooms, which is what lets a tail node survive the
|
||||
/// cancellation of the work it reports on. That falls out of the edge's own
|
||||
/// set rather than being a special case for particular node kinds.
|
||||
///
|
||||
/// Plus the structural edge: being a *child* of a doomed node. Its parent
|
||||
/// will never reach `Finishing`, so it was gated from ever starting, and
|
||||
/// leaving it pending would wedge the subtree non-terminal.
|
||||
fn cascade_cancel(&mut self, origin: NodeId) {
|
||||
let mut stack = vec![origin];
|
||||
while let Some(cur) = stack.pop() {
|
||||
let Some(outcome) = self.graph.node(cur).and_then(|n| n.state.terminal()) else {
|
||||
continue;
|
||||
};
|
||||
let doomed: Vec<NodeId> = self
|
||||
.graph
|
||||
.nodes()
|
||||
|
|
@ -333,7 +348,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
|
|||
n.state == State::Pending
|
||||
&& (n.parent == Some(cur)
|
||||
|| n.deps.iter().any(|d| {
|
||||
matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == cur)
|
||||
matches!(d, Dep::Node { id, when } if *id == cur && !when.accepts(outcome))
|
||||
}))
|
||||
})
|
||||
.map(|n| n.id)
|
||||
|
|
@ -413,6 +428,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{DepWhen, TerminalState};
|
||||
|
||||
fn res(name: &str) -> String {
|
||||
name.to_owned()
|
||||
|
|
@ -436,7 +452,7 @@ mod tests {
|
|||
fn after_ok(on: NodeId) -> Dep<String> {
|
||||
Dep::Node {
|
||||
id: on,
|
||||
when: DepWhen::AfterOk,
|
||||
when: DepWhen::AFTER_OK,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -727,7 +743,7 @@ mod tests {
|
|||
"weak",
|
||||
vec![Dep::Node {
|
||||
id: root,
|
||||
when: DepWhen::AfterAny,
|
||||
when: DepWhen::AFTER_ANY,
|
||||
}],
|
||||
None,
|
||||
)
|
||||
|
|
@ -739,6 +755,134 @@ mod tests {
|
|||
assert_eq!(s.settle(), vec![weak]);
|
||||
}
|
||||
|
||||
/// The direction only a *set* edge can express: a branch that runs solely on
|
||||
/// failure. Success has to rule it out, which means the cascade must fire on
|
||||
/// `Done` too — otherwise it sits `Pending` forever and wedges the graph.
|
||||
#[test]
|
||||
fn success_cancels_a_failure_only_branch() {
|
||||
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
||||
let root = s.append("root", vec![], None).expect("root");
|
||||
let on_fail = s
|
||||
.append(
|
||||
"compensate",
|
||||
vec![Dep::Node {
|
||||
id: root,
|
||||
when: DepWhen::of(&[TerminalState::Failed]),
|
||||
}],
|
||||
None,
|
||||
)
|
||||
.expect("compensate");
|
||||
assert_eq!(s.settle(), vec![root]);
|
||||
s.complete(root, Outcome::Done);
|
||||
assert_eq!(
|
||||
s.graph().node(on_fail).unwrap().state,
|
||||
State::Cancelled,
|
||||
"a Failed-only branch is unsatisfiable once its dep succeeds"
|
||||
);
|
||||
assert!(s.settle().is_empty(), "and nothing is left runnable");
|
||||
}
|
||||
|
||||
/// The mirror: the same branch is exactly what *does* run on failure, while
|
||||
/// an `AFTER_OK` sibling is cancelled. One edge set, both directions.
|
||||
#[test]
|
||||
fn failure_runs_the_failure_only_branch_and_cancels_the_ok_one() {
|
||||
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
||||
let root = s.append("root", vec![], None).expect("root");
|
||||
let on_ok = s
|
||||
.append("on_ok", vec![after_ok(root)], None)
|
||||
.expect("on_ok");
|
||||
let on_fail = s
|
||||
.append(
|
||||
"on_fail",
|
||||
vec![Dep::Node {
|
||||
id: root,
|
||||
when: DepWhen::of(&[TerminalState::Failed]),
|
||||
}],
|
||||
None,
|
||||
)
|
||||
.expect("on_fail");
|
||||
assert_eq!(s.settle(), vec![root]);
|
||||
s.complete(root, Outcome::Failed("boom".to_owned()));
|
||||
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Cancelled);
|
||||
assert_eq!(s.settle(), vec![on_fail]);
|
||||
}
|
||||
|
||||
/// A weak edge accepts cancellation, so a tail survives the cancellation of
|
||||
/// the work it reports on — the property hive-c0re's approval tails rely on,
|
||||
/// held here by the edge itself rather than by any node-kind special case.
|
||||
#[test]
|
||||
fn cancelled_dep_still_satisfies_a_weak_edge() {
|
||||
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
||||
let root = s.append("root", vec![], None).expect("root");
|
||||
let mid = s.append("mid", vec![after_ok(root)], None).expect("mid");
|
||||
let tail = s
|
||||
.append(
|
||||
"tail",
|
||||
vec![Dep::Node {
|
||||
id: mid,
|
||||
when: DepWhen::AFTER_ANY,
|
||||
}],
|
||||
None,
|
||||
)
|
||||
.expect("tail");
|
||||
assert_eq!(s.settle(), vec![root]);
|
||||
s.complete(root, Outcome::Failed("boom".to_owned()));
|
||||
assert_eq!(s.graph().node(mid).unwrap().state, State::Cancelled);
|
||||
assert_eq!(
|
||||
s.settle(),
|
||||
vec![tail],
|
||||
"the tail runs off a cancelled dependency"
|
||||
);
|
||||
}
|
||||
|
||||
/// Edges are **conjunctive**, so "any of these several nodes failed" is not
|
||||
/// directly expressible — a `{Failed}` edge on each would mean *all* failed.
|
||||
/// The composition that does work: the success branch depends `AFTER_OK` on
|
||||
/// every node (so it runs only if all succeeded, and is cancelled the moment
|
||||
/// one doesn't), and the failure branch hangs off *it* with `{Cancelled}`.
|
||||
/// The success branch becomes the aggregator, and exactly one of the two runs.
|
||||
#[test]
|
||||
fn ok_branch_aggregates_and_failure_branch_hangs_off_its_cancellation() {
|
||||
let build = || {
|
||||
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
|
||||
let a = s.append("a", vec![], None).expect("a");
|
||||
let b = s.append("b", vec![], None).expect("b");
|
||||
let on_ok = s
|
||||
.append("on_ok", vec![after_ok(a), after_ok(b)], None)
|
||||
.expect("on_ok");
|
||||
let on_fail = s
|
||||
.append(
|
||||
"on_fail",
|
||||
vec![Dep::Node {
|
||||
id: on_ok,
|
||||
when: DepWhen::of(&[TerminalState::Cancelled]),
|
||||
}],
|
||||
None,
|
||||
)
|
||||
.expect("on_fail");
|
||||
(s, a, b, on_ok, on_fail)
|
||||
};
|
||||
|
||||
// Everything succeeds: the ok branch runs, the failure branch is ruled out.
|
||||
let (mut s, a, b, on_ok, on_fail) = build();
|
||||
assert_eq!(s.settle(), vec![a, b], "both roots start; neither tail can");
|
||||
s.complete(a, Outcome::Done);
|
||||
s.complete(b, Outcome::Done);
|
||||
assert_eq!(s.settle(), vec![on_ok]);
|
||||
s.complete(on_ok, Outcome::Done);
|
||||
assert_eq!(s.graph().node(on_fail).unwrap().state, State::Cancelled);
|
||||
assert!(s.settle().is_empty());
|
||||
|
||||
// One of them fails: the ok branch is cancelled, which is precisely the
|
||||
// signal the failure branch waits on.
|
||||
let (mut s, a, b, on_ok, on_fail) = build();
|
||||
assert_eq!(s.settle(), vec![a, b]);
|
||||
s.complete(a, Outcome::Failed("boom".to_owned()));
|
||||
s.complete(b, Outcome::Done);
|
||||
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Cancelled);
|
||||
assert_eq!(s.settle(), vec![on_fail]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failed_parent_cancels_its_pending_children() {
|
||||
// A failed group node cancels its sub-nodes (they were gated from ever
|
||||
|
|
|
|||
Loading…
Reference in a new issue