diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 95ece62e..2258608a 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -580,35 +580,23 @@ fn build_approval_views(approvals: Vec) -> Vec { out } -/// `/api/jobq/graph` query string. Today's only field is `states`: a -/// comma-separated allow-list of `hive_jobq::State` names (`"Pending"`, -/// `"Running"`, ...). Empty / absent ⇒ no filter (current behaviour, every -/// visible root). Set ⇒ only **root** groups whose own state is named are -/// served — a root's state is already its subtree's rolled-up answer (see -/// `hive_jobq_wire`'s doc), so filtering the root filters the whole group. -/// Unknown tokens are silently ignored (an unrecognised name matches -/// nothing rather than erroring the whole request), mirroring -/// `DashboardStreamQuery::kinds` above. +/// `/api/jobq/graph` query string — the generic jobq/wire query shape (any +/// host serving a `GraphWire` projection over HTTP takes the same one; see +/// `hive_jobq_wire::parse_states`, which does the actual parsing this side +/// of the query string). Today's only field is `states`: a comma-separated +/// allow-list of `hive_jobq::State` names (`"Pending"`, `"Running"`, ...). +/// Empty / absent ⇒ no filter (current behaviour, every visible root). Set +/// ⇒ only **root** groups whose own state is named are served — a root's +/// state is already its subtree's rolled-up answer (see `hive_jobq_wire`'s +/// doc), so filtering the root filters the whole group. Unknown tokens are +/// silently ignored (an unrecognised name matches nothing rather than +/// erroring the whole request), mirroring `DashboardStreamQuery::kinds` +/// above. #[derive(Deserialize, Default, IntoParams)] pub(super) struct JobqGraphQuery { states: Option, } -/// Parses [`JobqGraphQuery::states`] into the list -/// [`crate::job_queue::JobQueue::graph_snapshot`] wants. `None` when absent -/// or when every token failed to parse — both mean "no filter" rather than -/// "match nothing", so an empty/garbled query reads as the unfiltered call -/// it replaces rather than an empty result set. -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) -} - #[utoipa::path( get, path = "/api/jobq/graph", @@ -617,10 +605,14 @@ fn parse_states(raw: Option<&str>) -> Option> { (status = 200, description = "every node of every retained job group, \ as generic `hive_jobq` graph nodes: identity, the parent tree, \ dependency edges with their accepted-outcome sets, lifecycle, and \ - one opaque per-node payload. Group roots ride as ordinary nodes \ - (`parent: null`) and `Done` nodes are not filtered by default — a \ - consumer renders the graph without knowing what any node means. \ - `?states=` narrows to root groups in the named states.", + one opaque per-node payload. This is the generic jobq/wire shape \ + (`hive_jobq_wire::GraphWire::wire_snapshot`), not a hive-c0re-only \ + projection — a `swarm-controller` serving its own graph responds \ + with the same shape at the same path. Group roots ride as \ + ordinary nodes (`parent: null`) and `Done` nodes are not \ + filtered by default — a consumer renders the graph without \ + knowing what any node means. `?states=` narrows to root groups \ + in the named states.", body = Vec), ), tag = "state_snapshot" @@ -629,7 +621,7 @@ pub(super) async fn jobq_graph( State(state): State, axum::extract::Query(q): axum::extract::Query, ) -> axum::Json> { - let states = parse_states(q.states.as_deref()); + let states = hive_jobq_wire::parse_states(q.states.as_deref()); axum::Json(state.coord.job_queue.graph_snapshot(states.as_deref())) } @@ -639,7 +631,9 @@ pub(super) async fn jobq_graph( 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 \ + triples — the generic jobq/wire roll-up shape \ + (`hive_jobq_wire::state_rollup`), same as `/api/jobq/graph` \ + above. 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 \ diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index e5fb5704..509bbac8 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -351,15 +351,16 @@ impl JobQueue { /// still-matching descendant one level higher rather than hiding or /// orphaning it. /// - /// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies - /// is *which* nodes to show — see [`visible_roots`] for why the graph - /// can't decide the root-visibility half of that for itself. + /// The projection and state filter are both [`hive_jobq_wire`]'s; this + /// layer only supplies *which roots* are in view (see [`visible_roots`] + /// for why the graph can't decide that itself) — an older, different + /// question than state filtering, and neither replaces the other. #[must_use] pub fn graph_snapshot(&self, states: Option<&[State]>) -> Vec { let inner = self.lock(); let roots = visible_roots(&inner); let nodes = inner.graph().wire_snapshot(roots); - filter_nodes_by_state(nodes, states) + hive_jobq_wire::filter_nodes_by_state(nodes, states) } /// Per-state counts over the **same** groups [`JobQueue::graph_snapshot`] @@ -414,26 +415,6 @@ fn find_node(sched: &Sched, id: u64) -> Option { .find_map(|n| (n.id.get() == id).then_some(n.id)) } -/// [`JobQueue::graph_snapshot`]'s `states` ask, applied to the already- -/// projected node list: keeps every node — root or descendant — whose own -/// `state` is named. -/// -/// Applied *after* [`GraphWire::wire_snapshot`] rather than as a root -/// pre-filter — narrowing which roots are visible at all is -/// [`visible_roots`]'s job (a different question: how much settled work is -/// retained, full stop); this is "of what's retained and live, which -/// individual nodes does the caller want shown right now." `None` (or an -/// unrecognised/absent query) is the identity filter. -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() -} - /// The visible **group** set for [`JobQueue::graph_snapshot`]: every live group /// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones. /// diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 4e27357a..11988883 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1527,60 +1527,6 @@ fn graph_snapshot_states_filters_by_node_state() { ); } -/// One hand-built node, bypassing the scheduler entirely — `filter_nodes_by_state` -/// is a pure `Vec -> Vec` transform, so this exercises it -/// directly rather than trying (and failing, per this module's own rule) to -/// drive a live group into a genuinely mixed state through the real queue. -fn node(id: u64, parent: Option, state: State) -> GraphNode { - GraphNode { - id, - parent, - state, - deps: Vec::new(), - created_at: Utc::now(), - started_at: None, - finished_at: None, - error: None, - payload: hive_jobq_wire::NodePayload { - label: id.to_string(), - data: serde_json::Value::Null, - }, - } -} - -/// The case `graph_snapshot_states_filters_by_node_state` above can't reach: -/// a still-live group (root `Running`) holding a mix of already-`Done` and -/// still-`Pending` steps. Filtering out `Done` must drop exactly the `Done` -/// node and nothing else — the root and the `Pending` sibling both stay, -/// even though the root itself isn't in the requested state set. -#[test] -fn filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree() { - let nodes = vec![ - node(1, None, State::Running), // root: whole group still live - node(2, Some(1), State::Done), // finished step, should be hidden - node(3, Some(1), State::Pending), // 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 mut unfiltered: Vec = filter_nodes_by_state(nodes, None) - .into_iter() - .map(|n| n.id) - .collect(); - unfiltered.sort_unstable(); - assert_eq!(unfiltered, vec![1, 2, 3], "None is the identity filter"); -} - #[test] fn error_truncation_cuts_on_a_char_boundary() { // `truncate_error` is a pure `&str -> String`. This used to submit a DAG,