a2edad715f
`Graph::insert` was public and validating, and the builder's insert loop called it with `?`. That made job-level atomicity an accident: the loop mutates as it goes, so a rejection at node `i` left `0..i` already in the graph — the error fired loudly *after* the corruption, not instead of it. It held only because `check_job_shape` happens to be exhaustive, with nothing in the types saying so. Make the guarantee structural instead. `Graph::insert` becomes `pub(crate)`; the builder drains into a new infallible `insert_unchecked` (via `Scheduler::append_unchecked`), so `insert_with`'s sink returns a bare `NodeId` and a half-built job is no longer expressible. Re-checking at the sink cannot add safety anyway — it can only report after the mutation it was meant to prevent. Drop `BuildError::Graph`: nothing in the builder path can produce a `GraphError` any more. Clippy could not see this (an unreachable variant of a `pub` enum is still constructible from outside the crate). `Scheduler::append` stays public and validating — hive-c0re inserts a DAG's container node through it. Folding that away means removing the container/DagView indirection, which is out of scope here.
967 lines
39 KiB
Rust
967 lines
39 KiB
Rust
//! `hive-jobq` — a persistent job-DAG scheduler, extracted from hive-c0re's
|
|
//! in-tree `job_queue` as a domain-agnostic library.
|
|
//!
|
|
//! # Model (v2)
|
|
//!
|
|
//! One **persistent graph** for the whole system, not a DAG per job. Enqueuing
|
|
//! inserts a self-contained sub-DAG of nodes and returns their ids; the
|
|
//! scheduler runs a continuous loop, starting every node whose [`Dep`]s are
|
|
//! satisfied:
|
|
//!
|
|
//! - **Resource** deps are named counting semaphores over a caller-chosen
|
|
//! type `R`: `build-slot` (cap N), `agent/<name>` (cap 1), or any name
|
|
//! (cap 1, created on use). A node acquires *all* its resource deps
|
|
//! atomically at start (all-or-nothing) — no hold-and-wait, no deadlock.
|
|
//! - **Node** deps wait on another node per [`DepWhen`]: `AfterOk` needs
|
|
//! success (a failed dep cancels the dependent), `AfterAny` only terminal.
|
|
//!
|
|
//! A node carries two independent axes: its [`Dep`]s (ordering + resource
|
|
//! needs) and its [`Node::parent`] (structural grouping) — the parent chain,
|
|
//! not the [`Dep::Node`] edges, is what the [`scheduler`] consults for resource
|
|
//! re-entrancy. A [`NodeId`] is opaque, stable, and monotonic (persisted). The
|
|
//! payload `N` is generic so the library stays container-agnostic.
|
|
//!
|
|
//! A resource unit is held for the acquiring node + its whole [`Node::parent`]
|
|
//! subtree; a node needing a resource an ancestor holds re-uses that grant (a
|
|
//! re-entrant borrow, one branch at a time). Single-threaded — the scheduler
|
|
//! owns the resource table and mutates it directly. See [`scheduler`].
|
|
|
|
pub mod builder;
|
|
pub mod resources;
|
|
pub mod scheduler;
|
|
|
|
pub use builder::{BuildError, JobBuilder, NodeGuid, NodeRef};
|
|
|
|
use chrono::{DateTime, Utc};
|
|
|
|
/// Opaque, stable, monotonic node identifier.
|
|
///
|
|
/// Assigned by the [`Graph`] on insert and persisted, so it is stable across
|
|
/// restarts.
|
|
///
|
|
/// The inner field is crate-private: an id can only originate from the graph's
|
|
/// monotonic counter (or deserialization of a persisted graph), never be
|
|
/// fabricated by a caller — that is what makes it opaque.
|
|
#[derive(
|
|
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
|
|
)]
|
|
pub struct NodeId(pub(crate) u64);
|
|
|
|
impl NodeId {
|
|
/// The underlying monotonic value, for carrying this id across a boundary
|
|
/// that cannot hold the opaque `NodeId` type — e.g. serializing it onto a
|
|
/// wire protocol. The inverse (fabricating a `NodeId` from a raw value)
|
|
/// stays impossible by construction: an id only ever originates from the
|
|
/// graph's counter, which is what makes it opaque.
|
|
#[must_use]
|
|
pub fn get(self) -> u64 {
|
|
self.0
|
|
}
|
|
}
|
|
|
|
/// 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)]
|
|
#[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 {
|
|
/// 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 {
|
|
dep_state.terminal().is_some_and(|t| self.accepts(t))
|
|
}
|
|
}
|
|
|
|
/// One dependency of a node. A node becomes runnable once every [`Dep::Node`]
|
|
/// edge it names is satisfied (per its [`DepWhen`]) *and* every [`Dep::Resource`]
|
|
/// it names can be acquired (all of them, atomically).
|
|
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
|
pub enum Dep<R> {
|
|
/// Depend on another node. Whether a *failed* dependency satisfies the edge
|
|
/// is decided by `when`: `AfterOk` requires success (and cancels this node
|
|
/// if the dep fails), `AfterAny` only requires the dep to be terminal.
|
|
Node {
|
|
/// The node depended on.
|
|
id: NodeId,
|
|
/// Strong (`AfterOk`) vs weak (`AfterAny`).
|
|
when: DepWhen,
|
|
},
|
|
/// Need `count` units of a named resource to run. Declared on every node
|
|
/// that needs it, even when a [`Node::parent`]-ancestor already holds it.
|
|
/// Acquired atomically with the node's other resource deps at start; the
|
|
/// acquired unit is held for the acquirer's whole subtree (released only
|
|
/// once the acquirer and all its sub-nodes are terminal). A node whose
|
|
/// parent-ancestor already holds this resource re-uses that grant (a
|
|
/// re-entrant borrow) instead of taking a fresh unit.
|
|
Resource {
|
|
/// The resource to acquire.
|
|
name: R,
|
|
/// How many units to hold (usually 1).
|
|
count: u32,
|
|
},
|
|
}
|
|
|
|
/// A node's lifecycle state.
|
|
///
|
|
/// This type *is* the wire representation — `hive-host-sock` hands it to
|
|
/// clients verbatim rather than mapping it through a parallel enum — so the
|
|
/// serialised names (`"Pending"`, `"Running"`, …) are what every consumer,
|
|
/// including the dashboard's JS, matches on.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
|
pub enum State {
|
|
/// Waiting on dependencies (node or resource).
|
|
Pending,
|
|
/// Dependencies satisfied, resources held, currently executing its own logic.
|
|
Running,
|
|
/// Own logic finished successfully, but the node is *not yet terminal*: it
|
|
/// waits here until all its sub-nodes ([`Node::parent`] children) are
|
|
/// terminal, then rolls up — `Failed` if any child failed, else `Cancelled`
|
|
/// if any was dropped, else `Done`. A node with no children never rests
|
|
/// here — it goes straight to a terminal state.
|
|
Finishing,
|
|
/// Completed successfully — own logic done *and* every sub-node `Done`.
|
|
Done,
|
|
/// Completed unsuccessfully — own logic failed, or a sub-node did.
|
|
Failed,
|
|
/// 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,
|
|
/// 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 | State::Skipped
|
|
)
|
|
}
|
|
}
|
|
|
|
/// Wall-clock UTC now — the source for node lifecycle timestamps
|
|
/// ([`Node::started_at`] / [`Node::finished_at`]). The graph stamps its own
|
|
/// timestamps rather than threading a clock through every call, so a node's
|
|
/// timing is self-contained. Derived from `SystemTime` (the workspace `chrono`
|
|
/// carries no `clock` feature, matching `hive_sh4re::wire_time`), truncated to
|
|
/// whole seconds; a pre-epoch or out-of-range clock clamps to the epoch.
|
|
fn now_utc() -> DateTime<Utc> {
|
|
let secs = std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
|
|
DateTime::<Utc>::from_timestamp(secs, 0).unwrap_or_default()
|
|
}
|
|
|
|
/// A node in the graph, carrying a caller-defined payload `N`.
|
|
///
|
|
/// The library schedules over `Node`s and resources without interpreting the
|
|
/// payload; the caller supplies `N` (its own node kind) and a runner to execute
|
|
/// a claimed node. A node carries two independent axes: its [`Dep`]s (ordering +
|
|
/// resource needs) and its [`Node::parent`] (structural grouping), both set by
|
|
/// the caller/submit layer.
|
|
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
|
pub struct Node<N, R> {
|
|
/// Stable identity, assigned on insert.
|
|
pub id: NodeId,
|
|
/// Structural grouping: the node this one is a sub-node of, or `None` for a
|
|
/// group root. Independent of [`Node::deps`] — grouping is *not* ordering.
|
|
/// The [`scheduler`] uses the parent chain to decide resource re-entrancy: a
|
|
/// node needing a resource a parent-ancestor holds re-uses that grant rather
|
|
/// than acquiring a fresh unit, and a held unit stays reserved for the
|
|
/// acquirer's whole subtree. A node's sub-nodes run *after* its own logic.
|
|
pub parent: Option<NodeId>,
|
|
/// Caller-defined payload (the node's kind / work description).
|
|
pub payload: N,
|
|
/// What must hold before this node runs (other nodes + resources).
|
|
pub deps: Vec<Dep<R>>,
|
|
/// Lifecycle state.
|
|
pub state: State,
|
|
/// UTC instant the node entered [`State::Running`] (`None` until it starts;
|
|
/// a cancelled node never ran, so it stays `None`). Stamped by the graph.
|
|
pub started_at: Option<DateTime<Utc>>,
|
|
/// UTC instant the node reached a terminal state (`Done` / `Failed` /
|
|
/// `Cancelled`). `None` while non-terminal. Stamped by the graph.
|
|
pub finished_at: Option<DateTime<Utc>>,
|
|
/// Failure reason for a `Failed` node, supplied by the runner via
|
|
/// [`scheduler::Outcome::Failed`]. `None` unless this node's own logic
|
|
/// failed (a node that rolled up `Failed` from a child, or was cancelled,
|
|
/// carries no error of its own).
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
/// An error from inserting into or loading a [`Graph`] with a dangling id.
|
|
///
|
|
/// A [`NodeId`] is only meaningful against the graph that minted it, so both
|
|
/// entry points — [`Graph::insert`] and deserialization — reject references to
|
|
/// nodes the graph does not contain. That is what lets internal iteration trust
|
|
/// every id the graph holds.
|
|
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
|
|
pub enum GraphError {
|
|
/// A node's dependency named an id not present in the graph.
|
|
#[error("dependency references unknown node {0:?}")]
|
|
UnknownDep(NodeId),
|
|
/// 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
|
|
/// group. Top-level nodes may only depend on top-level nodes.
|
|
#[error("dependency {dep:?} is outside the depender's parent group {parent:?}")]
|
|
DepOutsideParent {
|
|
/// The out-of-group dependency target.
|
|
dep: NodeId,
|
|
/// The depender's parent (the group the target had to be inside).
|
|
parent: Option<NodeId>,
|
|
},
|
|
/// A loaded graph's `next_id` counter is not past the largest existing id,
|
|
/// so the next minted id would collide with one already in the graph.
|
|
#[error("next_id {next_id} must exceed the largest existing node id {max_id}")]
|
|
NextIdTooSmall {
|
|
/// The persisted counter value.
|
|
next_id: u64,
|
|
/// The largest id already present.
|
|
max_id: u64,
|
|
},
|
|
}
|
|
|
|
/// The single persistent graph of all nodes.
|
|
///
|
|
/// New jobs are inserted as sub-DAGs of nodes; the scheduler walks this graph
|
|
/// filling open slots. Completed nodes are retained (no pruning in v1).
|
|
#[derive(Debug, serde::Serialize, serde::Deserialize)]
|
|
#[serde(
|
|
try_from = "GraphData<N, R>",
|
|
bound(
|
|
serialize = "N: serde::Serialize, R: serde::Serialize",
|
|
deserialize = "N: serde::Deserialize<'de>, R: serde::Deserialize<'de>"
|
|
)
|
|
)]
|
|
pub struct Graph<N, R> {
|
|
nodes: Vec<Node<N, R>>,
|
|
next_id: u64,
|
|
}
|
|
|
|
// Deserialization target: the raw fields, turned into a `Graph` by the `TryFrom`
|
|
// below — which runs [`Graph::validate`], so a loaded graph can never carry a
|
|
// dangling id reference (Serialize does not validate; Deserialize always does).
|
|
#[derive(serde::Deserialize)]
|
|
#[serde(bound(deserialize = "N: serde::Deserialize<'de>, R: serde::Deserialize<'de>"))]
|
|
struct GraphData<N, R> {
|
|
nodes: Vec<Node<N, R>>,
|
|
next_id: u64,
|
|
}
|
|
|
|
impl<N, R> TryFrom<GraphData<N, R>> for Graph<N, R> {
|
|
type Error = GraphError;
|
|
|
|
fn try_from(data: GraphData<N, R>) -> Result<Self, Self::Error> {
|
|
let graph = Graph {
|
|
nodes: data.nodes,
|
|
next_id: data.next_id,
|
|
};
|
|
graph.validate()?;
|
|
Ok(graph)
|
|
}
|
|
}
|
|
|
|
// A `derive(Default)` would wrongly require `N: Default` (an empty graph holds
|
|
// no payload); an empty `Vec<Node<N>>` needs no such bound, so impl it directly.
|
|
impl<N, R> Default for Graph<N, R> {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl<N, R> Graph<N, R> {
|
|
/// An empty graph.
|
|
#[must_use]
|
|
pub fn new() -> Self {
|
|
Self {
|
|
nodes: Vec::new(),
|
|
next_id: 0,
|
|
}
|
|
}
|
|
|
|
/// Mint the next stable node id.
|
|
fn mint_id(&mut self) -> NodeId {
|
|
let id = NodeId(self.next_id);
|
|
self.next_id += 1;
|
|
id
|
|
}
|
|
|
|
/// Insert a node with the given payload, deps, and `parent`, returning its
|
|
/// freshly-minted id. The node starts [`State::Pending`].
|
|
///
|
|
/// Every [`Dep::Node`] id and the `parent` id (when `Some`) must already
|
|
/// resolve to a node in the graph — an id is only meaningful against the
|
|
/// graph that minted it, so a dangling reference is rejected here rather than
|
|
/// surfacing as a broken edge later.
|
|
///
|
|
/// **Crate-private on purpose.** Nodes enter a graph only through
|
|
/// [`crate::builder::JobBuilder`] (via
|
|
/// [`crate::scheduler::Scheduler::insert_job`]) or
|
|
/// [`crate::scheduler::Scheduler::append`]. A second public way in would be
|
|
/// a second place validation has to agree with the builder's, and the two
|
|
/// would drift.
|
|
///
|
|
/// # Errors
|
|
/// Returns [`GraphError::UnknownDep`] / [`GraphError::UnknownParent`] for a
|
|
/// dangling dependency or parent id, or [`GraphError::DepOutsideParent`] if a
|
|
/// `Dep::Node` edge points outside the node's own parent group.
|
|
pub(crate) fn insert(
|
|
&mut self,
|
|
payload: N,
|
|
deps: Vec<Dep<R>>,
|
|
parent: Option<NodeId>,
|
|
) -> Result<NodeId, GraphError> {
|
|
if let Some(p) = parent
|
|
&& self.node(p).is_none()
|
|
{
|
|
return Err(GraphError::UnknownParent(p));
|
|
}
|
|
for dep in &deps {
|
|
if let Dep::Node { id, .. } = dep {
|
|
if self.node(*id).is_none() {
|
|
return Err(GraphError::UnknownDep(*id));
|
|
}
|
|
if !self.dep_target_in_group(parent, *id) {
|
|
return Err(GraphError::DepOutsideParent { dep: *id, parent });
|
|
}
|
|
}
|
|
}
|
|
Ok(self.insert_unchecked(payload, deps, parent))
|
|
}
|
|
|
|
/// Insert without re-validating — **only** for a node the builder has
|
|
/// already proved well-formed.
|
|
///
|
|
/// [`crate::builder::check_job_shape`] decides every rejection
|
|
/// [`Graph::insert`] could raise, before the first node lands. Re-checking
|
|
/// here would not add safety: the insert loop mutates as it goes, so a
|
|
/// rejection at node `i` would leave `0..i` in the graph — a loud error
|
|
/// *after* the corruption rather than instead of it. Making the sink
|
|
/// infallible is what turns the builder's atomicity from an accident of the
|
|
/// pre-pass being exhaustive into a property of the types.
|
|
pub(crate) fn insert_unchecked(
|
|
&mut self,
|
|
payload: N,
|
|
deps: Vec<Dep<R>>,
|
|
parent: Option<NodeId>,
|
|
) -> NodeId {
|
|
let id = self.mint_id();
|
|
self.nodes.push(Node {
|
|
id,
|
|
parent,
|
|
payload,
|
|
deps,
|
|
state: State::Pending,
|
|
started_at: None,
|
|
finished_at: None,
|
|
error: None,
|
|
});
|
|
id
|
|
}
|
|
|
|
/// Borrow a node by id.
|
|
#[must_use]
|
|
pub fn node(&self, id: NodeId) -> Option<&Node<N, R>> {
|
|
self.nodes.iter().find(|n| n.id == id)
|
|
}
|
|
|
|
/// Resolve a raw value back to the opaque [`NodeId`] it names — the inverse
|
|
/// of [`NodeId::get`], and the only way to perform that direction. A caller
|
|
/// holding a value that crossed a wire cannot fabricate an id from it (that
|
|
/// impossibility is the point of the type), so it has to be matched against
|
|
/// the graph, which is what makes this a search rather than a cast.
|
|
///
|
|
/// `None` when no node carries that value, which covers both a value that
|
|
/// was never an id and one whose node has since been reaped.
|
|
#[must_use]
|
|
pub fn resolve_id(&self, raw: u64) -> Option<NodeId> {
|
|
self.nodes
|
|
.iter()
|
|
.find_map(|n| (n.id.0 == raw).then_some(n.id))
|
|
}
|
|
|
|
/// Every node in the graph, in insertion order. The scheduler iterates
|
|
/// this to find runnable pending nodes.
|
|
pub fn nodes(&self) -> impl Iterator<Item = &Node<N, R>> {
|
|
self.nodes.iter()
|
|
}
|
|
|
|
/// Top of `id`'s [`Node::parent`] chain — the group root whose subtree `id`
|
|
/// lives in. Returns `id` itself when `id` is already a root, and `None`
|
|
/// only when `id` isn't in the graph.
|
|
#[must_use]
|
|
pub fn root_of(&self, id: NodeId) -> Option<NodeId> {
|
|
let mut cur = id;
|
|
loop {
|
|
match self.node(cur)?.parent {
|
|
Some(p) => cur = p,
|
|
None => return Some(cur),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Every node in `id`'s subtree, excluding `id` itself, in insertion order.
|
|
pub fn descendants(&self, id: NodeId) -> impl Iterator<Item = &Node<N, R>> {
|
|
self.nodes
|
|
.iter()
|
|
.filter(move |n| self.is_descendant(n.id, id))
|
|
}
|
|
|
|
/// Every group root — the nodes with no parent.
|
|
pub fn roots(&self) -> impl Iterator<Item = &Node<N, R>> {
|
|
self.nodes.iter().filter(|n| n.parent.is_none())
|
|
}
|
|
|
|
/// Whether `id` has settled, or `None` when there is no such node. A group
|
|
/// root's state is its subtree's roll-up, so for a root this answers "is
|
|
/// everything under it finished" — which is why callers don't scan the
|
|
/// subtree themselves.
|
|
///
|
|
/// `None` rather than `false` for an unknown id: "this node is not finished"
|
|
/// and "there is no such node" are different answers, and a caller that
|
|
/// conflates them keeps polling an id that will never settle.
|
|
#[must_use]
|
|
pub fn is_settled(&self, id: NodeId) -> Option<bool> {
|
|
self.node(id).map(|n| n.state.is_terminal())
|
|
}
|
|
|
|
/// Why `id`'s subtree failed: the error of the first `Failed` descendant
|
|
/// that carries one, in insertion order.
|
|
///
|
|
/// Skipping the ones without an error is the point, not an optimisation. A
|
|
/// node that rolled up `Failed` from a child holds no error of its own, and
|
|
/// such a node can sort before the child that actually broke — stopping at
|
|
/// the first `Failed` node would report `None` while the real reason sits
|
|
/// further down the subtree.
|
|
#[must_use]
|
|
pub fn first_error(&self, id: NodeId) -> Option<&str> {
|
|
self.descendants(id)
|
|
.filter(|n| matches!(n.state, State::Failed))
|
|
.find_map(|n| n.error.as_deref())
|
|
}
|
|
|
|
/// Whether `ancestor` lies on `node`'s [`Node::parent`] chain (i.e. `node` is
|
|
/// in `ancestor`'s subtree). `node` is not its own ancestor.
|
|
fn is_descendant(&self, node: NodeId, ancestor: NodeId) -> bool {
|
|
let mut cur = self.node(node).and_then(|n| n.parent);
|
|
while let Some(p) = cur {
|
|
if p == ancestor {
|
|
return true;
|
|
}
|
|
cur = self.node(p).and_then(|n| n.parent);
|
|
}
|
|
false
|
|
}
|
|
|
|
/// Whether a node whose parent is `node_parent` may depend on `target` — the
|
|
/// grouping rule: a [`Dep::Node`] edge must stay inside the depender's own
|
|
/// parent group. `target` must be a proper descendant of `node_parent` (a
|
|
/// sibling or a sibling's sub-node), never the parent itself (which would
|
|
/// deadlock: the parent stays [`State::Finishing`] until its children finish,
|
|
/// so a child that waited on the parent could never run). Top-level nodes
|
|
/// (`parent == None`) may only depend on other top-level nodes.
|
|
fn dep_target_in_group(&self, node_parent: Option<NodeId>, target: NodeId) -> bool {
|
|
match node_parent {
|
|
Some(p) => self.is_descendant(target, p),
|
|
None => self.node(target).is_some_and(|n| n.parent.is_none()),
|
|
}
|
|
}
|
|
|
|
/// Set a node's lifecycle state, returning `false` for an unknown id. The
|
|
/// scheduler drives every state transition — nothing else mutates state,
|
|
/// which is what keeps the resource guards + terminality in sync. This is
|
|
/// also where the node's lifecycle timestamps are stamped: `started_at` on
|
|
/// the first transition to [`State::Running`], `finished_at` on the first
|
|
/// transition to a terminal state (`Done` / `Failed` / `Cancelled`).
|
|
pub(crate) fn set_state(&mut self, id: NodeId, state: State) -> bool {
|
|
if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) {
|
|
node.state = state;
|
|
if state == State::Running {
|
|
if node.started_at.is_none() {
|
|
node.started_at = Some(now_utc());
|
|
}
|
|
} else if state.is_terminal() && node.finished_at.is_none() {
|
|
node.finished_at = Some(now_utc());
|
|
}
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
|
|
/// Record a node's failure reason ([`Node::error`]). No-op for an unknown
|
|
/// id. Called by the scheduler on an [`scheduler::Outcome::Failed`] before
|
|
/// the terminal state transition.
|
|
pub(crate) fn set_error(&mut self, id: NodeId, error: String) {
|
|
if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) {
|
|
node.error = Some(error);
|
|
}
|
|
}
|
|
|
|
/// Check that every id the graph holds resolves: every [`Dep::Node`] id
|
|
/// names a node present in the graph, and `next_id` is past the largest
|
|
/// existing id. Deserialization runs this, so a loaded graph is internally
|
|
/// consistent and internal iteration can trust its ids.
|
|
///
|
|
/// # Errors
|
|
/// Returns [`GraphError`] on a dangling dependency reference, or a `next_id`
|
|
/// that would remint an id already in the graph.
|
|
pub fn validate(&self) -> Result<(), GraphError> {
|
|
for node in &self.nodes {
|
|
if let Some(p) = node.parent
|
|
&& self.node(p).is_none()
|
|
{
|
|
return Err(GraphError::UnknownParent(p));
|
|
}
|
|
for dep in &node.deps {
|
|
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,
|
|
parent: node.parent,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if let Some(max_id) = self.nodes.iter().map(|n| n.id.0).max()
|
|
&& self.next_id <= max_id
|
|
{
|
|
return Err(GraphError::NextIdTooSmall {
|
|
next_id: self.next_id,
|
|
max_id,
|
|
});
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn insert_mints_stable_monotonic_ids() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let a = g.insert("sweep", vec![], None).unwrap();
|
|
let b = g
|
|
.insert(
|
|
"update",
|
|
vec![Dep::Node {
|
|
id: a,
|
|
when: DepWhen::AFTER_OK,
|
|
}],
|
|
None,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(a, NodeId(0));
|
|
assert_eq!(b, NodeId(1));
|
|
// The dep edge references the earlier node; ids are stable + monotonic.
|
|
assert!(matches!(
|
|
g.node(b).unwrap().deps.first(),
|
|
Some(Dep::Node { id, .. }) if *id == a
|
|
));
|
|
}
|
|
|
|
/// Borrow a node mutably so a test can force its lifecycle state. Looks the
|
|
/// node up by id rather than indexing, so it doesn't quietly depend on ids
|
|
/// and positions coinciding.
|
|
fn node_mut<'g>(
|
|
g: &'g mut Graph<&'static str, String>,
|
|
id: NodeId,
|
|
) -> &'g mut Node<&'static str, String> {
|
|
g.nodes
|
|
.iter_mut()
|
|
.find(|n| n.id == id)
|
|
.expect("node in graph")
|
|
}
|
|
|
|
/// `root_of` walks to the top of the parent chain; `descendants` is its
|
|
/// inverse and excludes the node itself.
|
|
#[test]
|
|
fn root_of_and_descendants_span_the_subtree() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let root = g.insert("root", vec![], None).unwrap();
|
|
let mid = g.insert("mid", vec![], Some(root)).unwrap();
|
|
let leaf = g.insert("leaf", vec![], Some(mid)).unwrap();
|
|
let other = g.insert("other-root", vec![], None).unwrap();
|
|
|
|
assert_eq!(g.root_of(leaf), Some(root), "walks the whole chain");
|
|
assert_eq!(g.root_of(root), Some(root), "a root is its own root");
|
|
assert_eq!(g.root_of(NodeId(99)), None, "unknown id");
|
|
|
|
let mut under_root: Vec<NodeId> = g.descendants(root).map(|n| n.id).collect();
|
|
under_root.sort_unstable();
|
|
assert_eq!(under_root, vec![mid, leaf], "excludes the node itself");
|
|
assert_eq!(g.descendants(other).count(), 0);
|
|
|
|
let roots: Vec<NodeId> = g.roots().map(|n| n.id).collect();
|
|
assert_eq!(roots, vec![root, other]);
|
|
}
|
|
|
|
/// The reason `first_error` looks for the first failed descendant **that
|
|
/// carries an error** rather than simply the first failed one: a node that
|
|
/// rolled its `Failed` up from a child holds no error of its own, and it
|
|
/// sorts *before* that child. Stopping at the first `Failed` node would
|
|
/// report `None` and lose the real reason.
|
|
#[test]
|
|
fn first_error_skips_a_rolled_up_failure_carrying_no_error() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let container = g.insert("dag", vec![], None).unwrap();
|
|
let rolled_up = g.insert("prebuild", vec![], Some(container)).unwrap();
|
|
let broke = g.insert("swap", vec![], Some(rolled_up)).unwrap();
|
|
|
|
node_mut(&mut g, rolled_up).state = State::Failed;
|
|
let broken = node_mut(&mut g, broke);
|
|
broken.state = State::Failed;
|
|
broken.error = Some("nix build exploded".to_owned());
|
|
|
|
assert_eq!(g.first_error(container), Some("nix build exploded"));
|
|
}
|
|
|
|
#[test]
|
|
fn is_settled_tracks_node_state() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let n = g.insert("n", vec![], None).unwrap();
|
|
assert_eq!(g.is_settled(n), Some(false), "Pending is not settled");
|
|
node_mut(&mut g, n).state = State::Finishing;
|
|
assert_eq!(
|
|
g.is_settled(n),
|
|
Some(false),
|
|
"Finishing still has children running"
|
|
);
|
|
node_mut(&mut g, n).state = State::Done;
|
|
assert_eq!(g.is_settled(n), Some(true));
|
|
assert_eq!(
|
|
g.is_settled(NodeId(99)),
|
|
None,
|
|
"an unknown id is not the same answer as `not settled`"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn resolve_id_inverts_get_and_rejects_a_value_that_was_never_an_id() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let a = g.insert("a", vec![], None).unwrap();
|
|
let b = g.insert("b", vec![], None).unwrap();
|
|
// Round-trips every id the graph handed out: this is the only way back
|
|
// from a raw value, since NodeId can't be constructed from one.
|
|
assert_eq!(g.resolve_id(a.get()), Some(a));
|
|
assert_eq!(g.resolve_id(b.get()), Some(b));
|
|
// A value that was never an id resolves to nothing, so a caller can't
|
|
// reach a node by guessing a number off the wire.
|
|
assert_eq!(g.resolve_id(u64::MAX), None);
|
|
}
|
|
|
|
#[test]
|
|
fn state_terminality() {
|
|
assert!(State::Done.is_terminal());
|
|
assert!(State::Failed.is_terminal());
|
|
assert!(State::Cancelled.is_terminal());
|
|
assert!(!State::Pending.is_terminal());
|
|
assert!(!State::Running.is_terminal());
|
|
// Finishing (logic done, children still running) is NOT terminal.
|
|
assert!(!State::Finishing.is_terminal());
|
|
}
|
|
|
|
#[test]
|
|
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::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::AFTER_OK.satisfied_by(State::Finishing));
|
|
assert!(!DepWhen::AFTER_ANY.satisfied_by(State::Finishing));
|
|
}
|
|
|
|
#[test]
|
|
fn insert_rejects_unknown_dep() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let bogus = NodeId(42);
|
|
let deps = vec![Dep::Node {
|
|
id: bogus,
|
|
when: DepWhen::AFTER_OK,
|
|
}];
|
|
assert_eq!(
|
|
g.insert("x", deps, None).unwrap_err(),
|
|
GraphError::UnknownDep(bogus)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn insert_rejects_unknown_parent() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let bogus = NodeId(7);
|
|
assert_eq!(
|
|
g.insert("x", vec![], Some(bogus)).unwrap_err(),
|
|
GraphError::UnknownParent(bogus)
|
|
);
|
|
// A resolvable parent is accepted and recorded.
|
|
let a = g.insert("a", vec![], None).unwrap();
|
|
let b = g.insert("b", vec![], Some(a)).unwrap();
|
|
assert_eq!(g.node(b).unwrap().parent, Some(a));
|
|
}
|
|
|
|
#[test]
|
|
fn valid_graph_round_trips_through_serde() {
|
|
let mut g: Graph<String, String> = Graph::new();
|
|
// `a` (top-level) and `b` (top-level, depends on its sibling `a`), plus
|
|
// `c` — a sub-node of `a` (grouping, no dep on its parent).
|
|
let a = g.insert("a".to_owned(), vec![], None).unwrap();
|
|
g.insert(
|
|
"b".to_owned(),
|
|
vec![Dep::Node {
|
|
id: a,
|
|
when: DepWhen::AFTER_ANY,
|
|
}],
|
|
None,
|
|
)
|
|
.unwrap();
|
|
let c = g.insert("c".to_owned(), vec![], Some(a)).unwrap();
|
|
let json = serde_json::to_string(&g).unwrap();
|
|
let back: Graph<String, String> = serde_json::from_str(&json).unwrap();
|
|
assert!(back.validate().is_ok());
|
|
assert_eq!(back.node(a).unwrap().payload, "a");
|
|
assert_eq!(back.node(c).unwrap().parent, Some(a));
|
|
}
|
|
|
|
#[test]
|
|
fn insert_rejects_dep_on_parent_and_cross_group() {
|
|
let mut g: Graph<&str, String> = Graph::new();
|
|
let root = g.insert("root", vec![], None).unwrap();
|
|
// A child cannot depend on its own parent (would deadlock under the
|
|
// roll-up model — the parent stays `Finishing` awaiting its children).
|
|
let on_parent = vec![Dep::Node {
|
|
id: root,
|
|
when: DepWhen::AFTER_OK,
|
|
}];
|
|
assert_eq!(
|
|
g.insert("child", on_parent, Some(root)).unwrap_err(),
|
|
GraphError::DepOutsideParent {
|
|
dep: root,
|
|
parent: Some(root),
|
|
}
|
|
);
|
|
// A sibling dep IS allowed: two children of `root`, the second on the first.
|
|
let c1 = g.insert("c1", vec![], Some(root)).unwrap();
|
|
let c2 = g
|
|
.insert("c2", vec![after_ok_dep(c1)], Some(root))
|
|
.expect("sibling dep is in-group");
|
|
assert_eq!(g.node(c2).unwrap().parent, Some(root));
|
|
// But a node in another group cannot be depended on across the boundary.
|
|
let other = g.insert("other", vec![], None).unwrap();
|
|
assert_eq!(
|
|
g.insert("x", vec![after_ok_dep(other)], Some(root))
|
|
.unwrap_err(),
|
|
GraphError::DepOutsideParent {
|
|
dep: other,
|
|
parent: Some(root),
|
|
}
|
|
);
|
|
}
|
|
|
|
fn after_ok_dep(on: NodeId) -> Dep<String> {
|
|
Dep::Node {
|
|
id: on,
|
|
when: DepWhen::AFTER_OK,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn deserialize_rejects_a_dangling_dependency() {
|
|
// Build a graph whose only node depends on a non-existent id, serialize
|
|
// it (Serialize does not validate), and confirm deserialize rejects it.
|
|
let bad = Graph::<String, String> {
|
|
nodes: vec![Node {
|
|
id: NodeId(0),
|
|
parent: None,
|
|
payload: "x".to_owned(),
|
|
deps: vec![Dep::Node {
|
|
id: NodeId(99),
|
|
when: DepWhen::AFTER_OK,
|
|
}],
|
|
state: State::Pending,
|
|
started_at: None,
|
|
finished_at: None,
|
|
error: None,
|
|
}],
|
|
next_id: 1,
|
|
};
|
|
let json = serde_json::to_string(&bad).unwrap();
|
|
let err = serde_json::from_str::<Graph<String, String>>(&json).unwrap_err();
|
|
assert!(err.to_string().contains("unknown node"));
|
|
}
|
|
|
|
#[test]
|
|
fn validate_rejects_next_id_that_would_remint() {
|
|
let bad = Graph::<&str, String> {
|
|
nodes: vec![Node {
|
|
id: NodeId(5),
|
|
parent: None,
|
|
payload: "x",
|
|
deps: vec![],
|
|
state: State::Pending,
|
|
started_at: None,
|
|
finished_at: None,
|
|
error: None,
|
|
}],
|
|
next_id: 3,
|
|
};
|
|
assert_eq!(
|
|
bad.validate().unwrap_err(),
|
|
GraphError::NextIdTooSmall {
|
|
next_id: 3,
|
|
max_id: 5,
|
|
}
|
|
);
|
|
}
|
|
}
|