From 10b0f640af003ab08befe5a62591691a7bca2587 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 18:47:22 +0200 Subject: [PATCH 1/4] hivectl: migrate dag_progress to hive-jobq-wire's generic GraphNode --- Cargo.lock | 2 + hive-c0re/src/job_queue/mod.rs | 24 ++ hive-c0re/src/job_queue/model.rs | 9 + hive-c0re/src/server.rs | 1 + hive-host-sock/Cargo.toml | 1 + hive-host-sock/src/lib.rs | 28 +++ hivectl/Cargo.toml | 1 + hivectl/src/dag_progress.rs | 381 +++++++++++++++++-------------- 8 files changed, 272 insertions(+), 175 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 93b98034..22171566 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1725,6 +1725,7 @@ version = "0.1.0" dependencies = [ "chrono", "hive-jobq", + "hive-jobq-wire", "hive-sh4re", "hive-types", "serde", @@ -1861,6 +1862,7 @@ dependencies = [ "clap-markdown", "clap_complete", "hive-host-sock", + "hive-jobq-wire", "hive-sh4re", "hive-types", "http-body-util", diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 6b2411df..e81a3bb6 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -378,6 +378,30 @@ impl JobQueue { .filter_map(|c| dag_view(&inner, c)) .collect() } + + /// One DAG's container node plus its live subtree, as generic wire + /// nodes — the `QueueNodes` polling surface behind `hivectl`'s + /// wait/progress loop. Sibling of [`Self::snapshot`] + /// (which serves the same graph through the typed `DagView`/`NodeView` + /// projection for the dashboard's `/api/state.rebuild_queue`), this one + /// goes through [`GraphWire::wire_snapshot`] instead — no `Done`-node + /// filtering, no roll-up field (the root's own `state` answers that, + /// see `hive_jobq_wire`'s doc comment). + /// + /// Empty when `dag_id` names no DAG container in the graph. Today that + /// only happens for a genuinely unknown id: nothing prunes the graph + /// yet (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so + /// a *completed* DAG's nodes keep riding here with a terminal `state` + /// rather than disappearing — callers watching for "done" should read + /// the root's `state`, not emptiness. + #[must_use] + pub fn dag_nodes(&self, dag_id: u64) -> Vec { + let inner = self.lock(); + let Some(root) = container(&inner, dag_id) else { + return Vec::new(); + }; + inner.graph().wire_snapshot([root]) + } } /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 9f45c3b1..a15e11d1 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -308,6 +308,15 @@ impl hive_jobq_wire::WireNode for NodeKind { { data.insert("inputs".to_owned(), inputs.clone().into()); } + // The DAG container's own metadata — nowhere else on the wire, since + // `GraphNode` carries no DAG-level fields (a group root is an + // ordinary node). `hivectl` needs `source` for its progress line; + // `reason` rides along for free rather than adding a second variant + // later for the one field the first pass missed. + if let NodeKind::Dag { source, reason, .. } = self { + data.insert("source".to_owned(), source.as_str().into()); + data.insert("reason".to_owned(), reason.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)) { diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 26803aa2..d187d0b8 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -160,6 +160,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { .collect(); HostResponse::dags(dags) } + HostRequest::QueueNodes { id } => HostResponse::nodes(coord.job_queue.dag_nodes(*id)), HostRequest::List => HostResponse::list(lifecycle::list().await?), // The agents root is ours and not world-traversable, so this // question is only answerable on this side of the socket — diff --git a/hive-host-sock/Cargo.toml b/hive-host-sock/Cargo.toml index ffd069fd..dd7fc555 100644 --- a/hive-host-sock/Cargo.toml +++ b/hive-host-sock/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] chrono.workspace = true hive-jobq.workspace = true +hive-jobq-wire.workspace = true hive-sh4re.workspace = true hive-types.workspace = true serde.workspace = true diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 2363e3db..560029a7 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -214,6 +214,14 @@ pub enum HostRequest { /// `hivectl`'s wait/progress loop. A multi-step op is a single DAG /// (its whole graph in `nodes`). Result: [`HostResponse::dags`]. QueueDag { id: u64 }, + /// Fetch one job-queue DAG's container node plus its live subtree, as + /// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. + /// Sibling of [`Self::QueueDag`]: same DAG, same + /// `id` (the container's own node id, what `queued_dags` already + /// carries), through the generic projection instead of the typed + /// `DagView`/`NodeView` (kept for `QueueDag`'s other consumer, + /// `/api/state.rebuild_queue`). Result: [`HostResponse::nodes`]. + QueueNodes { id: u64 }, /// List pending approval requests. Pending, /// Approve a pending request by id; the action runs immediately. @@ -539,6 +547,15 @@ pub struct HostResponse { /// been evicted from the queue's history tail. #[serde(default, skip_serializing_if = "Option::is_none")] pub dags: Option>, + /// `QueueNodes` result — the requested DAG's container node plus its + /// live subtree, as generic `hive-jobq-wire` nodes. `None` for every + /// other request kind. An empty `Vec` means `id` names no live DAG in + /// the graph (today: an unknown id — see `JobQueue::dag_nodes`'s doc + /// comment for why a *completed* DAG's nodes don't vanish the same way + /// `QueueDag`'s do); callers should read the root node's `state` for + /// terminality, not emptiness. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub nodes: Option>, /// Free-form operator-facing output lines the client prints verbatim /// (one per line). Carries results a request produced daemon-side that /// have no structured home — e.g. a freshly-minted matrix token, a @@ -639,6 +656,17 @@ impl HostResponse { } } + /// `QueueNodes` result — the polled DAG's container + subtree, as + /// generic wire nodes. + #[must_use] + pub fn nodes(nodes: Vec) -> Self { + Self { + ok: true, + nodes: Some(nodes), + ..Self::default() + } + } + /// A success carrying operator-facing output lines the client prints /// verbatim — the result shape for the `Matrix*` provisioning requests. #[must_use] diff --git a/hivectl/Cargo.toml b/hivectl/Cargo.toml index e282912d..baf32cf2 100644 --- a/hivectl/Cargo.toml +++ b/hivectl/Cargo.toml @@ -21,6 +21,7 @@ clap.workspace = true clap_complete.workspace = true clap-markdown = "0.1" hive-host-sock.workspace = true +hive-jobq-wire.workspace = true hive-sh4re.workspace = true hive-types.workspace = true http-body-util.workspace = true diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index aaca5710..fe4c3d4c 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -1,20 +1,30 @@ //! `hivectl` rebuild-queue progress rendering. //! //! Split out of `hivectl.rs` (which is already large): everything that -//! polls the daemon's DAG queue (`HostRequest::QueueDag`) and renders the +//! polls the daemon's DAG queue (`HostRequest::QueueNodes`) and renders the //! per-DAG / per-node progress lives here. [`wait_for_dags`] is the entry //! point the command handlers call; it dispatches to a live `indicatif` //! animation on a TTY and a plain line-on-change stream otherwise. +//! +//! Consumes `hive-jobq-wire`'s generic [`GraphNode`]/`NodePayload`: a +//! node's kind comes from `payload.label`, and node-specific extras +//! (`source`, `agent`) ride in `payload.data`'s opaque kvps — the shape +//! `hive-c0re`'s `impl WireNode for NodeKind` produces (see its doc +//! comment). There's no separate roll-up field on the wire: a node's own +//! `state` already reflects everything below it (see `hive_jobq_wire`'s +//! doc comment), so nothing here re-derives a roll-up or waits for every +//! node to go terminal before reading a result. use std::path::Path; use anyhow::{Context as _, Result, bail}; +use hive_host_sock::jobs::State; +use hive_jobq_wire::{GraphDep, GraphNode, WireId}; -/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~1s interval) and +/// Poll the submitted DAG ids (`HostRequest::QueueNodes`, ~1s interval) and /// render progress until they all reach a terminal state. Exits non-zero -/// (via the returned `Err`) when any DAG (or fan-out child) ends `failed`; -/// a `cancelled` DAG terminates the wait but is an operator action, not an -/// error. +/// (via the returned `Err`) when any DAG ends `failed`; a `cancelled` DAG +/// terminates the wait but is an operator action, not an error. /// /// # Errors /// @@ -34,6 +44,13 @@ pub(crate) async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) - } } +/// Find the DAG container among a `QueueNodes` response's nodes — the one +/// `GraphNode` with no structural parent. `wire_snapshot` hands back exactly +/// one per requested root, so this is a lookup, not a real search. +fn find_root(nodes: &[GraphNode]) -> Option<&GraphNode> { + nodes.iter().find(|n| n.parent.is_none()) +} + /// Non-TTY progress: print a fresh line whenever a DAG's rendered state /// changes. No cursor tricks, so it's clean in pipes and CI logs. async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { @@ -42,38 +59,31 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { let mut failed: Vec = Vec::new(); while !pending.is_empty() { for id in pending.clone() { - let resp = crate::client::request(socket, hive_host_sock::HostRequest::QueueDag { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let dags = resp.dags.unwrap_or_default(); - if dags.is_empty() { - // Evicted from the queue's history tail — it finished a - // while ago; nothing left to report on. + let resp = + crate::client::request(socket, hive_host_sock::HostRequest::QueueNodes { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let nodes = resp.nodes.unwrap_or_default(); + let Some(root) = find_root(&nodes) else { + // Unknown id — nothing prunes the graph yet (see + // `JobQueue::dag_nodes`'s doc comment), so an id that + // resolves to no container never named a real DAG. A + // *completed* DAG's nodes keep riding here instead, with a + // terminal root `state`, which is what the check below + // watches for. println!("job #{id}: gone from queue history"); pending.remove(&id); continue; + }; + let line = render_dag_line(root, &nodes); + if last.get(&id) != Some(&line) { + println!("{line}"); + last.insert(id, line); } - let mut all_terminal = true; - for d in &dags { - let line = render_dag_line(d); - if last.get(&d.id) != Some(&line) { - println!("{line}"); - last.insert(d.id, line); + if root.state.is_terminal() { + if root.state == State::Failed { + failed.push(format!("{} {}", dag_source(root), dag_agents(&nodes))); } - // Node-level terminality, not the roll-up: a DAG rolls - // up `failed` the moment one node fails while its - // after-any recovery node (rebuild's tail Reconcile) - // may still be running — keep watching so the operator - // sees whether the agent came back. - if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.rollup_state() == hive_host_sock::jobs::State::Failed { - failed.push(format!("{} {}", d.source.as_str(), dag_agents(d))); - } - } else { - all_terminal = false; - } - } - if all_terminal { pending.remove(&id); } } @@ -104,10 +114,9 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { // insertion order groups a DAG's nodes right under its header. let mut dag_bars: std::collections::HashMap = std::collections::HashMap::new(); - let mut node_bars: std::collections::HashMap<(u64, hive_host_sock::jobs::NodeId), ProgressBar> = + let mut node_bars: std::collections::HashMap<(u64, WireId), ProgressBar> = std::collections::HashMap::new(); - let mut node_done: std::collections::HashSet<(u64, hive_host_sock::jobs::NodeId)> = - std::collections::HashSet::new(); + let mut node_done: std::collections::HashSet<(u64, WireId)> = std::collections::HashSet::new(); let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); let mut failed: Vec = Vec::new(); @@ -115,62 +124,56 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { while !pending.is_empty() { let now = now_unix(); for id in pending.clone() { - let resp = crate::client::request(socket, hive_host_sock::HostRequest::QueueDag { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let dags = resp.dags.unwrap_or_default(); - if dags.is_empty() { + let resp = + crate::client::request(socket, hive_host_sock::HostRequest::QueueNodes { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let nodes = resp.nodes.unwrap_or_default(); + let Some(root) = find_root(&nodes) else { mp.println(format!("job #{id}: gone from queue history")) .ok(); pending.remove(&id); continue; - } - let mut all_terminal = true; - for d in &dags { - let hdr = dag_bars.entry(d.id).or_insert_with(|| { + }; + let hdr = dag_bars.entry(id).or_insert_with(|| { + let b = mp.add(ProgressBar::new_spinner()); + b.set_style(plain.clone()); + b + }); + hdr.set_message(format!( + "{} {} {} · {}", + state_glyph(root.state), + dag_source(root), + dag_agents(&nodes), + fmt_dur(node_elapsed(root, now)), + )); + for n in nodes.iter().filter(|n| n.id != root.id) { + let key = (id, n.id); + if node_done.contains(&key) { + continue; + } + let bar = node_bars.entry(key).or_insert_with(|| { let b = mp.add(ProgressBar::new_spinner()); - b.set_style(plain.clone()); + b.set_style(spinner.clone()); + b.enable_steady_tick(std::time::Duration::from_millis(120)); b }); - hdr.set_message(format!( - "{} {} {} · {}", - state_glyph(d.rollup_state()), - d.source.as_str(), - dag_agents(d), - fmt_dur(dag_elapsed(d, now)), - )); - for n in &d.nodes { - let key = (d.id, n.id); - if node_done.contains(&key) { - continue; - } - let bar = node_bars.entry(key).or_insert_with(|| { - let b = mp.add(ProgressBar::new_spinner()); - b.set_style(spinner.clone()); - b.enable_steady_tick(std::time::Duration::from_millis(120)); - b - }); - if n.state.is_terminal() { - bar.set_style(plain.clone()); - bar.finish_with_message(format!( - " {} {}", - state_glyph(n.state), - node_line(d, n, now) - )); - node_done.insert(key); - } else { - bar.set_message(node_line(d, n, now)); - } - } - if d.nodes.iter().all(|n| n.state.is_terminal()) { - if d.rollup_state() == hive_host_sock::jobs::State::Failed { - failed.push(format!("{} {}", d.source.as_str(), dag_agents(d))); - } + if n.state.is_terminal() { + bar.set_style(plain.clone()); + bar.finish_with_message(format!( + " {} {}", + state_glyph(n.state), + node_line(&nodes, n, now) + )); + node_done.insert(key); } else { - all_terminal = false; + bar.set_message(node_line(&nodes, n, now)); } } - if all_terminal { + if root.state.is_terminal() { + if root.state == State::Failed { + failed.push(format!("{} {}", dag_source(root), dag_agents(&nodes))); + } pending.remove(&id); } } @@ -202,14 +205,34 @@ fn finish_wait(mut failed: Vec) -> Result<()> { } } -/// Distinct agents across a DAG's nodes, comma-joined for display — the -/// per-node replacement for the old DAG-level `agent` field. Single-agent -/// DAGs render one name; a hive-wide DAG lists each. -fn dag_agents(d: &hive_host_sock::jobs::DagView) -> String { +/// The DAG's `source` tag (`"manual"`, `"meta_update"`, …), read from the +/// container node's `payload.data` — see `hive-c0re`'s `impl WireNode for +/// NodeKind`'s `NodeKind::Dag` arm, the one place it's put on the wire. +/// Falls back to the label when absent (defensive; every real DAG +/// container sets it). +fn dag_source(root: &GraphNode) -> &str { + root.payload + .data + .get("source") + .and_then(serde_json::Value::as_str) + .unwrap_or(&root.payload.label) +} + +/// Distinct agents across a DAG's nodes, comma-joined for display. Each +/// node's `agent` (when it targets one) rides in `payload.data["agent"]` — +/// an opaque kvp, not a typed field, since `GraphNode` carries nothing +/// domain-specific (see `hive_jobq_wire::WireNode::data`'s doc comment). +fn dag_agents(nodes: &[GraphNode]) -> String { let mut seen: Vec<&str> = Vec::new(); - for n in &d.nodes { - if !seen.contains(&n.agent.as_str()) { - seen.push(&n.agent); + for n in nodes { + if let Some(agent) = n + .payload + .data + .get("agent") + .and_then(serde_json::Value::as_str) + && !seen.contains(&agent) + { + seen.push(agent); } } seen.join(",") @@ -224,23 +247,13 @@ fn now_unix() -> i64 { .unwrap_or(0) } -/// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`) -/// through `finished_at` or `now`. The wire carries these as RFC3339 -/// `DateTime`; compare in unix seconds against `now`. -fn dag_elapsed(d: &hive_host_sock::jobs::DagView, now: i64) -> i64 { - let start = d - .started_at - .map_or_else(|| d.created_at.timestamp(), |t| t.timestamp()); - (d.finished_at.map_or(now, |t| t.timestamp()) - start).max(0) -} - -/// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0 -/// when it hasn't started. -fn node_elapsed(n: &hive_host_sock::jobs::NodeView, now: i64) -> i64 { - match n.started_at { - Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0), - None => 0, - } +/// Elapsed seconds for a node: `started_at` → `finished_at`/`now` once +/// it's run; `created_at` → `now` while it's still queued. `GraphNode` +/// carries no group-level timestamp separate from its own, so a header +/// line just reads this off whichever node it's summarizing. +fn node_elapsed(n: &GraphNode, now: i64) -> i64 { + let start = n.started_at.unwrap_or(n.created_at); + (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0) } /// Compact duration: `45s` under a minute, else `1m03s`. @@ -254,20 +267,26 @@ fn fmt_dur(secs: i64) -> String { } /// One animated node line: kind, an `(after …)` marker for a fan-in node -/// (>1 dep), its elapsed timer, and a truncated error tail. -fn node_line( - d: &hive_host_sock::jobs::DagView, - n: &hive_host_sock::jobs::NodeView, - now: i64, -) -> String { +/// (>1 node-dependency), its elapsed timer, and a truncated error tail. +/// Resource deps (`GraphDep::Resource`) don't name another node, so they're +/// filtered out of the fan-in count the same way `hive-c0re`'s own +/// `dag_view` projection already does. +fn node_line(nodes: &[GraphNode], n: &GraphNode, now: i64) -> String { use std::fmt::Write as _; - let mut s = n.kind.clone(); - if n.deps.len() > 1 { - let after: Vec<&str> = n - .deps + let mut s = n.payload.label.clone(); + let dep_ids: Vec = n + .deps + .iter() + .filter_map(|d| match d { + GraphDep::Node { id, .. } => Some(*id), + GraphDep::Resource { .. } => None, + }) + .collect(); + if dep_ids.len() > 1 { + let after: Vec<&str> = dep_ids .iter() - .filter_map(|dep| d.nodes.iter().find(|m| m.id == *dep)) - .map(|m| m.kind.as_str()) + .filter_map(|id| nodes.iter().find(|m| m.id == *id)) + .map(|m| m.payload.label.as_str()) .collect(); if !after.is_empty() { let _ = write!(s, " (after {})", after.join(", ")); @@ -284,38 +303,46 @@ fn node_line( s } -fn state_glyph(state: hive_host_sock::jobs::State) -> &'static str { +fn state_glyph(state: State) -> &'static str { match state { - hive_host_sock::jobs::State::Pending => "⏸", + State::Pending => "⏸", // `Finishing` is own-work-done with sub-nodes still going — in flight, // so it reads the same as running. - hive_host_sock::jobs::State::Running | hive_host_sock::jobs::State::Finishing => "▶", - hive_host_sock::jobs::State::Done => "✔", - hive_host_sock::jobs::State::Failed => "✖", - hive_host_sock::jobs::State::Cancelled => "⊘", + State::Running | State::Finishing => "▶", + State::Done => "✔", + State::Failed => "✖", + State::Cancelled => "⊘", // Distinct from cancelled: nothing went wrong, this branch just // wasn't the one the run took. - hive_host_sock::jobs::State::Skipped => "·", + State::Skipped => "·", } } /// One progress line for a DAG: roll-up glyph, `source`, agents, then the -/// node chain — the CLI twin of the dashboard's queue card. The header shows -/// what the backend sends (`source` + the raw node kinds); only the roll-up -/// state glyph is derived from the node set. Used by the plain (non-TTY) path. -fn render_dag_line(d: &hive_host_sock::jobs::DagView) -> String { +/// node chain — the CLI twin of the dashboard's queue card. The glyph and +/// `source` come off `root` (the one entry in `nodes` with no parent); +/// everything else comes straight off the wire. Used by the plain +/// (non-TTY) path. +/// +/// The chain shows every entry `wire_snapshot` sends, `Done` ones +/// included — that filtering was a dashboard-history-bounding concern, +/// not relevant to a single actively-watched job, so a completed step +/// keeps its checkmark instead of vanishing from the line, matching how +/// the animated path already behaves. +fn render_dag_line(root: &GraphNode, nodes: &[GraphNode]) -> String { use std::fmt::Write as _; let mut out = format!( "{} {} {:<12}", - state_glyph(d.rollup_state()), - d.source.as_str(), - dag_agents(d) + state_glyph(root.state), + dag_source(root), + dag_agents(nodes) ); - for (i, n) in d.nodes.iter().enumerate() { + let children: Vec<&GraphNode> = nodes.iter().filter(|n| n.id != root.id).collect(); + for (i, n) in children.iter().enumerate() { let sep = if i == 0 { " " } else { " → " }; - let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind); + let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.payload.label); } - if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) { + if let Some(err) = children.iter().find_map(|n| n.error.as_deref()) { let short: String = err.chars().take(120).collect(); let _ = write!(out, " — {short}"); } @@ -324,50 +351,61 @@ fn render_dag_line(d: &hive_host_sock::jobs::DagView) -> String { #[cfg(test)] mod tests { - use hive_host_sock::jobs::{DagView, NodeView, Source, State}; - use hive_sh4re::wire_time::from_secs; + use hive_host_sock::jobs::State; + use hive_jobq_wire::{GraphNode, NodePayload}; + use serde_json::json; use super::render_dag_line; - fn node(id: u64, agent: &str, kind: &str, state: State) -> NodeView { - NodeView { + fn dag_root(id: u64, source: &str, state: State) -> GraphNode { + GraphNode { id, parent: None, - agent: agent.to_owned(), - kind: kind.to_owned(), - deps: if id == 0 { vec![] } else { vec![id - 1] }, state, + deps: Vec::new(), + created_at: hive_sh4re::wire_time::from_secs(0), started_at: None, finished_at: None, error: None, - approval_id: None, - inputs: vec![], - build_log_id: None, + payload: NodePayload { + label: "dag".to_owned(), + data: json!({ "source": source }), + }, + } + } + + fn work_node(id: u64, root: u64, agent: &str, label: &str, state: State) -> GraphNode { + GraphNode { + id, + parent: Some(root), + state, + deps: Vec::new(), + created_at: hive_sh4re::wire_time::from_secs(0), + started_at: None, + finished_at: None, + error: None, + payload: NodePayload { + label: label.to_owned(), + data: json!({ "agent": agent }), + }, } } #[test] fn render_dag_line_shows_source_and_chain() { - // The header shows what the backend sends: the roll-up state glyph - // (derived — Running here) + the DAG `source` ("manual"); the operation - // is read off the node chain, not a client-side label. (`Done` nodes - // are included here to exercise glyph rendering; production filters - // them off.) - let dag = DagView { - id: 7, - source: Source::Manual, - reason: "manual".to_owned(), - created_at: from_secs(0), - started_at: Some(from_secs(1)), - finished_at: None, - nodes: vec![ - node(0, "alice", "prebuild", State::Done), - node(1, "alice", "stop_for_update", State::Done), - node(2, "alice", "swap", State::Running), - node(3, "alice", "reconcile", State::Pending), - ], - }; - let line = render_dag_line(&dag); + // The header shows what the backend sends: the glyph off `root`'s + // own `state` (`Running` here) + `source` ("manual"); the operation + // is read off the node chain, not a client-side label. `Done` + // nodes stay in the chain — nothing here filters them off. + let root = dag_root(7, "manual", State::Running); + let nodes = vec![ + root.clone(), + work_node(0, 7, "alice", "prebuild", State::Done), + work_node(1, 7, "alice", "stop_for_update", State::Done), + work_node(2, 7, "alice", "swap", State::Running), + work_node(3, 7, "alice", "reconcile", State::Pending), + ]; + let line = render_dag_line(&root, &nodes); assert!(line.starts_with("▶ manual alice"), "{line}"); assert!( line.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"), @@ -377,18 +415,11 @@ mod tests { #[test] fn render_dag_line_surfaces_first_node_error() { - let mut failed = node(0, "bob", "prebuild", State::Failed); + let root = dag_root(8, "manual", State::Failed); + let mut failed = work_node(0, 8, "bob", "prebuild", State::Failed); failed.error = Some("nix build exploded".to_owned()); - let dag = DagView { - id: 8, - source: Source::Manual, - reason: "manual".to_owned(), - created_at: from_secs(0), - started_at: Some(from_secs(1)), - finished_at: Some(from_secs(2)), - nodes: vec![failed], - }; - let line = render_dag_line(&dag); + let nodes = vec![root.clone(), failed]; + let line = render_dag_line(&root, &nodes); assert!(line.contains("✖ manual"), "{line}"); assert!(line.contains("— nix build exploded"), "{line}"); } From 309c1c91c6a393ef19bd2acecb7438224d6d6808 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 20:03:48 +0200 Subject: [PATCH 2/4] job_queue: drop the NodeKind::Dag/root filter from the QueueNodes lookup --- hive-c0re/src/job_queue/mod.rs | 50 ++++++++++++++++++---------------- hive-c0re/src/server.rs | 4 ++- hive-host-sock/src/lib.rs | 33 +++++++++++----------- hivectl/src/dag_progress.rs | 4 +-- 4 files changed, 48 insertions(+), 43 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index e81a3bb6..7ca0ce7f 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -289,8 +289,8 @@ impl JobQueue { #[must_use] pub fn first_error(&self, dag_id: u64) -> Option { let inner = self.lock(); - let container = container(&inner, dag_id)?; - inner.graph().first_error(container).map(ToOwned::to_owned) + let node = find_node(&inner, dag_id)?; + inner.graph().first_error(node).map(ToOwned::to_owned) } /// `(agent, label, takes_container_down)` for the live transient-pill set, @@ -379,38 +379,40 @@ impl JobQueue { .collect() } - /// One DAG's container node plus its live subtree, as generic wire - /// nodes — the `QueueNodes` polling surface behind `hivectl`'s - /// wait/progress loop. Sibling of [`Self::snapshot`] - /// (which serves the same graph through the typed `DagView`/`NodeView` - /// projection for the dashboard's `/api/state.rebuild_queue`), this one - /// goes through [`GraphWire::wire_snapshot`] instead — no `Done`-node - /// filtering, no roll-up field (the root's own `state` answers that, - /// see `hive_jobq_wire`'s doc comment). + /// A node plus its live subtree, as generic wire nodes — the + /// `QueueNodes` polling surface behind `hivectl`'s wait/progress loop. + /// Sibling of [`Self::snapshot`] (which serves the same graph through + /// the typed `DagView`/`NodeView` projection for the dashboard's + /// `/api/state.rebuild_queue`), this one goes through + /// [`GraphWire::wire_snapshot`] instead — no `Done`-node filtering, no + /// roll-up field (a node's own `state` answers that, see + /// `hive_jobq_wire`'s doc comment). Looks the id up by identity alone — + /// no assumption that it names a DAG container or a root; "just show + /// whatever the backend sends" for whatever id the caller asks about. /// - /// Empty when `dag_id` names no DAG container in the graph. Today that - /// only happens for a genuinely unknown id: nothing prunes the graph - /// yet (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so - /// a *completed* DAG's nodes keep riding here with a terminal `state` + /// Empty when `id` names no node in the graph. Today that only happens + /// for a genuinely unknown id: nothing prunes the graph yet + /// (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so a + /// *completed* DAG's nodes keep riding here with a terminal `state` /// rather than disappearing — callers watching for "done" should read /// the root's `state`, not emptiness. #[must_use] - pub fn dag_nodes(&self, dag_id: u64) -> Vec { + pub fn node_subtree(&self, id: u64) -> Vec { let inner = self.lock(); - let Some(root) = container(&inner, dag_id) else { + let Some(node) = find_node(&inner, id) else { return Vec::new(); }; - inner.graph().wire_snapshot([root]) + inner.graph().wire_snapshot([node]) } } -/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals -/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. -fn container(sched: &Sched, dag_id: u64) -> Option { - sched.graph().nodes().find_map(|n| { - (n.parent.is_none() && n.id.get() == dag_id && matches!(n.payload, NodeKind::Dag { .. })) - .then_some(n.id) - }) +/// The graph node whose id equals `id`, whatever its kind or depth. +/// `NodeId` is un-fabricable from a raw `u64`, so this is a search. +fn find_node(sched: &Sched, id: u64) -> Option { + sched + .graph() + .nodes() + .find_map(|n| (n.id.get() == id).then_some(n.id)) } /// Project a DAG into its wire [`DagView`]: a near-raw view of the diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index d187d0b8..4d22b800 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -160,7 +160,9 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { .collect(); HostResponse::dags(dags) } - HostRequest::QueueNodes { id } => HostResponse::nodes(coord.job_queue.dag_nodes(*id)), + HostRequest::QueueNodes { id } => { + HostResponse::nodes(coord.job_queue.node_subtree(*id)) + } HostRequest::List => HostResponse::list(lifecycle::list().await?), // The agents root is ours and not world-traversable, so this // question is only answerable on this side of the socket — diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 560029a7..363a3442 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -214,13 +214,14 @@ pub enum HostRequest { /// `hivectl`'s wait/progress loop. A multi-step op is a single DAG /// (its whole graph in `nodes`). Result: [`HostResponse::dags`]. QueueDag { id: u64 }, - /// Fetch one job-queue DAG's container node plus its live subtree, as - /// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. - /// Sibling of [`Self::QueueDag`]: same DAG, same - /// `id` (the container's own node id, what `queued_dags` already - /// carries), through the generic projection instead of the typed - /// `DagView`/`NodeView` (kept for `QueueDag`'s other consumer, - /// `/api/state.rebuild_queue`). Result: [`HostResponse::nodes`]. + /// Fetch one job-queue node plus its live subtree, as generic + /// `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. Sibling of + /// [`Self::QueueDag`]: same graph, same `id`, through the generic + /// projection instead of the typed `DagView`/`NodeView` (kept for + /// `QueueDag`'s other consumer, `/api/state.rebuild_queue`). No + /// assumption that `id` names a DAG container or root — whatever node + /// has that id, the backend hands back its subtree as-is. Result: + /// [`HostResponse::nodes`]. QueueNodes { id: u64 }, /// List pending approval requests. Pending, @@ -547,13 +548,13 @@ pub struct HostResponse { /// been evicted from the queue's history tail. #[serde(default, skip_serializing_if = "Option::is_none")] pub dags: Option>, - /// `QueueNodes` result — the requested DAG's container node plus its - /// live subtree, as generic `hive-jobq-wire` nodes. `None` for every - /// other request kind. An empty `Vec` means `id` names no live DAG in - /// the graph (today: an unknown id — see `JobQueue::dag_nodes`'s doc - /// comment for why a *completed* DAG's nodes don't vanish the same way - /// `QueueDag`'s do); callers should read the root node's `state` for - /// terminality, not emptiness. + /// `QueueNodes` result — the requested node plus its live subtree, as + /// generic `hive-jobq-wire` nodes. `None` for every other request kind. + /// An empty `Vec` means `id` names no live node in the graph (today: an + /// unknown id — see `JobQueue::node_subtree`'s doc comment for why a + /// *completed* DAG's nodes don't vanish the same way `QueueDag`'s do); + /// callers should read the root node's `state` for terminality, not + /// emptiness. #[serde(default, skip_serializing_if = "Option::is_none")] pub nodes: Option>, /// Free-form operator-facing output lines the client prints verbatim @@ -656,8 +657,8 @@ impl HostResponse { } } - /// `QueueNodes` result — the polled DAG's container + subtree, as - /// generic wire nodes. + /// `QueueNodes` result — the polled node + its subtree, as generic + /// wire nodes. #[must_use] pub fn nodes(nodes: Vec) -> Self { Self { diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index fe4c3d4c..917ba347 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -66,8 +66,8 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { let nodes = resp.nodes.unwrap_or_default(); let Some(root) = find_root(&nodes) else { // Unknown id — nothing prunes the graph yet (see - // `JobQueue::dag_nodes`'s doc comment), so an id that - // resolves to no container never named a real DAG. A + // `JobQueue::node_subtree`'s doc comment), so an id that + // resolves to no node never named a real DAG. A // *completed* DAG's nodes keep riding here instead, with a // terminal root `state`, which is what the check below // watches for. From 0c4d56a585cddfb902b8eb7ddc38129702b48fde Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 20:17:16 +0200 Subject: [PATCH 3/4] hivectl: batch QueueNodes polling by id set, drop remaining dag wording --- hive-c0re/src/job_queue/mod.rs | 42 ++++--- hive-c0re/src/server.rs | 4 +- hive-host-sock/src/lib.rs | 31 ++--- hivectl/src/agents.rs | 4 +- hivectl/src/dag_progress.rs | 208 ++++++++++++++++++++------------- hivectl/src/main.rs | 2 +- hivectl/src/power.rs | 8 +- hivectl/src/subvol.rs | 8 +- 8 files changed, 182 insertions(+), 125 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 7ca0ce7f..d4afeea0 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -379,30 +379,34 @@ impl JobQueue { .collect() } - /// A node plus its live subtree, as generic wire nodes — the - /// `QueueNodes` polling surface behind `hivectl`'s wait/progress loop. - /// Sibling of [`Self::snapshot`] (which serves the same graph through - /// the typed `DagView`/`NodeView` projection for the dashboard's - /// `/api/state.rebuild_queue`), this one goes through + /// One or more nodes plus their live subtrees, as generic wire nodes — + /// the `QueueNodes` polling surface behind `hivectl`'s wait/progress + /// loop. Sibling of [`Self::snapshot`] (which serves the same graph + /// through the typed `DagView`/`NodeView` projection for the + /// dashboard's `/api/state.rebuild_queue`), this one goes through /// [`GraphWire::wire_snapshot`] instead — no `Done`-node filtering, no /// roll-up field (a node's own `state` answers that, see - /// `hive_jobq_wire`'s doc comment). Looks the id up by identity alone — - /// no assumption that it names a DAG container or a root; "just show - /// whatever the backend sends" for whatever id the caller asks about. + /// `hive_jobq_wire`'s doc comment). Looks each id up by identity + /// alone — no assumption that it names a DAG container or a root; + /// "just show whatever the backend sends" for whatever ids the caller + /// asks about. Multiple ids in one call is the normal shape for a + /// batch op (e.g. restarting every agent submits one root per agent) — + /// callers should request the whole batch together rather than poll + /// one id per round-trip. /// - /// Empty when `id` names no node in the graph. Today that only happens - /// for a genuinely unknown id: nothing prunes the graph yet - /// (bounded-prune is a Stage-C follow-up, see [`visible_dags`]), so a - /// *completed* DAG's nodes keep riding here with a terminal `state` - /// rather than disappearing — callers watching for "done" should read - /// the root's `state`, not emptiness. + /// An id with no matching node in the graph is silently dropped from + /// the result rather than erroring the whole batch — some ids in a + /// batch may already be evicted while others are still live. Today + /// that only happens for a genuinely unknown id: nothing prunes the + /// graph yet (bounded-prune is a Stage-C follow-up, see + /// [`visible_dags`]), so a *completed* DAG's nodes keep riding here + /// with a terminal `state` rather than disappearing — callers + /// watching for "done" should read the root's `state`, not absence. #[must_use] - pub fn node_subtree(&self, id: u64) -> Vec { + pub fn node_subtrees(&self, ids: &[u64]) -> Vec { let inner = self.lock(); - let Some(node) = find_node(&inner, id) else { - return Vec::new(); - }; - inner.graph().wire_snapshot([node]) + let roots: Vec = ids.iter().filter_map(|id| find_node(&inner, *id)).collect(); + inner.graph().wire_snapshot(roots) } } diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 4d22b800..5427a7f6 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -160,8 +160,8 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { .collect(); HostResponse::dags(dags) } - HostRequest::QueueNodes { id } => { - HostResponse::nodes(coord.job_queue.node_subtree(*id)) + HostRequest::QueueNodes { ids } => { + HostResponse::nodes(coord.job_queue.node_subtrees(ids)) } HostRequest::List => HostResponse::list(lifecycle::list().await?), // The agents root is ours and not world-traversable, so this diff --git a/hive-host-sock/src/lib.rs b/hive-host-sock/src/lib.rs index 363a3442..caf7e3d9 100644 --- a/hive-host-sock/src/lib.rs +++ b/hive-host-sock/src/lib.rs @@ -214,15 +214,19 @@ pub enum HostRequest { /// `hivectl`'s wait/progress loop. A multi-step op is a single DAG /// (its whole graph in `nodes`). Result: [`HostResponse::dags`]. QueueDag { id: u64 }, - /// Fetch one job-queue node plus its live subtree, as generic - /// `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. Sibling of - /// [`Self::QueueDag`]: same graph, same `id`, through the generic + /// Fetch one or more job-queue nodes plus their live subtrees, as + /// generic `hive-jobq-wire` nodes — `hivectl`'s wait/progress loop. + /// Sibling of [`Self::QueueDag`]: same graph, through the generic /// projection instead of the typed `DagView`/`NodeView` (kept for /// `QueueDag`'s other consumer, `/api/state.rebuild_queue`). No - /// assumption that `id` names a DAG container or root — whatever node - /// has that id, the backend hands back its subtree as-is. Result: + /// assumption that an id names a DAG container or root — whatever + /// node has that id, the backend hands back its subtree as-is. A + /// batch op that submits several independent roots (e.g. one per + /// agent on a hive-wide restart) is a single request naming all of + /// them, not one request per id — the caller owns bundling `ids`, + /// this request just answers whatever it's asked. Result: /// [`HostResponse::nodes`]. - QueueNodes { id: u64 }, + QueueNodes { ids: Vec }, /// List pending approval requests. Pending, /// Approve a pending request by id; the action runs immediately. @@ -548,13 +552,14 @@ pub struct HostResponse { /// been evicted from the queue's history tail. #[serde(default, skip_serializing_if = "Option::is_none")] pub dags: Option>, - /// `QueueNodes` result — the requested node plus its live subtree, as - /// generic `hive-jobq-wire` nodes. `None` for every other request kind. - /// An empty `Vec` means `id` names no live node in the graph (today: an - /// unknown id — see `JobQueue::node_subtree`'s doc comment for why a - /// *completed* DAG's nodes don't vanish the same way `QueueDag`'s do); - /// callers should read the root node's `state` for terminality, not - /// emptiness. + /// `QueueNodes` result — the requested nodes plus their live subtrees, + /// as generic `hive-jobq-wire` nodes, all roots' subtrees combined in + /// one flat list. `None` for every other request kind. An id with no + /// live node in the graph is silently dropped rather than erroring + /// the whole batch — see `JobQueue::node_subtrees`'s doc comment for + /// why a *completed* DAG's nodes don't vanish the same way + /// `QueueDag`'s do; callers should read each root node's `state` for + /// terminality, not absence. #[serde(default, skip_serializing_if = "Option::is_none")] pub nodes: Option>, /// Free-form operator-facing output lines the client prints verbatim diff --git a/hivectl/src/agents.rs b/hivectl/src/agents.rs index c66f7e5d..776ace6f 100644 --- a/hivectl/src/agents.rs +++ b/hivectl/src/agents.rs @@ -13,7 +13,7 @@ use anyhow::{Context as _, Result, bail}; use hive_host_sock::HostRequest; use crate::cli::{AgentCmd, AgentQuotaCmd}; -use crate::dag_progress::wait_for_dags; +use crate::dag_progress::wait_for_nodes; use crate::util::render; async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()> { @@ -27,7 +27,7 @@ async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()> .with_context(|| format!("connect to daemon socket {}", socket.display()))?; if resp.ok { println!("restart queued: {name}"); - wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await + wait_for_nodes(socket, resp.queued_dags.unwrap_or_default(), no_wait).await } else { bail!( "restart {name}: {}", diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index 917ba347..7d137b52 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -1,10 +1,10 @@ //! `hivectl` rebuild-queue progress rendering. //! //! Split out of `hivectl.rs` (which is already large): everything that -//! polls the daemon's DAG queue (`HostRequest::QueueNodes`) and renders the -//! per-DAG / per-node progress lives here. [`wait_for_dags`] is the entry -//! point the command handlers call; it dispatches to a live `indicatif` -//! animation on a TTY and a plain line-on-change stream otherwise. +//! polls the daemon's node queue (`HostRequest::QueueNodes`) and renders +//! progress lives here. [`wait_for_nodes`] is the entry point the command +//! handlers call; it dispatches to a live `indicatif` animation on a TTY +//! and a plain line-on-change stream otherwise. //! //! Consumes `hive-jobq-wire`'s generic [`GraphNode`]/`NodePayload`: a //! node's kind comes from `payload.label`, and node-specific extras @@ -14,23 +14,31 @@ //! `state` already reflects everything below it (see `hive_jobq_wire`'s //! doc comment), so nothing here re-derives a roll-up or waits for every //! node to go terminal before reading a result. +//! +//! `ids` is always a **batch**, not a single id: a hive-wide op (e.g. +//! restarting every agent) submits one root per agent, and the whole +//! batch is polled together in a single `QueueNodes` request per tick — +//! [`group_by_root`] splits the combined response back into per-root +//! groups rather than issuing one round-trip per id. +use std::collections::{BTreeSet, HashMap, HashSet}; use std::path::Path; use anyhow::{Context as _, Result, bail}; use hive_host_sock::jobs::State; use hive_jobq_wire::{GraphDep, GraphNode, WireId}; -/// Poll the submitted DAG ids (`HostRequest::QueueNodes`, ~1s interval) and -/// render progress until they all reach a terminal state. Exits non-zero -/// (via the returned `Err`) when any DAG ends `failed`; a `cancelled` DAG -/// terminates the wait but is an operator action, not an error. +/// Poll the submitted ids (`HostRequest::QueueNodes`, ~1s interval, all +/// ids in one request per tick) and render progress until they all reach +/// a terminal state. Exits non-zero (via the returned `Err`) when any +/// node ends `failed`; a `cancelled` node terminates the wait but is an +/// operator action, not an error. /// /// # Errors /// -/// Returns an error if the daemon socket can't be reached, or if any polled -/// DAG finished in the `failed` state. -pub(crate) async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) -> Result<()> { +/// Returns an error if the daemon socket can't be reached, or if any +/// polled node finished in the `failed` state. +pub(crate) async fn wait_for_nodes(socket: &Path, ids: Vec, no_wait: bool) -> Result<()> { use std::io::IsTerminal as _; if no_wait || ids.is_empty() { return Ok(()); @@ -38,51 +46,71 @@ pub(crate) async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) - // Animate only on a real terminal. Piped / CI output falls back to the // plain line-on-change stream so logs stay free of spinner redraw noise. if std::io::stderr().is_terminal() { - wait_for_dags_animated(socket, ids).await + wait_for_nodes_animated(socket, ids).await } else { - wait_for_dags_plain(socket, ids).await + wait_for_nodes_plain(socket, ids).await } } -/// Find the DAG container among a `QueueNodes` response's nodes — the one -/// `GraphNode` with no structural parent. `wire_snapshot` hands back exactly -/// one per requested root, so this is a lookup, not a real search. -fn find_root(nodes: &[GraphNode]) -> Option<&GraphNode> { - nodes.iter().find(|n| n.parent.is_none()) +/// Split a combined `QueueNodes` response back into per-root groups, keyed +/// by each root's own id. A node with `parent: None` is a root and starts +/// its own group (keyed by its own id); everything else joins its direct +/// parent's group. Today's job shapes are root + flat children (no deeper +/// nesting), so a direct-parent lookup is enough — matches the flat +/// child-iteration every render helper below already assumed. +fn group_by_root(nodes: Vec) -> HashMap> { + let mut groups: HashMap> = HashMap::new(); + for n in nodes { + let key = n.parent.unwrap_or(n.id); + groups.entry(key).or_default().push(n); + } + groups } -/// Non-TTY progress: print a fresh line whenever a DAG's rendered state +/// This id's own root node within its group, found by id rather than +/// "no parent" — `group_by_root` already keyed each group by its root's +/// id, so the root is just the member whose id matches the key. +fn root_in_group(id: u64, nodes: &[GraphNode]) -> Option<&GraphNode> { + nodes.iter().find(|n| n.id == id) +} + +/// Non-TTY progress: print a fresh line whenever a node's rendered state /// changes. No cursor tricks, so it's clean in pipes and CI logs. -async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { - let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); - let mut last: std::collections::HashMap = std::collections::HashMap::new(); +async fn wait_for_nodes_plain(socket: &Path, ids: Vec) -> Result<()> { + let mut pending: BTreeSet = ids.into_iter().collect(); + let mut last: HashMap = HashMap::new(); let mut failed: Vec = Vec::new(); while !pending.is_empty() { + let batch: Vec = pending.iter().copied().collect(); + let resp = crate::client::request( + socket, + hive_host_sock::HostRequest::QueueNodes { ids: batch }, + ) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let groups = group_by_root(resp.nodes.unwrap_or_default()); for id in pending.clone() { - let resp = - crate::client::request(socket, hive_host_sock::HostRequest::QueueNodes { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let nodes = resp.nodes.unwrap_or_default(); - let Some(root) = find_root(&nodes) else { + let nodes = groups.get(&id); + let root = nodes.and_then(|n| root_in_group(id, n)); + let (Some(nodes), Some(root)) = (nodes, root) else { // Unknown id — nothing prunes the graph yet (see - // `JobQueue::node_subtree`'s doc comment), so an id that - // resolves to no node never named a real DAG. A - // *completed* DAG's nodes keep riding here instead, with a + // `JobQueue::node_subtrees`'s doc comment), so an id that + // resolves to no node never named a real job. A + // *completed* job's nodes keep riding here instead, with a // terminal root `state`, which is what the check below // watches for. println!("job #{id}: gone from queue history"); pending.remove(&id); continue; }; - let line = render_dag_line(root, &nodes); + let line = render_node_line(root, nodes); if last.get(&id) != Some(&line) { println!("{line}"); last.insert(id, line); } if root.state.is_terminal() { if root.state == State::Failed { - failed.push(format!("{} {}", dag_source(root), dag_agents(&nodes))); + failed.push(format!("{} {}", node_source(root), node_agents(nodes))); } pending.remove(&id); } @@ -95,11 +123,11 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec) -> Result<()> { } /// TTY progress: a live `indicatif` render — one braille-spinner line per -/// DAG node (grouped under a per-DAG header), each with its own elapsed +/// node (grouped under a per-root header), each with its own elapsed /// timer, plus an overall-elapsed footer. A node with more than one /// dependency (fan-in) gets its own row with an `(after …)` marker rather /// than being crammed onto a chain line. -async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { +async fn wait_for_nodes_animated(socket: &Path, ids: Vec) -> Result<()> { use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; let spinner = ProgressStyle::with_template(" {spinner} {msg}") @@ -110,32 +138,35 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { let mp = MultiProgress::new(); let started = std::time::Instant::now(); - // Per-DAG header bars + per-node bars, keyed so we update in place. The - // insertion order groups a DAG's nodes right under its header. - let mut dag_bars: std::collections::HashMap = - std::collections::HashMap::new(); - let mut node_bars: std::collections::HashMap<(u64, WireId), ProgressBar> = - std::collections::HashMap::new(); - let mut node_done: std::collections::HashSet<(u64, WireId)> = std::collections::HashSet::new(); + // Per-root header bars + per-node bars, keyed so we update in place. + // The insertion order groups a root's nodes right under its header. + let mut root_bars: HashMap = HashMap::new(); + let mut node_bars: HashMap<(u64, WireId), ProgressBar> = HashMap::new(); + let mut node_done: HashSet<(u64, WireId)> = HashSet::new(); - let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); + let mut pending: BTreeSet = ids.into_iter().collect(); let mut failed: Vec = Vec::new(); while !pending.is_empty() { let now = now_unix(); + let batch: Vec = pending.iter().copied().collect(); + let resp = crate::client::request( + socket, + hive_host_sock::HostRequest::QueueNodes { ids: batch }, + ) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let groups = group_by_root(resp.nodes.unwrap_or_default()); for id in pending.clone() { - let resp = - crate::client::request(socket, hive_host_sock::HostRequest::QueueNodes { id }) - .await - .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - let nodes = resp.nodes.unwrap_or_default(); - let Some(root) = find_root(&nodes) else { + let nodes = groups.get(&id); + let root = nodes.and_then(|n| root_in_group(id, n)); + let (Some(nodes), Some(root)) = (nodes, root) else { mp.println(format!("job #{id}: gone from queue history")) .ok(); pending.remove(&id); continue; }; - let hdr = dag_bars.entry(id).or_insert_with(|| { + let hdr = root_bars.entry(id).or_insert_with(|| { let b = mp.add(ProgressBar::new_spinner()); b.set_style(plain.clone()); b @@ -143,8 +174,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { hdr.set_message(format!( "{} {} {} · {}", state_glyph(root.state), - dag_source(root), - dag_agents(&nodes), + node_source(root), + node_agents(nodes), fmt_dur(node_elapsed(root, now)), )); for n in nodes.iter().filter(|n| n.id != root.id) { @@ -163,16 +194,16 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { bar.finish_with_message(format!( " {} {}", state_glyph(n.state), - node_line(&nodes, n, now) + node_line(nodes, n, now) )); node_done.insert(key); } else { - bar.set_message(node_line(&nodes, n, now)); + bar.set_message(node_line(nodes, n, now)); } } if root.state.is_terminal() { if root.state == State::Failed { - failed.push(format!("{} {}", dag_source(root), dag_agents(&nodes))); + failed.push(format!("{} {}", node_source(root), node_agents(nodes))); } pending.remove(&id); } @@ -183,7 +214,7 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec) -> Result<()> { } // Leave the final node lines on screen; drop the still-spinning headers // into their terminal state and print an overall-elapsed footer. - for hdr in dag_bars.values() { + for hdr in root_bars.values() { hdr.finish(); } mp.println(format!( @@ -205,12 +236,11 @@ fn finish_wait(mut failed: Vec) -> Result<()> { } } -/// The DAG's `source` tag (`"manual"`, `"meta_update"`, …), read from the -/// container node's `payload.data` — see `hive-c0re`'s `impl WireNode for -/// NodeKind`'s `NodeKind::Dag` arm, the one place it's put on the wire. -/// Falls back to the label when absent (defensive; every real DAG -/// container sets it). -fn dag_source(root: &GraphNode) -> &str { +/// A root's `source` tag (`"manual"`, `"meta_update"`, …), read from its +/// `payload.data` — see `hive-c0re`'s `impl WireNode for NodeKind`'s +/// `NodeKind::Dag` arm, the one place it's put on the wire. Falls back to +/// the label when absent (defensive; every real root sets it). +fn node_source(root: &GraphNode) -> &str { root.payload .data .get("source") @@ -218,11 +248,11 @@ fn dag_source(root: &GraphNode) -> &str { .unwrap_or(&root.payload.label) } -/// Distinct agents across a DAG's nodes, comma-joined for display. Each +/// Distinct agents across a root's nodes, comma-joined for display. Each /// node's `agent` (when it targets one) rides in `payload.data["agent"]` — /// an opaque kvp, not a typed field, since `GraphNode` carries nothing /// domain-specific (see `hive_jobq_wire::WireNode::data`'s doc comment). -fn dag_agents(nodes: &[GraphNode]) -> String { +fn node_agents(nodes: &[GraphNode]) -> String { let mut seen: Vec<&str> = Vec::new(); for n in nodes { if let Some(agent) = n @@ -318,24 +348,23 @@ fn state_glyph(state: State) -> &'static str { } } -/// One progress line for a DAG: roll-up glyph, `source`, agents, then the +/// One progress line for a root: roll-up glyph, `source`, agents, then the /// node chain — the CLI twin of the dashboard's queue card. The glyph and -/// `source` come off `root` (the one entry in `nodes` with no parent); -/// everything else comes straight off the wire. Used by the plain -/// (non-TTY) path. +/// `source` come off `root`; everything else comes straight off the wire. +/// Used by the plain (non-TTY) path. /// /// The chain shows every entry `wire_snapshot` sends, `Done` ones /// included — that filtering was a dashboard-history-bounding concern, /// not relevant to a single actively-watched job, so a completed step /// keeps its checkmark instead of vanishing from the line, matching how /// the animated path already behaves. -fn render_dag_line(root: &GraphNode, nodes: &[GraphNode]) -> String { +fn render_node_line(root: &GraphNode, nodes: &[GraphNode]) -> String { use std::fmt::Write as _; let mut out = format!( "{} {} {:<12}", state_glyph(root.state), - dag_source(root), - dag_agents(nodes) + node_source(root), + node_agents(nodes) ); let children: Vec<&GraphNode> = nodes.iter().filter(|n| n.id != root.id).collect(); for (i, n) in children.iter().enumerate() { @@ -355,9 +384,9 @@ mod tests { use hive_jobq_wire::{GraphNode, NodePayload}; use serde_json::json; - use super::render_dag_line; + use super::render_node_line; - fn dag_root(id: u64, source: &str, state: State) -> GraphNode { + fn root_node(id: u64, source: &str, state: State) -> GraphNode { GraphNode { id, parent: None, @@ -368,7 +397,7 @@ mod tests { finished_at: None, error: None, payload: NodePayload { - label: "dag".to_owned(), + label: "job".to_owned(), data: json!({ "source": source }), }, } @@ -392,12 +421,12 @@ mod tests { } #[test] - fn render_dag_line_shows_source_and_chain() { + fn render_node_line_shows_source_and_chain() { // The header shows what the backend sends: the glyph off `root`'s // own `state` (`Running` here) + `source` ("manual"); the operation // is read off the node chain, not a client-side label. `Done` // nodes stay in the chain — nothing here filters them off. - let root = dag_root(7, "manual", State::Running); + let root = root_node(7, "manual", State::Running); let nodes = vec![ root.clone(), work_node(0, 7, "alice", "prebuild", State::Done), @@ -405,7 +434,7 @@ mod tests { work_node(2, 7, "alice", "swap", State::Running), work_node(3, 7, "alice", "reconcile", State::Pending), ]; - let line = render_dag_line(&root, &nodes); + let line = render_node_line(&root, &nodes); assert!(line.starts_with("▶ manual alice"), "{line}"); assert!( line.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"), @@ -414,13 +443,32 @@ mod tests { } #[test] - fn render_dag_line_surfaces_first_node_error() { - let root = dag_root(8, "manual", State::Failed); + fn render_node_line_surfaces_first_node_error() { + let root = root_node(8, "manual", State::Failed); let mut failed = work_node(0, 8, "bob", "prebuild", State::Failed); failed.error = Some("nix build exploded".to_owned()); let nodes = vec![root.clone(), failed]; - let line = render_dag_line(&root, &nodes); + let line = render_node_line(&root, &nodes); assert!(line.contains("✖ manual"), "{line}"); assert!(line.contains("— nix build exploded"), "{line}"); } + + #[test] + fn group_by_root_splits_a_combined_batch_response() { + // Two independent roots' subtrees riding the same QueueNodes + // response (the whole point of batching): each node must land in + // its own root's group, not get mixed into the other's. + let nodes = vec![ + root_node(1, "manual", State::Running), + work_node(10, 1, "alice", "prebuild", State::Running), + root_node(2, "manual", State::Done), + work_node(20, 2, "bob", "prebuild", State::Done), + ]; + let groups = super::group_by_root(nodes); + assert_eq!(groups.len(), 2); + assert_eq!(groups[&1].len(), 2, "root 1's group must have its child"); + assert_eq!(groups[&2].len(), 2, "root 2's group must have its child"); + assert!(groups[&1].iter().any(|n| n.id == 10)); + assert!(groups[&2].iter().any(|n| n.id == 20)); + } } diff --git a/hivectl/src/main.rs b/hivectl/src/main.rs index 8c68ac70..95c4751b 100644 --- a/hivectl/src/main.rs +++ b/hivectl/src/main.rs @@ -24,7 +24,7 @@ mod cli; /// The host admin socket client (`request`), split out so it lives with /// hivectl rather than in the daemon crate. mod client; -/// Rebuild-queue DAG progress rendering (`wait_for_dags` + the spinner / +/// Rebuild-queue node progress rendering (`wait_for_nodes` + the spinner / /// plain renderers), split out to keep this file manageable. mod dag_progress; use cli::{Cli, Cmd, ForgeCmd, GatewayCmd, GithubCmd, WgCmd}; diff --git a/hivectl/src/power.rs b/hivectl/src/power.rs index f082c529..4966e248 100644 --- a/hivectl/src/power.rs +++ b/hivectl/src/power.rs @@ -5,7 +5,7 @@ use std::path::Path; use anyhow::{Context as _, Result}; -use crate::dag_progress::wait_for_dags; +use crate::dag_progress::wait_for_nodes; use crate::util::render_lifecycle; pub(crate) async fn stop( @@ -24,7 +24,7 @@ pub(crate) async fn stop( // watch the already-queued agent DAGs before surfacing the error — // they run regardless. let rendered = render_lifecycle(&resp, "stop queued"); - wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?; + wait_for_nodes(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?; rendered } @@ -37,7 +37,7 @@ pub(crate) async fn start( .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; let rendered = render_lifecycle(&resp, "start queued"); - wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?; + wait_for_nodes(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?; rendered } @@ -66,6 +66,6 @@ pub(crate) async fn restart( .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; let rendered = render_lifecycle(&resp, "restart queued"); - wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), false).await?; + wait_for_nodes(socket, resp.queued_dags.unwrap_or_default(), false).await?; rendered } diff --git a/hivectl/src/subvol.rs b/hivectl/src/subvol.rs index b1427732..f6740fb8 100644 --- a/hivectl/src/subvol.rs +++ b/hivectl/src/subvol.rs @@ -7,7 +7,7 @@ use std::path::Path; use anyhow::{Context as _, Result, bail}; use crate::cli::{SnapshotCmd, SubvolCmd}; -use crate::dag_progress::wait_for_dags; +use crate::dag_progress::wait_for_nodes; use crate::util::{agent_exists, daemon_request}; /// A [`LifecycleScope`](hive_host_sock::LifecycleScope) targeting exactly one @@ -78,10 +78,10 @@ async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> { stop_resp.error.as_deref().unwrap_or("unknown error") ); } - // The stop is a queued DAG now — the migration below snapshots + + // The stop is a queued job now — the migration below snapshots + // swaps the state dir and MUST NOT run under a live bind mount, so // wait for the stop to actually execute before touching anything. - wait_for_dags(socket, stop_resp.queued_dags.unwrap_or_default(), false) + wait_for_nodes(socket, stop_resp.queued_dags.unwrap_or_default(), false) .await .with_context(|| format!("waiting for {name} to stop before the migration"))?; @@ -130,7 +130,7 @@ async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> { start_resp.error.as_deref().unwrap_or("unknown error") ); } - wait_for_dags(socket, start_resp.queued_dags.unwrap_or_default(), false) + wait_for_nodes(socket, start_resp.queued_dags.unwrap_or_default(), false) .await .with_context(|| { format!( From b04e7d985d9928c015288da9effed1007b60f75d Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 20:25:47 +0200 Subject: [PATCH 4/4] job_queue: stop inventing wire data just to preserve the old source tag --- hive-c0re/src/job_queue/model.rs | 9 ------ hivectl/src/dag_progress.rs | 48 +++++++++++++------------------- 2 files changed, 19 insertions(+), 38 deletions(-) diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index a15e11d1..9f45c3b1 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -308,15 +308,6 @@ impl hive_jobq_wire::WireNode for NodeKind { { data.insert("inputs".to_owned(), inputs.clone().into()); } - // The DAG container's own metadata — nowhere else on the wire, since - // `GraphNode` carries no DAG-level fields (a group root is an - // ordinary node). `hivectl` needs `source` for its progress line; - // `reason` rides along for free rather than adding a second variant - // later for the one field the first pass missed. - if let NodeKind::Dag { source, reason, .. } = self { - data.insert("source".to_owned(), source.as_str().into()); - data.insert("reason".to_owned(), reason.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)) { diff --git a/hivectl/src/dag_progress.rs b/hivectl/src/dag_progress.rs index 7d137b52..fa25516b 100644 --- a/hivectl/src/dag_progress.rs +++ b/hivectl/src/dag_progress.rs @@ -7,10 +7,11 @@ //! and a plain line-on-change stream otherwise. //! //! Consumes `hive-jobq-wire`'s generic [`GraphNode`]/`NodePayload`: a -//! node's kind comes from `payload.label`, and node-specific extras -//! (`source`, `agent`) ride in `payload.data`'s opaque kvps — the shape -//! `hive-c0re`'s `impl WireNode for NodeKind` produces (see its doc -//! comment). There's no separate roll-up field on the wire: a node's own +//! node's kind (and, for a root, what submitted it) comes straight off +//! `payload.label`; node-specific extras (`agent`) ride in +//! `payload.data`'s opaque kvps — the shape `hive-c0re`'s +//! `impl WireNode for NodeKind` produces (see its doc comment). There's +//! no separate roll-up field on the wire: a node's own //! `state` already reflects everything below it (see `hive_jobq_wire`'s //! doc comment), so nothing here re-derives a roll-up or waits for every //! node to go terminal before reading a result. @@ -110,7 +111,7 @@ async fn wait_for_nodes_plain(socket: &Path, ids: Vec) -> Result<()> { } if root.state.is_terminal() { if root.state == State::Failed { - failed.push(format!("{} {}", node_source(root), node_agents(nodes))); + failed.push(format!("{} {}", root.payload.label, node_agents(nodes))); } pending.remove(&id); } @@ -174,7 +175,7 @@ async fn wait_for_nodes_animated(socket: &Path, ids: Vec) -> Result<()> { hdr.set_message(format!( "{} {} {} · {}", state_glyph(root.state), - node_source(root), + root.payload.label, node_agents(nodes), fmt_dur(node_elapsed(root, now)), )); @@ -203,7 +204,7 @@ async fn wait_for_nodes_animated(socket: &Path, ids: Vec) -> Result<()> { } if root.state.is_terminal() { if root.state == State::Failed { - failed.push(format!("{} {}", node_source(root), node_agents(nodes))); + failed.push(format!("{} {}", root.payload.label, node_agents(nodes))); } pending.remove(&id); } @@ -236,18 +237,6 @@ fn finish_wait(mut failed: Vec) -> Result<()> { } } -/// A root's `source` tag (`"manual"`, `"meta_update"`, …), read from its -/// `payload.data` — see `hive-c0re`'s `impl WireNode for NodeKind`'s -/// `NodeKind::Dag` arm, the one place it's put on the wire. Falls back to -/// the label when absent (defensive; every real root sets it). -fn node_source(root: &GraphNode) -> &str { - root.payload - .data - .get("source") - .and_then(serde_json::Value::as_str) - .unwrap_or(&root.payload.label) -} - /// Distinct agents across a root's nodes, comma-joined for display. Each /// node's `agent` (when it targets one) rides in `payload.data["agent"]` — /// an opaque kvp, not a typed field, since `GraphNode` carries nothing @@ -348,9 +337,9 @@ fn state_glyph(state: State) -> &'static str { } } -/// One progress line for a root: roll-up glyph, `source`, agents, then the +/// One progress line for a root: roll-up glyph, label, agents, then the /// node chain — the CLI twin of the dashboard's queue card. The glyph and -/// `source` come off `root`; everything else comes straight off the wire. +/// label come off `root`; everything else comes straight off the wire. /// Used by the plain (non-TTY) path. /// /// The chain shows every entry `wire_snapshot` sends, `Done` ones @@ -363,7 +352,7 @@ fn render_node_line(root: &GraphNode, nodes: &[GraphNode]) -> String { let mut out = format!( "{} {} {:<12}", state_glyph(root.state), - node_source(root), + root.payload.label, node_agents(nodes) ); let children: Vec<&GraphNode> = nodes.iter().filter(|n| n.id != root.id).collect(); @@ -386,7 +375,7 @@ mod tests { use super::render_node_line; - fn root_node(id: u64, source: &str, state: State) -> GraphNode { + fn root_node(id: u64, label: &str, state: State) -> GraphNode { GraphNode { id, parent: None, @@ -397,8 +386,8 @@ mod tests { finished_at: None, error: None, payload: NodePayload { - label: "job".to_owned(), - data: json!({ "source": source }), + label: label.to_owned(), + data: json!({}), }, } } @@ -421,11 +410,12 @@ mod tests { } #[test] - fn render_node_line_shows_source_and_chain() { + fn render_node_line_shows_label_and_chain() { // The header shows what the backend sends: the glyph off `root`'s - // own `state` (`Running` here) + `source` ("manual"); the operation - // is read off the node chain, not a client-side label. `Done` - // nodes stay in the chain — nothing here filters them off. + // own `state` (`Running` here) + `root`'s own `payload.label` + // ("manual"); the operation is read off the node chain, not any + // extra metadata. `Done` nodes stay in the chain — nothing here + // filters them off. let root = root_node(7, "manual", State::Running); let nodes = vec![ root.clone(),