vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Compare commits

base: vinzenz:7110a25cf6f0f9d863fbe27b6f9a4773fe3a9bc0
Branches Tags
vinzenz:main
...
compare: vinzenz:f48ba3ca0d428bd9525bfea8cf89fe20e541ae7a
Branches Tags
vinzenz:main
Author SHA1 Message Date
atlas
f48ba3ca0d build(#2772): declare enumflags2 at the workspace level 
`cargo add -p hive-jobq` wrote the version into the crate's own manifest
instead of `[workspace.dependencies]`, which is how every other shared
dep here is declared and the thing that stops two members drifting onto
different versions of the same crate.
 2026-07-27 19:06:27 +02:00
atlas
940c928fee refactor(#2772): enumflags2 for the edge set; drop re-export shims 
Review follow-ups on #2785.

`DepWhen` wraps `BitFlags<TerminalState>` instead of a hand-rolled `u8`,
so the bit manipulation belongs to the library and `TerminalState` gains
its flag value from `#[bitflags]` rather than a `bit()` match anyone
could get wrong. `of`/`accepts`/`is_empty` become one-liners over it.

Serialization is written out by hand rather than derived: clippy's
`unsafe_derive_deserialize` fires on deriving over a type with unsafe
internals, and the honest fix is to say what the wire form is. It is now
the list of accepted outcomes — `["done","failed"]` — which reads better
than a bitmask and survives the bits being renumbered.

Also drops the `pub use` re-export of `DepWhen` / `TerminalState` from
hive-c0re's `model`. It existed so that `use super::model::…` kept
compiling, which is a shim for one consumer's convenience; the sites
import from `hive_jobq` directly now.

And removes comments narrating what the code used to be. Git holds that.
 2026-07-27 19:06:27 +02:00
atlas
07078b76ef feat(#2772): branch on outcome in the graph, not inside the node 
Splits what was one `Cancelled` outcome into two, because they were two
different facts wearing one name:

- `Skipped` — the node's own edges ruled it out. Expected; the failure
  branch of a run that succeeded is `Skipped`. A parent's roll-up
  **ignores** it.
- `Cancelled` — the work was dropped before it could start. Still
  not-success for the roll-up, as before.

Without that split, branching on outcome defeats itself: exactly one
branch is always ruled out, `any_child_failed` counted it, and every DAG
containing a branch would have rolled up failed no matter how the run
went. Caught in review before it was written, not after.

`AFTER_ANY` becomes `{Done, Failed, Skipped}` — "anything except the work
being dropped". That is what it always meant; it only swept in
cancellation because cancellation wasn't distinguishable from
elimination. Audited every user rather than assuming, which is how the
one regression in my own proposal surfaced: `{Done, Failed}` would have
refused to run rebuild's recovery `Reconcile` after a failed `MetaSync`
(that eliminates `Prebuild`, so the tail's dep is `Skipped`, not
`Failed`) and left the container down.

With that, the templates stop computing outcomes and let the graph pick:

- `ResolveApproval { approval_id, outcome }` — one tail per outcome, each
  edged to accept only its own, so exactly one is ever runnable.
- `EmitRebuilt { agent, ok }` — a pair. `ok` is not derived, it is which
  of the two the graph let run.

Edges are conjunctive, so "any of these roots failed" is not directly
sayable. The composition: the success branch is `AFTER_OK` on every root
(so it is itself eliminated the moment one doesn't succeed), and the
failure branch keys off *that* elimination. The failure branch also
waits on every root — without it, a failed `Prebuild` eliminates the
success branch immediately and the failure would be announced while the
recovery `Reconcile` was still running. The tests caught that one.

Deletes, all of them #2770's host-side debt:

- `Claim.deps`, `DepOutcome`, `Claim::deps_state`, `Claim::deps_error`
  and the dep-snapshotting loop in `claim_ready`. Executors read their
  own variant now; nothing inspects anything.
- `NodeKind::is_tail()` and the `cancel` exemption built on it. Sparing
  is derived from the edges: `cancel` keeps a node iff one of its edges
  accepts `Cancelled`. An approval tail names it and survives to resolve
  the row; `Reconcile` doesn't and is cancelled with the rest. My earlier
  claim that this couldn't dissolve was only true while `AFTER_ANY`
  accepted cancellation.

`resolve_approval_dag` / `deploy_terminal_tag` now take `TerminalState`
rather than the wire `State`, so both matches are exhaustive instead of
ending in a catch-all.

Skipped nodes are filtered off the wire alongside `Done` ones. That costs
some dashboard detail on a failed rebuild — which steps were skipped —
and the tests say so with a pointer to the follow-up. Surfacing them as
`Cancelled` instead would be worse: the client roll-up ranks `Cancelled`
above `Running`, so a successful DAG with a not-taken branch would read
as cancelled.
 2026-07-27 19:06:27 +02:00
atlas
affedecaa5 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.
 2026-07-27 19:06:27 +02:00
11 changed files with 673 additions and 447 deletions
Download patch file Download diff file Expand all files Collapse all files

22
Cargo.lock generated
View file

@ -1116,6 +1116,27 @@ dependencies = [
"cfg-if",
]
[[package]]
name = "enumflags2"
version = "0.7.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1027f7680c853e056ebcec683615fb6fbbc07dbaa13b4d5d9442b146ded4ecef"
dependencies = [
"enumflags2_derive",
"serde",
]
[[package]]
name = "enumflags2_derive"
version = "0.7.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67c78a4d8fdf9953a5c9d458f9efe940fd97a0cab0941c075a813ac594733827"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "equivalent"
version = "1.0.2"
@ -1705,6 +1726,7 @@ name = "hive-jobq"
version = "0.1.0"
dependencies = [
"chrono",
"enumflags2",
"serde",
"serde_json",
"thiserror 2.0.18",

1
Cargo.toml
View file

@ -51,6 +51,7 @@ chrono = { version = "0.4", default-features = false, features = [
] }
clap = { version = "4", features = ["derive"] }
clap_complete = "4"
enumflags2 = { version = "0.7.12", features = ["serde"] }
indicatif = "0.18"
hive-sh4re = { path = "hive-sh4re" }
hive-agent-sock = { path = "hive-agent-sock" }

39
hive-c0re/src/actions.rs
View file

@ -507,20 +507,18 @@ async fn run_approval_schedule_prompt(
/// so the work's terminal state is the authoritative outcome and there's no
/// in-node resolution to skip around.
///
/// `state` / `error` come from the tail node's own dependency roll-up
/// ([`Claim::deps_state`] / [`Claim::deps_error`]), so this runs on the success,
/// failure **and cancel** paths alike.
/// `outcome` is the one the calling node was built to report — a template emits
/// one tail per outcome, so success, failure and cancel each arrive here from
/// their own node rather than from one node branching.
///
/// [`NodeKind::ResolveApproval`]: crate::job_queue::NodeKind::ResolveApproval
/// [`Claim::deps_state`]: crate::job_queue::Claim::deps_state
/// [`Claim::deps_error`]: crate::job_queue::Claim::deps_error
pub(crate) async fn resolve_approval_dag(
coord: &Arc<Coordinator>,
approval_id: i64,
state: crate::job_queue::State,
outcome: crate::job_queue::TerminalState,
error: Option<&str>,
) {
use crate::job_queue::State;
use crate::job_queue::TerminalState;
let approval = match coord.approvals.get(approval_id) {
Ok(Some(a)) => a,
Ok(None) => {
@ -532,10 +530,14 @@ pub(crate) async fn resolve_approval_dag(
return;
}
};
let result: Result<()> = match state {
State::Done => Ok(()),
State::Cancelled => Err(anyhow::anyhow!("cancelled before completion")),
_ => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))),
let result: Result<()> = match outcome {
TerminalState::Done => Ok(()),
// `Skipped` never reaches here — no template emits a tail for it — but it
// reads the same to an operator either way: the work did not happen.
TerminalState::Cancelled | TerminalState::Skipped => {
Err(anyhow::anyhow!("cancelled before completion"))
}
TerminalState::Failed => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))),
};
let mut terminal_tag = None;
match approval.kind {
@ -551,7 +553,7 @@ pub(crate) async fn resolve_approval_dag(
}
}
ApprovalKind::MergeConfigPr => {
terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, state).await;
terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, outcome).await;
// On a failed deploy, surface the failing build log back onto the
// PR so the manager sees why it was rejected without leaving the
// forge. Posted here rather than inside a node because this is the
@ -577,13 +579,14 @@ pub(crate) async fn resolve_approval_dag(
async fn deploy_terminal_tag(
agent: &str,
approval_id: i64,
state: crate::job_queue::State,
outcome: crate::job_queue::TerminalState,
) -> Option<String> {
use crate::job_queue::State;
let candidate = match state {
State::Done => format!("deployed/{approval_id}"),
State::Cancelled => return None,
_ => format!("failed/{approval_id}"),
use crate::job_queue::TerminalState;
let candidate = match outcome {
TerminalState::Done => format!("deployed/{approval_id}"),
// Nothing ran, so nothing was planted.
TerminalState::Cancelled | TerminalState::Skipped => return None,
TerminalState::Failed => format!("failed/{approval_id}"),
};
lifecycle::git_rev_parse(&crate::paths::applied_dir(agent), &candidate)
.await

82
hive-c0re/src/job_queue/exec.rs
View file

@ -11,7 +11,9 @@ use std::sync::Arc;
use anyhow::{Context as _, Result};
use super::Claim;
use super::model::{NodeKind, NodeSpec, State};
use hive_jobq::TerminalState;
use super::model::{NodeKind, NodeSpec};
use crate::coordinator::Coordinator;
use crate::power::{ReconcileAction, reconcile_action};
@ -95,10 +97,11 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await,
NodeKind::FinalizeDeploy { .. } => run_finalize_deploy(coord, claim).await,
NodeKind::DeployTail { .. } => run_deploy_tail(coord, claim).await,
NodeKind::ResolveApproval { approval_id, .. } => {
run_resolve_approval(coord, claim, *approval_id).await
}
NodeKind::EmitRebuilt { .. } => Ok(run_emit_rebuilt(coord, claim)),
NodeKind::ResolveApproval {
approval_id,
outcome,
} => run_resolve_approval(coord, claim, *approval_id, *outcome).await,
NodeKind::EmitRebuilt { ok, .. } => Ok(run_emit_rebuilt(coord, claim, *ok)),
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up),
// Pure grouping container — no work; completing it lets it reach
// `Finishing` so its child work nodes start. The DAG's terminal side
@ -107,58 +110,39 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
}
}
/// Resolve the DAG's approval row from how the work it follows ended. The
/// outcome comes off this node's own dependency roll-up, not from re-reading
/// the world. Best-effort: a resolution failure is logged inside
/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure —
/// the work already happened, and failing the tail would only misreport it.
/// Resolve the DAG's approval row the way this node's own `outcome` says.
///
/// Nothing is inspected: a template emits one of these per outcome, each edged to
/// accept only that one, so *which* node the scheduler let run already is the
/// answer. Best-effort — a resolution failure is logged inside
/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure,
/// since the work already happened and failing the tail would only misreport it.
async fn run_resolve_approval(
coord: &Arc<Coordinator>,
claim: &Claim,
approval_id: i64,
outcome: TerminalState,
) -> Result<NodeOutput> {
let reason = failure_reason(coord, claim);
crate::actions::resolve_approval_dag(coord, approval_id, claim.deps_state(), reason.as_deref())
.await;
let reason = (outcome == TerminalState::Failed)
.then(|| coord.job_queue.first_error(claim.dag_id))
.flatten();
crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await;
Ok(NodeOutput::default())
}
/// Why the work a tail node follows failed, as a human-readable string.
///
/// Prefers the tail's own dependency error, but a dep that is a **group root**
/// rolled up `Failed` from a child carries no error of its own (the reason lives
/// on the leaf that actually failed) — and a grafted subgraph's nodes can't be
/// edged statically anyway. So fall back to the DAG's first failing node. Still
/// the queue's own graph, not the outside world.
fn failure_reason(coord: &Arc<Coordinator>, claim: &Claim) -> Option<String> {
claim
.deps_error()
.map(str::to_owned)
.or_else(|| coord.job_queue.first_error(claim.dag_id))
}
/// Emit this agent's `Rebuilt` manager event — `ok` when the work it follows is
/// `Done`, `!ok` with the failure note when it `Failed`, and nothing at all when
/// it `Cancelled` (nothing ran, so there is no rebuild to report).
fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim) -> NodeOutput {
let agent = claim.agent.clone();
match claim.deps_state() {
State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent,
ok: true,
note: None,
sha: None,
tag: None,
}),
State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent,
ok: false,
note: failure_reason(coord, claim),
sha: None,
tag: None,
}),
_ => {}
}
/// Emit this agent's `Rebuilt` manager event. `ok` is not computed — it is which
/// of the tail pair the graph let run. The failure note comes from the DAG's
/// first failing node, since the branch knows *that* it failed but not *why*.
fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) -> NodeOutput {
coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: claim.agent.clone(),
ok,
note: (!ok)
.then(|| coord.job_queue.first_error(claim.dag_id))
.flatten(),
sha: None,
tag: None,
});
NodeOutput::default()
}

