vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#2282): split dag_progress out of hivectl.rs

Browse source
This commit is contained in:
damocles 2026-07-10 16:24:18 +02:00 • committed by mara
commit e0d5c98356
2 changed files with 394 additions and 374 deletions
Download patch file Download diff file Expand all files Collapse all files

382
hive-c0re/src/bin/hivectl.rs
View file

@ -22,6 +22,14 @@ use anyhow::{Context as _, Result, bail};
use clap::{Args, Parser, Subcommand};
use hive_c0re::coordinator::Coordinator;
/// Rebuild-queue DAG progress rendering (`wait_for_dags` + the spinner /
/// plain renderers), split out to keep this file manageable. `#[path]` keeps
/// the file under `bin/hivectl/` (a subdir cargo won't treat as its own
/// binary) rather than the sibling `bin/dag_progress.rs` a bare `mod` maps to.
#[path = "hivectl/dag_progress.rs"]
mod dag_progress;
use dag_progress::wait_for_dags;
#[derive(Parser)]
#[command(
name = "hivectl",
@ -1449,301 +1457,6 @@ async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()>
}
}
// ---------------------------------------------------------------------------
// Job-queue wait/progress loop — shared by every verb that submits DAGs
// ---------------------------------------------------------------------------
/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~1s interval)
/// and print a progress line whenever a DAG's rendered state changes —
/// including fan-out children that appear under a polled parent. Exits
/// non-zero when any DAG (or child) ends `failed`; a `cancelled` DAG
/// terminates the wait but is an operator action, not an error.
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 = hive_c0re::client::request(socket, hive_sh4re::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.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.kind.as_str(), d.agent));
}
} 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_sh4re::jobs::NodeId), ProgressBar> =
std::collections::HashMap::new();
let mut node_done: std::collections::HashSet<(u64, hive_sh4re::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 = hive_c0re::client::request(socket, hive_sh4re::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.state),
d.kind.as_str(),
d.agent,
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.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.kind.as_str(), d.agent));
}
} 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(", "))
}
}
/// 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 `enqueued_at`)
/// through `finished_at` or `now`.
fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 {
let start = d.started_at.unwrap_or(d.enqueued_at);
(d.finished_at.unwrap_or(now) - 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_sh4re::jobs::NodeView, now: i64) -> i64 {
match n.started_at {
Some(start) => (n.finished_at.unwrap_or(now) - start).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, live step, an `(after …)` marker for a
/// fan-in node (>1 dep), its elapsed timer, and a truncated error tail.
fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String {
use std::fmt::Write as _;
let mut s = n.kind.clone();
if n.state == hive_sh4re::jobs::State::Running
&& let Some(step) = &n.step
{
let _ = write!(s, " ({step})");
}
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_sh4re::jobs::State) -> &'static str {
match state {
hive_sh4re::jobs::State::Queued => "⏸",
hive_sh4re::jobs::State::Running => "▶",
hive_sh4re::jobs::State::Done => "✔",
hive_sh4re::jobs::State::Failed => "✖",
hive_sh4re::jobs::State::Cancelled => "⊘",
}
}
/// One progress line for a DAG: roll-up glyph, template, agent, then
/// the node chain with the running node's live step label — the CLI
/// twin of the dashboard's queue card.
fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
use std::fmt::Write as _;
let mut out = format!(
"{} {} {:<12}",
state_glyph(d.state),
d.kind.as_str(),
d.agent
);
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 n.state == hive_sh4re::jobs::State::Running
&& let Some(step) = &n.step
{
let _ = write!(out, " ({step})");
}
}
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
}
/// `hivectl agents list` — fetch the per-agent status roster from the
/// daemon (`HostRequest::AgentStatus`) and render it as a padded table,
/// or the raw JSON rows with `--json`. Reuses the dashboard's
@ -2021,82 +1734,3 @@ fn validate_htpasswd_username(username: &str) -> Result<()> {
}
Ok(())
}
#[cfg(test)]
mod tests {
use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template};
use super::render_dag_line;
fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView {
NodeView {
id,
kind: kind.to_owned(),
deps: if id == 0 { vec![] } else { vec![id - 1] },
state,
step: step.map(str::to_owned),
build_log_id: None,
started_at: None,
finished_at: None,
error: None,
}
}
#[test]
fn render_dag_line_shows_chain_and_running_step() {
let dag = DagView {
id: 7,
agent: "alice".to_owned(),
kind: Template::Rebuild,
state: State::Running,
source: Source::Manual,
parent_id: None,
reason: "manual".to_owned(),
enqueued_at: 0,
started_at: Some(1),
finished_at: None,
inputs: vec![],
approval_id: None,
perm_payload: None,
nodes: vec![
node(0, "prebuild", State::Done, None),
node(1, "stop_for_update", State::Done, None),
node(2, "swap", State::Running, Some("nixos-container update")),
node(3, "reconcile", State::Queued, None),
],
};
let line = render_dag_line(&dag);
assert!(line.starts_with("▶ rebuild alice"), "{line}");
assert!(
line.contains(
"✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile"
),
"{line}"
);
}
#[test]
fn render_dag_line_surfaces_first_node_error() {
let mut failed = node(0, "prebuild", State::Failed, None);
failed.error = Some("nix build exploded".to_owned());
let dag = DagView {
id: 8,
agent: "bob".to_owned(),
kind: Template::Rebuild,
state: State::Failed,
source: Source::Manual,
parent_id: None,
reason: "manual".to_owned(),
enqueued_at: 0,
started_at: Some(1),
finished_at: Some(2),
inputs: vec![],
approval_id: None,
perm_payload: None,
nodes: vec![failed],
};
let line = render_dag_line(&dag);
assert!(line.contains("✖ rebuild"), "{line}");
assert!(line.contains("— nix build exploded"), "{line}");
}
}

