778 lines
31 KiB
Rust
778 lines
31 KiB
Rust
//! A [`hive_jobq`] graph, serialised without knowing what its nodes mean.
|
|
//!
|
|
//! A [`Graph`] is generic over its node payload `N` and resource name `R`, so
|
|
//! the scheduler can carry a host's domain types without interpreting them.
|
|
//! This crate extends that to the wire: [`GraphNode`] carries what a [`Node`]
|
|
//! itself carries — identity, the parent tree, dependency edges, lifecycle —
|
|
//! and pushes everything domain-specific into one opaque [`NodePayload::data`]
|
|
//! slot the consumer renders without branching on.
|
|
//!
|
|
//! **Why a separate crate.** `hive-jobq` is logic; this is presentation. Held
|
|
//! together they remix, and every consumer of the scheduler ends up carrying a
|
|
//! JSON vocabulary it may never serve.
|
|
//!
|
|
//! A host says how its `N` and `R` render by implementing [`WireNode`] and
|
|
//! [`WireResource`]. [`GraphWire`] is then blanket-implemented for any
|
|
//! `Graph<N, R>` whose two type parameters do — so the projection exists
|
|
//! exactly when both have answered, and a payload that has never said how it
|
|
//! displays cannot reach a viewer at all. The alternative (a projection written
|
|
//! next to each host's endpoint) lets the second host ship a graph of
|
|
//! unlabelled integers.
|
|
//!
|
|
//! There is deliberately **no roll-up field**. A group root's own [`State`]
|
|
//! *is* its subtree's answer: `Finishing` means "own logic done, children still
|
|
//! running", and the terminal states are the rolled-up outcome. A separate
|
|
//! field would be a lossier copy of a value already on the wire — lossier
|
|
//! because it would have to flatten `Running` and `Finishing` together.
|
|
//!
|
|
//! [`Node`]: hive_jobq::Node
|
|
|
|
use chrono::{DateTime, Utc};
|
|
use enumflags2::BitFlags;
|
|
use hive_jobq::{Dep, Graph, NodeId, State, TerminalState};
|
|
use serde::{Deserialize, Serialize};
|
|
use utoipa::ToSchema;
|
|
|
|
/// `OpenAPI` mirror of [`hive_jobq::State`].
|
|
///
|
|
/// Exists **only** so the generated spec can enumerate the states: `State` is
|
|
/// a foreign type, so neither `ToSchema` nor a newtype around it can be
|
|
/// implemented here, and `utoipa` is deliberately not a dependency of the
|
|
/// scheduler crate. Referenced via `#[schema(value_type = …)]`; nothing is ever
|
|
/// serialised through it, so it cannot change the wire.
|
|
///
|
|
/// [`StateSchema::of`] is what keeps it honest — an exhaustive match, so adding
|
|
/// a variant upstream **fails to compile here** instead of quietly leaving the
|
|
/// documented enum short.
|
|
#[derive(Debug, Clone, Copy, ToSchema)]
|
|
pub enum StateSchema {
|
|
Pending,
|
|
Running,
|
|
Finishing,
|
|
Done,
|
|
Failed,
|
|
Cancelled,
|
|
Skipped,
|
|
}
|
|
|
|
impl StateSchema {
|
|
/// The mirror variant for `state`. Public because the mirror type is: a
|
|
/// consumer that renders the documented enum needs the same mapping, and a
|
|
/// constructor nobody outside can call is a schema nobody can check.
|
|
#[must_use]
|
|
pub fn of(state: State) -> Self {
|
|
match state {
|
|
State::Pending => Self::Pending,
|
|
State::Running => Self::Running,
|
|
State::Finishing => Self::Finishing,
|
|
State::Done => Self::Done,
|
|
State::Failed => Self::Failed,
|
|
State::Cancelled => Self::Cancelled,
|
|
State::Skipped => Self::Skipped,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `OpenAPI` mirror of [`hive_jobq::TerminalState`] — same rationale and same
|
|
/// exhaustive-match guard as [`StateSchema`].
|
|
#[derive(Debug, Clone, Copy, ToSchema)]
|
|
pub enum TerminalStateSchema {
|
|
Done,
|
|
Failed,
|
|
Cancelled,
|
|
Skipped,
|
|
}
|
|
|
|
impl TerminalStateSchema {
|
|
/// The mirror variant for `outcome`. Public for the same reason as
|
|
/// [`StateSchema::of`].
|
|
#[must_use]
|
|
pub fn of(outcome: TerminalState) -> Self {
|
|
match outcome {
|
|
TerminalState::Done => Self::Done,
|
|
TerminalState::Failed => Self::Failed,
|
|
TerminalState::Cancelled => Self::Cancelled,
|
|
TerminalState::Skipped => Self::Skipped,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Node id, carried verbatim from [`hive_jobq::NodeId`]: globally unique across
|
|
/// the whole graph, not per group. Opaque to consumers — they group by
|
|
/// [`GraphNode::parent`] and match dependency edges, nothing more.
|
|
pub type WireId = u64;
|
|
|
|
/// How a node payload `N` describes itself to a generic viewer.
|
|
///
|
|
/// Half of [`GraphWire`]'s bound, which is the enforcement: no impl, no
|
|
/// projection — a payload that has not answered these two questions has no way
|
|
/// onto the wire.
|
|
pub trait WireNode {
|
|
/// Short tag the consumer displays verbatim. **Not** for matching on: a
|
|
/// generic viewer that branches on this has stopped being generic.
|
|
fn label(&self) -> String;
|
|
|
|
/// Domain data for this node, rendered generically (key/value, a details
|
|
/// pane, a tooltip — the consumer's choice). Return
|
|
/// [`serde_json::Value::Null`] when there is nothing to add; it is then
|
|
/// omitted from the wire entirely.
|
|
///
|
|
/// Takes the node's `id` because a host's extra data is not always *in* the
|
|
/// payload — it may be keyed on node identity in a side table (a build log,
|
|
/// an artifact store) that only the host can resolve.
|
|
fn data(&self, id: WireId) -> serde_json::Value;
|
|
}
|
|
|
|
/// How a resource name `R` rides the wire.
|
|
///
|
|
/// Resource deps are named rather than typed on the wire: the resource
|
|
/// vocabulary belongs to the host, and a viewer only ever groups by the string.
|
|
pub trait WireResource {
|
|
/// Stable display name for this resource.
|
|
fn name(&self) -> String;
|
|
}
|
|
|
|
/// One node, serialised near-raw from [`hive_jobq::Node`].
|
|
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
|
|
pub struct GraphNode {
|
|
/// Unique across the whole graph, not per group.
|
|
pub id: WireId,
|
|
/// Structural parent, or `None` for a group root.
|
|
///
|
|
/// A group root is an **ordinary node here** — nothing is hidden, so a
|
|
/// consumer needs no special case for "the container", and the root's own
|
|
/// `state` answers "how is this whole group doing".
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub parent: Option<WireId>,
|
|
/// Lifecycle state. On a group root this is also the subtree's answer:
|
|
/// `Finishing` = own logic done, children still running; a terminal value
|
|
/// is the rolled-up outcome.
|
|
#[schema(value_type = StateSchema)]
|
|
pub state: State,
|
|
/// What must hold before this node runs. Omitted when empty.
|
|
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
|
pub deps: Vec<GraphDep>,
|
|
/// When the node was inserted into the graph. Always present — a node that
|
|
/// exists was created, so unlike the two below this is not an `Option`.
|
|
pub created_at: DateTime<Utc>,
|
|
/// When the node entered `Running`. `None` until it starts; a node that
|
|
/// never ran keeps `None`.
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub started_at: Option<DateTime<Utc>>,
|
|
/// When the node reached a terminal state. `None` while non-terminal.
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub finished_at: Option<DateTime<Utc>>,
|
|
/// Failure reason, set only when this node's *own* logic failed — a node
|
|
/// that rolled up `Failed` from a child carries none of its own.
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub error: Option<String>,
|
|
pub payload: NodePayload,
|
|
}
|
|
|
|
/// How much of the graph is in one [`State`], counted two ways.
|
|
///
|
|
/// A pair, not a map: JSON object keys are strings, so a map would spell the
|
|
/// state twice — once as the key and once in whatever the consumer parses it
|
|
/// back into — and give the wire no ordering. A list of pairs keeps the state a
|
|
/// state.
|
|
///
|
|
/// **Both counts, because "how many things are running" has two honest answers
|
|
/// and which one a viewer wants is its own business.** One rebuild is ~7
|
|
/// `nodes` and 1 `roots`; a summary line that means *operations* wants the
|
|
/// latter, a progress bar over steps wants the former. Reporting one would make
|
|
/// this crate decide what counts as a job — the domain question it exists not
|
|
/// to answer — and reporting both costs a `u64`.
|
|
#[derive(Debug, Clone, Copy, Serialize, Deserialize, ToSchema)]
|
|
pub struct StateCount {
|
|
#[schema(value_type = StateSchema)]
|
|
pub state: State,
|
|
/// Every node in this state, at any depth.
|
|
pub nodes: u64,
|
|
/// Only the roots the caller asked for, in this state.
|
|
///
|
|
/// A root's own state is already its subtree's answer (a group root does
|
|
/// not reach `Done` before its children), so this is "how many of the
|
|
/// visible groups are in this state" without the caller re-deriving it.
|
|
pub roots: u64,
|
|
}
|
|
|
|
/// Every [`State`], in the order [`ALL_STATES`] declares, so a consumer can
|
|
/// index the roll-up positionally and never has to handle a missing key.
|
|
///
|
|
/// Zero counts are **included** on purpose: a summary that renders "3 running"
|
|
/// needs to know the other buckets are empty rather than absent, and the whole
|
|
/// vector is seven entries regardless of graph size.
|
|
const ALL_STATES: [State; 7] = [
|
|
State::Pending,
|
|
State::Running,
|
|
State::Finishing,
|
|
State::Done,
|
|
State::Failed,
|
|
State::Cancelled,
|
|
State::Skipped,
|
|
];
|
|
|
|
/// What a node *is*, in terms the graph layer does not interpret.
|
|
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
|
|
pub struct NodePayload {
|
|
/// From [`WireNode::label`].
|
|
pub label: String,
|
|
/// From [`WireNode::data`]. Everything a specific host wants to say about a
|
|
/// node beyond its label lives here, so adding a field costs the wire type
|
|
/// nothing.
|
|
#[serde(default, skip_serializing_if = "serde_json::Value::is_null")]
|
|
pub data: serde_json::Value,
|
|
}
|
|
|
|
/// What must hold before a node runs.
|
|
///
|
|
/// Externally tagged on `kind`, whose values are the **variant names verbatim**
|
|
/// (`"Node"`, `"Resource"`) — no `rename_all`. A rename is a second spelling of
|
|
/// the same name that has to be kept in agreement with the Rust one by hand.
|
|
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
|
|
#[serde(tag = "kind")]
|
|
pub enum GraphDep {
|
|
/// Depend on another node finishing acceptably.
|
|
Node {
|
|
/// The node depended on. May name a node the consumer has not been sent
|
|
/// (a filtered view); treat an absent target as satisfied.
|
|
id: WireId,
|
|
/// **The outcomes that satisfy this edge, as a set** — not a
|
|
/// strong/weak flag.
|
|
///
|
|
/// A host routinely emits several tails edged on the *same* upstream
|
|
/// node, distinguished only by which outcomes each accepts (one for
|
|
/// `Done`, one for `Failed`/`Cancelled`, …). Collapsing this to a
|
|
/// boolean renders those as identical nodes.
|
|
///
|
|
/// Every accepted outcome is **named**, so a client never shifts bits
|
|
/// to read an edge.
|
|
#[schema(value_type = Vec<TerminalStateSchema>)]
|
|
accepts: Vec<TerminalState>,
|
|
},
|
|
/// Need `count` units of a named resource, per [`WireResource::name`].
|
|
Resource { name: String, count: u32 },
|
|
}
|
|
|
|
/// Serving a [`Graph`] to a viewer.
|
|
///
|
|
/// Blanket-implemented for every `Graph<N, R>` whose payload and resource types
|
|
/// implement [`WireNode`] / [`WireResource`], and for no others — an extension
|
|
/// trait rather than an inherent method because the graph lives in the
|
|
/// scheduler crate and this projection deliberately does not.
|
|
pub trait GraphWire {
|
|
/// `roots` and all their descendants as [`GraphNode`]s, ordered by root id
|
|
/// with each root immediately followed by its own subtree.
|
|
///
|
|
/// Takes the roots rather than reading [`Graph::roots`] itself, because
|
|
/// *which* groups to show is the host's policy and the graph has no opinion
|
|
/// on it — nothing in a `Graph` is ever removed, so a host that keeps
|
|
/// finished work bounded must bound it at this call. Pass `graph.roots()`
|
|
/// ids to serve everything.
|
|
///
|
|
/// What is *not* the host's business is picking those roots by matching on
|
|
/// `N`: a root is a node with no parent, which is structure. Keying a
|
|
/// generic view on one domain's node vocabulary is exactly what this crate
|
|
/// exists to avoid.
|
|
fn wire_snapshot(&self, roots: impl IntoIterator<Item = NodeId>) -> Vec<GraphNode>;
|
|
}
|
|
|
|
impl<N: WireNode, R: WireResource> GraphWire for Graph<N, R> {
|
|
fn wire_snapshot(&self, roots: impl IntoIterator<Item = NodeId>) -> Vec<GraphNode> {
|
|
let mut roots: Vec<NodeId> = roots.into_iter().collect();
|
|
roots.sort_unstable_by_key(|id| id.get());
|
|
roots
|
|
.into_iter()
|
|
.flat_map(|root| {
|
|
self.node(root)
|
|
.into_iter()
|
|
.chain(self.descendants(root))
|
|
.map(wire_node)
|
|
})
|
|
.collect()
|
|
}
|
|
}
|
|
|
|
/// Count `roots` and their subtrees by [`State`], straight off the graph.
|
|
///
|
|
/// For a consumer that wants "how much is in flight" without carrying the
|
|
/// graph: a summary line, a badge, a health check. Every consumer re-deriving
|
|
/// the same tally is the thing this replaces.
|
|
///
|
|
/// **Free of `WireNode`/`WireResource`, unlike [`GraphWire`]** — a node's state
|
|
/// is a scheduler concept, so counting by state needs to know nothing about
|
|
/// what the payload or the resource *are*. Bounding it like the projection
|
|
/// would make a host implement two display traits to be allowed to count, which
|
|
/// is a requirement about rendering imposed on arithmetic.
|
|
///
|
|
/// Takes the same `roots` as [`GraphWire::wire_snapshot`] and for the same
|
|
/// reason — *which* groups are in view is the host's policy — so passing the
|
|
/// same set makes the roll-up describe exactly the graph beside it. Shares the
|
|
/// caveat too: overlapping roots double-count.
|
|
#[must_use]
|
|
pub fn state_rollup<N, R>(
|
|
graph: &Graph<N, R>,
|
|
roots: impl IntoIterator<Item = NodeId>,
|
|
) -> Vec<StateCount> {
|
|
// Tallied positionally against `ALL_STATES` rather than into a map: the
|
|
// output order is then the declared one for free, and adding a `State`
|
|
// upstream fails the exhaustive match in `state_index` instead of silently
|
|
// dropping a bucket.
|
|
let mut nodes = [0u64; ALL_STATES.len()];
|
|
let mut roots_by_state = [0u64; ALL_STATES.len()];
|
|
for root in roots {
|
|
if let Some(node) = graph.node(root) {
|
|
roots_by_state[state_index(node.state)] += 1;
|
|
}
|
|
for node in graph.node(root).into_iter().chain(graph.descendants(root)) {
|
|
nodes[state_index(node.state)] += 1;
|
|
}
|
|
}
|
|
ALL_STATES
|
|
.iter()
|
|
.zip(nodes)
|
|
.zip(roots_by_state)
|
|
.map(|((&state, nodes), roots)| StateCount {
|
|
state,
|
|
nodes,
|
|
roots,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Position of `state` in [`ALL_STATES`]. Exhaustive on purpose — a new
|
|
/// upstream variant must not silently land in an existing bucket.
|
|
fn state_index(state: State) -> usize {
|
|
match state {
|
|
State::Pending => 0,
|
|
State::Running => 1,
|
|
State::Finishing => 2,
|
|
State::Done => 3,
|
|
State::Failed => 4,
|
|
State::Cancelled => 5,
|
|
State::Skipped => 6,
|
|
}
|
|
}
|
|
|
|
/// Parses a `?states=` query value (comma-separated [`State`] names, e.g.
|
|
/// `"Running,Pending"`) into the filter [`filter_nodes_by_state`] wants.
|
|
///
|
|
/// Absent or entirely-unparseable input is "no filter," never "match
|
|
/// nothing" — an empty/garbled query reads as the unfiltered call it
|
|
/// replaces rather than as a request for zero results. Unrecognised tokens
|
|
/// are silently dropped rather than erroring the whole request.
|
|
///
|
|
/// Lives here rather than in each host's own dashboard/API layer because
|
|
/// every host that serves [`GraphWire::wire_snapshot`] over HTTP wants the
|
|
/// same query shape — a second, independently-typed copy is exactly the
|
|
/// kind of drift this crate exists to prevent (see the module doc comment).
|
|
#[must_use]
|
|
pub fn parse_states(raw: Option<&str>) -> Option<Vec<State>> {
|
|
let states: Vec<State> = raw?
|
|
.split(',')
|
|
.map(str::trim)
|
|
.filter(|s| !s.is_empty())
|
|
.filter_map(|s| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok())
|
|
.collect();
|
|
(!states.is_empty()).then_some(states)
|
|
}
|
|
|
|
/// Keeps only the nodes whose own `state` is named in `states` — a flat
|
|
/// filter on [`GraphNode::state`], no special-casing for a root vs. a
|
|
/// descendant (a root's own state already *is* its subtree's rolled-up
|
|
/// answer, per the module doc comment, so filtering the root filters the
|
|
/// whole group). `None` is a no-op, not "match nothing," mirroring
|
|
/// [`parse_states`]'s own absent-input rule.
|
|
#[must_use]
|
|
pub fn filter_nodes_by_state(nodes: Vec<GraphNode>, states: Option<&[State]>) -> Vec<GraphNode> {
|
|
let Some(states) = states else {
|
|
return nodes;
|
|
};
|
|
nodes
|
|
.into_iter()
|
|
.filter(|n| states.contains(&n.state))
|
|
.collect()
|
|
}
|
|
|
|
fn wire_node<N: WireNode, R: WireResource>(node: &hive_jobq::Node<N, R>) -> GraphNode {
|
|
let id = node.id.get();
|
|
GraphNode {
|
|
id,
|
|
parent: node.parent.map(NodeId::get),
|
|
state: node.state,
|
|
deps: node.deps.iter().map(wire_dep).collect(),
|
|
created_at: node.created_at,
|
|
started_at: node.started_at,
|
|
finished_at: node.finished_at,
|
|
error: node.error.clone(),
|
|
payload: NodePayload {
|
|
label: node.payload.label(),
|
|
data: node.payload.data(id),
|
|
},
|
|
}
|
|
}
|
|
|
|
fn wire_dep<R: WireResource>(dep: &Dep<R>) -> GraphDep {
|
|
match dep {
|
|
Dep::Node { id, when } => GraphDep::Node {
|
|
id: id.get(),
|
|
// Every outcome the type knows about, not a hand-listed set: a
|
|
// variant added to `TerminalState` is then named on the wire
|
|
// automatically, where a literal array would have silently
|
|
// dropped it from every edge that accepts it.
|
|
accepts: BitFlags::<TerminalState>::ALL
|
|
.iter()
|
|
.filter(|outcome| when.accepts(*outcome))
|
|
.collect(),
|
|
},
|
|
Dep::Resource { name, count } => GraphDep::Resource {
|
|
name: name.name(),
|
|
count: *count,
|
|
},
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use hive_jobq::Graph;
|
|
use hive_jobq::resources::ResourceTable;
|
|
use hive_jobq::scheduler::Scheduler;
|
|
|
|
use super::{
|
|
ALL_STATES, BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema,
|
|
TerminalState, TerminalStateSchema, WireId, WireNode, filter_nodes_by_state, parse_states,
|
|
state_rollup,
|
|
};
|
|
|
|
/// The mirrors document what the wire actually says — a mirror that
|
|
/// disagreed would be worse than none, since the spec is all a consumer
|
|
/// has.
|
|
///
|
|
/// `StateSchema::of` is the *compile-time* half (exhaustive, so a new
|
|
/// upstream variant breaks the build). This is the *value* half: the
|
|
/// documented variant name has to equal the serialised one. Both are
|
|
/// needed — an exhaustive match still compiles if the names diverge, which
|
|
/// is exactly what a stray `rename_all` upstream would do.
|
|
#[test]
|
|
fn the_openapi_mirrors_name_states_exactly_as_the_wire_does() {
|
|
let states = [
|
|
State::Pending,
|
|
State::Running,
|
|
State::Finishing,
|
|
State::Done,
|
|
State::Failed,
|
|
State::Cancelled,
|
|
State::Skipped,
|
|
];
|
|
for state in states {
|
|
let on_the_wire = serde_json::to_string(&state).expect("serialises");
|
|
let documented = format!("{:?}", StateSchema::of(state));
|
|
assert_eq!(
|
|
on_the_wire,
|
|
format!("\"{documented}\""),
|
|
"{state:?} is documented as {documented} but serialises as {on_the_wire}"
|
|
);
|
|
}
|
|
|
|
for outcome in BitFlags::<TerminalState>::ALL {
|
|
let on_the_wire = serde_json::to_string(&outcome).expect("serialises");
|
|
let documented = format!("{:?}", TerminalStateSchema::of(outcome));
|
|
assert_eq!(
|
|
on_the_wire,
|
|
format!("\"{documented}\""),
|
|
"{outcome:?} is documented as {documented} but serialises as {on_the_wire}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// A scheduler over the test payloads, built through the same public API a
|
|
/// host uses.
|
|
///
|
|
/// These tests originally reached for `Graph::insert` — which is
|
|
/// `pub(crate)` to `hive-jobq`, so they only ever compiled while this
|
|
/// module still lived *inside* that crate. That they now have to go through
|
|
/// `insert_job` is the point rather than an inconvenience: it proves the
|
|
/// projection needs no privileged access to a graph it does not own.
|
|
fn sched() -> Scheduler<&'static str, String> {
|
|
Scheduler::new(Graph::new(), ResourceTable::new())
|
|
}
|
|
|
|
// The test payloads. `&str` labels itself and has nothing extra to say;
|
|
// `String` is its own resource name.
|
|
impl WireNode for &str {
|
|
fn label(&self) -> String {
|
|
(*self).to_owned()
|
|
}
|
|
fn data(&self, id: WireId) -> serde_json::Value {
|
|
// Keyed on identity, not on the payload — the shape a host with a
|
|
// side table (build logs, artifacts) needs.
|
|
serde_json::json!({ "seen_id": id })
|
|
}
|
|
}
|
|
|
|
impl super::WireResource for String {
|
|
fn name(&self) -> String {
|
|
self.clone()
|
|
}
|
|
}
|
|
|
|
/// The bound is the *caller's*: `wire_snapshot` serialises the roots it is
|
|
/// handed and their subtrees, and nothing else. Nothing in a `Graph` is
|
|
/// ever removed, so a host that couldn't restrict this would have no way to
|
|
/// stop the payload growing for its whole uptime.
|
|
#[test]
|
|
fn only_the_requested_roots_and_their_descendants_ride() {
|
|
let mut sched = sched();
|
|
let shown = sched
|
|
.insert_job(None, |job| {
|
|
let root = job.node("dag");
|
|
let child = job.node("prebuild").part_of(root);
|
|
let grandchild = job.node("swap").part_of(child);
|
|
vec![root.guid(), child.guid(), grandchild.guid()]
|
|
})
|
|
.expect("job inserts");
|
|
let evicted = sched
|
|
.insert_job(None, |job| vec![job.node("old-dag").guid()])
|
|
.expect("job inserts");
|
|
|
|
let wire = sched.graph().wire_snapshot([shown[0]]);
|
|
|
|
let ids: Vec<WireId> = wire.iter().map(|n| n.id).collect();
|
|
assert_eq!(
|
|
ids,
|
|
shown.iter().map(|id| id.get()).collect::<Vec<_>>(),
|
|
"the root, then its whole subtree"
|
|
);
|
|
assert!(
|
|
!ids.contains(&evicted[0].get()),
|
|
"a root the host did not ask for stays off the wire"
|
|
);
|
|
assert_eq!(wire[0].parent, None, "a group root is parentless");
|
|
assert_eq!(wire[1].parent, Some(shown[0].get()));
|
|
}
|
|
|
|
/// `label` / `data` / `name` all come from the host's impls — the graph
|
|
/// layer never invents a rendering for a payload it cannot read.
|
|
#[test]
|
|
fn payload_and_resource_rendering_come_from_the_host_impls() {
|
|
let mut sched = sched();
|
|
let ids = sched
|
|
.insert_job(None, |job| {
|
|
let first = job.node("reconcile");
|
|
let second = job
|
|
.node("swap")
|
|
.after_ok(first)
|
|
.needs_units("build-slot".to_owned(), 2);
|
|
vec![first.guid(), second.guid()]
|
|
})
|
|
.expect("job inserts");
|
|
let (first, second) = (ids[0], ids[1]);
|
|
|
|
let wire = sched.graph().wire_snapshot([first, second]);
|
|
|
|
assert_eq!(wire[0].payload.label, "reconcile");
|
|
assert_eq!(wire[0].payload.data["seen_id"], first.get());
|
|
match &wire[1].deps[..] {
|
|
[
|
|
GraphDep::Node { id, accepts },
|
|
GraphDep::Resource { name, count },
|
|
] => {
|
|
assert_eq!(*id, first.get());
|
|
assert_eq!(accepts, &vec![TerminalState::Done]);
|
|
assert_eq!(name, "build-slot", "the host's WireResource::name");
|
|
assert_eq!(*count, 2);
|
|
}
|
|
other => panic!("expected a node dep then a resource dep, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
fn node(id: WireId, parent: Option<WireId>, state: State, deps: Vec<GraphDep>) -> GraphNode {
|
|
GraphNode {
|
|
id,
|
|
parent,
|
|
state,
|
|
deps,
|
|
created_at: chrono::DateTime::<chrono::Utc>::default(),
|
|
started_at: None,
|
|
finished_at: None,
|
|
error: None,
|
|
payload: NodePayload {
|
|
label: "reconcile".to_owned(),
|
|
data: serde_json::Value::Null,
|
|
},
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_node_round_trips_through_json() {
|
|
let before = node(
|
|
7,
|
|
Some(3),
|
|
State::Running,
|
|
vec![
|
|
GraphDep::Node {
|
|
id: 3,
|
|
accepts: vec![TerminalState::Done],
|
|
},
|
|
GraphDep::Resource {
|
|
name: "build-slot".to_owned(),
|
|
count: 1,
|
|
},
|
|
],
|
|
);
|
|
let json = serde_json::to_string(&before).expect("serialises");
|
|
let after: GraphNode = serde_json::from_str(&json).expect("round trips");
|
|
assert_eq!(format!("{after:?}"), format!("{before:?}"));
|
|
}
|
|
|
|
#[test]
|
|
fn two_tails_on_one_upstream_are_distinguishable_by_their_accepted_set() {
|
|
// The reason `accepts` is a set and not a strong/weak bool: these two
|
|
// are the same shape and the same upstream, and *only* the outcome
|
|
// set tells them apart.
|
|
let ok_tail = node(
|
|
10,
|
|
None,
|
|
State::Pending,
|
|
vec![GraphDep::Node {
|
|
id: 4,
|
|
accepts: vec![TerminalState::Done],
|
|
}],
|
|
);
|
|
let fail_tail = node(
|
|
11,
|
|
None,
|
|
State::Pending,
|
|
vec![GraphDep::Node {
|
|
id: 4,
|
|
accepts: vec![TerminalState::Failed, TerminalState::Cancelled],
|
|
}],
|
|
);
|
|
let render = |n: &GraphNode| match n.deps.first() {
|
|
Some(GraphDep::Node { accepts, .. }) => format!("{accepts:?}"),
|
|
_ => "none".to_owned(),
|
|
};
|
|
assert_ne!(render(&ok_tail), render(&fail_tail));
|
|
}
|
|
|
|
#[test]
|
|
fn a_group_root_is_an_ordinary_node_and_carries_the_groups_answer() {
|
|
// No roll-up field: `Finishing` on the root says "own logic done,
|
|
// children still running", and a terminal root state is the roll-up.
|
|
let root = node(1, None, State::Finishing, Vec::new());
|
|
assert!(root.parent.is_none(), "a group root is just parentless");
|
|
assert_eq!(root.state, State::Finishing);
|
|
|
|
let done_root = node(2, None, State::Failed, Vec::new());
|
|
assert_eq!(
|
|
done_root.state,
|
|
State::Failed,
|
|
"the root's own state is the subtree's rolled-up outcome"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn parse_states_covers_absent_malformed_and_valid() {
|
|
assert_eq!(parse_states(None), None, "no query is no filter");
|
|
assert_eq!(
|
|
parse_states(Some("")),
|
|
None,
|
|
"an empty query is no filter, not a request for zero results"
|
|
);
|
|
assert_eq!(
|
|
parse_states(Some("not-a-state,also-not-one")),
|
|
None,
|
|
"every token failing to parse is still no filter"
|
|
);
|
|
assert_eq!(
|
|
parse_states(Some(" Running , Pending ,bogus")),
|
|
Some(vec![State::Running, State::Pending]),
|
|
"trims whitespace, keeps the valid tokens, drops the unrecognised one"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree() {
|
|
let nodes = vec![
|
|
node(1, None, State::Running, Vec::new()), // root: whole group still live
|
|
node(2, Some(1), State::Done, Vec::new()), // finished step, should be hidden
|
|
node(3, Some(1), State::Pending, Vec::new()), // not-yet-run step, should stay
|
|
];
|
|
|
|
let mut kept: Vec<WireId> =
|
|
filter_nodes_by_state(nodes.clone(), Some(&[State::Running, State::Pending]))
|
|
.into_iter()
|
|
.map(|n| n.id)
|
|
.collect();
|
|
kept.sort_unstable();
|
|
assert_eq!(
|
|
kept,
|
|
vec![1, 3],
|
|
"Done is filtered out even though its still-live parent (root) isn't"
|
|
);
|
|
|
|
let unfiltered = filter_nodes_by_state(nodes, None);
|
|
assert_eq!(unfiltered.len(), 3, "no states given is a no-op, not empty");
|
|
}
|
|
|
|
#[test]
|
|
fn an_empty_payload_slot_is_omitted_from_the_wire() {
|
|
// The opaque slot costs nothing when a host has nothing to say.
|
|
let json =
|
|
serde_json::to_string(&node(1, None, State::Pending, Vec::new())).expect("serialises");
|
|
assert!(!json.contains("\"data\""), "null data is skipped: {json}");
|
|
assert!(json.contains("\"label\""), "the label always rides: {json}");
|
|
}
|
|
|
|
/// The roll-up counts **the same visible set** the snapshot serialises,
|
|
/// both ways, and always emits every state.
|
|
///
|
|
/// Four assertions in one because they are the same promise from four
|
|
/// sides: a consumer indexes the vector positionally (fixed order), never
|
|
/// handles a missing bucket (zeros present), gets a node tally that
|
|
/// includes the group root like any other node (3, not 1), and gets a root
|
|
/// tally that is the *group* count for the same states (1, not 3) — the two
|
|
/// numbers this endpoint exists to stop consumers from confusing.
|
|
#[test]
|
|
fn the_rollup_counts_nodes_and_roots_and_names_every_state() {
|
|
let mut sched = sched();
|
|
let shown = sched
|
|
.insert_job(None, |job| {
|
|
let root = job.node("dag");
|
|
let child = job.node("prebuild").part_of(root);
|
|
let _grandchild = job.node("swap").part_of(child);
|
|
vec![root.guid()]
|
|
})
|
|
.expect("job inserts");
|
|
// A second group deliberately left out of `roots` — the roll-up is
|
|
// bounded by the caller exactly as `wire_snapshot` is, so an unshown
|
|
// group must not leak into the counts.
|
|
let _hidden = sched
|
|
.insert_job(None, |job| vec![job.node("other-dag").guid()])
|
|
.expect("job inserts");
|
|
|
|
let rollup = state_rollup(sched.graph(), shown.iter().copied());
|
|
|
|
assert_eq!(
|
|
rollup.iter().map(|c| c.state).collect::<Vec<_>>(),
|
|
ALL_STATES.to_vec(),
|
|
"every state rides, in the declared order, so a consumer can index it"
|
|
);
|
|
assert_eq!(
|
|
rollup.iter().map(|c| c.nodes).sum::<u64>(),
|
|
3,
|
|
"root + child + grandchild — every node, and the hidden group is excluded"
|
|
);
|
|
assert_eq!(
|
|
rollup.iter().map(|c| c.roots).sum::<u64>(),
|
|
1,
|
|
"one group, counted once — the same subtree the node tally reads as 3"
|
|
);
|
|
let pending = rollup
|
|
.iter()
|
|
.find(|c| matches!(c.state, State::Pending))
|
|
.expect("Pending is always present");
|
|
assert_eq!(pending.nodes, 3, "nothing has run yet");
|
|
assert_eq!(pending.roots, 1, "and its group is pending with it");
|
|
}
|
|
}
|