diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs index d3a06761..efe09319 100644 --- a/hive-jobq-wire/src/lib.rs +++ b/hive-jobq-wire/src/lib.rs @@ -354,6 +354,46 @@ fn state_index(state: State) -> usize { } } +/// 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 { @@ -400,7 +440,8 @@ mod tests { use super::{ ALL_STATES, BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, - TerminalState, TerminalStateSchema, WireId, WireNode, state_rollup, + TerminalState, TerminalStateSchema, WireId, WireNode, filter_nodes_by_state, parse_states, + state_rollup, }; /// The mirrors document what the wire actually says — a mirror that @@ -630,6 +671,50 @@ mod tests { ); } + #[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. diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 6b768d6d..a42aaa95 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -34,15 +34,18 @@ use utoipa_axum::{router::OpenApiRouter, routes}; mod status; -/// Placeholder node payload — uninhabited on purpose. This wires the graph -/// and its read-only endpoints; giving it real variants waits on there -/// being an actual job (agent creation) to run. `WireNode` is trivially -/// satisfiable on an empty enum (`match *self {}`), so the wire machinery -/// below is real and typechecked today, with nothing yet to put in it. +/// Placeholder node payload for the swarm-level job graph — uninhabited on +/// purpose, and named `Swarm*` rather than the bare `NodeKind`/`Resource` +/// `hive-c0re::job_queue::model` already uses, so a grep for either doesn't +/// land on both crates. This wires the graph and its read-only endpoints; +/// giving it real variants waits on there being an actual job (agent +/// creation) to run. `WireNode` is trivially satisfiable on an empty enum +/// (`match *self {}`), so the wire machinery below is real and typechecked +/// today, with nothing yet to put in it. #[derive(Clone, Debug)] -enum NodeKind {} +enum SwarmNodeKind {} -impl hive_jobq_wire::WireNode for NodeKind { +impl hive_jobq_wire::WireNode for SwarmNodeKind { fn label(&self) -> String { match *self {} } @@ -53,11 +56,11 @@ impl hive_jobq_wire::WireNode for NodeKind { } /// Placeholder resource name — same rationale and same "no variants until a -/// real node needs one" shape as [`NodeKind`]. +/// real node needs one" shape as [`SwarmNodeKind`]. #[derive(Clone, Debug, PartialEq, Eq, Hash)] -enum ResourceKind {} +enum SwarmResourceKind {} -impl hive_jobq_wire::WireResource for ResourceKind { +impl hive_jobq_wire::WireResource for SwarmResourceKind { fn name(&self) -> String { match *self {} } @@ -140,11 +143,10 @@ struct AppState { /// `async-nats` reconnects underneath it. status: Option>, /// The swarm-level job graph. `std::sync::Mutex`, not `tokio`'s — every - /// lock scope below is synchronous (no `.await` while held), same - /// choice `hive-c0re::job_queue::JobQueue` makes for the same reason. - /// Always `Some` graph, never gated on the swarm queue: this is process - /// state, not something read over the network. - jobq: Arc>>, + /// lock scope below is synchronous (no `.await` while held). Always a + /// graph, never gated on the swarm queue: this is process state, not + /// something read over the network. + jobq: Arc>>, } /// Env var the controller's NixOS module sets from @@ -288,53 +290,24 @@ async fn get_hives_status( } } -/// Query params for `GET /api/jobq/graph` — same shape and same rationale as -/// `hive-c0re::dashboard::state_snapshot::JobqGraphQuery`. +/// Query params for `GET /api/jobq/graph` — `?states=` narrows to root +/// groups in the named states, same shape `hive_jobq_wire::parse_states` +/// parses. #[derive(Deserialize, utoipa::IntoParams)] struct JobqGraphQuery { states: Option, } -/// Same parsing rules as `hive-c0re`'s own `parse_states`: an absent or -/// entirely-unparseable query is "no filter," never "match nothing." -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) -} - -/// Same shape as `hive-c0re::job_queue::filter_nodes_by_state`: a flat -/// filter on each node's own state, no special-casing for a root vs. a -/// descendant. `None` (no query given) is a no-op, not "match nothing." -fn filter_nodes_by_state( - nodes: Vec, - states: Option<&[hive_jobq::State]>, -) -> Vec { - let Some(states) = states else { - return nodes; - }; - nodes - .into_iter() - .filter(|n| states.contains(&n.state)) - .collect() -} - -/// Every node of every root group in the swarm-level job graph — same shape -/// as `hive-c0re`'s `/api/jobq/graph`. Nothing filters "done and old" here -/// the way `hive-c0re::job_queue::visible_roots` does, because nothing is -/// old yet: this wires the graph with no nodes, so every root the graph -/// has is worth showing. +/// Every node of every root group in the swarm-level job graph. Nothing +/// filters "done and old" here — with zero nodes ever submitted there is +/// nothing to bound yet, so `graph.roots()` (everything) is the whole +/// roster passed to [`hive_jobq_wire::GraphWire::wire_snapshot`]. #[utoipa::path( get, path = "/api/jobq/graph", params(JobqGraphQuery), responses((status = 200, description = "every node of every root group, as generic \ - `hive_jobq` graph nodes — same shape as hive-c0re's own `/api/jobq/graph`. \ - `?states=` narrows to root groups in the named states.", + `hive_jobq` graph nodes. `?states=` narrows to root groups in the named states.", body = Vec)), tag = "jobq" )] @@ -342,23 +315,24 @@ async fn get_jobq_graph( State(state): State, axum::extract::Query(q): axum::extract::Query, ) -> Json> { - let states = parse_states(q.states.as_deref()); + let states = hive_jobq_wire::parse_states(q.states.as_deref()); let graph = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let roots: Vec = graph.roots().map(|n| n.id).collect(); let nodes = graph.wire_snapshot(roots); - Json(filter_nodes_by_state(nodes, states.as_deref())) + Json(hive_jobq_wire::filter_nodes_by_state( + nodes, + states.as_deref(), + )) } -/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves — -/// same shape as `hive-c0re`'s own `/api/jobq/rollup`. +/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves. #[utoipa::path( get, path = "/api/jobq/rollup", - responses((status = 200, description = "counts by lifecycle state, same shape as \ - hive-c0re's own `/api/jobq/rollup`", body = Vec)), + responses((status = 200, description = "counts by lifecycle state", body = Vec)), tag = "jobq" )] async fn get_jobq_rollup(State(state): State) -> Json> {