151
hive-c0re/src/job_queue/mod.rs
View file

@ -18,10 +18,9 @@
//! by a subtree root and borrowed by its descendants (continuity);
//! - per-DAG terminal work is an ordinary **tail node**
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
//! appends in [`templates`], weak-edged (`AfterAny`) onto the DAG's other group
//! roots so it runs on success, failure and cancel alike. It reads how the work
//! went off its own [`Claim::deps`] — no inline hook fired from outside the
//! graph, no drained event stream.
//! appends in [`templates`], edged onto the DAG's other group roots by the
//! outcome it reports. Templates emit one tail per outcome and the graph runs
//! exactly one, so nothing branches at runtime.
//!
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
//! state is re-derived by the reconcile sweep. A single scheduler task
@ -44,12 +43,13 @@ 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 hive_jobq::TerminalState;
pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State};
use resource::Resource;
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
@ -60,31 +60,6 @@ const MAX_HISTORY_DAGS: usize = 50;
/// Cap on stored node error strings.
const MAX_ERROR_LEN: usize = 2_000;
/// How one of a claimed node's dependencies finished, snapshotted at claim time.
///
/// A node only starts once its edges are satisfied, so every dep named here is
/// already terminal: an `AfterOk` edge means [`State::Done`], an `AfterAny` edge
/// means any of `Done` / `Failed` / `Cancelled`.
///
/// This is what lets a tail node be an ordinary node. A tail that must report
/// how the work below it went — "emit `Rebuilt { ok }`", "resolve this approval
/// with the failure note" — reads its deps' outcomes off its own claim, rather
/// than re-deriving them from the world after the fact. The alternative in this
/// codebase is `DeployTail`, which infers success by re-reading *git state*; that
/// works only because a deploy happens to write its result somewhere durable, and
/// it is not a pattern to copy.
///
/// Carries no node id: a tail acts on *how* its dependencies ended, never on
/// which one it was, so an id here would be a field with no reader.
#[derive(Debug, Clone)]
pub struct DepOutcome {
/// Its terminal state, in wire terms.
pub state: State,
/// Its failure reason, when it failed with one of its own. A node that
/// rolled up `Failed` from a child, or was cancelled, carries no error.
pub error: Option<String>,
}
/// A node claimed for execution — everything the executor needs, snapshotted at
/// claim time.
#[derive(Debug, Clone)]
@ -101,38 +76,6 @@ pub struct Claim {
/// pill is currently shown is derived from live lease ownership
/// ([`JobQueue::held_transients`]), not a per-claim edge.
pub transient: Option<TransientKind>,
/// How each node this one depends on finished. Empty for a head node.
/// See [`DepOutcome`] — this is how a tail node learns the outcome of the
/// work it follows without going back to the world to ask.
pub deps: Vec<DepOutcome>,
}
impl Claim {
/// Roll this claim's dependencies up into one outcome — what a weak-edged
/// tail node acts on. `Done` only when every dep succeeded; `Failed` when any
/// failed; otherwise `Cancelled` (all terminal, none failed, so the work was
/// dropped before it ran).
///
/// A head node has no deps and rolls up `Done` — vacuously true, and never
/// reached in practice since only tail kinds consult this.
pub fn deps_state(&self) -> State {
if self.deps.iter().all(|d| d.state == State::Done) {
State::Done
} else if self.deps.iter().any(|d| d.state == State::Failed) {
State::Failed
} else {
State::Cancelled
}
}
/// The first failed dependency's error, for reporting *why* the work ended
/// badly. `None` when nothing failed.
pub fn deps_error(&self) -> Option<&str> {
self.deps
.iter()
.find(|d| d.state == State::Failed)
.and_then(|d| d.error.as_deref())
}
}
/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle
@ -190,23 +133,21 @@ 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`).
///
/// `Skipped` has no wire counterpart and folds into `Cancelled`: to a reader
/// both mean "this never ran". The distinction is a *scheduling* one — it
/// decides whether a parent's roll-up counts the node — and the wire carries no
/// roll-up input, only display state. In practice a client never sees it either:
/// `dag_view` drops skipped nodes from the snapshot along with `Done` ones.
fn to_wire_state(state: JobState) -> State {
match state {
JobState::Pending => State::Queued,
JobState::Running | JobState::Finishing => State::Running,
JobState::Done => State::Done,
JobState::Failed => State::Failed,
JobState::Cancelled => State::Cancelled,
JobState::Cancelled | JobState::Skipped => State::Cancelled,
}
}
@ -241,7 +182,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 {
@ -373,24 +314,6 @@ impl JobQueue {
};
let kind = node.payload.clone();
let agent = node.payload.agent().to_owned();
let dep_ids: Vec<NodeId> = node
.deps
.iter()
.filter_map(|d| match d {
Dep::Node { id, .. } => Some(*id),
Dep::Resource { .. } => None,
})
.collect();
let deps: Vec<DepOutcome> = dep_ids
.into_iter()
.filter_map(|dep| {
let n = inner.sched.graph().node(dep)?;
Some(DepOutcome {
state: to_wire_state(n.state),
error: n.error.clone(),
})
})
.collect();
let Some(container) = inner.dag_of(id) else {
continue;
};
@ -405,7 +328,6 @@ impl JobQueue {
approval_id: meta.approval_id,
inputs: meta.inputs,
transient: meta.transient,
deps,
});
// `started_at` is stamped on the graph `Node` by the scheduler's
// transition to `Running` — no host-side copy needed.
@ -438,17 +360,17 @@ impl JobQueue {
/// so each is cancelled. `false` once any work node is running or terminal —
/// an in-flight nix build isn't interruptible.
///
/// **Tail nodes are spared** ([`NodeKind::is_tail`]). They are weak-edged
/// (`AfterAny`), and a `Cancelled` dep satisfies a weak edge, so sparing one
/// leaves it *ready* rather than stranded: the scheduler claims it on the next
/// pass, its [`Claim::deps_state`] reads `Cancelled`, and it resolves the
/// approval as "cancelled before completion". That is what stops a queued
/// approval DAG the operator cancelled from dangling its approval forever —
/// the job the inline hook used to do from outside the graph.
/// **Nodes that explicitly observe cancellation are spared** — a node whose
/// edge names [`hive_jobq::TerminalState::Cancelled`] is asking to run when
/// the work it follows was dropped, which is exactly what an approval tail
/// needs: cancel the work, and the tail still fires to resolve the approval
/// row rather than leaving it dangling forever.
///
/// Cancelling the tail too would be the bug: `cancel_node` only cascades along
/// `AfterOk` edges and parent links, so nothing else would reach it, and the
/// approval row would simply never be touched.
/// Nothing is special-cased by node kind. `AFTER_ANY` deliberately does *not*
/// accept `Cancelled`, so an ordinary weak-edged step (rebuild's `Reconcile`,
/// say) is cancelled along with everything else — there is nothing to converge
/// when no node ever ran. Only a node that named `Cancelled` survives, and it
/// survives because it asked to.
pub fn cancel(&self, dag_id: u64) -> bool {
let mut inner = self.lock();
let Some(container) = inner.container(dag_id) else {
@ -466,12 +388,7 @@ impl JobQueue {
return false;
}
for id in work {
if inner
.sched
.graph()
.node(id)
.is_some_and(|n| n.payload.is_tail())
{
if inner.observes_cancellation(id) {
continue;
}
inner.sched.cancel_node(id);
@ -650,6 +567,18 @@ impl QueueInner {
.is_some_and(|n| n.state.is_terminal())
}
/// Whether `id` has an edge that accepts a **dropped** dependency — i.e. the
/// node exists to report on work that may never run. Used by
/// [`JobQueue::cancel`] to decide what to spare, so the decision comes from
/// the node's own declared edges rather than a hardcoded list of kinds.
fn observes_cancellation(&self, id: NodeId) -> bool {
self.sched.graph().node(id).is_some_and(|n| {
n.deps.iter().any(|d| {
matches!(d, Dep::Node { when, .. } if when.accepts(hive_jobq::TerminalState::Cancelled))
})
})
}
/// First failed work node's error (read off the graph `Node`), for the
/// dashboard's DAG-level error line.
fn dag_first_error(&self, container: NodeId) -> Option<String> {
@ -690,7 +619,11 @@ impl QueueInner {
if let Some(f) = node.finished_at {
finished.push(f);
}
if node.state == JobState::Done {
// `Done` nodes drop off the wire (a finished step isn't interesting),
// and so do `Skipped` ones: a branch that was never taken is noise on
// the dashboard, and surfacing it would also drag the client-side
// roll-up toward `Cancelled` for a run that went fine.
if matches!(node.state, JobState::Done | JobState::Skipped) {
continue;
}
let deps: Vec<u64> = node

57
hive-c0re/src/job_queue/model.rs
View file

@ -17,18 +17,7 @@ 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,
}
use hive_jobq::{DepWhen, TerminalState};
/// A dependency edge (intra-DAG only — cross-DAG ordering comes from
/// the per-agent lease + dedup, never from edges between DAGs).
@ -230,7 +219,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
@ -240,18 +229,30 @@ pub enum NodeKind {
/// carrying one here would be a second copy free to drift. Like
/// [`NodeKind::MetaLock`] it reports `""` from [`NodeKind::agent`] and takes
/// no lease — which is also what lets one close a multi-agent DAG.
///
/// [`Claim::deps`]: super::Claim::deps
ResolveApproval { approval_id: i64 },
ResolveApproval {
approval_id: i64,
/// Which outcome this node reports. A template emits **one per outcome**,
/// each edged to accept only that one, so exactly one is ever runnable
/// and the executor has nothing to decide — it resolves the row the way
/// its own variant says. The `Cancelled` one is also the node that
/// [`super::JobQueue::cancel`] spares, since its edge is the only one
/// that accepts a dropped dependency.
outcome: TerminalState,
},
/// Tail node of a rebuild / perm-change: emit this agent's `Rebuilt` manager
/// event — `ok` when its deps are `Done`, `!ok` carrying the failure note when
/// they `Failed`, and **nothing at all** when they `Cancelled` (a cancelled DAG
/// never ran, so there is no rebuild to report).
///
/// One node **per agent**, unlike the DAG-wide hook it replaces: a multi-agent
/// DAG now reports each agent's own outcome instead of painting every agent with
/// the whole DAG's roll-up.
EmitRebuilt { agent: String },
/// One node per **agent** — a multi-agent DAG reports each agent's own
/// outcome rather than painting all of them with the whole DAG's roll-up —
/// and one per **outcome**: `ok` isn't computed here, it's which of the pair
/// the graph let run.
///
/// No cancel variant, deliberately: a DAG dropped before it started has no
/// rebuild to report, and neither tail's edge accepts `Cancelled`, so both
/// are cancelled with the rest and nothing is emitted.
EmitRebuilt { agent: String, ok: bool },
/// Write the agent's durable power intent (`wanted = Up` when `up`, else
/// `Offline`) as a first-class DAG node, at the head of a power-op
/// template so the downstream `Reconcile` reads it. Replaces the old
@ -340,7 +341,7 @@ impl NodeKind {
| NodeKind::DeployApply { agent }
| NodeKind::FinalizeDeploy { agent }
| NodeKind::DeployTail { agent }
| NodeKind::EmitRebuilt { agent }
| NodeKind::EmitRebuilt { agent, .. }
| NodeKind::SetWanted { agent, .. } => agent,
NodeKind::MetaLock { .. }
| NodeKind::Reparent { .. }
@ -349,20 +350,6 @@ impl NodeKind {
}
}
/// Whether this is a DAG's **tail** — a node that reports how the rest of the
/// DAG ended rather than doing work of its own.
///
/// The one place this matters is [`super::JobQueue::cancel`], which spares
/// tails so they still run (and report `Cancelled`) on a cancelled DAG. Note
/// [`NodeKind::DeployTail`] is *not* one: despite the name it does real
/// compensating work, and on a cancelled DAG there is nothing to compensate.
pub fn is_tail(&self) -> bool {
matches!(
self,
NodeKind::ResolveApproval { .. } | NodeKind::EmitRebuilt { .. }
)
}
/// Nix-heavy kinds hold one of the `buildSlots` semaphore permits
/// for the node's duration.
pub fn needs_build_slot(&self) -> bool {

164
hive-c0re/src/job_queue/templates.rs
View file

@ -30,7 +30,9 @@
use anyhow::{Result, bail};
use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source};
use hive_jobq::{DepWhen, TerminalState};
use super::model::{DagSpec, Dep, NodeKind, NodeSpec, PermPayload, Source};
use crate::coordinator::TransientKind;
/// After-ok edge on the previous node — the common chain link. Shared with
@ -39,27 +41,124 @@ use crate::coordinator::TransientKind;
pub(crate) fn after_ok(on: u64) -> Vec<Dep> {
vec![Dep {
on,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}]
}
/// Weak edges onto every one of a DAG's other **group-roots** — how a tail node
/// (`ResolveApproval` / `EmitRebuilt`) sees the whole DAG's outcome.
/// `AfterOk` edges onto every one of a DAG's **group-roots** — the success
/// branch of a per-outcome tail pair, and the aggregator the failure branch
/// keys off.
///
/// Group-roots are the right granularity, not "every node": a root's state *is*
/// its subtree's roll-up, so edging the roots covers every descendant while
/// keeping the tail's dep list small and stable as subtrees grow. `AfterAny`
/// throughout, so the tail runs on success, failure and cancel alike and decides
/// from [`super::Claim::deps_state`].
/// keeping the dep list small and stable as subtrees grow. Because every edge is
/// `AFTER_OK`, this node runs only if *all* of them succeeded — and is ruled out
/// ([`TerminalState::Skipped`]) the moment one doesn't, which is precisely the
/// signal [`on_elimination_of`] waits for.
pub(crate) fn after_ok_all(ons: &[u64]) -> Vec<Dep> {
ons.iter()
.map(|&on| Dep {
on,
when: DepWhen::AFTER_OK,
})
.collect()
}
/// `AFTER_ANY` edges onto every group-root — "wait for all of these to finish,
/// however they went". Ordering only; it accepts any outcome except the DAG
/// being dropped.
pub(crate) fn after_any_all(ons: &[u64]) -> Vec<Dep> {
ons.iter()
.map(|&on| Dep {
on,
when: DepWhen::AfterAny,
when: DepWhen::AFTER_ANY,
})
.collect()
}
/// A single edge satisfied only when `on` was **ruled out** by its own edges.
///
/// Dependency edges are conjunctive, so "any one of these several nodes failed"
/// cannot be written directly. This is the composition that expresses it: point
/// the success branch at every root with [`after_ok_all`], then hang the failure
/// branch off *that* node's elimination. Exactly one of the pair ever runs.
///
/// Note it accepts `Skipped` and **not** `Cancelled`: if the whole DAG was
/// dropped before it started, the success branch is marked `Cancelled` directly
/// and this branch is ruled out too — a job nobody ran reports nothing.
pub(crate) fn on_elimination_of(on: u64) -> Vec<Dep> {
vec![Dep {
on,
when: DepWhen::of(&[TerminalState::Skipped]),
}]
}
/// A single edge satisfied only by the listed outcomes of `on` — for the
/// one-tail-per-outcome shape an approval DAG uses.
pub(crate) fn on_outcome(on: u64, outcomes: &[TerminalState]) -> Vec<Dep> {
vec![Dep {
on,
when: DepWhen::of(outcomes),
}]
}
/// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node
/// gated on every group-root in `roots`, and the failure node gated on *its*
/// elimination. `base` is the spec index the pair starts at.
///
/// Exactly one runs on a DAG that executed, and neither runs on one the operator
/// dropped — see [`on_elimination_of`].
fn emit_rebuilt_tails(agent: &str, roots: &[u64], base: u64) -> Vec<NodeSpec> {
// The failure branch needs *both*: the ok branch being ruled out (that is the
// "something went wrong" signal) **and** every root actually finished. The
// second half is easy to forget and gets the ordering wrong without it — a
// failed `Prebuild` eliminates the ok branch immediately, while the recovery
// `Reconcile` is still bringing the container back up, so reporting straight
// off the elimination would announce the failure mid-recovery.
let mut on_fail = after_any_all(roots);
on_fail.extend(on_elimination_of(base));
vec![
node(
NodeKind::EmitRebuilt {
agent: agent.to_owned(),
ok: true,
},
after_ok_all(roots),
),
node(
NodeKind::EmitRebuilt {
agent: agent.to_owned(),
ok: false,
},
on_fail,
),
]
}
/// The approval-resolving tails for an approval-carrying DAG: one per outcome of
/// the DAG's single group-root `root`, each accepting only its own.
///
/// The `Cancelled` node is what keeps a dropped approval DAG from dangling its
/// row forever — its edge is the only one [`super::JobQueue::cancel`] spares.
fn resolve_approval_tails(approval_id: i64, root: u64) -> Vec<NodeSpec> {
[
TerminalState::Done,
TerminalState::Failed,
TerminalState::Cancelled,
]
.into_iter()
.map(|outcome| {
node(
NodeKind::ResolveApproval {
approval_id,
outcome,
},
on_outcome(root, &[outcome]),
)
})
.collect()
}
/// Build one **top-level (group-root)** node — `parent = None`. `kind` carries
/// the agent it targets ([`NodeKind`] is the payload directly). Shared with
/// `submit.rs`'s dynamic power-op builders. A root owns whatever resource it
@ -135,7 +234,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 +273,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,
},
],
));
@ -198,12 +297,7 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str) -> Vec<NodeSpec> {
/// it reaches `Done` even after a failed swap and the tail would report success.
pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec {
let mut nodes = rebuild_nodes(agent, relock, 0);
nodes.push(node(
NodeKind::EmitRebuilt {
agent: agent.to_owned(),
},
after_any_all(&[0, 1, 5]),
));
nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 5], 6));
DagSpec {
source,
reason,
@ -256,14 +350,13 @@ 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(
NodeKind::ResolveApproval { approval_id },
after_any_all(&[0]),
),
],
]
.into_iter()
.chain(resolve_approval_tails(approval_id, 0))
.collect(),
}
}
@ -319,11 +412,10 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec {
child(0, NodeKind::Create { agent: a() }, Vec::new()),
child(1, NodeKind::WriteDropin { agent: a() }, Vec::new()),
child(1, NodeKind::Reconcile { agent: a() }, after_ok(2)),
node(
NodeKind::ResolveApproval { approval_id },
after_any_all(&[0]),
),
]
.into_iter()
.chain(resolve_approval_tails(approval_id, 0))
.collect()
},
}
}
@ -342,12 +434,7 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay
Vec::new(),
)];
nodes.extend(rebuild_nodes(agent, true, 1));
nodes.push(node(
NodeKind::EmitRebuilt {
agent: agent.to_owned(),
},
after_any_all(&[0, 1, 2, 6]),
));
nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 2, 6], 7));
DagSpec {
source,
reason,
@ -382,14 +469,11 @@ pub fn meta_update(
Vec::new(),
)];
// The bump itself has no side effect, so an operator-driven one ends at the
// `MetaLock`; an approval-driven one still has its row to resolve and gets a
// tail edged onto that single group-root — whose roll-up covers the rebuild
// subgraphs `MetaLock` grows into itself.
// `MetaLock`; an approval-driven one still has its row to resolve and gets the
// per-outcome tails edged onto that single group-root — whose roll-up covers
// the rebuild subgraphs `MetaLock` grows into itself.
if let Some(approval_id) = approval_id {
nodes.push(node(
NodeKind::ResolveApproval { approval_id },
after_any_all(&[0]),
));
nodes.extend(resolve_approval_tails(approval_id, 0));
}
DagSpec {
source,

197
hive-c0re/src/job_queue/tests.rs
View file

@ -6,7 +6,9 @@
//! scheduler's async loop is a thin claim/complete pump over the same
//! methods exercised here.
use super::model::{Dep, DepWhen, NodeKind, NodeSpec};
use hive_jobq::DepWhen;
use super::model::{Dep, NodeKind, NodeSpec};
use super::*;
fn submit(q: &JobQueue, spec: DagSpec) -> u64 {
@ -48,35 +50,34 @@ fn claim_one(q: &JobQueue) -> Claim {
claims.pop().expect("one claim")
}
/// Claim an approval DAG's `ResolveApproval` tail, assert which approval it
/// carries and what it will report to that row, then complete it. Replaces the
/// old `terminal_summary()` assertions: the outcome is no longer a struct handed
/// to a hook, it's what this node reads off its own deps.
fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: State) {
/// Claim an approval DAG's `ResolveApproval` tail and complete it, asserting it
/// is the one built for `expect`.
///
/// A template emits one tail per outcome and the graph runs exactly one, so the
/// assertion is on *which node was claimed* — that alone says what the approval
/// row is about to be resolved as. Nothing computes it.
fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: TerminalState) {
let tail = claim_one(q);
assert!(
matches!(tail.kind, NodeKind::ResolveApproval { approval_id: got } if got == approval_id),
"expected the ResolveApproval tail for #{approval_id}, got {:?}",
matches!(
tail.kind,
NodeKind::ResolveApproval { approval_id: got, outcome }
if got == approval_id && outcome == expect
),
"expected the {expect:?} ResolveApproval tail for #{approval_id}, got {:?}",
tail.kind
);
assert_eq!(
tail.deps_state(),
expect,
"outcome the tail reports to approval #{approval_id}"
);
q.complete_node(dag_id, tail.node_id, Ok(()));
}
/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim a rebuild /
/// perm-change DAG's tail, assert the `Rebuilt` event it will emit, complete it.
fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect: State) {
/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim the tail the
/// graph let run and assert it's the `ok` one expected.
fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect_ok: bool) {
let tail = claim_one(q);
assert_eq!(tail.kind.as_str(), "emit_rebuilt");
assert_eq!(tail.agent, agent, "`Rebuilt` is emitted per agent");
assert_eq!(
tail.deps_state(),
expect,
"ok-ness of the emitted `Rebuilt`"
assert!(
matches!(&tail.kind, NodeKind::EmitRebuilt { agent: a, ok } if a == agent && *ok == expect_ok),
"expected the ok={expect_ok} EmitRebuilt tail for {agent}, got {:?}",
tail.kind
);
q.complete_node(dag_id, tail.node_id, Ok(()));
}
@ -154,7 +155,7 @@ fn cyclic_dag_is_rejected_at_submit() {
},
deps: vec![Dep {
on: 1,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}],
parent: None,
},
@ -164,7 +165,7 @@ fn cyclic_dag_is_rejected_at_submit() {
},
deps: vec![Dep {
on: 0,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}],
parent: None,
},
@ -183,7 +184,7 @@ fn unknown_dep_is_rejected_at_submit() {
},
deps: vec![Dep {
on: 9,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}],
parent: None,
}];
@ -229,7 +230,7 @@ fn rebuild_chain_claims_in_dep_order() {
);
q.complete_node(id, c.node_id, Ok(()));
}
settle_rebuild_tail(&q, id, "agent-a", State::Done);
settle_rebuild_tail(&q, id, "agent-a", true);
assert_eq!(state_of(&q, id), State::Done);
}
@ -736,73 +737,6 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() {
);
}
// ---- dep outcomes on the claim ----
/// Every claimed node reports how each node it depends on finished, and those
/// deps are always already terminal — that is what a node's edges being
/// satisfied *means*. A tail node reads its `deps` instead of going back to the
/// world to find out how the work below it went.
#[test]
fn claim_carries_terminal_dep_outcomes() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
let mut saw_a_dep = false;
loop {
let mut claims = q.claim_ready();
let Some(claim) = claims.pop() else { break };
assert!(claims.is_empty(), "one build slot ⇒ one claim at a time");
for dep in &claim.deps {
saw_a_dep = true;
assert!(
dep.state.is_terminal(),
"{} was claimed with a non-terminal dep ({:?}) — a node's edges \
being satisfied is exactly the claim that its deps have finished",
claim.kind.as_str(),
dep.state
);
assert_eq!(
dep.state,
State::Done,
"on the happy path every dep of {} finished Done",
claim.kind.as_str()
);
assert_eq!(dep.error, None, "a Done dep carries no error");
}
q.complete_node(id, claim.node_id, Ok(()));
}
assert!(saw_a_dep, "the rebuild DAG has at least one dependent node");
assert_eq!(state_of(&q, id), State::Done);
}
/// The failure direction, which is the whole point of carrying outcomes at all:
/// `Reconcile` hangs off `Prebuild` with `AfterAny`, so a failed prebuild
/// cancel-cascades `StopForUpdate`/`Swap`/`PostSwap` and `Reconcile` still runs
/// — and its claim hands it the failure, including the reason, rather than
/// leaving the executor to go and re-derive it from the world.
#[test]
fn claim_dep_outcome_reports_a_failed_dep_with_its_error() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
let meta_sync = claim_one(&q);
q.complete_node(id, meta_sync.node_id, Ok(()));
let prebuild = claim_one(&q);
assert_eq!(prebuild.kind.as_str(), "prebuild");
q.complete_node(id, prebuild.node_id, Err("nix build exploded".to_owned()));
let reconcile = claim_one(&q);
assert_eq!(reconcile.kind.as_str(), "reconcile");
assert_eq!(
reconcile
.deps
.iter()
.map(|d| (d.state, d.error.as_deref()))
.collect::<Vec<_>>(),
vec![(State::Failed, Some("nix build exploded"))],
"reconcile's claim carries the failed prebuild and its reason"
);
assert_eq!(reconcile.deps_state(), State::Failed);
assert_eq!(reconcile.deps_error(), Some("nix build exploded"));
}
// ---- failure: cancel-downstream + AfterAny ----
#[test]
@ -831,9 +765,19 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
.state
};
assert_eq!(by_kind("prebuild"), State::Failed);
assert_eq!(by_kind("stop_for_update"), State::Cancelled);
assert_eq!(by_kind("swap"), State::Cancelled);
assert_eq!(by_kind("post_swap"), State::Cancelled);
// `StopForUpdate` / `Swap` / `PostSwap` were *ruled out* by the failed
// `Prebuild` — `Skipped`, and skipped nodes are filtered off the wire along
// with `Done` ones. The failure itself is still visible (the `prebuild` row
// above, and the roll-up), which is the part an operator acts on.
// Restoring that detail wants a real `Skipped` wire state the client renders
// as "not run" — surfacing them as `Cancelled` instead would make a
// *successful* DAG with a not-taken branch read as cancelled.
for ruled_out in ["stop_for_update", "swap", "post_swap"] {
assert!(
dag.nodes.iter().all(|n| n.kind != ruled_out),
"{ruled_out} was ruled out, so it is off the wire"
);
}
// The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`,
// and `Done` nodes are excluded from the wire, so it's absent here.
assert!(
@ -872,14 +816,18 @@ fn swap_failure_still_runs_reconcile() {
q.complete_node(id, reconcile.node_id, Ok(()));
let all_dags = q.snapshot();
let dag = all_dags.iter().find(|d| d.id == id).expect("dag");
assert!(
dag.nodes.iter().all(|n| n.kind != "post_swap"),
"PostSwap is ruled out by the failed Swap (`Skipped`, so off the wire)"
);
assert_eq!(
dag.nodes
.iter()
.find(|n| n.kind == "post_swap")
.expect("post_swap node")
.find(|n| n.kind == "swap")
.expect("swap node")
.state,
State::Cancelled,
"PostSwap must cancel-cascade when Swap fails"
State::Failed,
"and the failure that ruled it out is still on the wire"
);
assert_eq!(state_of(&q, id), State::Failed);
}
@ -911,7 +859,7 @@ fn swap_ok_runs_post_swap_before_reconcile() {
let reconcile = claim_one(&q);
assert_eq!(reconcile.kind.as_str(), "reconcile");
q.complete_node(id, reconcile.node_id, Ok(()));
settle_rebuild_tail(&q, id, "agent-a", State::Done);
settle_rebuild_tail(&q, id, "agent-a", true);
assert_eq!(state_of(&q, id), State::Done);
}
@ -939,18 +887,14 @@ fn cancel_clears_queued_dag() {
// operator who just cancelled it (the dashboard renders this roll-up from
// the snapshot `post_rebuild_queue_cancel` emits synchronously).
assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap");
// Every work node is `Cancelled`, but the tail is spared so it can still
// report the cancellation — so it is the one thing left to claim.
let tail = claim_one(&q);
assert_eq!(tail.kind.as_str(), "emit_rebuilt");
assert_eq!(
tail.deps_state(),
State::Cancelled,
"the spared tail sees its deps cancelled, so it emits no `Rebuilt`"
// Neither `EmitRebuilt` tail accepts a *dropped* dependency — the ok one is
// `AFTER_OK`, the failure one keys on elimination — so both are cancelled
// with the work and **nothing is emitted** for a rebuild that never ran.
assert!(
q.claim_ready().is_empty(),
"a dropped rebuild reports nothing"
);
q.complete_node(id, tail.node_id, Ok(()));
assert_eq!(state_of(&q, id), State::Cancelled);
assert!(q.claim_ready().is_empty());
}
#[test]
@ -1065,21 +1009,10 @@ fn cancelled_dag_still_runs_its_approval_tail() {
templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
);
assert!(q.cancel(id), "fully-queued dag cancels");
let tail = claim_one(&q);
assert_eq!(tail.dag_id, id);
assert_eq!(tail.kind.as_str(), "resolve_approval");
assert!(
matches!(tail.kind, NodeKind::ResolveApproval { approval_id: 7 }),
"the tail carries the approval to resolve, got {:?}",
tail.kind
);
assert_eq!(
tail.deps_state(),
State::Cancelled,
"so the executor resolves the approval as cancelled, not as a failure"
);
assert_eq!(tail.deps_error(), None, "a cancelled dep carries no error");
q.complete_node(id, tail.node_id, Ok(()));
// The `Cancelled` tail is the only node whose edge accepts a dropped
// dependency, so it is the only one `cancel` spares — and claiming it *is*
// the assertion that the approval gets resolved as cancelled.
settle_approval_tail(&q, id, 7, TerminalState::Cancelled);
assert_eq!(state_of(&q, id), State::Cancelled);
// Unrelated later activity doesn't disturb the settled DAG.
let other = submit(&q, rebuild("agent-b", "r"));
@ -1130,7 +1063,7 @@ fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() {
);
q.complete_node(id, tail.node_id, Ok(()));
settle_approval_tail(&q, id, 7, State::Failed);
settle_approval_tail(&q, id, 7, TerminalState::Failed);
assert_eq!(
state_of(&q, id),
State::Failed,
@ -1203,7 +1136,7 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
assert!(matches!(tail.kind, NodeKind::DeployTail { .. }));
q.complete_node(id, tail.node_id, Ok(()));
settle_approval_tail(&q, id, 11, State::Done);
settle_approval_tail(&q, id, 11, TerminalState::Done);
assert_eq!(state_of(&q, id), State::Done);
}
@ -1254,7 +1187,7 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() {
);
q.complete_node(id, tail.node_id, Ok(()));
settle_approval_tail(&q, id, 13, State::Failed);
settle_approval_tail(&q, id, 13, TerminalState::Failed);
assert_eq!(state_of(&q, id), State::Failed);
assert_eq!(
q.first_error(id).as_deref(),
@ -1292,7 +1225,7 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() {
);
q.complete_node(id, tail.node_id, Ok(()));
settle_approval_tail(&q, id, 9, State::Failed);
settle_approval_tail(&q, id, 9, TerminalState::Failed);
assert_eq!(state_of(&q, id), State::Failed);
}
@ -1432,7 +1365,7 @@ fn spawn_shape_provision_create_dropin_reconcile() {
assert_eq!(c.approval_id, Some(7));
q.complete_node(id, c.node_id, Ok(()));
}
settle_approval_tail(&q, id, 7, State::Done);
settle_approval_tail(&q, id, 7, TerminalState::Done);
assert_eq!(state_of(&q, id), State::Done);
}
@ -1464,7 +1397,7 @@ fn perm_change_shape_prefixes_rebuild_chain() {
assert_eq!(c.kind.as_str(), expected);
q.complete_node(id, c.node_id, Ok(()));
}
settle_rebuild_tail(&q, id, "agent-a", State::Done);
settle_rebuild_tail(&q, id, "agent-a", true);
assert_eq!(state_of(&q, id), State::Done);
}

