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(#2591): read node lifecycle off the jobq Node; slim DagView build (WIP)

Browse source 
hive-c0re side of the raw-graph wire: dag_view now projects the slim
DagView, reading started_at/finished_at/error straight off the
hive_jobq Node (removes the node_rt double-write from #2645), excludes
Done nodes, and rides approval_id/inputs on the owning node. Template
moved into hive-c0re (model.rs) — no longer on the wire. Still WIP:
build-log endpoint + hivectl derive + compile fixes to follow.
This commit is contained in:
atlas 2026-07-23 14:30:21 +02:00 • committed by mara
commit c574948d7c
2 changed files with 112 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -96,15 +96,14 @@ pub struct TerminalDag {
pub error: Option<String>,
}
/// Per-node runtime metadata the crate graph doesn't carry (kind + agent live
/// in the node payload; state lives in the node).
/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle
/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::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)]
struct NodeRuntime {
step: Option<String>,
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`]).
@ -386,9 +385,8 @@ impl JobQueue {
inputs: meta.inputs,
transient: meta.transient,
});
if let Some(rt) = inner.node_rt.get_mut(&id) {
rt.started_at = Some(now);
}
// `started_at` is stamped on the graph `Node` by the scheduler's
// transition to `Running` — no host-side copy needed.
}
claims
}
@ -405,23 +403,15 @@ impl JobQueue {
result: Result<(), String>,
) -> Option<TerminalDag> {
let mut inner = self.lock();
let now = now_unix();
let (error, outcome) = match result {
Ok(()) => (None, Outcome::Done),
Err(e) => {
// The reason rides the crate `Outcome::Failed` (stamped onto the
// 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))
}
// The failure reason + `finished_at` are stamped onto the graph `Node`
// by the scheduler (the reason rides `Outcome::Failed`); no host-side
// copy. We only clear the live sub-step label here.
let outcome = match result {
Ok(()) => Outcome::Done,
Err(e) => Outcome::Failed(truncate_error(&e)),
};
if let Some(rt) = inner.node_rt.get_mut(&node_id) {
rt.finished_at = Some(now);
rt.step = None;
if let Some(e) = error {
rt.error = Some(e);
}
}
let container = inner.dag_of(node_id);
inner.sched.complete(node_id, outcome);
@ -721,15 +711,13 @@ impl QueueInner {
seen
}
/// First failed work node's stored error, for the roll-up `error` field.
/// First failed work node's error (read off the graph `Node`), for the
/// terminal roll-up summary the inline hook consumes.
fn dag_first_error(&self, container: NodeId) -> Option<String> {
for id in self.subtree(container) {
if self
.sched
.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())
if let Some(n) = self.sched.graph().node(id)
&& n.state == JobState::Failed
&& let Some(e) = n.error.clone()
{
return Some(e);
}
@ -749,19 +737,24 @@ impl QueueInner {
})
}
/// Rebuild the wire [`DagView`] for a DAG from its container metadata + work
/// nodes + per-node runtime.
/// Project a DAG into its wire [`DagView`]: a near-raw view of the
/// container's work nodes, with `Done` nodes excluded. Lifecycle
/// (`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> {
let meta = self.dag_meta(container)?;
let node_ids = self.subtree(container);
let mut nodes = Vec::with_capacity(node_ids.len());
let mut started: Vec<i64> = Vec::new();
let mut finished: Vec<i64> = Vec::new();
for &id in &node_ids {
let mut nodes = Vec::new();
for id in self.subtree(container) {
let Some(node) = self.sched.graph().node(id) else {
continue;
};
let rt = self.node_rt.get(&id);
if node.state == JobState::Done {
continue;
}
let deps: Vec<u64> = node
.deps
.iter()
@ -770,51 +763,49 @@ impl QueueInner {
Dep::Resource { .. } => None,
})
.collect();
if let Some(s) = rt.and_then(|r| r.started_at) {
started.push(s);
}
if let Some(fin) = rt.and_then(|r| r.finished_at) {
finished.push(fin);
}
// Non-derivable per-node payload rides the node that owns it.
let approval_id = matches!(node.payload, NodeKind::ApprovalDeploy { .. })
.then_some(meta.approval_id)
.flatten();
let inputs = if matches!(node.payload, NodeKind::MetaLock { .. }) {
meta.inputs.clone()
} else {
Vec::new()
};
nodes.push(NodeView {
id: id.get(),
agent: node.payload.agent().to_owned(),
kind: node.payload.as_str().to_owned(),
deps,
state: to_wire_state(node.state),
step: rt.and_then(|r| r.step.clone()),
build_log_id: rt.and_then(|r| r.build_log_id),
started_at: rt.and_then(|r| r.started_at),
finished_at: rt.and_then(|r| r.finished_at),
error: rt.and_then(|r| r.error.clone()),
started_at: node.started_at,
finished_at: node.finished_at,
error: node.error.clone(),
approval_id,
inputs,
});
}
let is_terminal = self.dag_is_terminal(container);
if nodes.is_empty() {
return None;
}
Some(DagView {
id: container.get(),
kind: meta.template,
state: self.dag_rollup(container),
source: meta.source,
reason: meta.reason.clone(),
enqueued_at: meta.created_at,
started_at: started.into_iter().min(),
finished_at: if is_terminal {
finished.into_iter().max()
} else {
None
},
inputs: meta.inputs.clone(),
approval_id: meta.approval_id,
created_at: hive_sh4re::wire_time::from_secs(meta.created_at),
nodes,
})
}
/// When a DAG's work node finishes on `finished_at` — the max over its
/// subtree, for the history cap ordering.
/// subtree (read off the graph `Node`, as unix seconds), for the history
/// cap ordering.
fn dag_finished_at(&self, container: NodeId) -> i64 {
self.subtree(container)
.iter()
.filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at))
.filter_map(|id| self.sched.graph().node(*id))
.filter_map(|n| n.finished_at)
.map(|t| t.timestamp())
.max()
.unwrap_or(0)
}

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

@ -11,11 +11,69 @@
//! DAG can span agents). See `docs/coordinator.md::Job queue` for the
//! full design.
pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State, Template};
pub use hive_sh4re::jobs::{DagView, NodeId, PermPayload, Source, State};
use serde::Serialize;
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,
/// 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.
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.
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",
}
}
}
/// When a dependency edge is considered satisfied.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]