From 9e7a2002d112a245c46c2af2d6ec8b46319dd58c Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 3 Aug 2026 19:17:50 +0200 Subject: [PATCH] jobq: a generic per-state roll-up, served beside the graph MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A consumer that wants "how much is in flight" — a summary line, a badge, a health check — had to fetch the whole graph and tally it client-side, on every poll, in every consumer. `hive_jobq_wire::state_rollup` counts `roots` and their subtrees by state, straight off a `Graph` with **no bound on either parameter**. 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 does would make a host implement two display traits to be allowed to count, which is a requirement about rendering imposed on arithmetic. It takes the roots for the same reason `wire_snapshot` does — which groups are in view is the host's policy, and nothing is ever removed from a graph — so passing the same set makes the roll-up describe exactly the graph beside it. Each entry carries BOTH counts: `nodes` (the whole subtree) and `roots` (just the group tops). One rebuild is ~7 nodes and 1 root, so a summary meaning *operations* and one meaning *steps* are different numbers over the same queue, and picking one here would make this crate decide what counts as a job — the domain question it exists not to answer. It reports both structural facts; the viewer chooses. A pair, not a map: JSON object keys are strings, so a map would spell the state twice and give the wire no ordering. Every state rides with its zeros in a fixed order, so a consumer can index positionally and never handles a missing bucket. Tallying positionally against `ALL_STATES` means a new upstream `State` fails the exhaustive match in `state_index` rather than silently landing in an existing bucket. hive-c0re serves it at `GET /api/jobq/rollup`. The queue-side method is a call site, not an implementation: it supplies the lock and the same `visible_roots` as `graph_snapshot`, so the summary cannot describe a different visible set than the graph it summarises. --- hive-c0re/src/dashboard/mod.rs | 4 +- hive-c0re/src/dashboard/state_snapshot.rs | 22 +++ hive-c0re/src/job_queue/mod.rs | 12 ++ hive-jobq-wire/src/lib.rs | 160 +++++++++++++++++++++- 4 files changed, 195 insertions(+), 3 deletions(-) diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index edc20408..9fc491b2 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -206,6 +206,7 @@ pub async fn serve( .routes(routes!(state_snapshot::dashboard_stream)) .routes(routes!(state_snapshot::dashboard_history)) .routes(routes!(state_snapshot::jobq_graph)) + .routes(routes!(state_snapshot::jobq_rollup)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from // the nix store (see the module doc comment above `ApiDoc`). `api` @@ -450,6 +451,7 @@ mod router_build_probe { .routes(routes!(build_logs::get_build_log_stream)) .routes(routes!(state_snapshot::dashboard_stream)) .routes(routes!(state_snapshot::dashboard_history)) - .routes(routes!(state_snapshot::jobq_graph)); + .routes(routes!(state_snapshot::jobq_graph)) + .routes(routes!(state_snapshot::jobq_rollup)); } } diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 97fb208a..ccf04b78 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -678,6 +678,28 @@ pub(super) async fn jobq_graph( axum::Json(state.coord.job_queue.graph_snapshot()) } +#[utoipa::path( + get, + path = "/api/jobq/rollup", + responses( + (status = 200, description = "counts by lifecycle state over the same \ + groups `/api/jobq/graph` serves, as `(state, nodes, roots)` \ + triples. Every state is present, zero counts included, in a fixed \ + order — a consumer renders a summary (\"3 running · 2 queued\") \ + without fetching the graph and without re-deriving the tally. \ + `roots` counts groups, `nodes` counts every step at any depth: one \ + rebuild is 1 root and ~7 nodes, so a summary meaning *operations* \ + reads `roots` and one meaning *steps* reads `nodes`.", + body = Vec), + ), + tag = "state_snapshot" +)] +pub(super) async fn jobq_rollup( + State(state): State, +) -> axum::Json> { + axum::Json(state.coord.job_queue.state_rollup()) +} + #[utoipa::path( get, path = "/api/dashboard/history", diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index d4afeea0..c996582b 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -368,6 +368,18 @@ impl JobQueue { inner.graph().wire_snapshot(visible_roots(&inner)) } + /// Per-state counts over the **same** groups [`Queue::graph_snapshot`] + /// serves. + /// + /// Supplies the same two things and nothing else: the lock, and + /// [`visible_roots`]. The counting is [`hive_jobq_wire::state_rollup`]'s and + /// is generic over the payload — this is a call site, not an implementation. + #[must_use] + pub fn state_rollup(&self) -> Vec { + let inner = self.lock(); + hive_jobq_wire::state_rollup(inner.graph(), visible_roots(&inner)) + } + /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. #[must_use] pub fn snapshot(&self) -> Vec { diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs index 437782ea..d3a06761 100644 --- a/hive-jobq-wire/src/lib.rs +++ b/hive-jobq-wire/src/lib.rs @@ -169,6 +169,49 @@ pub struct GraphNode { 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 { @@ -250,6 +293,67 @@ impl GraphWire for Graph { } } +/// 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( + 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, + } +} + fn wire_node(node: &hive_jobq::Node) -> GraphNode { let id = node.id.get(); GraphNode { @@ -295,8 +399,8 @@ mod tests { use hive_jobq::scheduler::Scheduler; use super::{ - BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, TerminalState, - TerminalStateSchema, WireId, WireNode, + ALL_STATES, BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, + TerminalState, TerminalStateSchema, WireId, WireNode, state_rollup, }; /// The mirrors document what the wire actually says — a mirror that @@ -534,4 +638,56 @@ mod tests { 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"); + } }