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

Compare commits

base: vinzenz:0ae078008991172f674e054dc68c230317257ee0
Branches Tags
vinzenz:main
..
compare: vinzenz:a7f0f3d2318d662afeaacc4bc20c550a9dcf3e0c
Branches Tags
vinzenz:main
8 changed files with 224 additions and 368 deletions
Download patch file Download diff file Expand all files Collapse all files

36
hive-c0re/src/dashboard/build_logs.rs
View file

@ -90,42 +90,6 @@ pub(super) async fn get_build_log_full(
} }
} }
/// `GET /api/build-log/{node_id}` — the build log for a **queue node**,
/// resolved node id → log-row id → full log. Same `BuildLogFull` JSON
/// (`stdout` / `stderr` + header) as `get_build_log_full`; HTTP 404 when the
/// node has no linked log (the client gates the request on `NodeView.has_log`,
/// but a vacuum race can still 404). This is the on-demand log fetch the
/// raw-graph dashboard uses instead of an inline `build_log_id` on the wire.
pub(super) async fn get_build_log_for_node(
State(state): State<AppState>,
AxumPath(node_id): AxumPath<u64>,
) -> Response {
match state.coord.job_queue.build_log_id_of(node_id) {
Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await,
None => (
StatusCode::NOT_FOUND,
format!("node #{node_id} has no build log"),
)
.into_response(),
}
}
/// `GET /api/build-log/{node_id}/raw` — the node's build log as `text/plain`
/// for download (delegates to `get_build_log_raw` after resolving the node id).
pub(super) async fn get_build_log_raw_for_node(
State(state): State<AppState>,
AxumPath(node_id): AxumPath<u64>,
) -> Response {
match state.coord.job_queue.build_log_id_of(node_id) {
Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await,
None => (
StatusCode::NOT_FOUND,
format!("node #{node_id} has no build log"),
)
.into_response(),
}
}
/// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel. /// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel.
/// `stdout_append` / `stderr_append` carry only the new bytes since the /// `stdout_append` / `stderr_append` carry only the new bytes since the
/// last frame; `done = true` means the build finished and the stream /// last frame; `done = true` means the build finished and the stream

8
hive-c0re/src/dashboard/mod.rs
View file

