From c574948d7cb5998a64a7d8e635ae0472e4e5216d Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 23 Jul 2026 14:30:21 +0200 Subject: [PATCH] refactor(#2591): read node lifecycle off the jobq Node; slim DagView build (WIP) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- hive-c0re/src/job_queue/mod.rs | 115 ++++++++++++++----------------- hive-c0re/src/job_queue/model.rs | 60 +++++++++++++++- 2 files changed, 112 insertions(+), 63 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 2e5a987a..ad1a4ff2 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -96,15 +96,14 @@ pub struct TerminalDag { pub error: Option, } -/// 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, build_log_id: Option, - started_at: Option, - finished_at: Option, - error: Option, } /// 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 { 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 { 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 { 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 = Vec::new(); - let mut finished: Vec = 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 = 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) } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index bfc66aa0..15159ea9 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -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")]