386
hive-c0re/src/bin/hivectl/dag_progress.rs Normal file
View file

@ -0,0 +1,386 @@
//! `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 = hive_c0re::client::request(socket, hive_sh4re::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.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.kind.as_str(), d.agent));
}
} 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_sh4re::jobs::NodeId), ProgressBar> =
std::collections::HashMap::new();
let mut node_done: std::collections::HashSet<(u64, hive_sh4re::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 = hive_c0re::client::request(socket, hive_sh4re::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.state),
d.kind.as_str(),
d.agent,
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.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.kind.as_str(), d.agent));
}
} 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(", "))
}
}
/// 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 `enqueued_at`)
/// through `finished_at` or `now`.
fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 {
let start = d.started_at.unwrap_or(d.enqueued_at);
(d.finished_at.unwrap_or(now) - 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_sh4re::jobs::NodeView, now: i64) -> i64 {
match n.started_at {
Some(start) => (n.finished_at.unwrap_or(now) - start).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, live step, an `(after …)` marker for a
/// fan-in node (>1 dep), its elapsed timer, and a truncated error tail.
fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String {
use std::fmt::Write as _;
let mut s = n.kind.clone();
if n.state == hive_sh4re::jobs::State::Running
&& let Some(step) = &n.step
{
let _ = write!(s, " ({step})");
}
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_sh4re::jobs::State) -> &'static str {
match state {
hive_sh4re::jobs::State::Queued => "⏸",
hive_sh4re::jobs::State::Running => "▶",
hive_sh4re::jobs::State::Done => "✔",
hive_sh4re::jobs::State::Failed => "✖",
hive_sh4re::jobs::State::Cancelled => "⊘",
}
}
/// One progress line for a DAG: roll-up glyph, template, agent, then
/// the node chain with the running node's live step label — the CLI
/// twin of the dashboard's queue card. Used by the plain (non-TTY) path.
fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
use std::fmt::Write as _;
let mut out = format!(
"{} {} {:<12}",
state_glyph(d.state),
d.kind.as_str(),
d.agent
);
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 n.state == hive_sh4re::jobs::State::Running
&& let Some(step) = &n.step
{
let _ = write!(out, " ({step})");
}
}
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_sh4re::jobs::{DagView, NodeView, Source, State, Template};
use super::render_dag_line;
fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView {
NodeView {
id,
kind: kind.to_owned(),
deps: if id == 0 { vec![] } else { vec![id - 1] },
state,
step: step.map(str::to_owned),
build_log_id: None,
started_at: None,
finished_at: None,
error: None,
}
}
#[test]
fn render_dag_line_shows_chain_and_running_step() {
let dag = DagView {
id: 7,
agent: "alice".to_owned(),
kind: Template::Rebuild,
state: State::Running,
source: Source::Manual,
parent_id: None,
reason: "manual".to_owned(),
enqueued_at: 0,
started_at: Some(1),
finished_at: None,
inputs: vec![],
approval_id: None,
perm_payload: None,
nodes: vec![
node(0, "prebuild", State::Done, None),
node(1, "stop_for_update", State::Done, None),
node(2, "swap", State::Running, Some("nixos-container update")),
node(3, "reconcile", State::Queued, None),
],
};
let line = render_dag_line(&dag);
assert!(line.starts_with("▶ rebuild alice"), "{line}");
assert!(
line.contains(
"✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile"
),
"{line}"
);
}
#[test]
fn render_dag_line_surfaces_first_node_error() {
let mut failed = node(0, "prebuild", State::Failed, None);
failed.error = Some("nix build exploded".to_owned());
let dag = DagView {
id: 8,
agent: "bob".to_owned(),
kind: Template::Rebuild,
state: State::Failed,
source: Source::Manual,
parent_id: None,
reason: "manual".to_owned(),
enqueued_at: 0,
started_at: Some(1),
finished_at: Some(2),
inputs: vec![],
approval_id: None,
perm_payload: None,
nodes: vec![failed],
};
let line = render_dag_line(&dag);
assert!(line.contains("✖ rebuild"), "{line}");
assert!(line.contains("— nix build exploded"), "{line}");
}
}