@ -100,14 +100,6 @@ pub async fn serve(
) )
.route("/api/audit-log", get(misc_api::api_audit_log)) .route("/api/audit-log", get(misc_api::api_audit_log))
.route("/api/build-logs", get(build_logs::get_build_logs_all)) .route("/api/build-logs", get(build_logs::get_build_logs_all))
.route(
"/api/build-log/{node_id}",
get(build_logs::get_build_log_for_node),
)
.route(
"/api/build-log/{node_id}/raw",
get(build_logs::get_build_log_raw_for_node),
)
.route( .route(
"/api/build-logs/{agent}", "/api/build-logs/{agent}",
get(build_logs::get_build_logs_agent), get(build_logs::get_build_logs_agent),

144
hive-c0re/src/job_queue/mod.rs
View file

@ -38,7 +38,6 @@ mod tests;
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Mutex; use std::sync::Mutex;
use chrono::{DateTime, Utc};
use hive_jobq::resources::ResourceTable; use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState}; use hive_jobq::{Dep, DepWhen as JobDepWhen, Graph, NodeId, State as JobState};
@ -97,14 +96,15 @@ pub struct TerminalDag {
pub error: Option<String>, pub error: Option<String>,
} }
/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle /// Per-node runtime metadata the crate graph doesn't carry (kind + agent live
/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node` /// in the node payload; state lives in the node).
/// itself now, so only the two host-side extras remain: the live sub-step
/// label and the build-log row link (the client fetches the log by node id).
#[derive(Debug, Default, Clone)] #[derive(Debug, Default, Clone)]
struct NodeRuntime { struct NodeRuntime {
step: Option<String>, step: Option<String>,
build_log_id: Option<i64>, build_log_id: Option<i64>,
started_at: Option<i64>,
finished_at: Option<i64>,
error: Option<String>,
} }
/// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]). /// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
@ -362,6 +362,7 @@ impl JobQueue {
let mut inner = self.lock(); let mut inner = self.lock();
let inner = &mut *inner; let inner = &mut *inner;
let started = inner.sched.settle(); let started = inner.sched.settle();
let now = now_unix();
let mut claims = Vec::with_capacity(started.len()); let mut claims = Vec::with_capacity(started.len());
for id in started { for id in started {
let Some(node) = inner.sched.graph().node(id) else { let Some(node) = inner.sched.graph().node(id) else {
@ -385,8 +386,9 @@ impl JobQueue {
inputs: meta.inputs, inputs: meta.inputs,
transient: meta.transient, transient: meta.transient,
}); });
// `started_at` is stamped on the graph `Node` by the scheduler's if let Some(rt) = inner.node_rt.get_mut(&id) {
// transition to `Running` — no host-side copy needed. rt.started_at = Some(now);
}
} }
claims claims
} }
@ -403,15 +405,23 @@ impl JobQueue {
result: Result<(), String>, result: Result<(), String>,
) -> Option<TerminalDag> { ) -> Option<TerminalDag> {
let mut inner = self.lock(); let mut inner = self.lock();
// The failure reason + `finished_at` are stamped onto the graph `Node` let now = now_unix();
// by the scheduler (the reason rides `Outcome::Failed`); no host-side let (error, outcome) = match result {
// copy. We only clear the live sub-step label here. Ok(()) => (None, Outcome::Done),
let outcome = match result { Err(e) => {
Ok(()) => Outcome::Done, // The reason rides the crate `Outcome::Failed` (stamped onto the
Err(e) => Outcome::Failed(truncate_error(&e)), // graph `Node`); the `node_rt` copy stays for now until the wire
// reads it off the node directly.
let msg = truncate_error(&e);
(Some(msg.clone()), Outcome::Failed(msg))
}
}; };
if let Some(rt) = inner.node_rt.get_mut(&node_id) { if let Some(rt) = inner.node_rt.get_mut(&node_id) {
rt.finished_at = Some(now);
rt.step = None; rt.step = None;
if let Some(e) = error {
rt.error = Some(e);
}
} }
let container = inner.dag_of(node_id); let container = inner.dag_of(node_id);
inner.sched.complete(node_id, outcome); inner.sched.complete(node_id, outcome);
@ -511,21 +521,6 @@ impl JobQueue {
true true
} }
/// The `build_logs` row id linked to the wire node id `node_id`, if any —
/// the lookup behind the `GET /api/build-log/<node_id>` query endpoint (the
/// client fetches a node's captured build output on demand rather than
/// receiving it inline). Takes the raw wire `u64` (the endpoint's path
/// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the
/// matching id — the map is small (live + recently-terminal nodes).
#[must_use]
pub fn build_log_id_of(&self, node_id: u64) -> Option<i64> {
self.lock()
.node_rt
.iter()
.find(|(nid, _)| nid.get() == node_id)
.and_then(|(_, rt)| rt.build_log_id)
}
/// A DAG's terminal roll-up summary, computed on demand from its container. /// A DAG's terminal roll-up summary, computed on demand from its container.
/// `None` if the DAG id is unknown. Test-only — production reads the summary /// `None` if the DAG id is unknown. Test-only — production reads the summary
/// `complete_node` returns when the container rolls up terminal. /// `complete_node` returns when the container rolls up terminal.
@ -726,13 +721,15 @@ impl QueueInner {
seen seen
} }
/// First failed work node's error (read off the graph `Node`), for the /// First failed work node's stored error, for the roll-up `error` field.
/// terminal roll-up summary the inline hook consumes.
fn dag_first_error(&self, container: NodeId) -> Option<String> { fn dag_first_error(&self, container: NodeId) -> Option<String> {
for id in self.subtree(container) { for id in self.subtree(container) {
if let Some(n) = self.sched.graph().node(id) if self
&& n.state == JobState::Failed .sched
&& let Some(e) = n.error.clone() .graph()
.node(id)
.is_some_and(|n| n.state == JobState::Failed)
&& let Some(e) = self.node_rt.get(&id).and_then(|r| r.error.clone())
{ {
return Some(e); return Some(e);
} }
@ -752,35 +749,19 @@ impl QueueInner {
}) })
} }
/// Project a DAG into its wire [`DagView`]: a near-raw view of the /// Rebuild the wire [`DagView`] for a DAG from its container metadata + work
/// container's work nodes, with `Done` nodes excluded. Lifecycle /// nodes + per-node runtime.
/// (`state` / `started_at` / `finished_at` / `error`) is read straight
/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up
/// state, and DAG timestamps from the node set. Non-derivable per-node
/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns
/// `None` when every work node is `Done` — a fully-completed DAG drops
/// out of the snapshot entirely (a `Failed` one lingers until aged out).
fn dag_view(&self, container: NodeId) -> Option<DagView> { fn dag_view(&self, container: NodeId) -> Option<DagView> {
let meta = self.dag_meta(container)?; let meta = self.dag_meta(container)?;
let mut nodes = Vec::new(); let node_ids = self.subtree(container);
// DAG-level timestamps are taken over *all* subtree nodes (including the let mut nodes = Vec::with_capacity(node_ids.len());
// `Done` ones excluded from the wire) — the client can't derive them let mut started: Vec<i64> = Vec::new();
// from a `Done`-filtered node set, so the host computes them here. let mut finished: Vec<i64> = Vec::new();
let mut started: Vec<DateTime<Utc>> = Vec::new(); for &id in &node_ids {
let mut finished: Vec<DateTime<Utc>> = Vec::new();
for id in self.subtree(container) {
let Some(node) = self.sched.graph().node(id) else { let Some(node) = self.sched.graph().node(id) else {
continue; continue;
}; };
if let Some(s) = node.started_at { let rt = self.node_rt.get(&id);
started.push(s);
}
if let Some(f) = node.finished_at {
finished.push(f);
}
if node.state == JobState::Done {
continue;
}
let deps: Vec<u64> = node let deps: Vec<u64> = node
.deps .deps
.iter() .iter()
@ -789,54 +770,51 @@ impl QueueInner {
Dep::Resource { .. } => None, Dep::Resource { .. } => None,
}) })
.collect(); .collect();
// Non-derivable per-node payload rides the node that owns it. if let Some(s) = rt.and_then(|r| r.started_at) {
let approval_id = matches!(node.payload, NodeKind::ApprovalDeploy { .. }) started.push(s);
.then_some(meta.approval_id) }
.flatten(); if let Some(fin) = rt.and_then(|r| r.finished_at) {
let inputs = if matches!(node.payload, NodeKind::MetaLock { .. }) { finished.push(fin);
meta.inputs.clone() }
} else {
Vec::new()
};
let has_log = self.node_rt.get(&id).and_then(|r| r.build_log_id).is_some();
nodes.push(NodeView { nodes.push(NodeView {
id: id.get(), id: id.get(),
agent: node.payload.agent().to_owned(), agent: node.payload.agent().to_owned(),
kind: node.payload.as_str().to_owned(), kind: node.payload.as_str().to_owned(),
deps, deps,
state: to_wire_state(node.state), state: to_wire_state(node.state),
started_at: node.started_at, step: rt.and_then(|r| r.step.clone()),
finished_at: node.finished_at, build_log_id: rt.and_then(|r| r.build_log_id),
error: node.error.clone(), started_at: rt.and_then(|r| r.started_at),
approval_id, finished_at: rt.and_then(|r| r.finished_at),
inputs, error: rt.and_then(|r| r.error.clone()),
has_log,
}); });
} }
if nodes.is_empty() {
return None;
}
let is_terminal = self.dag_is_terminal(container); let is_terminal = self.dag_is_terminal(container);
Some(DagView { Some(DagView {
id: container.get(), id: container.get(),
kind: meta.template,
state: self.dag_rollup(container),
source: meta.source, source: meta.source,
reason: meta.reason.clone(), reason: meta.reason.clone(),
created_at: hive_sh4re::wire_time::from_secs(meta.created_at), enqueued_at: meta.created_at,
started_at: started.into_iter().min(), started_at: started.into_iter().min(),
finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), finished_at: if is_terminal {
finished.into_iter().max()
} else {
None
},
inputs: meta.inputs.clone(),
approval_id: meta.approval_id,
nodes, nodes,
}) })
} }
/// When a DAG's work node finishes on `finished_at` — the max over its /// When a DAG's work node finishes on `finished_at` — the max over its
/// subtree (read off the graph `Node`, as unix seconds), for the history /// subtree, for the history cap ordering.
/// cap ordering.
fn dag_finished_at(&self, container: NodeId) -> i64 { fn dag_finished_at(&self, container: NodeId) -> i64 {
self.subtree(container) self.subtree(container)
.iter() .iter()
.filter_map(|id| self.sched.graph().node(*id)) .filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at))
.filter_map(|n| n.finished_at)
.map(|t| t.timestamp())
.max() .max()
.unwrap_or(0) .unwrap_or(0)
} }

53
hive-c0re/src/job_queue/model.rs
View file

@ -11,62 +11,11 @@
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! DAG can span agents). See `docs/coordinator.md::Job queue` for the
//! full design. //! full design.
pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State}; pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State, Template};
use serde::Serialize; use serde::Serialize;
use crate::coordinator::TransientKind; use crate::coordinator::TransientKind;
/// What a DAG *means* — the request-level shape. Internal to the queue now:
/// it drives the terminal-hook dispatch ([`crate::job_queue`]'s `dag_hook`)
/// and the meta-update dedup key, and is **no longer sent on the wire** — the
/// dashboard derives a DAG's label from its node kinds (see `DagView`). The
/// `NodeKind::Dag` container carries it in its payload.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum Template {
/// Rebuild one agent's container (prebuild → stop → profile-swap →
/// reconcile).
Rebuild,
/// Bump meta flake locks; grows a rebuild subgraph per affected
/// agent into the same DAG on completion.
MetaUpdate,
/// First-deploy spawn (approval-driven).
Spawn,
/// Mechanical stop + converge to `wanted = Up` (a restart).
Restart,
/// Signal → drain → mechanical stop → converge to `wanted = Up` — a
/// graceful restart as one atomic DAG.
GracefulRestart,
/// Perm-file commit followed by the rebuild subgraph.
PermChange,
/// Quiesce the harness, drain, then stop (`wanted = Offline`).
GracefulStop,
/// Converge to `wanted = Up`.
Start,
/// Converge to `wanted = Offline`.
Stop,
/// Boot-time config sweep as one DAG.
Boot,
}
impl Template {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Template::Rebuild => "rebuild",
Template::MetaUpdate => "meta_update",
Template::Spawn => "spawn",
Template::Restart => "restart",
Template::GracefulRestart => "graceful_restart",
Template::PermChange => "perm_change",
Template::GracefulStop => "graceful_stop",
Template::Start => "start",
Template::Stop => "stop",
Template::Boot => "boot",
}
}
}
/// When a dependency edge is considered satisfied. /// When a dependency edge is considered satisfied.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]

41
hive-c0re/src/job_queue/tests.rs
View file

@ -45,14 +45,11 @@ fn claim_one(q: &JobQueue) -> Claim {
} }
fn state_of(q: &JobQueue, dag_id: u64) -> State { fn state_of(q: &JobQueue, dag_id: u64) -> State {
// A fully-`Done` DAG drops out of the snapshot (its nodes are all
// excluded) — absence is the completion signal, so map it to `Done`.
// Otherwise derive the roll-up from the node set, exactly as every wire
// consumer does.
q.snapshot() q.snapshot()
.iter() .iter()
.find(|d| d.id == dag_id) .find(|d| d.id == dag_id)
.map_or(State::Done, DagView::rollup_state) .expect("dag present")
.state
} }
// ---- submit (dedup removed — every submit is a fresh DAG) ---- // ---- submit (dedup removed — every submit is a fresh DAG) ----
@ -667,7 +664,7 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
q.complete_node(id, reconcile.node_id, Ok(())); q.complete_node(id, reconcile.node_id, Ok(()));
let snap = q.snapshot(); let snap = q.snapshot();
let dag = snap.iter().find(|d| d.id == id).expect("dag"); let dag = snap.iter().find(|d| d.id == id).expect("dag");
assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed"); assert_eq!(dag.state, State::Failed, "roll-up failed");
let by_kind = |k: &str| { let by_kind = |k: &str| {
dag.nodes dag.nodes
.iter() .iter()
@ -679,12 +676,7 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
assert_eq!(by_kind("stop_for_update"), State::Cancelled); assert_eq!(by_kind("stop_for_update"), State::Cancelled);
assert_eq!(by_kind("swap"), State::Cancelled); assert_eq!(by_kind("swap"), State::Cancelled);
assert_eq!(by_kind("post_swap"), State::Cancelled); assert_eq!(by_kind("post_swap"), State::Cancelled);
// The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`, assert_eq!(by_kind("reconcile"), State::Done);
// and `Done` nodes are excluded from the wire, so it's absent here.
assert!(
dag.nodes.iter().all(|n| n.kind != "reconcile"),
"the completed (Done) reconcile is filtered off the wire"
);
assert_eq!( assert_eq!(
dag.nodes dag.nodes
.iter() .iter()
@ -880,10 +872,10 @@ fn set_step_only_on_running_and_signals_change() {
); );
assert!(q.set_step(id, c.node_id, "next phase")); assert!(q.set_step(id, c.node_id, "next phase"));
assert!(q.set_step_running(id, "via running lookup")); assert!(q.set_step_running(id, "via running lookup"));
// `step` is host-side only now (off the wire); completion clears it
// internally, but there's no wire field to observe — the return-value
// contract above (running-gating + change signalling) is the behaviour.
q.complete_node(id, c.node_id, Ok(())); q.complete_node(id, c.node_id, Ok(()));
let snap = q.snapshot();
let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
assert_eq!(node.step, None, "step cleared on completion");
} }
#[test] #[test]
@ -898,13 +890,9 @@ fn set_build_log_id_links_running_node() {
assert!(q.set_build_log_id(id, c.node_id, 42)); assert!(q.set_build_log_id(id, c.node_id, 42));
assert!(q.set_build_log_id_running(id, 43)); assert!(q.set_build_log_id_running(id, 43));
q.complete_node(id, c.node_id, Ok(())); q.complete_node(id, c.node_id, Ok(()));
// The log id is fetched by node id (the `GET /api/build-log/<id>` lookup), let snap = q.snapshot();
// not carried on the wire — it survives completion in the node runtime. let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
assert_eq!( assert_eq!(node.build_log_id, Some(43), "log id survives completion");
q.build_log_id_of(c.node_id.get()),
Some(43),
"log id survives completion"
);
} }
#[test] #[test]
@ -922,12 +910,9 @@ fn history_evicts_old_terminals_per_template() {
), ),
); );
let c = claim_one(&q); let c = claim_one(&q);
// Fail the single work node so the DAG *lingers*: a fully-`Done` DAG // Completing the single work node rolls the container up terminal (its
// drops off the wire entirely, but a `Failed` one is retained (+ // inline hook fires off the returned summary — no terminal-hook node).
// history-capped) so the operator can still triage it. Completing the q.complete_node(id, c.node_id, Ok(()));
// node rolls the container up terminal (its inline hook fires off the
// returned summary — no terminal-hook node).
q.complete_node(id, c.node_id, Err("boom".to_owned()));
} }
// Fresh terminals are inside the grace window: nothing evicts yet, // Fresh terminals are inside the grace window: nothing evicts yet,
// so a ~1s QueueDag poller can still observe every terminal state // so a ~1s QueueDag poller can still observe every terminal state

11
hive-c0re/src/server.rs
View file

@ -795,14 +795,9 @@ async fn await_dags(coord: &Arc<Coordinator>, ids: &[u64], timeout: std::time::D
let deadline = std::time::Instant::now() + timeout; let deadline = std::time::Instant::now() + timeout;
loop { loop {
let snap = coord.job_queue.snapshot(); let snap = coord.job_queue.snapshot();
// A DAG has settled when it's either gone from the snapshot (fully let pending = ids
// `Done` DAGs drop out) or still present but with every node terminal .iter()
// (a `Failed`/`Cancelled` DAG lingers). It's pending only while it has .any(|id| snap.iter().any(|d| d.id == *id && !d.state.is_terminal()));
// a non-terminal node.
let pending = ids.iter().any(|id| {
snap.iter()
.any(|d| d.id == *id && d.nodes.iter().any(|n| !n.state.is_terminal()))
});
if !pending { if !pending {
return; return;
} }

196
hive-sh4re/src/jobs.rs
View file

@ -6,9 +6,68 @@
//! live in `hive-c0re::job_queue`; these are the serialized views it //! live in `hive-c0re::job_queue`; these are the serialized views it
//! produces. Semantics: `docs/coordinator.md::Job queue`. //! produces. Semantics: `docs/coordinator.md::Job queue`.
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
/// What a DAG *means* — the request-level shape. Wire strings match
/// the pre-DAG queue's `kind` values so dashboards key off the same
/// tags.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Template {
/// Rebuild one agent's container (prebuild → stop → profile-swap →
/// reconcile).
Rebuild,
/// Bump meta flake locks; grows a rebuild subgraph per affected
/// agent into the same DAG on completion.
MetaUpdate,
/// First-deploy spawn (approval-driven).
Spawn,
/// Reserved for a future destroy integration.
Destroy,
/// Mechanical stop + converge to `wanted = Up` (a restart).
Restart,
/// Signal → drain → mechanical stop → converge to `wanted = Up` — a
/// graceful restart as one atomic DAG (drains the harness before the
/// stop, same as `GracefulStop`, but then reconciles back up instead
/// of staying down).
GracefulRestart,
/// Perm-file commit followed by the rebuild subgraph.
PermChange,
/// Quiesce the harness, drain, then stop (`wanted = Offline`).
GracefulStop,
/// Converge to `wanted = Up`.
Start,
/// Converge to `wanted = Offline`.
Stop,
/// Bare converge of observed power state to the persisted intent
/// (boot reconcile).
Reconcile,
/// Boot-time config sweep as one DAG: a hyperhive lock bump that grows
/// a rebuild subgraph per stale agent, plus a `Reconcile` per drifted
/// agent — all in a single DAG (no anchor node, no child DAGs).
Boot,
}
impl Template {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Template::Rebuild => "rebuild",
Template::MetaUpdate => "meta_update",
Template::Spawn => "spawn",
Template::Destroy => "destroy",
Template::Restart => "restart",
Template::GracefulRestart => "graceful_restart",
Template::PermChange => "perm_change",
Template::GracefulStop => "graceful_stop",
Template::Start => "start",
Template::Stop => "stop",
Template::Reconcile => "reconcile",
Template::Boot => "boot",
}
}
}
/// Where the submit request originated — drives the "why" chip on the /// Where the submit request originated — drives the "why" chip on the
/// dashboard. /// dashboard.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@ -81,12 +140,9 @@ pub enum PermPayload {
/// so the widening from the old dag-local `u32` is transparent. /// so the widening from the old dag-local `u32` is transparent.
pub type NodeId = u64; pub type NodeId = u64;
/// One node of a queued DAG, serialized near-raw from the scheduler /// One node of a queued DAG, as serialized. Step labels, build-log
/// graph. Lifecycle (`state` / `started_at` / `finished_at` / `error`) /// links, errors, and timestamps are per-node; the DAG-level `state`
/// comes straight off the `hive_jobq::Node`. The client derives DAG-level /// is a roll-up.
/// roll-ups (label, state, timestamps) from the node set — nothing is
/// rolled up host-side. Build logs are fetched on demand by node id
/// (`GET /api/build-log/<id>`), not carried inline.
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeView { pub struct NodeView {
pub id: NodeId, pub id: NodeId,
@ -100,124 +156,44 @@ pub struct NodeView {
/// `"drain"`, `"write_dropin"`, `"write_perm_file"`, /// `"drain"`, `"write_dropin"`, `"write_perm_file"`,
/// `"approval_deploy"`. /// `"approval_deploy"`.
pub kind: String, pub kind: String,
/// Ids of the nodes this one waits for. May reference an already-`Done` /// Ids of the nodes this one waits for.
/// node that's been filtered out of the wire — the client treats a dep
/// on an absent node as satisfied.
#[serde(default)] #[serde(default)]
pub deps: Vec<NodeId>, pub deps: Vec<NodeId>,
pub state: State, pub state: State,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<DateTime<Utc>>, pub step: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub finished_at: Option<DateTime<Utc>>, pub build_log_id: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished_at: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>, pub error: Option<String>,
/// Approval-queue row id — present only on the `approval_deploy` node.
/// The client links a DAG to its pending approval through this (it is
/// not derivable from the graph, so it rides the node that owns it).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub approval_id: Option<i64>,
/// Meta-flake inputs being bumped — present only on the `meta_lock`
/// node. Display-only payload, not derivable from the graph.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub inputs: Vec<String>,
/// Whether this node has a captured build log fetchable at
/// `GET /api/build-log/<id>`. Only the nix-heavy nodes that stream build
/// output set one; the client gates its log link on this so lock / noop /
/// store-only nodes don't render a link that 404s.
#[serde(default)]
pub has_log: bool,
} }
/// A queued / running / failed DAG — a thin projection of one container /// A queued/running/recent DAG. `kind` = template string, roll-up
/// node plus its (non-`Done`) subtree from the scheduler graph. Only /// `state`; everything per-node appears exactly once, inside `nodes`.
/// non-derivable facts live here: `id`, `source`, `reason`, `created_at`, /// There is no DAG-level `agent` — a DAG can span agents, so agent lives
/// and the node set. The client derives the card label, roll-up state, and /// on each [`NodeView`]; consumers group nodes by `NodeView::agent`.
/// DAG timestamps from `nodes` (per-node `kind` + lifecycle) — nothing is
/// rolled up host-side. There is no DAG-level `agent`: a DAG can span
/// agents, so agent is per-[`NodeView`]; consumers group by `NodeView::agent`.
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DagView { pub struct DagView {
pub id: u64, pub id: u64,
/// Template wire string — same values the old `kind` field used.
pub kind: Template,
/// Roll-up: `failed` if any node failed, else `running` /
/// `queued` / `cancelled` / `done`.
pub state: State,
pub source: Source, pub source: Source,
pub reason: String, pub reason: String,
/// When the DAG was enqueued. pub enqueued_at: i64,
pub created_at: DateTime<Utc>,
/// When the DAG's first node started (min over *all* its nodes) — computed
/// host-side, **not** derived on the client: `Done` nodes are excluded from
/// `nodes` below, so the earliest-started node is usually absent from the
/// wire and the client can't take the min itself. `None` until a node runs.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<DateTime<Utc>>, pub started_at: Option<i64>,
/// When the DAG finished (max `finished_at` over all its nodes), set only
/// once the DAG has settled terminal. Host-computed for the same reason as
/// `started_at`. `None` while the DAG is still live.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub finished_at: Option<DateTime<Utc>>, pub finished_at: Option<i64>,
/// Nodes of this DAG with `Done` ones excluded. A DAG whose nodes are #[serde(default, skip_serializing_if = "Vec::is_empty")]
/// all `Done` is omitted from the snapshot entirely; a `Failed` DAG pub inputs: Vec<String>,
/// lingers until aged out by the history cap. #[serde(default, skip_serializing_if = "Option::is_none")]
pub approval_id: Option<i64>,
pub nodes: Vec<NodeView>, pub nodes: Vec<NodeView>,
} }
impl DagView {
/// Roll-up state derived from the node set — the shared derivation every
/// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's
/// JS render and the host agree: `Failed` if any node failed, else
/// `Running` if any running, else `Queued` if any queued, else
/// `Cancelled` if any cancelled, else `Done`. `Done` nodes are excluded
/// from the wire, so a DAG that is *entirely* done isn't sent at all —
/// its absence from the snapshot is what signals completion.
#[must_use]
pub fn rollup_state(&self) -> State {
let mut any_running = false;
let mut any_queued = false;
let mut any_cancelled = false;
for n in &self.nodes {
match n.state {
State::Failed => return State::Failed,
State::Running => any_running = true,
State::Queued => any_queued = true,
State::Cancelled => any_cancelled = true,
State::Done => {}
}
}
if any_running {
State::Running
} else if any_queued {
State::Queued
} else if any_cancelled {
State::Cancelled
} else {
State::Done
}
}
/// A short human label for the DAG, derived from its node kinds — the
/// shared replacement for the old `Template` wire string now that the kind
/// isn't sent. Priority-ordered so the most distinctive node wins (an
/// approval deploy reads as "deploy" even though it also rebuilds). Purely
/// cosmetic (progress/queue display); consumers that need exactness inspect
/// the node kinds directly.
#[must_use]
pub fn label(&self) -> &'static str {
let has = |k: &str| self.nodes.iter().any(|n| n.kind == k);
if has("approval_deploy") {
"deploy"
} else if has("meta_lock") {
"meta-update"
} else if has("create") || has("provision") {
"spawn"
} else if has("swap") || has("prebuild") || has("post_swap") {
"rebuild"
} else if has("write_perm_file") {
"perm-change"
} else if has("signal") || has("drain") {
"graceful"
} else if has("set_wanted") || has("stop_for_update") {
"power"
} else {
"reconcile"
}
}
}

