b9aab7e923
`hive_host_sock::jobs::State` was a hand-maintained copy of `hive_jobq::State` — five variants spelled the same in both, kept in sync by whoever remembered. Adding `Skipped` last week meant adding it twice. The wire crate now re-exports the scheduler's enum and `to_wire_state` is gone. Two states that were hidden now reach clients. `to_wire_state` renamed `Pending` to `Queued` and folded `Finishing` into `Running`, so the dashboard could not distinguish a node waiting on its dependencies from one whose own work is done while its sub-nodes still run. Both are now visible, and consumers say which they mean. Every consumer had to move with it, and only the Rust ones said so: the exhaustive matches in `hivectl` and `DagView::rollup_state` failed to compile, while the dashboard's fourteen string comparisons would have gone quietly wrong — a `finishing` node no longer counting as running, a `pending` node no longer as queued. The frontend also builds CSS class names out of the state string (`rqe-` + state, `rqe-node-` + state) and keys its glyph map on it, all lowercase. Those go through a `stateSlug` helper now; comparisons use the wire spelling, presentation lowercases. Without that split every queue entry and node chip would have silently lost its styling. Dropping the `State as JobState` alias in hive-c0re falls out of this: the alias only existed to tell two `State` types apart, and there is one now.
395 lines
15 KiB
Rust
395 lines
15 KiB
Rust
//! `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
|
|
//! 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.
|
|
|
|
use std::path::Path;
|
|
|
|
use anyhow::{Context as _, Result, bail};
|
|
|
|
/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~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.
|
|
///
|
|
/// # 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<u64>, no_wait: bool) -> Result<()> {
|
|
use std::io::IsTerminal as _;
|
|
if no_wait || ids.is_empty() {
|
|
return Ok(());
|
|
}
|
|
// 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
|
|
} else {
|
|
wait_for_dags_plain(socket, ids).await
|
|
}
|
|
}
|
|
|
|
/// 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<u64>) -> Result<()> {
|
|
let mut pending: std::collections::BTreeSet<u64> = ids.into_iter().collect();
|
|
let mut last: std::collections::HashMap<u64, String> = std::collections::HashMap::new();
|
|
let mut failed: Vec<String> = 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.
|
|
println!("job #{id}: gone from queue history");
|
|
pending.remove(&id);
|
|
continue;
|
|
}
|
|
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);
|
|
}
|
|
// 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);
|
|
}
|
|
}
|
|
if !pending.is_empty() {
|
|
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
|
}
|
|
}
|
|
finish_wait(failed)
|
|
}
|
|
|
|
/// TTY progress: a live `indicatif` render — one braille-spinner line per
|
|
/// DAG node (grouped under a per-DAG 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<u64>) -> Result<()> {
|
|
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
|
|
|
|
let spinner = ProgressStyle::with_template(" {spinner} {msg}")
|
|
.unwrap_or_else(|_| ProgressStyle::default_spinner())
|
|
.tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏", "·"]);
|
|
let plain =
|
|
ProgressStyle::with_template("{msg}").unwrap_or_else(|_| ProgressStyle::default_bar());
|
|
|
|
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<u64, ProgressBar> =
|
|
std::collections::HashMap::new();
|
|
let mut node_bars: std::collections::HashMap<(u64, hive_host_sock::jobs::NodeId), ProgressBar> =
|
|
std::collections::HashMap::new();
|
|
let mut node_done: std::collections::HashSet<(u64, hive_host_sock::jobs::NodeId)> =
|
|
std::collections::HashSet::new();
|
|
|
|
let mut pending: std::collections::BTreeSet<u64> = ids.into_iter().collect();
|
|
let mut failed: Vec<String> = Vec::new();
|
|
|
|
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() {
|
|
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 b = mp.add(ProgressBar::new_spinner());
|
|
b.set_style(plain.clone());
|
|
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)));
|
|
}
|
|
} else {
|
|
all_terminal = false;
|
|
}
|
|
}
|
|
if all_terminal {
|
|
pending.remove(&id);
|
|
}
|
|
}
|
|
if !pending.is_empty() {
|
|
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
|
}
|
|
}
|
|
// 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() {
|
|
hdr.finish();
|
|
}
|
|
mp.println(format!(
|
|
"done · {} elapsed",
|
|
fmt_dur(started.elapsed().as_secs().try_into().unwrap_or(0))
|
|
))
|
|
.ok();
|
|
finish_wait(failed)
|
|
}
|
|
|
|
/// Shared tail: succeed when nothing failed, else bail listing the failures.
|
|
fn finish_wait(mut failed: Vec<String>) -> Result<()> {
|
|
if failed.is_empty() {
|
|
Ok(())
|
|
} else {
|
|
failed.sort();
|
|
failed.dedup();
|
|
bail!("queued job(s) failed: {}", failed.join(", "))
|
|
}
|
|
}
|
|
|
|
/// 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 {
|
|
let mut seen: Vec<&str> = Vec::new();
|
|
for n in &d.nodes {
|
|
if !seen.contains(&n.agent.as_str()) {
|
|
seen.push(&n.agent);
|
|
}
|
|
}
|
|
seen.join(",")
|
|
}
|
|
|
|
/// Current unix time in seconds (0 on the impossible pre-epoch error).
|
|
fn now_unix() -> i64 {
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.ok()
|
|
.and_then(|d| i64::try_from(d.as_secs()).ok())
|
|
.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<Utc>`; 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,
|
|
}
|
|
}
|
|
|
|
/// Compact duration: `45s` under a minute, else `1m03s`.
|
|
fn fmt_dur(secs: i64) -> String {
|
|
let s = secs.max(0);
|
|
if s < 60 {
|
|
format!("{s}s")
|
|
} else {
|
|
format!("{}m{:02}s", s / 60, s % 60)
|
|
}
|
|
}
|
|
|
|
/// 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 {
|
|
use std::fmt::Write as _;
|
|
let mut s = n.kind.clone();
|
|
if n.deps.len() > 1 {
|
|
let after: Vec<&str> = n
|
|
.deps
|
|
.iter()
|
|
.filter_map(|dep| d.nodes.iter().find(|m| m.id == *dep))
|
|
.map(|m| m.kind.as_str())
|
|
.collect();
|
|
if !after.is_empty() {
|
|
let _ = write!(s, " (after {})", after.join(", "));
|
|
}
|
|
}
|
|
let el = node_elapsed(n, now);
|
|
if el > 0 {
|
|
let _ = write!(s, " · {}", fmt_dur(el));
|
|
}
|
|
if let Some(err) = &n.error {
|
|
let short: String = err.chars().take(100).collect();
|
|
let _ = write!(s, " — {short}");
|
|
}
|
|
s
|
|
}
|
|
|
|
fn state_glyph(state: hive_host_sock::jobs::State) -> &'static str {
|
|
match state {
|
|
hive_host_sock::jobs::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 => "⊘",
|
|
// Distinct from cancelled: nothing went wrong, this branch just
|
|
// wasn't the one the run took.
|
|
hive_host_sock::jobs::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 {
|
|
use std::fmt::Write as _;
|
|
let mut out = format!(
|
|
"{} {} {:<12}",
|
|
state_glyph(d.rollup_state()),
|
|
d.source.as_str(),
|
|
dag_agents(d)
|
|
);
|
|
for (i, n) in d.nodes.iter().enumerate() {
|
|
let sep = if i == 0 { " " } else { " → " };
|
|
let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind);
|
|
}
|
|
if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) {
|
|
let short: String = err.chars().take(120).collect();
|
|
let _ = write!(out, " — {short}");
|
|
}
|
|
out
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use hive_host_sock::jobs::{DagView, NodeView, Source, State};
|
|
use hive_sh4re::wire_time::from_secs;
|
|
|
|
use super::render_dag_line;
|
|
|
|
fn node(id: u64, agent: &str, kind: &str, state: State) -> NodeView {
|
|
NodeView {
|
|
id,
|
|
parent: None,
|
|
agent: agent.to_owned(),
|
|
kind: kind.to_owned(),
|
|
deps: if id == 0 { vec![] } else { vec![id - 1] },
|
|
state,
|
|
started_at: None,
|
|
finished_at: None,
|
|
error: None,
|
|
approval_id: None,
|
|
inputs: vec![],
|
|
has_log: false,
|
|
}
|
|
}
|
|
|
|
#[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);
|
|
assert!(line.starts_with("▶ manual alice"), "{line}");
|
|
assert!(
|
|
line.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"),
|
|
"{line}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn render_dag_line_surfaces_first_node_error() {
|
|
let mut failed = node(0, "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);
|
|
assert!(line.contains("✖ manual"), "{line}");
|
|
assert!(line.contains("— nix build exploded"), "{line}");
|
|
}
|
|
}
|