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

feat(hivectl): queue-routed lifecycle verbs with wait + DAG progress

Browse source 
every agent lifecycle verb on the admin socket (rebuild / restart /
restart-all / kill / stop / start) now submits job-queue DAGs and
returns their ids; hivectl polls the new HostRequest::QueueDag and
prints a live node-chain progress line per DAG (fan-out children
included), exiting non-zero on failure — --no-wait opts out. DagView
and the queue wire enums move to hive_sh4re::jobs (wire types live in
the shared crate); the last fused rebuild path (lifecycle::rebuild)
is gone. tracker: #2166
This commit is contained in:
müde 2026-07-06 22:30:49 +02:00
commit b489454dc2
9 changed files with 641 additions and 443 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -146,6 +146,10 @@ enum Cmd {
/// hard stop). All drains overlap. Applies to agents only.
#[arg(long)]
graceful: bool,
/// Return immediately after the stop DAGs are queued instead of
/// waiting for them with live per-node progress.
#[arg(long)]
no_wait: bool,
},
/// Start containers hive-wide — the inverse of `hivectl stop`. Bare
/// `hivectl start` starts everything back up; the same scope flags as
@ -154,6 +158,10 @@ enum Cmd {
Start {
#[command(flatten)]
scope: ScopeArgs,
/// Return immediately after the start DAGs are queued instead
/// of waiting for them with live per-node progress.
#[arg(long)]
no_wait: bool,
},
/// Restart containers hive-wide — `stop` then `start` over the same
/// scope. Bare `hivectl restart` restarts **everything** (all sub-agents
@ -531,15 +539,23 @@ enum AgentsCmd {
/// Stop and start a single agent container without rebuilding config.
/// Useful for "kick the container" when the process is stuck or the
/// container needs a clean restart without changing the NixOS config.
/// Rides the job queue (serialized against in-flight rebuilds for
/// the same agent); waits with live progress unless `--no-wait`.
Restart {
/// Agent name (e.g. `damocles`, `ruth`).
name: String,
/// Return immediately after the restart DAG is queued.
#[arg(long)]
no_wait: bool,
},
/// Restart ALL managed agent containers via one restart DAG each —
/// unrelated agents overlap, each serializes on its own lease.
/// Waits for the whole set with live progress unless `--no-wait`.
RestartAll {
/// Return immediately after the restart DAGs are queued.
#[arg(long)]
no_wait: bool,
},
/// Stop and restart ALL managed agent containers in sequence.
/// Iterates the live container list and restarts each one. Any per-agent
/// failure is reported at the end rather than stopping mid-run, so all
/// containers get a restart attempt.
RestartAll,
}
#[derive(Subcommand)]
@ -602,8 +618,8 @@ async fn main() -> Result<()> {
},
Cmd::Agents { cmd } => match cmd {
AgentsCmd::List { json } => agents_list(&socket, json).await,
AgentsCmd::Restart { name } => agents_restart(&socket, &name).await,
AgentsCmd::RestartAll => agents_restart_all(&socket).await,
AgentsCmd::Restart { name, no_wait } => agents_restart(&socket, &name, no_wait).await,
AgentsCmd::RestartAll { no_wait } => agents_restart_all(&socket, no_wait).await,
},
Cmd::Wg { cmd } => match cmd {
WgCmd::Init { address } => wg_init(&socket, address.as_deref()).await,
@ -626,8 +642,12 @@ async fn main() -> Result<()> {
peer_config(&domain, wg_address.as_deref(), wg_endpoint.as_deref());
Ok(())
}
Cmd::Stop { scope, graceful } => stop(&socket, scope.to_scope(), graceful).await,
Cmd::Start { scope } => start(&socket, scope.to_scope()).await,
Cmd::Stop {
scope,
graceful,
no_wait,
} => stop(&socket, scope.to_scope(), graceful, no_wait).await,
Cmd::Start { scope, no_wait } => start(&socket, scope.to_scope(), no_wait).await,
Cmd::Restart { scope, graceful } => restart(&socket, scope.to_scope(), graceful).await,
Cmd::Subvol { cmd } => match cmd {
SubvolCmd::Upgrade { name, yes } => subvol_upgrade(&socket, &name, yes).await,
@ -1406,7 +1426,7 @@ fn gateway_list_users(file: &Path) -> Result<()> {
// Agent management helpers (require daemon via host admin socket)
// ---------------------------------------------------------------------------
async fn agents_restart(socket: &Path, name: &str) -> Result<()> {
async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()> {
let resp = hive_c0re::client::request(
socket,
hive_sh4re::HostRequest::Restart {
@ -1416,8 +1436,8 @@ async fn agents_restart(socket: &Path, name: &str) -> Result<()> {
.await
.with_context(|| format!("connect to daemon socket {}", socket.display()))?;
if resp.ok {
println!("restarted: {name}");
Ok(())
println!("restart queued: {name}");
wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
} else {
bail!(
"restart {name}: {}",
@ -1426,6 +1446,104 @@ async fn agents_restart(socket: &Path, name: &str) -> 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<()> {
if no_wait || ids.is_empty() {
return Ok(());
}
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);
}
match d.state {
hive_sh4re::jobs::State::Failed => {
failed.push(format!("{} {}", d.kind.as_str(), d.agent));
}
hive_sh4re::jobs::State::Done | hive_sh4re::jobs::State::Cancelled => {}
_ => all_terminal = false,
}
}
if all_terminal {
pending.remove(&id);
}
}
if !pending.is_empty() {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
}
if failed.is_empty() {
Ok(())
} else {
failed.sort();
failed.dedup();
bail!("queued job(s) failed: {}", failed.join(", "))
}
}
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
@ -1502,7 +1620,7 @@ async fn agents_list(socket: &Path, json: bool) -> Result<()> {
Ok(())
}
async fn agents_restart_all(socket: &Path) -> Result<()> {
async fn agents_restart_all(socket: &Path, no_wait: bool) -> Result<()> {
let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::RestartAll)
.await
.with_context(|| format!("connect to daemon socket {}", socket.display()))?;
@ -1511,7 +1629,7 @@ async fn agents_restart_all(socket: &Path) -> Result<()> {
println!("restart-all: no managed containers found");
} else {
for a in agents {
println!("restarted: {a}");
println!("restart queued: {a}");
}
}
if !resp.ok {
@ -1520,22 +1638,29 @@ async fn agents_restart_all(socket: &Path) -> Result<()> {
resp.error.as_deref().unwrap_or("unknown error")
);
}
Ok(())
wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
}
async fn stop(socket: &Path, scope: hive_sh4re::LifecycleScope, graceful: bool) -> Result<()> {
async fn stop(
socket: &Path,
scope: hive_sh4re::LifecycleScope,
graceful: bool,
no_wait: bool,
) -> Result<()> {
let resp =
hive_c0re::client::request(socket, hive_sh4re::HostRequest::Stop { scope, graceful })
.await
.with_context(|| format!("connect to daemon socket {}", socket.display()))?;
render_lifecycle(&resp, "stopped")
render_lifecycle(&resp, "stop queued")?;
wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
}
async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope) -> Result<()> {
async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope, no_wait: bool) -> Result<()> {
let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::Start { scope })
.await
.with_context(|| format!("connect to daemon socket {}", socket.display()))?;
render_lifecycle(&resp, "started")
render_lifecycle(&resp, "start queued")?;
wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
}
/// Restart = `stop` then `start` over the same scope, composed client-side
@ -1543,9 +1668,13 @@ async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope) -> Result<()> {
/// `--graceful`; if it reports a failure (`stop` returns `Err`) the `?`
/// short-circuits before the start phase, so a half-stopped hive isn't
/// blindly started over — the operator sees the stop errors and can recover.
///
/// No `--no-wait` here on purpose: the stop DAGs must complete before
/// the start submits, otherwise the start's `wanted = Up` write would
/// land before the queued stops execute and turn them into noops.
async fn restart(socket: &Path, scope: hive_sh4re::LifecycleScope, graceful: bool) -> Result<()> {
stop(socket, scope.clone(), graceful).await?;
start(socket, scope).await
stop(socket, scope.clone(), graceful, false).await?;
start(socket, scope, false).await
}
/// A [`LifecycleScope`](hive_sh4re::LifecycleScope) targeting exactly one
@ -1673,3 +1802,82 @@ 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}");
}
}