103
hivectl/src/dag_progress.rs
View file

@ -66,8 +66,8 @@ async fn wait_for_dags_plain(socket: &Path, ids: Vec<u64>) -> Result<()> {
// may still be running — keep watching so the operator // may still be running — keep watching so the operator
// sees whether the agent came back. // sees whether the agent came back.
if d.nodes.iter().all(|n| n.state.is_terminal()) { if d.nodes.iter().all(|n| n.state.is_terminal()) {
if d.rollup_state() == hive_sh4re::jobs::State::Failed { if d.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.label(), dag_agents(d))); failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d)));
} }
} else { } else {
all_terminal = false; all_terminal = false;
@ -134,8 +134,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
}); });
hdr.set_message(format!( hdr.set_message(format!(
"{} {} {} · {}", "{} {} {} · {}",
state_glyph(d.rollup_state()), state_glyph(d.state),
d.label(), d.kind.as_str(),
dag_agents(d), dag_agents(d),
fmt_dur(dag_elapsed(d, now)), fmt_dur(dag_elapsed(d, now)),
)); ));
@ -163,8 +163,8 @@ async fn wait_for_dags_animated(socket: &Path, ids: Vec<u64>) -> Result<()> {
} }
} }
if d.nodes.iter().all(|n| n.state.is_terminal()) { if d.nodes.iter().all(|n| n.state.is_terminal()) {
if d.rollup_state() == hive_sh4re::jobs::State::Failed { if d.state == hive_sh4re::jobs::State::Failed {
failed.push(format!("{} {}", d.label(), dag_agents(d))); failed.push(format!("{} {}", d.kind.as_str(), dag_agents(d)));
} }
} else { } else {
all_terminal = false; all_terminal = false;
@ -224,21 +224,18 @@ fn now_unix() -> i64 {
.unwrap_or(0) .unwrap_or(0)
} }
/// Elapsed seconds for a DAG: `started_at` (falling back to `created_at`) /// Elapsed seconds for a DAG: `started_at` (falling back to `enqueued_at`)
/// through `finished_at` or `now`. The wire carries these as RFC3339 /// through `finished_at` or `now`.
/// `DateTime<Utc>`; compare in unix seconds against `now`.
fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 { fn dag_elapsed(d: &hive_sh4re::jobs::DagView, now: i64) -> i64 {
let start = d let start = d.started_at.unwrap_or(d.enqueued_at);
.started_at (d.finished_at.unwrap_or(now) - start).max(0)
.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 /// Elapsed seconds for a node: `started_at` → `finished_at`/`now`, or 0
/// when it hasn't started. /// when it hasn't started.
fn node_elapsed(n: &hive_sh4re::jobs::NodeView, now: i64) -> i64 { fn node_elapsed(n: &hive_sh4re::jobs::NodeView, now: i64) -> i64 {
match n.started_at { match n.started_at {
Some(start) => (n.finished_at.map_or(now, |t| t.timestamp()) - start.timestamp()).max(0), Some(start) => (n.finished_at.unwrap_or(now) - start).max(0),
None => 0, None => 0,
} }
} }
@ -253,11 +250,16 @@ fn fmt_dur(secs: i64) -> String {
} }
} }
/// One animated node line: kind, an `(after …)` marker for a fan-in node /// One animated node line: kind, live step, an `(after …)` marker for a
/// (>1 dep), its elapsed timer, and a truncated error tail. /// 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 { fn node_line(d: &hive_sh4re::jobs::DagView, n: &hive_sh4re::jobs::NodeView, now: i64) -> String {
use std::fmt::Write as _; use std::fmt::Write as _;
let mut s = n.kind.clone(); 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 { if n.deps.len() > 1 {
let after: Vec<&str> = n let after: Vec<&str> = n
.deps .deps
@ -290,21 +292,25 @@ fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str {
} }
} }
/// One progress line for a DAG: roll-up glyph, derived label, agents, then /// One progress line for a DAG: roll-up glyph, template, agent, then
/// the node chain — the CLI twin of the dashboard's queue card. Both the /// the node chain with the running node's live step label — the CLI
/// roll-up state and the label are derived from the node set (the wire no /// twin of the dashboard's queue card. Used by the plain (non-TTY) path.
/// longer carries them). Used by the plain (non-TTY) path.
fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
use std::fmt::Write as _; use std::fmt::Write as _;
let mut out = format!( let mut out = format!(
"{} {} {:<12}", "{} {} {:<12}",
state_glyph(d.rollup_state()), state_glyph(d.state),
d.label(), d.kind.as_str(),
dag_agents(d) dag_agents(d)
); );
for (i, n) in d.nodes.iter().enumerate() { for (i, n) in d.nodes.iter().enumerate() {
let sep = if i == 0 { " " } else { " → " }; 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.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()) { if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) {
let short: String = err.chars().take(120).collect(); let short: String = err.chars().take(120).collect();
@ -315,65 +321,76 @@ fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use hive_sh4re::jobs::{DagView, NodeView, Source, State}; use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template};
use hive_sh4re::wire_time::from_secs;
use super::render_dag_line; use super::render_dag_line;
fn node(id: u64, agent: &str, kind: &str, state: State) -> NodeView { fn node(id: u64, agent: &str, kind: &str, state: State, step: Option<&str>) -> NodeView {
NodeView { NodeView {
id, id,
agent: agent.to_owned(), agent: agent.to_owned(),
kind: kind.to_owned(), kind: kind.to_owned(),
deps: if id == 0 { vec![] } else { vec![id - 1] }, deps: if id == 0 { vec![] } else { vec![id - 1] },
state, state,
step: step.map(str::to_owned),
build_log_id: None,
started_at: None, started_at: None,
finished_at: None, finished_at: None,
error: None, error: None,
approval_id: None,
inputs: vec![],
has_log: false,
} }
} }
#[test] #[test]
fn render_dag_line_shows_chain_with_derived_label_and_state() { fn render_dag_line_shows_chain_and_running_step() {
// Roll-up state (Running) + label ("rebuild") are derived from the node
// set — the wire no longer carries them. (`Done` nodes are included
// here to exercise glyph rendering; production filters them off.)
let dag = DagView { let dag = DagView {
id: 7, id: 7,
kind: Template::Rebuild,
state: State::Running,
source: Source::Manual, source: Source::Manual,
reason: "manual".to_owned(), reason: "manual".to_owned(),
created_at: from_secs(0), enqueued_at: 0,
started_at: Some(from_secs(1)), started_at: Some(1),
finished_at: None, finished_at: None,
inputs: vec![],
approval_id: None,
nodes: vec![ nodes: vec![
node(0, "alice", "prebuild", State::Done), node(0, "alice", "prebuild", State::Done, None),
node(1, "alice", "stop_for_update", State::Done), node(1, "alice", "stop_for_update", State::Done, None),
node(2, "alice", "swap", State::Running), node(
node(3, "alice", "reconcile", State::Queued), 2,
"alice",
"swap",
State::Running,
Some("nixos-container update"),
),
node(3, "alice", "reconcile", State::Queued, None),
], ],
}; };
let line = render_dag_line(&dag); let line = render_dag_line(&dag);
assert!(line.starts_with("▶ rebuild alice"), "{line}"); assert!(line.starts_with("▶ rebuild alice"), "{line}");
assert!( assert!(
line.contains("✔ prebuild → ✔ stop_for_update → ▶ swap → ⏸ reconcile"), line.contains(
"✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile"
),
"{line}" "{line}"
); );
} }
#[test] #[test]
fn render_dag_line_surfaces_first_node_error() { fn render_dag_line_surfaces_first_node_error() {
let mut failed = node(0, "bob", "prebuild", State::Failed); let mut failed = node(0, "bob", "prebuild", State::Failed, None);
failed.error = Some("nix build exploded".to_owned()); failed.error = Some("nix build exploded".to_owned());
let dag = DagView { let dag = DagView {
id: 8, id: 8,
kind: Template::Rebuild,
state: State::Failed,
source: Source::Manual, source: Source::Manual,
reason: "manual".to_owned(), reason: "manual".to_owned(),
created_at: from_secs(0), enqueued_at: 0,
started_at: Some(from_secs(1)), started_at: Some(1),
finished_at: Some(from_secs(2)), finished_at: Some(2),
inputs: vec![],
approval_id: None,
nodes: vec![failed], nodes: vec![failed],
}; };
let line = render_dag_line(&dag); let line = render_dag_line(&dag);