c0re: serve the jobq graph generically at /api/jobq/graph
`graph_node` projects a `hive_jobq::Node` onto the wire type from the parent commit: everything the crate records, with hive-c0re's own fields (`agent`, `approval_id`, `inputs`, `build_log_id`) collected into the opaque payload slot instead of standing as named columns. A node with nothing domain-specific to say serialises no `data` key at all, so the slot costs nothing when it is unused. `graph_dep` turns a `DepWhen` into the set of outcomes that satisfy the edge by asking it about each of the four terminal states, rather than leaking the bitflags representation onto the wire. `Resource::wire_name` gives the resource vocabulary a string form — hive-jobq is generic over the resource type, so a viewer that renders any graph cannot be handed this enum. The `agent:` prefix keeps per-agent leases from colliding with a global resource sharing an agent's name. `graph_snapshot` deliberately reuses `visible_dags` for retention: every live group plus the newest terminal ones. Serving the raw graph would grow without bound — evicted groups' nodes linger until bounded pruning lands. Within a retained group nothing is filtered: group roots ride as ordinary nodes, and `Done` nodes stay, which is the projection defect behind the "rebuild shows a single node" report. The endpoint lands in the same commit rather than after it. Without a consumer the whole projection is dead code, and a wire type nobody produces cannot be reviewed for whether it says the right things. Its OpenAPI body is `serde_json::Value`, matching `api_state`: no type in `hive-host-sock` derives `ToSchema`, and that crate stays dependency-lean on purpose.
This commit is contained in:
parent
792d7cb304
commit
f357f98867
4 changed files with 145 additions and 1 deletions
|
|
@ -205,6 +205,7 @@ pub async fn serve(
|
||||||
.routes(routes!(build_logs::get_build_log_stream))
|
.routes(routes!(build_logs::get_build_log_stream))
|
||||||
.routes(routes!(state_snapshot::dashboard_stream))
|
.routes(routes!(state_snapshot::dashboard_stream))
|
||||||
.routes(routes!(state_snapshot::dashboard_history))
|
.routes(routes!(state_snapshot::dashboard_history))
|
||||||
|
.routes(routes!(state_snapshot::jobq_graph))
|
||||||
.split_for_parts();
|
.split_for_parts();
|
||||||
// Just the JSON, not the UI — Swagger UI itself is nginx-hosted from
|
// Just the JSON, not the UI — Swagger UI itself is nginx-hosted from
|
||||||
// the nix store (see the module doc comment above `ApiDoc`). `api`
|
// the nix store (see the module doc comment above `ApiDoc`). `api`
|
||||||
|
|
@ -448,6 +449,7 @@ mod router_build_probe {
|
||||||
.routes(routes!(meta_inputs::post_meta_update))
|
.routes(routes!(meta_inputs::post_meta_update))
|
||||||
.routes(routes!(build_logs::get_build_log_stream))
|
.routes(routes!(build_logs::get_build_log_stream))
|
||||||
.routes(routes!(state_snapshot::dashboard_stream))
|
.routes(routes!(state_snapshot::dashboard_stream))
|
||||||
.routes(routes!(state_snapshot::dashboard_history));
|
.routes(routes!(state_snapshot::dashboard_history))
|
||||||
|
.routes(routes!(state_snapshot::jobq_graph));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -651,6 +651,27 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
out
|
out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[utoipa::path(
|
||||||
|
get,
|
||||||
|
path = "/api/jobq/graph",
|
||||||
|
responses(
|
||||||
|
(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 — a consumer \
|
||||||
|
renders the graph without knowing what any node means. Each element \
|
||||||
|
is a `hive_host_sock::graph::GraphNode`",
|
||||||
|
body = serde_json::Value),
|
||||||
|
),
|
||||||
|
tag = "state_snapshot"
|
||||||
|
)]
|
||||||
|
pub(super) async fn jobq_graph(
|
||||||
|
State(state): State<AppState>,
|
||||||
|
) -> axum::Json<Vec<hive_host_sock::graph::GraphNode>> {
|
||||||
|
axum::Json(state.coord.job_queue.graph_snapshot())
|
||||||
|
}
|
||||||
|
|
||||||
#[utoipa::path(
|
#[utoipa::path(
|
||||||
get,
|
get,
|
||||||
path = "/api/dashboard/history",
|
path = "/api/dashboard/history",
|
||||||
|
|
|
||||||
|
|
@ -39,6 +39,7 @@ mod tests;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
|
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
|
use hive_host_sock::graph::{GraphDep, GraphNode, NodePayload};
|
||||||
use hive_host_sock::jobs::NodeView;
|
use hive_host_sock::jobs::NodeView;
|
||||||
use hive_jobq::resources::ResourceTable;
|
use hive_jobq::resources::ResourceTable;
|
||||||
use hive_jobq::scheduler::{Outcome, Scheduler};
|
use hive_jobq::scheduler::{Outcome, Scheduler};
|
||||||
|
|
@ -354,6 +355,36 @@ impl JobQueue {
|
||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Every node of every visible group, as generic graph nodes.
|
||||||
|
///
|
||||||
|
/// **Nothing is hidden.** Group roots ride as ordinary nodes (so a
|
||||||
|
/// consumer needs no special case for "the container" and reads the
|
||||||
|
/// root's own `state` as the group's answer), and `Done` nodes stay (so a
|
||||||
|
/// finished step is visible rather than vanishing from the payload, which
|
||||||
|
/// is what makes a fast rebuild render as a single node).
|
||||||
|
///
|
||||||
|
/// Retention is the *same* bound as [`Self::snapshot`] — every live group
|
||||||
|
/// plus the newest terminal ones, via [`visible_dags`]. Serving the raw
|
||||||
|
/// graph instead would grow without limit: evicted groups' nodes linger
|
||||||
|
/// until bounded pruning lands.
|
||||||
|
#[must_use]
|
||||||
|
pub fn graph_snapshot(&self) -> Vec<GraphNode> {
|
||||||
|
let inner = self.lock();
|
||||||
|
let mut roots = visible_dags(&inner);
|
||||||
|
roots.sort_unstable_by_key(|root| root.get());
|
||||||
|
roots
|
||||||
|
.into_iter()
|
||||||
|
.flat_map(|root| {
|
||||||
|
inner
|
||||||
|
.graph()
|
||||||
|
.node(root)
|
||||||
|
.into_iter()
|
||||||
|
.chain(inner.graph().descendants(root))
|
||||||
|
.map(graph_node)
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
|
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn snapshot(&self) -> Vec<DagView> {
|
pub fn snapshot(&self) -> Vec<DagView> {
|
||||||
|
|
@ -488,6 +519,79 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Project one graph node onto the generic wire shape.
|
||||||
|
///
|
||||||
|
/// Deliberately near-total: everything `hive_jobq` records is carried, and
|
||||||
|
/// the only editorial decision is what goes in the opaque payload. That is
|
||||||
|
/// the inverse of [`dag_view`], which decides what a consumer is allowed to
|
||||||
|
/// see — a viewer that can render *any* graph cannot have that decided for it.
|
||||||
|
fn graph_node(node: &hive_jobq::Node<NodeKind, Resource>) -> GraphNode {
|
||||||
|
GraphNode {
|
||||||
|
id: node.id.get(),
|
||||||
|
parent: node.parent.map(NodeId::get),
|
||||||
|
state: node.state,
|
||||||
|
deps: node.deps.iter().map(graph_dep).collect(),
|
||||||
|
started_at: node.started_at,
|
||||||
|
finished_at: node.finished_at,
|
||||||
|
error: node.error.clone(),
|
||||||
|
payload: NodePayload {
|
||||||
|
label: node.payload.as_str().to_owned(),
|
||||||
|
data: node_data(node.id, &node.payload),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn graph_dep(dep: &Dep<Resource>) -> GraphDep {
|
||||||
|
match dep {
|
||||||
|
Dep::Node { id, when } => GraphDep::Node {
|
||||||
|
id: id.get(),
|
||||||
|
accepts: [
|
||||||
|
TerminalState::Done,
|
||||||
|
TerminalState::Failed,
|
||||||
|
TerminalState::Cancelled,
|
||||||
|
TerminalState::Skipped,
|
||||||
|
]
|
||||||
|
.into_iter()
|
||||||
|
.filter(|outcome| when.accepts(*outcome))
|
||||||
|
.collect(),
|
||||||
|
},
|
||||||
|
Dep::Resource { name, count } => GraphDep::Resource {
|
||||||
|
name: name.wire_name(),
|
||||||
|
count: *count,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// hive-c0re's domain data for one node, as the opaque payload slot.
|
||||||
|
///
|
||||||
|
/// Every field here used to be a named column on `NodeView`, meaningful for
|
||||||
|
/// one node kind and `null` on all the others. As free-form data it costs the
|
||||||
|
/// wire type nothing and a generic consumer renders it without knowing what
|
||||||
|
/// any of it means.
|
||||||
|
fn node_data(id: NodeId, kind: &NodeKind) -> serde_json::Value {
|
||||||
|
let mut data = serde_json::Map::new();
|
||||||
|
let agent = kind.agent();
|
||||||
|
if !agent.is_empty() {
|
||||||
|
data.insert("agent".to_owned(), agent.into());
|
||||||
|
}
|
||||||
|
if let NodeKind::DeployWindow { approval_id, .. } = kind {
|
||||||
|
data.insert("approval_id".to_owned(), (*approval_id).into());
|
||||||
|
}
|
||||||
|
if let NodeKind::MetaLock { inputs, .. } = kind
|
||||||
|
&& !inputs.is_empty()
|
||||||
|
{
|
||||||
|
data.insert("inputs".to_owned(), inputs.clone().into());
|
||||||
|
}
|
||||||
|
if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())) {
|
||||||
|
data.insert("build_log_id".to_owned(), log.into());
|
||||||
|
}
|
||||||
|
if data.is_empty() {
|
||||||
|
serde_json::Value::Null
|
||||||
|
} else {
|
||||||
|
serde_json::Value::Object(data)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Which of a DAG's work nodes ride the wire, by index into `states` — or
|
/// Which of a DAG's work nodes ride the wire, by index into `states` — or
|
||||||
/// `None` when the DAG has nothing left worth showing and drops out of the
|
/// `None` when the DAG has nothing left worth showing and drops out of the
|
||||||
/// snapshot entirely.
|
/// snapshot entirely.
|
||||||
|
|
|
||||||
|
|
@ -57,3 +57,20 @@ pub enum Resource {
|
||||||
/// held across the holder's whole subtree).
|
/// held across the holder's whole subtree).
|
||||||
MetaWindow,
|
MetaWindow,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl Resource {
|
||||||
|
/// This resource's name on the generic graph wire.
|
||||||
|
///
|
||||||
|
/// `hive_jobq` is generic over the resource type, so a viewer that can
|
||||||
|
/// render any graph gets a string here rather than this enum. The
|
||||||
|
/// `agent:` prefix keeps the per-agent leases from colliding with a
|
||||||
|
/// hypothetical global resource that happens to share an agent's name.
|
||||||
|
#[must_use]
|
||||||
|
pub fn wire_name(&self) -> String {
|
||||||
|
match self {
|
||||||
|
Resource::BuildSlot => "build-slot".to_owned(),
|
||||||
|
Resource::Agent(agent) => format!("agent:{agent}"),
|
||||||
|
Resource::MetaWindow => "meta-window".to_owned(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue