diff --git a/Cargo.lock b/Cargo.lock index f0956b19..bae5012b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4563,6 +4563,8 @@ dependencies = [ "async-nats", "axum", "futures-util", + "hive-jobq", + "hive-jobq-wire", "serde", "serde_json", "swarm-queue-client", diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 8448176c..cb5f2f54 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -18,6 +18,14 @@ 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 3c9e675e..6b768d6d 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -23,16 +23,46 @@ use std::os::unix::fs::PermissionsExt as _; use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; 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 — 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. +#[derive(Clone, Debug)] +enum NodeKind {} + +impl hive_jobq_wire::WireNode for NodeKind { + 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 [`NodeKind`]. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +enum ResourceKind {} + +impl hive_jobq_wire::WireResource for ResourceKind { + 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 @@ -68,6 +98,7 @@ 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; @@ -108,6 +139,12 @@ 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), 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>>, } /// Env var the controller's NixOS module sets from @@ -251,6 +288,88 @@ async fn get_hives_status( } } +/// Query params for `GET /api/jobq/graph` — same shape and same rationale as +/// `hive-c0re::dashboard::state_snapshot::JobqGraphQuery`. +#[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. +#[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.", + body = Vec)), + tag = "jobq" +)] +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 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())) +} + +/// Counts by lifecycle state over the same groups `/api/jobq/graph` serves — +/// same shape as `hive-c0re`'s own `/api/jobq/rollup`. +#[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)), + 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() @@ -334,6 +453,7 @@ 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()) @@ -341,6 +461,8 @@ 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