From 7966d5eb660ca21311a5f1abb8444b153968a773 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 23:52:32 +0200 Subject: [PATCH] jobq-wire: move the generic graph projection into its own crate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wire types were in hive-host-sock, which is the host *socket* crate — so anything living there is core-shaped by construction, and the projection had quietly grown two core dependencies to match: it selected roots by matching NodeKind::Dag, and rendered payloads through free functions in hive-c0re that nothing obliged a second host to write. hive-jobq is the wrong home too. That crate is the scheduler — logic — and folding presentation in means every consumer of it carries a JSON vocabulary it may never serve. So: a new hive-jobq-wire. A host implements WireNode for its payload N and WireResource for its resource name R; GraphWire::wire_snapshot is blanket-implemented for Graph when both hold, and for nothing else. A payload that has never said how it displays has no way onto the wire. wire_snapshot takes the roots to serve rather than reading Graph::roots itself. Nothing is ever removed from a Graph, so retention is a policy only the host can hold; hive-c0re passes visible_roots(), which is the existing MAX_HISTORY_DAGS bound selected structurally (a root is a node with no parent) instead of by node kind. --- CLAUDE.md | 7 + Cargo.lock | 12 +- Cargo.toml | 2 + hive-c0re/Cargo.toml | 1 + hive-c0re/src/dashboard/state_snapshot.rs | 4 +- hive-c0re/src/job_queue/mod.rs | 145 +++----- hive-c0re/src/job_queue/model.rs | 38 ++ hive-c0re/src/job_queue/resource.rs | 17 +- hive-host-sock/Cargo.toml | 1 - hive-host-sock/src/graph.rs | 192 ---------- hive-host-sock/src/lib.rs | 1 - hive-jobq-wire/Cargo.toml | 14 + hive-jobq-wire/README.md | 16 + hive-jobq-wire/src/lib.rs | 411 ++++++++++++++++++++++ 14 files changed, 564 insertions(+), 297 deletions(-) delete mode 100644 hive-host-sock/src/graph.rs create mode 100644 hive-jobq-wire/Cargo.toml create mode 100644 hive-jobq-wire/README.md create mode 100644 hive-jobq-wire/src/lib.rs diff --git a/CLAUDE.md b/CLAUDE.md index b88d9d3b..20201153 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,6 +63,13 @@ hand-maintained per-file tree drifts out of sync with the code. remaining `job_queue/` module is the c0re-specific layer *over* this crate, and is being removed in favour of it — new scheduler-shaped code belongs here, not there. +- **`hive-jobq-wire/`** — wire types for serving a `hive-jobq` graph to a + viewer, plus the `WireNode` / `WireResource` traits a host implements to + say how its `N` and `R` render. Deliberately *not* part of `hive-jobq`: + that crate is logic, this is presentation, and folded together they + remix. `GraphWire::wire_snapshot` is blanket-implemented for any + `Graph` whose parameters implement both — so a payload that has + never said how it displays cannot reach a viewer at all. - **`hive-screen-mcp/`** — stdio MCP bridge for GUI agents (`hyperhive.gui.enable`): `screenshot` via `grim`, `type_text` / `key_press` via `wtype` (Wayland virtual-keyboard protocol), and diff --git a/Cargo.lock b/Cargo.lock index 00118c83..ddbeb5ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1636,6 +1636,7 @@ dependencies = [ "hive-core-agent-sock", "hive-host-sock", "hive-jobq", + "hive-jobq-wire", "hive-priv-sock", "hive-sh4re", "hive-sock-client", @@ -1727,7 +1728,6 @@ dependencies = [ "hive-sh4re", "hive-types", "serde", - "serde_json", ] [[package]] @@ -1742,6 +1742,16 @@ dependencies = [ "uuid", ] +[[package]] +name = "hive-jobq-wire" +version = "0.1.0" +dependencies = [ + "chrono", + "hive-jobq", + "serde", + "serde_json", +] + [[package]] name = "hive-matrix-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 56a33612..50181767 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "hive-forge-notify", "hive-host-sock", "hive-jobq", + "hive-jobq-wire", "hive-matrix-mcp", "hive-metric", "hive-priv", @@ -56,6 +57,7 @@ indicatif = "0.18" hive-sh4re = { path = "hive-sh4re" } hive-agent-sock = { path = "hive-agent-sock" } hive-jobq = { path = "hive-jobq" } +hive-jobq-wire = { path = "hive-jobq-wire" } hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-claude = "0.1" hive-host-sock = { path = "hive-host-sock" } diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index b87de707..54e97769 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -36,6 +36,7 @@ hive-core-agent-sock.workspace = true hive-sh4re.workspace = true hive-host-sock.workspace = true hive-jobq.workspace = true +hive-jobq-wire.workspace = true hive-priv-sock.workspace = true hive-agent-sock.workspace = true hive-sock-client.workspace = true diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 09763b60..f6c58682 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -661,14 +661,14 @@ fn build_approval_views(approvals: Vec) -> Vec { 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`", + is a `hive_jobq_wire::GraphNode`", body = serde_json::Value), ), tag = "state_snapshot" )] pub(super) async fn jobq_graph( State(state): State, -) -> axum::Json> { +) -> axum::Json> { axum::Json(state.coord.job_queue.graph_snapshot()) } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index f42da480..561d8c67 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -39,11 +39,11 @@ mod tests; use std::sync::{Arc, Mutex}; use chrono::{DateTime, Utc}; -use hive_host_sock::graph::{GraphDep, GraphNode, NodePayload}; use hive_host_sock::jobs::NodeView; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq::{Dep, Graph, NodeId}; +use hive_jobq_wire::{GraphNode, GraphWire}; use tokio::sync::Notify; pub use hive_jobq::TerminalState; @@ -363,26 +363,13 @@ impl JobQueue { /// 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. + /// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies + /// is *which* groups to show — see [`visible_roots`] for why the graph + /// can't decide that for itself. #[must_use] pub fn graph_snapshot(&self) -> Vec { 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() + inner.graph().wire_snapshot(visible_roots(&inner)) } /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. @@ -519,79 +506,6 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option { }) } -/// 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) -> 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) -> 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 /// `None` when the DAG has nothing left worth showing and drops out of the /// snapshot entirely. @@ -665,6 +579,55 @@ fn visible_dags(sched: &Sched) -> Vec { retain_history(live, terminal, MAX_HISTORY_DAGS) } +/// The visible **group** set for [`Queue::graph_snapshot`]: every live group +/// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones. +/// +/// Same policy as [`visible_dags`], selected *structurally* — a root is a node +/// with no parent. The `DagView` path next door keys on `NodeKind::Dag` +/// instead, which is fine for a projection that already only means anything to +/// hive-c0re, but would make the generic endpoint depend on one node kind that +/// is itself slated for removal. +/// +/// **This bound is load-bearing, not tidiness.** Nothing ever removes a node +/// from the graph (bounded pruning is a Stage-C follow-up), so serving +/// `graph.roots()` directly would grow the payload without limit for the whole +/// uptime of the daemon. +fn visible_roots(sched: &Sched) -> Vec { + let roots: Vec = sched.graph().roots().map(|n| n.id).collect(); + let mut live: Vec = Vec::new(); + let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new(); + for root in roots { + if sched.graph().is_settled(root) == Some(true) { + terminal.push((root, group_finished_at(sched, root), root.get())); + } else { + live.push(root); + } + } + retain_history(live, terminal, MAX_HISTORY_DAGS) +} + +/// When a whole group last finished: the newest `finished_at` across the root +/// **and** its descendants. Unlike [`dag_finished_at`] the root itself counts, +/// because a generic group root can be an ordinary node with no children at +/// all — reading only descendants would date every such group to the epoch and +/// evict it first. +fn group_finished_at(sched: &Sched, root: NodeId) -> i64 { + sched + .graph() + .node(root) + .and_then(|n| n.finished_at) + .into_iter() + .chain( + sched + .graph() + .descendants(root) + .filter_map(|n| n.finished_at), + ) + .map(|t| t.timestamp()) + .max() + .unwrap_or(0) +} + /// [`visible_dags`]'s policy, split from the graph it reads: keep every live /// DAG, plus the newest `cap` terminal ones. /// diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index b95f89e8..263ccf2f 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -283,6 +283,44 @@ pub enum NodeKind { }, } +/// How a hive-c0re node describes itself to a generic graph viewer. +/// +/// Every field in [`WireNode::data`] 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. +impl hive_jobq_wire::WireNode for NodeKind { + fn label(&self) -> String { + self.as_str().to_owned() + } + + fn data(&self, id: hive_jobq_wire::WireId) -> serde_json::Value { + let mut data = serde_json::Map::new(); + let agent = self.agent(); + if !agent.is_empty() { + data.insert("agent".to_owned(), agent.into()); + } + if let NodeKind::DeployWindow { approval_id, .. } = self { + data.insert("approval_id".to_owned(), (*approval_id).into()); + } + if let NodeKind::MetaLock { inputs, .. } = self + && !inputs.is_empty() + { + data.insert("inputs".to_owned(), inputs.clone().into()); + } + // Not in the payload at all — the build log is keyed on node identity + // in a side table, which is why `data` is handed the id. + if let Some(log) = crate::build_logs::global().and_then(|h| h.id_for_node(id)) { + data.insert("build_log_id".to_owned(), log.into()); + } + if data.is_empty() { + serde_json::Value::Null + } else { + serde_json::Value::Object(data) + } + } +} + impl NodeKind { /// Wire string for `NodeView.kind`. pub fn as_str(&self) -> &'static str { diff --git a/hive-c0re/src/job_queue/resource.rs b/hive-c0re/src/job_queue/resource.rs index c8500983..dbb3fdc0 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -58,15 +58,14 @@ pub enum Resource { 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 { +/// 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. +impl hive_jobq_wire::WireResource for Resource { + fn name(&self) -> String { match self { Resource::BuildSlot => "build-slot".to_owned(), Resource::Agent(agent) => format!("agent:{agent}"), diff --git a/hive-host-sock/Cargo.toml b/hive-host-sock/Cargo.toml index 425d890d..ffd069fd 100644 --- a/hive-host-sock/Cargo.toml +++ b/hive-host-sock/Cargo.toml @@ -13,4 +13,3 @@ hive-jobq.workspace = true hive-sh4re.workspace = true hive-types.workspace = true serde.workspace = true -serde_json.workspace = true diff --git a/hive-host-sock/src/graph.rs b/hive-host-sock/src/graph.rs deleted file mode 100644 index d98853ed..00000000 --- a/hive-host-sock/src/graph.rs +++ /dev/null @@ -1,192 +0,0 @@ -//! Generic jobq graph wire types — a `hive_jobq` graph serialised without -//! knowing what its nodes mean. -//! -//! [`super::jobs`]'s `DagView` / `NodeView` are hive-c0re's *domain* -//! projection: they carry `approval_id`, `inputs`, `build_log_id` and an -//! `agent`, each meaningful for a subset of one specific node kind. Anything -//! built on those can only ever display hive-c0re's queue. -//! -//! These types carry what `hive_jobq::Node` itself carries — identity, the -//! parent tree, dependency edges, lifecycle — and push everything -//! domain-specific into one opaque [`NodePayload::data`] slot the consumer -//! renders without branching on. That is the crate boundary made visible: -//! `hive-jobq` owns structure, its host owns meaning. -//! -//! There is deliberately **no roll-up field**. A group root's own [`State`] -//! *is* its subtree's answer: `Finishing` means "own logic done, children -//! still running", and the terminal states are the rolled-up outcome. A -//! separate field would be a lossier copy of a value already on the wire — -//! lossier because it would have to flatten `Running` and `Finishing` together. - -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; - -pub use hive_jobq::{State, TerminalState}; - -/// Node id, carried verbatim from `hive_jobq::NodeId`: globally unique across -/// the whole graph, not per group. Opaque to consumers — they group by -/// [`GraphNode::parent`] and match dependency edges, nothing more. -pub type NodeId = u64; - -/// One node, serialised near-raw from `hive_jobq::Node`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct GraphNode { - pub id: NodeId, - /// Structural parent, or `None` for a group root. - /// - /// A group root is an **ordinary node here** — nothing is hidden, so a - /// consumer needs no special case for "the container", and the root's - /// own `state` answers "how is this whole group doing". - #[serde(default, skip_serializing_if = "Option::is_none")] - pub parent: Option, - pub state: State, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub deps: Vec, - /// When the node entered `Running`. `None` until it starts; a node that - /// never ran keeps `None`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option>, - /// When the node reached a terminal state. `None` while non-terminal. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option>, - /// Failure reason, set only when this node's *own* logic failed — a node - /// that rolled up `Failed` from a child carries none of its own. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub error: Option, - pub payload: NodePayload, -} - -/// What a node *is*, in terms the graph layer does not interpret. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NodePayload { - /// Short tag the consumer displays verbatim. **Not** for matching on: a - /// generic viewer that branches on this has stopped being generic. - pub label: String, - /// Domain data, rendered generically (as key/value, a details pane, a - /// tooltip — the consumer's choice). Everything a specific host wants to - /// say about a node beyond its label lives here, so adding a field costs - /// the wire type nothing. - #[serde(default, skip_serializing_if = "serde_json::Value::is_null")] - pub data: serde_json::Value, -} - -/// What must hold before a node runs. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "snake_case", tag = "kind")] -pub enum GraphDep { - /// Depend on another node finishing acceptably. - Node { - /// The node depended on. May name a node the consumer has not been - /// sent (a filtered view); treat an absent target as satisfied. - id: NodeId, - /// **The outcomes that satisfy this edge, as a set** — not a - /// strong/weak flag. - /// - /// A template routinely emits several tails edged on the *same* - /// upstream node, distinguished only by which outcomes each accepts - /// (one for `Done`, one for `Failed`/`Cancelled`, …). Collapsing this - /// to a boolean renders those as identical nodes. - accepts: Vec, - }, - /// Need `count` units of a named resource. Named, not typed: the resource - /// vocabulary belongs to the host, so it rides as a string. - Resource { name: String, count: u32 }, -} - -#[cfg(test)] -mod tests { - use super::{GraphDep, GraphNode, NodePayload, State, TerminalState}; - - fn node(id: u64, parent: Option, state: State, deps: Vec) -> GraphNode { - GraphNode { - id, - parent, - state, - deps, - started_at: None, - finished_at: None, - error: None, - payload: NodePayload { - label: "reconcile".to_owned(), - data: serde_json::Value::Null, - }, - } - } - - #[test] - fn a_node_round_trips_through_json() { - let before = node( - 7, - Some(3), - State::Running, - vec![ - GraphDep::Node { - id: 3, - accepts: vec![TerminalState::Done], - }, - GraphDep::Resource { - name: "build-slot".to_owned(), - count: 1, - }, - ], - ); - let json = serde_json::to_string(&before).expect("serialises"); - let after: GraphNode = serde_json::from_str(&json).expect("round trips"); - assert_eq!(format!("{after:?}"), format!("{before:?}")); - } - - #[test] - fn two_tails_on_one_upstream_are_distinguishable_by_their_accepted_set() { - // The reason `accepts` is a set and not a strong/weak bool: these two - // are the same shape and the same upstream, and *only* the outcome - // set tells them apart. - let ok_tail = node( - 10, - None, - State::Pending, - vec![GraphDep::Node { - id: 4, - accepts: vec![TerminalState::Done], - }], - ); - let fail_tail = node( - 11, - None, - State::Pending, - vec![GraphDep::Node { - id: 4, - accepts: vec![TerminalState::Failed, TerminalState::Cancelled], - }], - ); - let render = |n: &GraphNode| match n.deps.first() { - Some(GraphDep::Node { accepts, .. }) => format!("{accepts:?}"), - _ => "none".to_owned(), - }; - assert_ne!(render(&ok_tail), render(&fail_tail)); - } - - #[test] - fn a_group_root_is_an_ordinary_node_and_carries_the_groups_answer() { - // No roll-up field: `Finishing` on the root says "own logic done, - // children still running", and a terminal root state is the roll-up. - let root = node(1, None, State::Finishing, Vec::new()); - assert!(root.parent.is_none(), "a group root is just parentless"); - assert_eq!(root.state, State::Finishing); - - let done_root = node(2, None, State::Failed, Vec::new()); - assert_eq!( - done_root.state, - State::Failed, - "the root's own state is the subtree's rolled-up outcome" - ); - } - - #[test] - fn an_empty_payload_slot_is_omitted_from_the_wire() { - // The opaque slot costs nothing when a host has nothing to say. - let json = - serde_json::to_string(&node(1, None, State::Pending, Vec::new())).expect("serialises"); - assert!(!json.contains("\"data\""), "null data is skipped: {json}"); - assert!(json.contains("\"label\""), "the label always rides: {json}"); - } -} diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 8524d14f..2363e3db 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -16,7 +16,6 @@ use hive_sh4re::{AgentStatusRow, Approval}; use hive_types::Ident; use serde::{Deserialize, Serialize}; -pub mod graph; pub mod jobs; // ── Shared hive layout facts ────────────────────────────────────────────── diff --git a/hive-jobq-wire/Cargo.toml b/hive-jobq-wire/Cargo.toml new file mode 100644 index 00000000..6a0606fe --- /dev/null +++ b/hive-jobq-wire/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "hive-jobq-wire" +edition.workspace = true +version.workspace = true +readme = "README.md" + +[lints] +workspace = true + +[dependencies] +chrono = { workspace = true } +hive-jobq = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } diff --git a/hive-jobq-wire/README.md b/hive-jobq-wire/README.md new file mode 100644 index 00000000..78031a4b --- /dev/null +++ b/hive-jobq-wire/README.md @@ -0,0 +1,16 @@ +# hive-jobq-wire + +Wire types for serving a [`hive-jobq`](../hive-jobq) graph to a viewer, plus +the traits a host implements to say how its graph renders. + +**Why this is not part of `hive-jobq`.** The scheduler crate is logic: a graph, +a resource pool, a run loop. Presentation is a different concern with a +different audience, and folding it in means every consumer of the scheduler +also carries a JSON vocabulary it may never serve — the two get remixed and +stay that way. A separate crate keeps that boundary where it can be seen. + +A host implements `WireNode` for its node payload `N` and `WireResource` for +its resource name `R`. `GraphWire::wire_snapshot` is then blanket-implemented +for `hive_jobq::Graph` — so the projection exists exactly when both types +have said how they render, and a payload that hasn't cannot reach a viewer at +all. diff --git a/hive-jobq-wire/src/lib.rs b/hive-jobq-wire/src/lib.rs new file mode 100644 index 00000000..fcd18724 --- /dev/null +++ b/hive-jobq-wire/src/lib.rs @@ -0,0 +1,411 @@ +//! A [`hive_jobq`] graph, serialised without knowing what its nodes mean. +//! +//! A [`Graph`] is generic over its node payload `N` and resource name `R`, so +//! the scheduler can carry a host's domain types without interpreting them. +//! This crate extends that to the wire: [`GraphNode`] carries what a [`Node`] +//! itself carries — identity, the parent tree, dependency edges, lifecycle — +//! and pushes everything domain-specific into one opaque [`NodePayload::data`] +//! slot the consumer renders without branching on. +//! +//! **Why a separate crate.** `hive-jobq` is logic; this is presentation. Held +//! together they remix, and every consumer of the scheduler ends up carrying a +//! JSON vocabulary it may never serve. +//! +//! A host says how its `N` and `R` render by implementing [`WireNode`] and +//! [`WireResource`]. [`GraphWire`] is then blanket-implemented for any +//! `Graph` whose two type parameters do — so the projection exists +//! exactly when both have answered, and a payload that has never said how it +//! displays cannot reach a viewer at all. The alternative (a projection written +//! next to each host's endpoint) lets the second host ship a graph of +//! unlabelled integers. +//! +//! There is deliberately **no roll-up field**. A group root's own [`State`] +//! *is* its subtree's answer: `Finishing` means "own logic done, children still +//! running", and the terminal states are the rolled-up outcome. A separate +//! field would be a lossier copy of a value already on the wire — lossier +//! because it would have to flatten `Running` and `Finishing` together. +//! +//! [`Node`]: hive_jobq::Node + +use chrono::{DateTime, Utc}; +use hive_jobq::{Dep, Graph, NodeId, State, TerminalState}; +use serde::{Deserialize, Serialize}; + +/// Node id, carried verbatim from [`hive_jobq::NodeId`]: globally unique across +/// the whole graph, not per group. Opaque to consumers — they group by +/// [`GraphNode::parent`] and match dependency edges, nothing more. +pub type WireId = u64; + +/// How a node payload `N` describes itself to a generic viewer. +/// +/// Half of [`GraphWire`]'s bound, which is the enforcement: no impl, no +/// projection — a payload that has not answered these two questions has no way +/// onto the wire. +pub trait WireNode { + /// Short tag the consumer displays verbatim. **Not** for matching on: a + /// generic viewer that branches on this has stopped being generic. + fn label(&self) -> String; + + /// Domain data for this node, rendered generically (key/value, a details + /// pane, a tooltip — the consumer's choice). Return + /// [`serde_json::Value::Null`] when there is nothing to add; it is then + /// omitted from the wire entirely. + /// + /// Takes the node's `id` because a host's extra data is not always *in* the + /// payload — it may be keyed on node identity in a side table (a build log, + /// an artifact store) that only the host can resolve. + fn data(&self, id: WireId) -> serde_json::Value; +} + +/// How a resource name `R` rides the wire. +/// +/// Resource deps are named rather than typed on the wire: the resource +/// vocabulary belongs to the host, and a viewer only ever groups by the string. +pub trait WireResource { + /// Stable display name for this resource. + fn name(&self) -> String; +} + +/// One node, serialised near-raw from [`hive_jobq::Node`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GraphNode { + pub id: WireId, + /// Structural parent, or `None` for a group root. + /// + /// A group root is an **ordinary node here** — nothing is hidden, so a + /// consumer needs no special case for "the container", and the root's own + /// `state` answers "how is this whole group doing". + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent: Option, + pub state: State, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub deps: Vec, + /// When the node entered `Running`. `None` until it starts; a node that + /// never ran keeps `None`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at: Option>, + /// When the node reached a terminal state. `None` while non-terminal. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at: Option>, + /// Failure reason, set only when this node's *own* logic failed — a node + /// that rolled up `Failed` from a child carries none of its own. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + pub payload: NodePayload, +} + +/// What a node *is*, in terms the graph layer does not interpret. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NodePayload { + /// From [`WireNode::label`]. + pub label: String, + /// From [`WireNode::data`]. Everything a specific host wants to say about a + /// node beyond its label lives here, so adding a field costs the wire type + /// nothing. + #[serde(default, skip_serializing_if = "serde_json::Value::is_null")] + pub data: serde_json::Value, +} + +/// What must hold before a node runs. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", tag = "kind")] +pub enum GraphDep { + /// Depend on another node finishing acceptably. + Node { + /// The node depended on. May name a node the consumer has not been sent + /// (a filtered view); treat an absent target as satisfied. + id: WireId, + /// **The outcomes that satisfy this edge, as a set** — not a + /// strong/weak flag. + /// + /// A host routinely emits several tails edged on the *same* upstream + /// node, distinguished only by which outcomes each accepts (one for + /// `Done`, one for `Failed`/`Cancelled`, …). Collapsing this to a + /// boolean renders those as identical nodes. + accepts: Vec, + }, + /// Need `count` units of a named resource, per [`WireResource::name`]. + Resource { name: String, count: u32 }, +} + +/// Serving a [`Graph`] to a viewer. +/// +/// Blanket-implemented for every `Graph` whose payload and resource types +/// implement [`WireNode`] / [`WireResource`], and for no others — an extension +/// trait rather than an inherent method because the graph lives in the +/// scheduler crate and this projection deliberately does not. +pub trait GraphWire { + /// `roots` and all their descendants as [`GraphNode`]s, ordered by root id + /// with each root immediately followed by its own subtree. + /// + /// Takes the roots rather than reading [`Graph::roots`] itself, because + /// *which* groups to show is the host's policy and the graph has no opinion + /// on it — nothing in a `Graph` is ever removed, so a host that keeps + /// finished work bounded must bound it at this call. Pass `graph.roots()` + /// ids to serve everything. + /// + /// What is *not* the host's business is picking those roots by matching on + /// `N`: a root is a node with no parent, which is structure. Keying a + /// generic view on one domain's node vocabulary is exactly what this crate + /// exists to avoid. + fn wire_snapshot(&self, roots: impl IntoIterator) -> Vec; +} + +impl GraphWire for Graph { + fn wire_snapshot(&self, roots: impl IntoIterator) -> Vec { + let mut roots: Vec = roots.into_iter().collect(); + roots.sort_unstable_by_key(|id| id.get()); + roots + .into_iter() + .flat_map(|root| { + self.node(root) + .into_iter() + .chain(self.descendants(root)) + .map(wire_node) + }) + .collect() + } +} + +fn wire_node(node: &hive_jobq::Node) -> GraphNode { + let id = node.id.get(); + GraphNode { + id, + parent: node.parent.map(NodeId::get), + state: node.state, + deps: node.deps.iter().map(wire_dep).collect(), + started_at: node.started_at, + finished_at: node.finished_at, + error: node.error.clone(), + payload: NodePayload { + label: node.payload.label(), + data: node.payload.data(id), + }, + } +} + +fn wire_dep(dep: &Dep) -> 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.name(), + count: *count, + }, + } +} + +#[cfg(test)] +mod tests { + use hive_jobq::Graph; + use hive_jobq::resources::ResourceTable; + use hive_jobq::scheduler::Scheduler; + + use super::{ + GraphDep, GraphNode, GraphWire, NodePayload, State, TerminalState, WireId, WireNode, + }; + + /// A scheduler over the test payloads, built through the same public API a + /// host uses. + /// + /// These tests originally reached for `Graph::insert` — which is + /// `pub(crate)` to `hive-jobq`, so they only ever compiled while this + /// module still lived *inside* that crate. That they now have to go through + /// `insert_job` is the point rather than an inconvenience: it proves the + /// projection needs no privileged access to a graph it does not own. + fn sched() -> Scheduler<&'static str, String> { + Scheduler::new(Graph::new(), ResourceTable::new()) + } + + // The test payloads. `&str` labels itself and has nothing extra to say; + // `String` is its own resource name. + impl WireNode for &str { + fn label(&self) -> String { + (*self).to_owned() + } + fn data(&self, id: WireId) -> serde_json::Value { + // Keyed on identity, not on the payload — the shape a host with a + // side table (build logs, artifacts) needs. + serde_json::json!({ "seen_id": id }) + } + } + + impl super::WireResource for String { + fn name(&self) -> String { + self.clone() + } + } + + /// The bound is the *caller's*: `wire_snapshot` serialises the roots it is + /// handed and their subtrees, and nothing else. Nothing in a `Graph` is + /// ever removed, so a host that couldn't restrict this would have no way to + /// stop the payload growing for its whole uptime. + #[test] + fn only_the_requested_roots_and_their_descendants_ride() { + let mut sched = sched(); + let shown = sched + .insert_job(None, |job| { + let root = job.node("dag"); + let child = job.node("prebuild").part_of(root); + let grandchild = job.node("swap").part_of(child); + vec![root.guid(), child.guid(), grandchild.guid()] + }) + .expect("job inserts"); + let evicted = sched + .insert_job(None, |job| vec![job.node("old-dag").guid()]) + .expect("job inserts"); + + let wire = sched.graph().wire_snapshot([shown[0]]); + + let ids: Vec = wire.iter().map(|n| n.id).collect(); + assert_eq!( + ids, + shown.iter().map(|id| id.get()).collect::>(), + "the root, then its whole subtree" + ); + assert!( + !ids.contains(&evicted[0].get()), + "a root the host did not ask for stays off the wire" + ); + assert_eq!(wire[0].parent, None, "a group root is parentless"); + assert_eq!(wire[1].parent, Some(shown[0].get())); + } + + /// `label` / `data` / `name` all come from the host's impls — the graph + /// layer never invents a rendering for a payload it cannot read. + #[test] + fn payload_and_resource_rendering_come_from_the_host_impls() { + let mut sched = sched(); + let ids = sched + .insert_job(None, |job| { + let first = job.node("reconcile"); + let second = job + .node("swap") + .after_ok(first) + .needs_units("build-slot".to_owned(), 2); + vec![first.guid(), second.guid()] + }) + .expect("job inserts"); + let (first, second) = (ids[0], ids[1]); + + let wire = sched.graph().wire_snapshot([first, second]); + + assert_eq!(wire[0].payload.label, "reconcile"); + assert_eq!(wire[0].payload.data["seen_id"], first.get()); + match &wire[1].deps[..] { + [ + GraphDep::Node { id, accepts }, + GraphDep::Resource { name, count }, + ] => { + assert_eq!(*id, first.get()); + assert_eq!(accepts, &vec![TerminalState::Done]); + assert_eq!(name, "build-slot", "the host's WireResource::name"); + assert_eq!(*count, 2); + } + other => panic!("expected a node dep then a resource dep, got {other:?}"), + } + } + + fn node(id: WireId, parent: Option, state: State, deps: Vec) -> GraphNode { + GraphNode { + id, + parent, + state, + deps, + started_at: None, + finished_at: None, + error: None, + payload: NodePayload { + label: "reconcile".to_owned(), + data: serde_json::Value::Null, + }, + } + } + + #[test] + fn a_node_round_trips_through_json() { + let before = node( + 7, + Some(3), + State::Running, + vec![ + GraphDep::Node { + id: 3, + accepts: vec![TerminalState::Done], + }, + GraphDep::Resource { + name: "build-slot".to_owned(), + count: 1, + }, + ], + ); + let json = serde_json::to_string(&before).expect("serialises"); + let after: GraphNode = serde_json::from_str(&json).expect("round trips"); + assert_eq!(format!("{after:?}"), format!("{before:?}")); + } + + #[test] + fn two_tails_on_one_upstream_are_distinguishable_by_their_accepted_set() { + // The reason `accepts` is a set and not a strong/weak bool: these two + // are the same shape and the same upstream, and *only* the outcome + // set tells them apart. + let ok_tail = node( + 10, + None, + State::Pending, + vec![GraphDep::Node { + id: 4, + accepts: vec![TerminalState::Done], + }], + ); + let fail_tail = node( + 11, + None, + State::Pending, + vec![GraphDep::Node { + id: 4, + accepts: vec![TerminalState::Failed, TerminalState::Cancelled], + }], + ); + let render = |n: &GraphNode| match n.deps.first() { + Some(GraphDep::Node { accepts, .. }) => format!("{accepts:?}"), + _ => "none".to_owned(), + }; + assert_ne!(render(&ok_tail), render(&fail_tail)); + } + + #[test] + fn a_group_root_is_an_ordinary_node_and_carries_the_groups_answer() { + // No roll-up field: `Finishing` on the root says "own logic done, + // children still running", and a terminal root state is the roll-up. + let root = node(1, None, State::Finishing, Vec::new()); + assert!(root.parent.is_none(), "a group root is just parentless"); + assert_eq!(root.state, State::Finishing); + + let done_root = node(2, None, State::Failed, Vec::new()); + assert_eq!( + done_root.state, + State::Failed, + "the root's own state is the subtree's rolled-up outcome" + ); + } + + #[test] + fn an_empty_payload_slot_is_omitted_from_the_wire() { + // The opaque slot costs nothing when a host has nothing to say. + let json = + serde_json::to_string(&node(1, None, State::Pending, Vec::new())).expect("serialises"); + assert!(!json.contains("\"data\""), "null data is skipped: {json}"); + assert!(json.contains("\"label\""), "the label always rides: {json}"); + } +}