1
hive-jobq/Cargo.toml
View file

@ -9,6 +9,7 @@ workspace = true
[dependencies]
chrono = { workspace = true }
enumflags2.workspace = true
serde = { workspace = true }
thiserror = { workspace = true }

201
hive-jobq/src/lib.rs
View file

@ -56,29 +56,122 @@ 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.
#[enumflags2::bitflags]
#[repr(u8)]
#[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 because the work was **dropped** before it could start — the
/// caller cancelled the whole group while it was still queued. Counts as
/// not-success when a parent rolls up.
Cancelled,
/// Never ran because its own **edges ruled it out**: a dependency settled on
/// an outcome the edge doesn't accept. Expected, not a problem — the failure
/// branch of a run that succeeded is `Skipped`.
///
/// A parent's roll-up **ignores** `Skipped` children entirely. Without that,
/// branching on outcome would be self-defeating: exactly one branch is always
/// ruled out, so every group containing one would roll up failed.
Skipped,
}
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::Skipped => Some(TerminalState::Skipped),
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)]
pub struct DepWhen(enumflags2::BitFlags<TerminalState>);
/// Serialised as the list of outcomes it accepts (`["done","failed"]`) rather
/// than the underlying bitmask, so the wire form stays readable and survives the
/// bits being renumbered.
impl serde::Serialize for DepWhen {
fn serialize<S: serde::Serializer>(&self, ser: S) -> Result<S::Ok, S::Error> {
serde::Serialize::serialize(&self.0.iter().collect::<Vec<_>>(), ser)
}
}
impl<'de> serde::Deserialize<'de> for DepWhen {
fn deserialize<D: serde::Deserializer<'de>>(de: D) -> Result<Self, D::Error> {
let outcomes = Vec::<TerminalState>::deserialize(de)?;
Ok(Self(outcomes.into_iter().collect()))
}
}
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(enumflags2::make_bitflags!(TerminalState::{Done}));
/// Anything **except the work being dropped** — `Done`, `Failed` or
/// `Skipped`. For steps that must converge regardless of how the run went,
/// e.g. `Reconcile` bringing a container back up even when the preceding
/// `Swap` failed *or* was itself ruled out by a failed `MetaSync`.
///
/// Deliberately excludes [`TerminalState::Cancelled`]: if the group never
/// started at all there is nothing to converge, and running the recovery
/// step anyway would act on work that provably never happened. A node that
/// must report a cancellation names `Cancelled` explicitly.
pub const AFTER_ANY: Self =
Self(enumflags2::make_bitflags!(TerminalState::{Done | Failed | Skipped}));
/// An edge satisfied by exactly the listed outcomes.
#[must_use]
pub fn of(outcomes: &[TerminalState]) -> Self {
Self(outcomes.iter().copied().collect())
}
/// Whether `outcome` satisfies this edge.
#[must_use]
pub fn accepts(self, outcome: TerminalState) -> bool {
self.0.contains(outcome)
}
/// An edge no outcome can satisfy — rejected at [`Graph::validate`].
#[must_use]
pub fn is_empty(self) -> bool {
self.0.is_empty()
}
/// 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))
}
}
@ -128,18 +221,24 @@ pub enum State {
Done,
/// Completed unsuccessfully — own logic failed, or a sub-node did.
Failed,
/// Never ran: an `AfterOk` dependency failed, so this node (and the rest of
/// its strong-dependent chain) is cancelled rather than run.
/// Never ran: the work was dropped while still queued. See
/// [`TerminalState::Cancelled`].
Cancelled,
/// Never ran: its own edges ruled it out. See [`TerminalState::Skipped`] —
/// notably, a parent's roll-up ignores these.
Skipped,
}
impl State {
/// A node is *terminal* once it has finished — successfully, unsuccessfully,
/// or cancelled — which is when its resources are released and dependents
/// are re-evaluated.
/// dropped, or ruled out — which is when its resources are released and
/// dependents are re-evaluated.
#[must_use]
pub fn is_terminal(self) -> bool {
matches!(self, State::Done | State::Failed | State::Cancelled)
matches!(
self,
State::Done | State::Failed | State::Cancelled | State::Skipped
)
}
}
@ -207,6 +306,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 +536,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 +580,7 @@ mod tests {
"update",
vec![Dep::Node {
id: a,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}],
None,
)
@ -494,19 +609,25 @@ 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));
// 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_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));
assert!(!DepWhen::AFTER_OK.satisfied_by(State::Skipped));
// AfterAny: the dep reached a terminal state *some other way than being
// dropped* — success, failure, or ruled out by its own edges.
assert!(DepWhen::AFTER_ANY.satisfied_by(State::Done));
assert!(DepWhen::AFTER_ANY.satisfied_by(State::Failed));
assert!(DepWhen::AFTER_ANY.satisfied_by(State::Skipped));
assert!(
!DepWhen::AFTER_ANY.satisfied_by(State::Cancelled),
"a dropped dep does not converge a weak dependent — nothing ever ran"
);
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 +636,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 +668,7 @@ mod tests {
"b".to_owned(),
vec![Dep::Node {
id: a,
when: DepWhen::AfterAny,
when: DepWhen::AFTER_ANY,
}],
None,
)
@ -568,7 +689,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 +719,7 @@ mod tests {
fn after_ok_dep(on: NodeId) -> Dep<String> {
Dep::Node {
id: on,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}
}
@ -613,7 +734,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,

205
hive-jobq/src/scheduler.rs
View file

@ -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`].
///
@ -235,6 +235,9 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// Whether any direct child of `id` ended `Failed`/`Cancelled` — the roll-up
/// failure condition for the parent.
/// `Skipped` children are **not** counted: being ruled out by an edge is the
/// expected fate of every branch not taken, so counting it would make any
/// group that branches on outcome roll up failed no matter how the run went.
fn any_child_failed(&self, id: NodeId) -> bool {
self.graph
.nodes()
@ -244,6 +247,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 +262,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 +285,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 +324,33 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
.collect()
}
/// Propagate cancellation out from a just-failed/cancelled `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.
/// Propagate elimination out from a just-terminal `origin`: every
/// still-`Pending` node that can no longer run gets marked
/// [`State::Skipped`], transitively. Skipped nodes were `Pending`, so they
/// hold no resources.
///
/// `Skipped`, not `Cancelled`: these nodes were *ruled out by their edges*,
/// which is a normal outcome, not a dropped job. `Cancelled` is reserved for
/// work the caller abandoned before it started ([`Scheduler::cancel_node`]),
/// and the two are distinguished precisely so a parent's roll-up can ignore
/// the former while still treating the latter as not-success.
///
/// 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,13 +358,13 @@ 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)
.collect();
for d in doomed {
self.graph.set_state(d, State::Cancelled);
self.graph.set_state(d, State::Skipped);
stack.push(d);
}
}
@ -413,6 +438,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 +462,7 @@ mod tests {
fn after_ok(on: NodeId) -> Dep<String> {
Dep::Node {
id: on,
when: DepWhen::AfterOk,
when: DepWhen::AFTER_OK,
}
}
@ -489,10 +515,10 @@ mod tests {
assert_eq!(n.error.as_deref(), Some("boom"));
assert!(n.finished_at.is_some());
// The `AfterOk`-cancelled node: cascade-cancelled, so finished_at is set,
// The `AFTER_OK` dependent: ruled out by its edge, so finished_at is set,
// but it never ran (no started_at) and carries no error of its own.
let n = s.graph().node(downstream).unwrap();
assert_eq!(n.state, State::Cancelled);
assert_eq!(n.state, State::Skipped);
assert!(n.started_at.is_none());
assert!(n.finished_at.is_some());
assert_eq!(n.error, None);
@ -727,18 +753,148 @@ mod tests {
"weak",
vec![Dep::Node {
id: root,
when: DepWhen::AfterAny,
when: DepWhen::AFTER_ANY,
}],
None,
)
.expect("weak");
assert_eq!(s.settle(), vec![root]);
s.complete(root, Outcome::Failed(String::new()));
assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped);
assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped);
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::Skipped,
"a Failed-only branch is unsatisfiable once its dep succeeds — and it is \
`Skipped`, not `Cancelled`, so the parent roll-up ignores it"
);
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::Skipped);
assert_eq!(s.settle(), vec![on_fail]);
}
/// A weak edge accepts a dependency that was *ruled out*, so a tail still
/// runs when the work it reports on never happened — held by the edge itself
/// rather than by any node-kind special case.
#[test]
fn eliminated_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::Skipped);
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 ruled out the moment
/// one doesn't), and the failure branch hangs off *it* with `{Skipped}` —
/// "run when the success branch was ruled out". The success branch is the
/// aggregator, and exactly one of the two runs.
#[test]
fn ok_branch_aggregates_and_failure_branch_hangs_off_its_elimination() {
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::Skipped]),
}],
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::Skipped);
assert!(s.settle().is_empty());
// One of them fails: the ok branch is ruled out, 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::Skipped);
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
@ -749,8 +905,8 @@ mod tests {
let grandchild = s.append("gc", vec![], Some(child)).expect("gc");
assert_eq!(s.settle(), vec![root]);
s.complete(root, Outcome::Failed(String::new()));
assert_eq!(s.graph().node(child).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(grandchild).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(child).unwrap().state, State::Skipped);
assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped);
}
#[test]
@ -759,8 +915,9 @@ mod tests {
let a = s.append("a", vec![], None).expect("a");
let b = s.append("b", vec![after_ok(a)], None).expect("b");
assert!(s.cancel_node(a));
// `a` was dropped by the caller; `b` was merely ruled out by its edge.
assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(b).unwrap().state, State::Skipped);
let c = s.append("c", vec![], None).expect("c");
assert_eq!(s.settle(), vec![c]);
assert!(!s.cancel_node(c));