24cd7f6d65
`Node` carried two of the three lifecycle timestamps; the third lived on hive-c0re's `NodeKind::Dag` container payload, a core-specific wrapper the graph knows nothing about. Give it its real home so the container's copy becomes redundant rather than load-bearing. Not an `Option` like its neighbours: starting and finishing are events that may never happen, but a node that exists was created. Modelling it as optional would encode a state the graph cannot be in. `hive-jobq-wire::GraphNode` gains the field in the same commit — it already carries the other two, and without this one the value cannot reach a viewer when the container's copy is deleted.
537 lines
21 KiB
Rust
537 lines
21 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,
|
|
}
|
|
|
|
/// 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()
|
|
}
|
|
}
|
|
|
|
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::{
|
|
BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, TerminalState,
|
|
TerminalStateSchema, WireId, WireNode,
|
|
};
|
|
|
|
/// 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 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}");
|
|
}
|
|
}
|