vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

address review: move parse_states/filter_nodes_by_state to hive-jobq-wire, rename placeholder enums, trim core-mirroring framing

Browse source
This commit is contained in:
damocles 2026-08-16 16:34:17 +02:00 • committed by mara
commit 08efd7875e
2 changed files with 118 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

87
hive-jobq-wire/src/lib.rs
View file

@ -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<Vec<State>> {
let states: Vec<State> = 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<GraphNode>, states: Option<&[State]>) -> Vec<GraphNode> {
let Some(states) = states else {
return nodes;
};
nodes
.into_iter()
.filter(|n| states.contains(&n.state))
.collect()
}
fn wire_node<N: WireNode, R: WireResource>(node: &hive_jobq::Node<N, R>) -> 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<WireId> =
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.

90
swarm-controller/src/main.rs
View file

@ -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<Arc<status::StatusReader>>,
/// 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<Mutex<hive_jobq::Graph<NodeKind, ResourceKind>>>,
/// 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<Mutex<hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>>>,
}
/// 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<String>,
}
/// 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<Vec<hive_jobq::State>> {
let states: Vec<hive_jobq::State> = 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<hive_jobq_wire::GraphNode>,
states: Option<&[hive_jobq::State]>,
) -> Vec<hive_jobq_wire::GraphNode> {
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<hive_jobq_wire::GraphNode>)),
tag = "jobq"
)]
@ -342,23 +315,24 @@ async fn get_jobq_graph(
State(state): State<AppState>,
axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
) -> Json<Vec<hive_jobq_wire::GraphNode>> {
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<hive_jobq::NodeId> = 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<hive_jobq_wire::StateCount>)),
responses((status = 200, description = "counts by lifecycle state", body = Vec<hive_jobq_wire::StateCount>)),
tag = "jobq"
)]
async fn get_jobq_rollup(State(state): State<AppState>) -> Json<Vec<hive_jobq_wire::StateCount>> {