//! 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` renders by implementing [`WireNode`]. //! [`GraphWire`] is then blanket-implemented for any `Graph` whose payload //! does — so the projection exists exactly when `N` has 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. //! //! Resource edges (`Dep::Resource`) are scheduling detail and do not ride the //! wire: no viewer reads them. //! //! 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 hive_jobq::{Dep, Graph, NodeId, State}; 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, } } } /// 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; } /// 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, /// 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, /// 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, 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"`) — 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, }, /// Any other `kind`. Only deserialising produces it: a sender from before /// resource edges left the wire still emits `"Resource"`, and a viewer /// must skip it rather than reject the whole graph. Nothing serialises /// it; remove once no such sender runs. #[serde(other)] Unknown, } /// Serving a [`Graph`] to a viewer. /// /// Blanket-implemented for every `Graph` whose payload implements /// [`WireNode`], 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) -> Vec; } impl GraphWire for Graph { fn wire_snapshot(&self, roots: impl IntoIterator) -> Vec { let mut roots: Vec = 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`, unlike [`GraphWire`]** — a node's state /// is a scheduler concept, so counting by state needs to know nothing about /// what the payload *is*. 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( graph: &Graph, roots: impl IntoIterator, ) -> Vec { // 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> { let states: Vec = 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, states: Option<&[State]>) -> Vec { let Some(states) = states else { return nodes; }; nodes .into_iter() .filter(|n| states.contains(&n.state)) .collect() } fn wire_node(node: &hive_jobq::Node) -> GraphNode { let id = node.id.get(); GraphNode { id, parent: node.parent.map(NodeId::get), state: node.state, deps: node.deps.iter().filter_map(wire_dep).collect(), error: node.error.clone(), payload: NodePayload { label: node.payload.label(), data: node.payload.data(id), }, } } fn wire_dep(dep: &Dep) -> Option { match dep { Dep::Node { id, .. } => Some(GraphDep::Node { id: id.get() }), Dep::Resource { .. } => None, } } #[cfg(test)] mod tests { use hive_jobq::Graph; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::Scheduler; use super::{ ALL_STATES, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, 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}" ); } } /// 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; 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 }) } } /// 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 = wire.iter().map(|n| n.id).collect(); assert_eq!( ids, shown.iter().map(|id| id.get()).collect::>(), "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` come from the host's impl — the graph layer never /// invents a rendering for a payload it cannot read. A resource edge is in /// the scheduler's graph but not on the wire. #[test] fn payload_rendering_comes_from_the_host_impl_and_resource_edges_stay_off_the_wire() { 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!( sched.graph().node(second).expect("node").deps.len(), 2, "control: the scheduler holds the node dep and the resource dep" ); assert_eq!(wire[0].payload.label, "reconcile"); assert_eq!(wire[0].payload.data["seen_id"], first.get()); match &wire[1].deps[..] { [GraphDep::Node { id }] => { assert_eq!(*id, first.get()); } other => panic!("expected only the node dep, got {other:?}"), } } fn node(id: WireId, parent: Option, state: State, deps: Vec) -> GraphNode { GraphNode { id, parent, state, deps, 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 }]); 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:?}")); } /// A sender from before resource edges left the wire still emits them; the /// receiver keeps the node and reads the edge as `Unknown` instead of /// rejecting the graph. #[test] fn an_old_senders_resource_edge_is_tolerated() { let mut json = serde_json::to_value(node( 7, None, State::Pending, vec![GraphDep::Node { id: 3 }], )) .expect("serialises"); json["deps"] .as_array_mut() .expect("deps") .push(serde_json::json!({ "kind": "Resource", "name": "build-slot", "count": 1 })); let after: GraphNode = serde_json::from_value(json).expect("tolerated"); assert_eq!(after.deps.len(), 2); assert!( matches!(after.deps[0], GraphDep::Node { id: 3 }), "control: a node edge still parses" ); assert!(matches!(after.deps[1], GraphDep::Unknown)); } /// The serialised node carries exactly these keys (`parent`, `deps` and /// `error` are omitted when empty, `payload.data` when null). #[test] fn a_node_serialises_exactly_its_read_fields() { let full = GraphNode { error: Some("boom".to_owned()), ..node(7, Some(3), State::Failed, vec![GraphDep::Node { id: 3 }]) }; let value = serde_json::to_value(&full).expect("serialises"); let mut keys: Vec<&str> = value .as_object() .unwrap() .keys() .map(String::as_str) .collect(); keys.sort_unstable(); assert_eq!(keys, ["deps", "error", "id", "parent", "payload", "state"]); assert_eq!( value["deps"][0], serde_json::json!({ "kind": "Node", "id": 3 }) ); } #[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 = 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::>(), 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::(), 3, "root + child + grandchild — every node, and the hidden group is excluded" ); assert_eq!( rollup.iter().map(|c| c.roots).sum::(), 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"); } }