diff --git a/Cargo.lock b/Cargo.lock index bae5012b..f0956b19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4563,8 +4563,6 @@ dependencies = [ "async-nats", "axum", "futures-util", - "hive-jobq", - "hive-jobq-wire", "serde", "serde_json", "swarm-queue-client", diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 2258608a..95ece62e 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -580,23 +580,35 @@ fn build_approval_views(approvals: Vec) -> Vec { out } -/// `/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. +/// `/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. #[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", @@ -605,14 +617,10 @@ pub(super) struct JobqGraphQuery { (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. 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.", + 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.", body = Vec), ), tag = "state_snapshot" @@ -621,7 +629,7 @@ pub(super) async fn jobq_graph( State(state): State, axum::extract::Query(q): axum::extract::Query, ) -> axum::Json> { - let states = hive_jobq_wire::parse_states(q.states.as_deref()); + let states = parse_states(q.states.as_deref()); axum::Json(state.coord.job_queue.graph_snapshot(states.as_deref())) } @@ -631,9 +639,7 @@ 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 — 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 \ + 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 \ diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 509bbac8..e5fb5704 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -351,16 +351,15 @@ impl JobQueue { /// still-matching descendant one level higher rather than hiding or /// orphaning it. /// - /// 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. + /// 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. #[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); - hive_jobq_wire::filter_nodes_by_state(nodes, states) + filter_nodes_by_state(nodes, states) } /// Per-state counts over the **same** groups [`JobQueue::graph_snapshot`] @@ -415,6 +414,26 @@ 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 11988883..4e27357a 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1527,6 +1527,60 @@ 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, diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs index efe09319..d3a06761 100644 --- a/hive-jobq-wire/src/lib.rs +++ b/hive-jobq-wire/src/lib.rs @@ -354,46 +354,6 @@ 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 { @@ -440,8 +400,7 @@ mod tests { use super::{ ALL_STATES, BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, - TerminalState, TerminalStateSchema, WireId, WireNode, filter_nodes_by_state, parse_states, - state_rollup, + TerminalState, TerminalStateSchema, WireId, WireNode, state_rollup, }; /// The mirrors document what the wire actually says — a mirror that @@ -671,50 +630,6 @@ 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/Cargo.toml b/swarm-controller/Cargo.toml index cb5f2f54..8448176c 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -18,14 +18,6 @@ anyhow.workspace = true async-nats = { workspace = true, features = ["kv"] } axum.workspace = true futures-util.workspace = true -# The graph itself, held directly rather than behind a c0re-style wrapper -# module — that layering (`hive-c0re::job_queue`) is partially legacy (predates -# `hive-jobq`'s extraction into its own crate) and this daemon does not need it -# repeated. No `Scheduler` yet either: nothing here submits or executes a job, -# so there is nothing to schedule — just a `Graph` for the read-only endpoints -# to serve. -hive-jobq.workspace = true -hive-jobq-wire.workspace = true serde.workspace = true serde_json.workspace = true # The queue connect (token mint + auth callback + reconnect) is shared with diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index a42aaa95..3c9e675e 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -23,49 +23,16 @@ use std::os::unix::fs::PermissionsExt as _; use std::path::PathBuf; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use anyhow::{Context, Result}; use axum::{Json, extract::State, routing::get}; -use hive_jobq_wire::GraphWire as _; use serde::{Deserialize, Serialize}; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; mod status; -/// 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 SwarmNodeKind {} - -impl hive_jobq_wire::WireNode for SwarmNodeKind { - fn label(&self) -> String { - match *self {} - } - - fn data(&self, _id: hive_jobq_wire::WireId) -> serde_json::Value { - match *self {} - } -} - -/// Placeholder resource name — same rationale and same "no variants until a -/// real node needs one" shape as [`SwarmNodeKind`]. -#[derive(Clone, Debug, PartialEq, Eq, Hash)] -enum SwarmResourceKind {} - -impl hive_jobq_wire::WireResource for SwarmResourceKind { - fn name(&self) -> String { - match *self {} - } -} - /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. /// /// A compiled-in default is legitimate here and is *not* the mistake that @@ -101,7 +68,6 @@ fn socket_path() -> PathBuf { (name = "health", description = "liveness probe"), (name = "hives", description = "the swarm's hive directory"), (name = "links", description = "swarm service quick links"), - (name = "jobq", description = "the swarm-level job graph"), ) )] struct ApiDoc; @@ -142,11 +108,6 @@ struct AppState { /// that is merely *unreachable* still yields a reader, because /// `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). 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 @@ -290,60 +251,6 @@ async fn get_hives_status( } } -/// 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, -} - -/// 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. `?states=` narrows to root groups in the named states.", - body = Vec)), - tag = "jobq" -)] -async fn get_jobq_graph( - State(state): State, - axum::extract::Query(q): axum::extract::Query, -) -> Json> { - 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(hive_jobq_wire::filter_nodes_by_state( - nodes, - states.as_deref(), - )) -} - -/// 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", body = Vec)), - tag = "jobq" -)] -async fn get_jobq_rollup(State(state): State) -> Json> { - let graph = state - .jobq - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let roots: Vec = graph.roots().map(|n| n.id).collect(); - Json(hive_jobq_wire::state_rollup(&graph, roots)) -} - #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -427,7 +334,6 @@ async fn main() -> Result<()> { hives: Arc::new(load_hives()), links: Arc::new(load_links()), status, - jobq: Arc::new(Mutex::new(hive_jobq::Graph::new())), }; let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi()) @@ -435,8 +341,6 @@ async fn main() -> Result<()> { .routes(routes!(get_hives)) .routes(routes!(get_hives_status)) .routes(routes!(get_links)) - .routes(routes!(get_jobq_graph)) - .routes(routes!(get_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). `api` is