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): model a DAG as a container node — grouping side-tables become graph walks

Browse source
This commit is contained in:
atlas 2026-07-20 22:33:43 +02:00 • committed by mara
commit a78280feed
4 changed files with 362 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

103
hive-c0re/src/job_queue/exec.rs
View file

@ -100,76 +100,69 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
NodeKind::WritePermFile => run_write_perm_file(coord, claim, &ctx).await, NodeKind::WritePermFile => run_write_perm_file(coord, claim, &ctx).await,
NodeKind::ApprovalDeploy => run_approval_deploy(coord, claim).await, NodeKind::ApprovalDeploy => run_approval_deploy(coord, claim).await,
NodeKind::SetWanted { up } => run_set_wanted(coord, claim, *up), NodeKind::SetWanted { up } => run_set_wanted(coord, claim, *up),
NodeKind::ResolveApproval => run_resolve_approval(coord, claim).await,
NodeKind::EmitRebuilt => Ok(run_emit_rebuilt(coord, claim)),
NodeKind::RevertIntent => run_revert_intent(coord, claim).await,
// Pure grouping container — no work; completing it lets it reach // Pure grouping container — no work; completing it lets it reach
// `Finishing` so its child template nodes start. The DAG's terminal // `Finishing` so its child template nodes start. The DAG's terminal
// hook fires when the container itself rolls up terminal. // hook fires (inline, via `run_terminal_hook`) when the container itself
// rolls up terminal — not as a scheduled node.
NodeKind::Dag { .. } => Ok(NodeOutput::default()), NodeKind::Dag { .. } => Ok(NodeOutput::default()),
} }
} }
/// Terminal hook (approval DAGs — spawn / opaque deploy): resolve the DAG's /// Run a settled DAG's inline terminal hook, dispatched off its rolled-up
/// approval row from its rolled-up outcome. Its own graph node, weak-dep on the /// summary — the container-terminal replacement for the old per-DAG hook node.
/// DAG tails, so it runs once everything has settled (any outcome, incl. a /// Always best-effort: a hook failure is logged inside, never surfaced.
/// cancel before starting — the fallback that resolves a queued-then-cancelled pub(super) async fn run_terminal_hook(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
/// approval whose node never ran). Always succeeds — a hook failure is logged match super::terminal_hook(terminal.template, terminal.approval_id) {
/// inside, not surfaced as a node failure. Some(super::HookKind::ResolveApproval) => {
async fn run_resolve_approval(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> { crate::actions::resolve_approval_dag(coord, terminal).await;
if let Some(terminal) = coord.job_queue.terminal_summary(claim.dag_id) { }
crate::actions::resolve_approval_dag(coord, &terminal).await; Some(super::HookKind::EmitRebuilt) => emit_rebuilt(coord, terminal),
Some(super::HookKind::RevertIntent) => revert_intent(coord, terminal).await,
None => {}
} }
Ok(NodeOutput::default())
} }
/// Terminal hook (rebuild / perm-change DAGs): emit one `Rebuilt` manager event /// Rebuild / perm-change hook: emit one `Rebuilt` manager event per targeted
/// per targeted agent — `ok` on `Done`, `!ok` on `Failed`, none on cancel. /// agent — `ok` on `Done`, `!ok` on `Failed`, none on cancel.
fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim) -> NodeOutput { fn emit_rebuilt(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
if let Some(terminal) = coord.job_queue.terminal_summary(claim.dag_id) { for agent in &terminal.agents {
for agent in &terminal.agents { match terminal.state {
match terminal.state { State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { agent: agent.clone(),
agent: agent.clone(), ok: true,
ok: true, note: None,
note: None, sha: None,
sha: None, tag: None,
tag: None, }),
}), State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { agent: agent.clone(),
agent: agent.clone(), ok: false,
ok: false, note: terminal.error.clone(),
note: terminal.error.clone(), sha: None,
sha: None, tag: None,
tag: None, }),
}), _ => {}
_ => {}
}
} }
} }
NodeOutput::default()
} }
/// Terminal hook (power-op DAGs): on a *cancelled* DAG, revert each targeted /// Power-op hook: on a *cancelled* DAG, revert each targeted agent's `wanted`
/// agent's `wanted` intent to its observed state — the operator's cancel means /// intent to its observed state — the operator's cancel means "don't do it", so
/// "don't do it", so the intent snaps back instead of the flip executing as a /// the intent snaps back instead of the flip executing as a surprise side effect
/// surprise side effect of some later reconcile. Noop on any non-cancelled /// of some later reconcile. Noop on any non-cancelled outcome.
/// outcome. Always succeeds — a revert failure is logged, not surfaced. async fn revert_intent(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
async fn run_revert_intent(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> { if terminal.state != State::Cancelled {
if let Some(terminal) = coord.job_queue.terminal_summary(claim.dag_id) return;
&& terminal.state == State::Cancelled }
{ for agent in &terminal.agents {
for agent in &terminal.agents { let running = crate::lifecycle::is_running(agent).await;
let running = crate::lifecycle::is_running(agent).await; if let Err(e) = coord
if let Err(e) = coord .power
.power .set(agent, crate::power::Wanted::from_running(running))
.set(agent, crate::power::Wanted::from_running(running)) {
{ tracing::warn!(%agent, error = ?e, "agent_power: cancel revert failed");
tracing::warn!(%agent, error = ?e, "agent_power: cancel revert failed");
}
} }
} }
Ok(NodeOutput::default())
} }
/// Write the agent's durable power intent — the DAG-node form of the old /// Write the agent's durable power intent — the DAG-node form of the old

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

@ -97,19 +97,6 @@ pub struct TerminalDag {
pub error: Option<String>, pub error: Option<String>,
} }
/// Per-DAG metadata that isn't a node — the group's display + hook inputs.
#[derive(Debug, Clone)]
struct GroupMeta {
template: Template,
source: Source,
reason: String,
approval_id: Option<i64>,
inputs: Vec<String>,
perm_payload: Option<PermPayload>,
transient: Option<TransientKind>,
created_at: i64,
}
/// Per-node runtime metadata the crate graph doesn't carry (kind + agent live /// Per-node runtime metadata the crate graph doesn't carry (kind + agent live
/// in the node payload; state lives in the node). /// in the node payload; state lives in the node).
#[derive(Debug, Default, Clone)] #[derive(Debug, Default, Clone)]
@ -121,23 +108,31 @@ struct NodeRuntime {
error: Option<String>, error: Option<String>,
} }
/// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
/// Derived on read from the container node — the data has a single home (the
/// node payload); this is not a stored side-table.
struct DagMeta {
template: Template,
source: Source,
reason: String,
transient: Option<TransientKind>,
approval_id: Option<i64>,
inputs: Vec<String>,
perm_payload: Option<PermPayload>,
created_at: i64,
}
/// The mutable queue state behind the mutex: the crate scheduler plus the /// The mutable queue state behind the mutex: the crate scheduler plus the
/// host-side grouping side-tables (`dag_id` ↔ nodes, per-DAG meta, per-node /// per-node runtime metadata the graph can't carry. A **DAG is a single
/// runtime metadata). One shared crate [`Graph`] holds every DAG's nodes. /// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the
/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state
/// is the DAG state, and there are no grouping side-tables: membership + meta
/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] /
/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG.
struct QueueInner { struct QueueInner {
sched: Scheduler<JobPayload, Resource>, sched: Scheduler<JobPayload, Resource>,
next_dag: u64, /// Per-node runtime metadata (build-log id, step, timestamps, error) —
/// Per-DAG metadata, keyed by DAG id. /// mutable after insert, so it can't ride the immutable node payload.
dag_meta: HashMap<u64, GroupMeta>,
/// Work nodes of each DAG, in insert order (drives rollup + the view).
/// Excludes the internal terminal node.
dag_nodes: HashMap<u64, Vec<NodeId>>,
/// The terminal-hook node id of each DAG that has one (approval-resolve /
/// rebuilt-emit / intent-revert). Hook-less DAGs are absent from the map.
terminal_node: HashMap<u64, NodeId>,
/// Reverse lookup: crate node id → its owning DAG id.
node_dag: HashMap<NodeId, u64>,
/// Per-node runtime metadata.
node_rt: HashMap<NodeId, NodeRuntime>, node_rt: HashMap<NodeId, NodeRuntime>,
} }
@ -169,22 +164,32 @@ fn to_crate_when(when: DepWhen) -> JobDepWhen {
} }
} }
/// The terminal-hook node kind a DAG needs, from its template + approval id — /// The inline terminal-hook a settled DAG fires — dispatched off its container's
/// or `None` for a DAG with no terminal side effect (meta-update, boot, bare /// template + approval id when the container rolls up terminal (no hook node).
/// reconcile). Each concern is its own focused node rather than one node that #[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// branches on metadata: an approval DAG resolves its approval, a rebuild / pub enum HookKind {
/// perm-change emits `Rebuilt`, a power-op reverts its `wanted` intent on cancel. /// Approval-driven DAG (spawn / opaque deploy): resolve the approval row.
fn terminal_kind(template: Template, approval_id: Option<i64>) -> Option<NodeKind> { ResolveApproval,
/// Rebuild / perm-change: emit one `Rebuilt` manager event per agent.
EmitRebuilt,
/// Power-op: on a *cancelled* DAG, revert each agent's `wanted` intent.
RevertIntent,
}
/// The terminal hook a DAG needs, from its template + approval id — or `None`
/// for a DAG with no terminal side effect (meta-update, boot, bare reconcile).
#[must_use]
pub fn terminal_hook(template: Template, approval_id: Option<i64>) -> Option<HookKind> {
if approval_id.is_some() { if approval_id.is_some() {
return Some(NodeKind::ResolveApproval); return Some(HookKind::ResolveApproval);
} }
match template { match template {
Template::Rebuild | Template::PermChange => Some(NodeKind::EmitRebuilt), Template::Rebuild | Template::PermChange => Some(HookKind::EmitRebuilt),
Template::Start Template::Start
| Template::Stop | Template::Stop
| Template::GracefulStop | Template::GracefulStop
| Template::Restart | Template::Restart
| Template::GracefulRestart => Some(NodeKind::RevertIntent), | Template::GracefulRestart => Some(HookKind::RevertIntent),
_ => None, _ => None,
} }
} }
@ -212,22 +217,20 @@ fn to_wire_state(state: JobState) -> State {
/// keeps a resource continuous across a subtree (a root owns it, descendants /// keeps a resource continuous across a subtree (a root owns it, descendants
/// borrow it). Independent group roots (multiple `parent = None` nodes) carry no /// borrow it). Independent group roots (multiple `parent = None` nodes) carry no
/// cross-links, so a multi-agent DAG's per-agent subgraphs run concurrently, each /// cross-links, so a multi-agent DAG's per-agent subgraphs run concurrently, each
/// on its own lease. Records per-node bookkeeping (`node_dag`, `node_rt`); the /// on its own lease. Records per-node `node_rt`. Returns the inserted ids
/// caller owns `dag_nodes`. Returns the inserted ids (index-aligned with `nodes`) /// (index-aligned with `nodes`). A node with `parent = None` is re-parented to
/// and the group roots (the `parent = None` nodes). A node's `parent` / dep /// `group_parent` (the DAG container for a template, or the emitting node for a
/// targets must precede it in `nodes` (submit-time `validate` enforces density + /// runtime-appended subgraph); a node's `parent` / dep targets must precede it
/// acyclicity). /// in `nodes` (submit-time `validate` enforces density + acyclicity).
/// ///
/// # Errors /// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group( fn insert_group(
inner: &mut QueueInner, inner: &mut QueueInner,
dag_id: u64,
nodes: &[NodeSpec], nodes: &[NodeSpec],
group_parent: Option<NodeId>, group_parent: Option<NodeId>,
) -> anyhow::Result<(Vec<NodeId>, Vec<NodeId>)> { ) -> anyhow::Result<Vec<NodeId>> {
let mut ids: Vec<NodeId> = Vec::with_capacity(nodes.len()); let mut ids: Vec<NodeId> = Vec::with_capacity(nodes.len());
let mut roots: Vec<NodeId> = Vec::new();
for ns in nodes { for ns in nodes {
let payload = JobPayload { let payload = JobPayload {
kind: ns.kind.clone(), kind: ns.kind.clone(),
@ -248,14 +251,10 @@ fn insert_group(
.sched .sched
.append(payload, deps, parent) .append(payload, deps, parent)
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
if ns.parent.is_none() {
roots.push(id);
}
ids.push(id); ids.push(id);
inner.node_dag.insert(id, dag_id);
inner.node_rt.insert(id, NodeRuntime::default()); inner.node_rt.insert(id, NodeRuntime::default());
} }
Ok((ids, roots)) Ok(ids)
} }
impl JobQueue { impl JobQueue {
@ -269,11 +268,6 @@ impl JobQueue {
Self { Self {
inner: Mutex::new(QueueInner { inner: Mutex::new(QueueInner {
sched: Scheduler::new(Graph::new(), table), sched: Scheduler::new(Graph::new(), table),
next_dag: 0,
dag_meta: HashMap::new(),
dag_nodes: HashMap::new(),
terminal_node: HashMap::new(),
node_dag: HashMap::new(),
node_rt: HashMap::new(), node_rt: HashMap::new(),
}), }),
notify: Notify::new(), notify: Notify::new(),
@ -284,69 +278,43 @@ impl JobQueue {
self.inner.lock().expect("job_queue mutex poisoned") self.inner.lock().expect("job_queue mutex poisoned")
} }
/// Submit a DAG. Validates the spec (cycle rejection), inserts its nodes /// Submit a DAG. Validates the spec, inserts a [`NodeKind::Dag`] **container
/// into the shared graph (chain deps + resource edges), appends the per-DAG /// node** carrying the group's metadata, then inserts the template's nodes as
/// terminal node (weak-depending on the DAG's tail nodes), and returns the /// its subtree (their roots re-parented to the container). Returns the
/// new DAG id. /// container's id as the DAG id — its rolled-up state is the DAG state and it
/// reaching terminal fires the DAG's inline hook.
/// ///
/// # Errors /// # Errors
/// Propagates the spec-validation error (empty / cyclic) or a graph-insert /// Propagates the spec-validation error (empty / cyclic / bad parent) or a
/// error (a spec whose dependencies aren't dependency-topological). /// graph-insert error (dependencies that aren't dependency-topological).
pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> { pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> {
templates::validate(&spec)?; templates::validate(&spec)?;
let mut inner = self.lock(); let mut inner = self.lock();
inner.next_dag += 1; let container = inner
let dag_id = inner.next_dag; .sched
.append(
let (work, roots) = insert_group(&mut inner, dag_id, &spec.nodes, None)?; JobPayload {
kind: NodeKind::Dag {
// Append the DAG's terminal-hook node — but only if it needs one. It template: spec.template,
// weak-depends (`AfterAny`) on every group **root**, each of which the source: spec.source,
// crate rolls up terminal only once its whole subtree — the entire op, reason: spec.reason,
// including any runtime-appended subgraphs (children of nodes inside the transient: spec.transient,
// group) — has settled. So the hook runs exactly when the DAG is done, approval_id: spec.approval_id,
// with no `add_dep` wiring. Hook-less DAGs (meta-update, boot, reconcile) inputs: spec.inputs,
// get none. perm_payload: spec.perm_payload,
if let Some(kind) = terminal_kind(spec.template, spec.approval_id) { created_at: now_unix(),
let term = inner
.sched
.append(
JobPayload {
kind,
agent: String::new(),
}, },
roots agent: String::new(),
.iter() },
.map(|&id| Dep::Node { Vec::new(),
id, None,
when: JobDepWhen::AfterAny, )
}) .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
.collect(), inner.node_rt.insert(container, NodeRuntime::default());
None, insert_group(&mut inner, &spec.nodes, Some(container))?;
)
.map_err(|e| anyhow::anyhow!("job_queue: terminal node insert failed: {e}"))?;
inner.terminal_node.insert(dag_id, term);
inner.node_dag.insert(term, dag_id);
inner.node_rt.insert(term, NodeRuntime::default());
}
inner.dag_nodes.insert(dag_id, work);
inner.dag_meta.insert(
dag_id,
GroupMeta {
template: spec.template,
source: spec.source,
reason: spec.reason,
approval_id: spec.approval_id,
inputs: spec.inputs,
perm_payload: spec.perm_payload,
transient: spec.transient,
created_at: now_unix(),
},
);
drop(inner); drop(inner);
self.notify.notify_one(); self.notify.notify_one();
Ok(dag_id) Ok(container.get())
} }
/// Append a whole *subgraph* into a live DAG at runtime — the single /// Append a whole *subgraph* into a live DAG at runtime — the single
@ -364,16 +332,17 @@ impl JobQueue {
return Vec::new(); return Vec::new();
} }
let mut inner = self.lock(); let mut inner = self.lock();
if !inner.dag_meta.contains_key(&dag_id) { if inner.container(dag_id).is_none() {
return Vec::new(); return Vec::new();
} }
// Insert the subgraph as a group rooted under the emitting node: the // Insert the subgraph as a group rooted under the emitting node: the
// subgraph's own root becomes a child of `dep_on`, its steps children of // subgraph's own root becomes a child of `dep_on`, its steps children of
// that root. No terminal-node wiring — roll-up carries terminality: the // that root. No terminal-node wiring — roll-up carries terminality: the
// emitter stays `Finishing` until this appended subtree settles, and the // emitter stays `Finishing` until this appended subtree settles, and the
// DAG's terminal node deps on the top root, so the hook waits for free. // container node rolls up terminal only once its whole subtree (incl. this
let ids = match insert_group(&mut inner, dag_id, nodes, Some(dep_on)) { // appended work) has settled, so the DAG hook waits for free.
Ok((ids, _roots)) => ids, let ids = match insert_group(&mut inner, nodes, Some(dep_on)) {
Ok(ids) => ids,
Err(e) => { Err(e) => {
tracing::error!( tracing::error!(
dag = dag_id, dag = dag_id,
@ -383,9 +352,6 @@ impl JobQueue {
return Vec::new(); return Vec::new();
} }
}; };
if let Some(list) = inner.dag_nodes.get_mut(&dag_id) {
list.extend(ids.iter().copied());
}
drop(inner); drop(inner);
self.notify.notify_one(); self.notify.notify_one();
ids ids
@ -393,9 +359,9 @@ impl JobQueue {
/// Claim every currently-runnable node, acquiring its resources, and mark it /// Claim every currently-runnable node, acquiring its resources, and mark it
/// `Running`. Delegates readiness + resource acquisition to the crate's /// `Running`. Delegates readiness + resource acquisition to the crate's
/// settle loop; builds a [`Claim`] per started node from its payload + the /// settle loop; builds a [`Claim`] per started node from its payload + its
/// DAG's metadata. The internal terminal node is claimed like any other /// DAG container's metadata. The container node itself is claimed like any
/// (its executor runs the terminal hooks). /// other (its executor is an instant no-op that lets its subtree start).
pub fn claim_ready(&self) -> Vec<Claim> { pub fn claim_ready(&self) -> Vec<Claim> {
let mut inner = self.lock(); let mut inner = self.lock();
let inner = &mut *inner; let inner = &mut *inner;
@ -408,21 +374,21 @@ impl JobQueue {
}; };
let kind = node.payload.kind.clone(); let kind = node.payload.kind.clone();
let agent = node.payload.agent.clone(); let agent = node.payload.agent.clone();
let Some(&dag_id) = inner.node_dag.get(&id) else { let Some(container) = inner.dag_of(id) else {
continue; continue;
}; };
let Some(meta) = inner.dag_meta.get(&dag_id) else { let Some(meta) = inner.dag_meta(container) else {
continue; continue;
}; };
claims.push(Claim { claims.push(Claim {
dag_id, dag_id: container.get(),
node_id: id, node_id: id,
kind, kind,
agent, agent,
template: meta.template, template: meta.template,
approval_id: meta.approval_id, approval_id: meta.approval_id,
inputs: meta.inputs.clone(), inputs: meta.inputs,
perm_payload: meta.perm_payload.clone(), perm_payload: meta.perm_payload,
transient: meta.transient, transient: meta.transient,
}); });
if let Some(rt) = inner.node_rt.get_mut(&id) { if let Some(rt) = inner.node_rt.get_mut(&id) {
@ -434,9 +400,15 @@ impl JobQueue {
/// Mark a claimed node terminal, recording its outcome + (truncated) error. /// Mark a claimed node terminal, recording its outcome + (truncated) error.
/// The crate releases the node's build slot immediately and cascades the /// The crate releases the node's build slot immediately and cascades the
/// `AfterOk` failure cancellation + subtree lease release; terminal hooks /// `AfterOk` failure cancellation + subtree lease release. Returns the DAG's
/// run later as the terminal node (no drain). /// terminal summary **iff** this completion rolled its container terminal —
pub fn complete_node(&self, _dag_id: u64, node_id: NodeId, result: Result<(), String>) { /// the scheduler runs the DAG's inline hook off it.
pub fn complete_node(
&self,
_dag_id: u64,
node_id: NodeId,
result: Result<(), String>,
) -> Option<TerminalDag> {
let mut inner = self.lock(); let mut inner = self.lock();
let now = now_unix(); let now = now_unix();
let (error, outcome) = match result { let (error, outcome) = match result {
@ -450,24 +422,33 @@ impl JobQueue {
rt.error = Some(e); rt.error = Some(e);
} }
} }
let container = inner.dag_of(node_id);
inner.sched.complete(node_id, outcome); inner.sched.complete(node_id, outcome);
inner.trim_history(now - HISTORY_GRACE_SECS); // If this completion rolled the DAG's container up to a terminal state,
// hand its summary back so the scheduler fires the inline hook once.
let terminal = container
.filter(|&c| c != node_id && inner.dag_is_terminal(c))
.and_then(|c| inner.terminal_dag(c));
drop(inner); drop(inner);
self.notify.notify_one(); self.notify.notify_one();
terminal
} }
/// Cancel a DAG that hasn't started yet: every work node is still `Queued`, /// Cancel a DAG that hasn't started yet: the container + every work node is
/// so each is cancelled. No-op (`false`) once any node is running or /// still `Pending`, so each is cancelled. No-op (`false`) once any node is
/// terminal — an in-flight nix build isn't interruptible. The terminal node /// running or terminal — an in-flight nix build isn't interruptible. The
/// (a weak edge) still runs afterwards, so the cancel's terminal hooks /// cancel rolls the container up to `Cancelled`, so its inline hook (approval
/// (approval resolution, power-intent revert) fire. /// resolution, power-intent revert) still fires.
pub fn cancel(&self, dag_id: u64) -> bool { pub fn cancel(&self, dag_id: u64) -> bool {
let mut inner = self.lock(); let mut inner = self.lock();
let inner = &mut *inner; let inner = &mut *inner;
let Some(nodes) = inner.dag_nodes.get(&dag_id) else { let Some(container) = inner.container(dag_id) else {
return false; return false;
}; };
let all_pending = nodes.iter().all(|&id| { let ids: Vec<NodeId> = std::iter::once(container)
.chain(inner.subtree(container))
.collect();
let all_pending = ids.iter().all(|&id| {
inner inner
.sched .sched
.graph() .graph()
@ -477,7 +458,6 @@ impl JobQueue {
if !all_pending { if !all_pending {
return false; return false;
} }
let ids: Vec<NodeId> = nodes.clone();
for id in ids { for id in ids {
inner.sched.cancel_node(id); inner.sched.cancel_node(id);
} }
@ -488,7 +468,7 @@ impl JobQueue {
/// Set the step label on a `Running` node. Returns `true` when it changed. /// Set the step label on a `Running` node. Returns `true` when it changed.
pub fn set_step(&self, dag_id: u64, node_id: NodeId, step: &str) -> bool { pub fn set_step(&self, dag_id: u64, node_id: NodeId, step: &str) -> bool {
let mut inner = self.lock(); let mut inner = self.lock();
if inner.node_dag.get(&node_id) != Some(&dag_id) || !inner.node_running(node_id) { if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) {
return false; return false;
} }
let rt = inner.node_rt.entry(node_id).or_default(); let rt = inner.node_rt.entry(node_id).or_default();
@ -517,7 +497,7 @@ impl JobQueue {
/// Link a `build_logs` row to a specific `Running` node. /// Link a `build_logs` row to a specific `Running` node.
pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool { pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool {
let mut inner = self.lock(); let mut inner = self.lock();
if inner.node_dag.get(&node_id) != Some(&dag_id) || !inner.node_running(node_id) { if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) {
return false; return false;
} }
inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id); inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
@ -535,20 +515,15 @@ impl JobQueue {
true true
} }
/// A DAG's terminal roll-up summary, computed on demand — a terminal hook /// A DAG's terminal roll-up summary, computed on demand from its container.
/// node's executor calls this to run its side effect. `None` if the DAG /// `None` if the DAG id is unknown. Test-only — production reads the summary
/// is unknown (already history-trimmed). /// `complete_node` returns when the container rolls up terminal.
#[cfg(test)]
#[must_use] #[must_use]
pub fn terminal_summary(&self, dag_id: u64) -> Option<TerminalDag> { pub(crate) fn terminal_summary(&self, dag_id: u64) -> Option<TerminalDag> {
let inner = self.lock(); let inner = self.lock();
let meta = inner.dag_meta.get(&dag_id)?; let container = inner.container(dag_id)?;
Some(TerminalDag { inner.terminal_dag(container)
template: meta.template,
agents: inner.dag_agents(dag_id),
approval_id: meta.approval_id,
state: inner.dag_rollup(dag_id),
error: inner.dag_first_error(dag_id),
})
} }
/// The `(dag_id, agent, kind)` triples for every per-agent lease currently /// The `(dag_id, agent, kind)` triples for every per-agent lease currently
@ -566,9 +541,9 @@ impl JobQueue {
let Resource::Agent(agent) = res else { let Resource::Agent(agent) = res else {
return None; return None;
}; };
let dag = *inner.node_dag.get(&holder)?; let container = inner.dag_of(holder)?;
let kind = inner.dag_meta.get(&dag)?.transient?; let kind = inner.dag_meta(container)?.transient?;
Some((dag, agent, kind)) Some((container.get(), agent, kind))
}) })
.collect() .collect()
} }
@ -576,10 +551,16 @@ impl JobQueue {
/// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
#[must_use] #[must_use]
pub fn snapshot(&self) -> Vec<DagView> { pub fn snapshot(&self) -> Vec<DagView> {
self.snapshot_capped(now_unix() - HISTORY_GRACE_SECS)
}
/// Snapshot the visible DAG set (live + newest-per-template terminal, terminal
/// ones after `grace_cutoff` always kept), sorted by container id.
fn snapshot_capped(&self, grace_cutoff: i64) -> Vec<DagView> {
let inner = self.lock(); let inner = self.lock();
let mut ids: Vec<u64> = inner.dag_meta.keys().copied().collect(); let mut ids = inner.visible_dags(grace_cutoff);
ids.sort_unstable(); ids.sort_unstable_by_key(|c| c.get());
ids.into_iter().filter_map(|d| inner.dag_view(d)).collect() ids.into_iter().filter_map(|c| inner.dag_view(c)).collect()
} }
/// Number of live (non-terminal) DAGs — tests + diagnostics. /// Number of live (non-terminal) DAGs — tests + diagnostics.
@ -588,17 +569,18 @@ impl JobQueue {
pub fn live_count(&self) -> usize { pub fn live_count(&self) -> usize {
let inner = self.lock(); let inner = self.lock();
inner inner
.dag_meta .containers()
.keys() .into_iter()
.filter(|&&d| !inner.dag_is_terminal(d)) .filter(|&c| !inner.dag_is_terminal(c))
.count() .count()
} }
/// Test hook: trim history with the grace window disabled. /// Test hook: snapshot with the history grace window disabled, so the
/// per-template cap applies to just-finished terminal DAGs too.
#[cfg(test)] #[cfg(test)]
pub(crate) fn trim_ignoring_grace(&self) { #[must_use]
let mut inner = self.lock(); pub(crate) fn snapshot_no_grace(&self) -> Vec<DagView> {
inner.trim_history(i64::MAX); self.snapshot_capped(i64::MAX)
} }
} }
@ -611,27 +593,88 @@ impl QueueInner {
.is_some_and(|n| n.state == JobState::Running) .is_some_and(|n| n.state == JobState::Running)
} }
/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals
/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search.
fn container(&self, dag_id: u64) -> Option<NodeId> {
self.sched.graph().nodes().find_map(|n| {
(n.parent.is_none()
&& n.id.get() == dag_id
&& matches!(n.payload.kind, NodeKind::Dag { .. }))
.then_some(n.id)
})
}
/// The DAG container a node belongs to — walk its parent chain to the root
/// (`parent == None`), which is the container. Returns `id` itself for a
/// container node.
fn dag_of(&self, id: NodeId) -> Option<NodeId> {
let mut cur = id;
loop {
match self.sched.graph().node(cur)?.parent {
Some(p) => cur = p,
None => return Some(cur),
}
}
}
/// The DAG's work nodes — its `container`'s subtree, excluding the container.
fn subtree(&self, container: NodeId) -> Vec<NodeId> {
self.sched
.graph()
.nodes()
.filter(|n| n.id != container && self.dag_of(n.id) == Some(container))
.map(|n| n.id)
.collect()
}
/// The container's carried domain metadata as an owned read-view. The data
/// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
/// not a stored side-table.
fn dag_meta(&self, container: NodeId) -> Option<DagMeta> {
let NodeKind::Dag {
template,
source,
reason,
transient,
approval_id,
inputs,
perm_payload,
created_at,
} = &self.sched.graph().node(container)?.payload.kind
else {
return None;
};
Some(DagMeta {
template: *template,
source: *source,
reason: reason.clone(),
transient: *transient,
approval_id: *approval_id,
inputs: inputs.clone(),
perm_payload: perm_payload.clone(),
created_at: *created_at,
})
}
/// The DAG's currently-running work node, if any (the opaque approval /// The DAG's currently-running work node, if any (the opaque approval
/// pipeline's single-node DAGs make this exact). /// pipeline's single-node DAGs make this exact).
fn running_node_of(&self, dag_id: u64) -> Option<NodeId> { fn running_node_of(&self, dag_id: u64) -> Option<NodeId> {
self.dag_nodes let container = self.container(dag_id)?;
.get(&dag_id)? self.subtree(container)
.iter() .into_iter()
.copied()
.find(|&id| self.node_running(id)) .find(|&id| self.node_running(id))
} }
/// Roll-up state over a DAG's work nodes: `Failed` if any failed; else /// Roll-up state over a DAG's work nodes: `Failed` if any failed; else
/// `Running` if any running; else `Queued` if any queued; else `Cancelled` /// `Running` if any running; else `Queued` if any queued; else `Cancelled`
/// if any cancelled; else `Done`. /// if any cancelled; else `Done`. (Kept eager over the subtree — a failed
fn dag_rollup(&self, dag_id: u64) -> State { /// child shows `Failed` immediately, before the container finishes rolling
let Some(nodes) = self.dag_nodes.get(&dag_id) else { /// up — matching the pre-container behaviour.)
return State::Done; fn dag_rollup(&self, container: NodeId) -> State {
};
let mut any_running = false; let mut any_running = false;
let mut any_queued = false; let mut any_queued = false;
let mut any_cancelled = false; let mut any_cancelled = false;
for &id in nodes { for id in self.subtree(container) {
match self.sched.graph().node(id).map(|n| n.state) { match self.sched.graph().node(id).map(|n| n.state) {
Some(JobState::Failed) => return State::Failed, Some(JobState::Failed) => return State::Failed,
Some(JobState::Running | JobState::Finishing) => any_running = true, Some(JobState::Running | JobState::Finishing) => any_running = true,
@ -651,28 +694,23 @@ impl QueueInner {
} }
} }
/// True when every work node of the DAG is terminal. /// True when the DAG has settled — its container has rolled up terminal
fn dag_is_terminal(&self, dag_id: u64) -> bool { /// (equivalent to every work node being terminal).
self.dag_nodes.get(&dag_id).is_some_and(|nodes| { fn dag_is_terminal(&self, container: NodeId) -> bool {
nodes.iter().all(|&id| { self.sched
self.sched .graph()
.graph() .node(container)
.node(id) .is_some_and(|n| n.state.is_terminal())
.is_some_and(|n| n.state.is_terminal())
})
})
} }
/// Distinct agents a DAG's work nodes target, in first-seen order. /// Distinct agents a DAG's work nodes target, in first-seen order.
fn dag_agents(&self, dag_id: u64) -> Vec<String> { fn dag_agents(&self, container: NodeId) -> Vec<String> {
let mut seen: Vec<String> = Vec::new(); let mut seen: Vec<String> = Vec::new();
if let Some(nodes) = self.dag_nodes.get(&dag_id) { for id in self.subtree(container) {
for &id in nodes { if let Some(n) = self.sched.graph().node(id) {
if let Some(n) = self.sched.graph().node(id) { let agent = &n.payload.agent;
let agent = &n.payload.agent; if !agent.is_empty() && !seen.iter().any(|s| s == agent) {
if !agent.is_empty() && !seen.iter().any(|s| s == agent) { seen.push(agent.clone());
seen.push(agent.clone());
}
} }
} }
} }
@ -680,9 +718,8 @@ impl QueueInner {
} }
/// First failed work node's stored error, for the roll-up `error` field. /// First failed work node's stored error, for the roll-up `error` field.
fn dag_first_error(&self, dag_id: u64) -> Option<String> { fn dag_first_error(&self, container: NodeId) -> Option<String> {
let nodes = self.dag_nodes.get(&dag_id)?; for id in self.subtree(container) {
for &id in nodes {
if self if self
.sched .sched
.graph() .graph()
@ -696,15 +733,27 @@ impl QueueInner {
None None
} }
/// Rebuild the wire [`DagView`] for a DAG from its metadata + work nodes + /// A DAG's terminal roll-up summary — the input to its inline hook.
/// per-node runtime. The internal terminal node is excluded. fn terminal_dag(&self, container: NodeId) -> Option<TerminalDag> {
fn dag_view(&self, dag_id: u64) -> Option<DagView> { let meta = self.dag_meta(container)?;
let meta = self.dag_meta.get(&dag_id)?; Some(TerminalDag {
let node_ids = self.dag_nodes.get(&dag_id)?; template: meta.template,
agents: self.dag_agents(container),
approval_id: meta.approval_id,
state: self.dag_rollup(container),
error: self.dag_first_error(container),
})
}
/// Rebuild the wire [`DagView`] for a DAG from its container metadata + work
/// nodes + per-node runtime.
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 nodes = Vec::with_capacity(node_ids.len());
let mut started: Vec<i64> = Vec::new(); let mut started: Vec<i64> = Vec::new();
let mut finished: Vec<i64> = Vec::new(); let mut finished: Vec<i64> = Vec::new();
for &id in node_ids { for &id in &node_ids {
let Some(node) = self.sched.graph().node(id) else { let Some(node) = self.sched.graph().node(id) else {
continue; continue;
}; };
@ -736,11 +785,11 @@ impl QueueInner {
error: rt.and_then(|r| r.error.clone()), error: rt.and_then(|r| r.error.clone()),
}); });
} }
let is_terminal = self.dag_is_terminal(dag_id); let is_terminal = self.dag_is_terminal(container);
Some(DagView { Some(DagView {
id: dag_id, id: container.get(),
kind: meta.template, kind: meta.template,
state: self.dag_rollup(dag_id), state: self.dag_rollup(container),
source: meta.source, source: meta.source,
reason: meta.reason.clone(), reason: meta.reason.clone(),
enqueued_at: meta.created_at, enqueued_at: meta.created_at,
@ -757,56 +806,55 @@ impl QueueInner {
}) })
} }
/// Keep only the newest [`MAX_HISTORY_PER_TEMPLATE`] terminal DAGs per /// When a DAG's work node finishes on `finished_at` — the max over its
/// template, evicting older ones' side-tables (their now-terminal crate /// subtree, for the history cap ordering.
/// nodes linger harmlessly in the graph — a bounded-prune primitive is a fn dag_finished_at(&self, container: NodeId) -> i64 {
/// tracked follow-up). Terminal DAGs finished after `grace_cutoff` are self.subtree(container)
/// exempt (never counted). .iter()
fn trim_history(&mut self, grace_cutoff: i64) { .filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at))
let mut terminal: Vec<(u64, Template, i64)> = self .max()
.dag_meta .unwrap_or(0)
.keys() }
.copied()
.filter(|&d| self.dag_is_terminal(d)) /// Every DAG container node id in the graph.
.map(|d| { fn containers(&self) -> Vec<NodeId> {
let finished = self self.sched
.dag_nodes .graph()
.get(&d) .nodes()
.into_iter() .filter(|n| n.parent.is_none() && matches!(n.payload.kind, NodeKind::Dag { .. }))
.flatten() .map(|n| n.id)
.filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at)) .collect()
.max() }
.unwrap_or(0);
(d, self.dag_meta[&d].template, finished) /// The **visible** DAG set for the snapshot: every live (non-terminal) DAG,
}) /// plus the newest [`MAX_HISTORY_PER_TEMPLATE`] terminal DAGs per template
.collect(); /// (terminal DAGs finished after `grace_cutoff` are always kept). Crate nodes
// Newest first, so the cap keeps the most recent per template. /// for evicted DAGs linger in the graph (bounded-prune is a Stage-C
terminal.sort_by(|a, b| b.2.cmp(&a.2).then(b.0.cmp(&a.0))); /// follow-up); this filter is what bounds what the dashboard sees.
let mut counts: HashMap<Template, usize> = HashMap::new(); fn visible_dags(&self, grace_cutoff: i64) -> Vec<NodeId> {
let mut evict: Vec<u64> = Vec::new(); let mut live: Vec<NodeId> = Vec::new();
for (d, template, finished) in terminal { let mut terminal: Vec<(NodeId, Template, i64)> = Vec::new();
if finished > grace_cutoff { for c in self.containers() {
continue; if self.dag_is_terminal(c) {
} if let Some(meta) = self.dag_meta(c) {
let c = counts.entry(template).or_insert(0); terminal.push((c, meta.template, self.dag_finished_at(c)));
*c += 1;
if *c > MAX_HISTORY_PER_TEMPLATE {
evict.push(d);
}
}
for d in evict {
if let Some(nodes) = self.dag_nodes.remove(&d) {
for id in nodes {
self.node_dag.remove(&id);
self.node_rt.remove(&id);
} }
} else {
live.push(c);
} }
if let Some(term) = self.terminal_node.remove(&d) {
self.node_dag.remove(&term);
self.node_rt.remove(&term);
}
self.dag_meta.remove(&d);
} }
// Newest first so the per-template cap keeps the most recent.
terminal.sort_by(|a, b| b.2.cmp(&a.2).then(b.0.get().cmp(&a.0.get())));
let mut counts: HashMap<Template, usize> = HashMap::new();
let mut kept = live;
for (c, template, finished) in terminal {
let n = counts.entry(template).or_insert(0);
*n += 1;
if *n <= MAX_HISTORY_PER_TEMPLATE || finished > grace_cutoff {
kept.push(c);
}
}
kept
} }
} }

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

@ -141,21 +141,6 @@ pub enum NodeKind {
/// `Prebuild`, but that's a no-op there — the agent is down, so prebuild /// `Prebuild`, but that's a no-op there — the agent is down, so prebuild
/// is skipped.) /// is skipped.)
SetWanted { up: bool }, SetWanted { up: bool },
/// Per-DAG terminal hook (approval-driven DAGs — spawn / opaque deploy):
/// resolve the DAG's approval row from the rolled-up outcome. Appended once
/// with a weak (`AfterAny`) edge on the DAG's tails, so it runs exactly when
/// the DAG has settled (any outcome, including a cancel before starting).
/// Build-slot- and lease-exempt; always runs (weak edge ⇒ never cascaded).
ResolveApproval,
/// Per-DAG terminal hook (rebuild / perm-change DAGs): emit one `Rebuilt`
/// manager event per targeted agent — `ok` on `Done`, `!ok` on `Failed`,
/// none on cancel. Appended weak-dep on the tails; slot/lease-exempt.
EmitRebuilt,
/// Per-DAG terminal hook (power-op DAGs): on a *cancelled* DAG, revert each
/// agent's `wanted` intent to its observed state — the operator's cancel
/// means "don't do it". Noop on any non-cancelled outcome. Appended weak-dep
/// on the tails; slot/lease-exempt.
RevertIntent,
/// The **DAG container** node: one per submitted DAG, carrying the group's /// The **DAG container** node: one per submitted DAG, carrying the group's
/// domain metadata. Every template node hangs *under* it (its subtree), so /// domain metadata. Every template node hangs *under* it (its subtree), so
/// the container's `NodeId` **is** the DAG id, its rolled-up state **is** the /// the container's `NodeId` **is** the DAG id, its rolled-up state **is** the
@ -196,9 +181,6 @@ impl NodeKind {
NodeKind::WritePermFile => "write_perm_file", NodeKind::WritePermFile => "write_perm_file",
NodeKind::ApprovalDeploy => "approval_deploy", NodeKind::ApprovalDeploy => "approval_deploy",
NodeKind::SetWanted { .. } => "set_wanted", NodeKind::SetWanted { .. } => "set_wanted",
NodeKind::ResolveApproval => "resolve_approval",
NodeKind::EmitRebuilt => "emit_rebuilt",
NodeKind::RevertIntent => "revert_intent",
NodeKind::Dag { .. } => "dag", NodeKind::Dag { .. } => "dag",
} }
} }

19
hive-c0re/src/job_queue/scheduler.rs
View file

@ -106,9 +106,10 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
.job_queue .job_queue
.append_subgraph(claim.dag_id, subgraph, claim.node_id); .append_subgraph(claim.dag_id, subgraph, claim.node_id);
} }
coord let terminal = coord
.job_queue .job_queue
.complete_node(claim.dag_id, claim.node_id, Ok(())); .complete_node(claim.dag_id, claim.node_id, Ok(()));
fire_terminal_hook(coord, terminal);
} }
Err(e) => { Err(e) => {
let msg = format!("{e:#}"); let msg = format!("{e:#}");
@ -120,9 +121,10 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
error = %msg, error = %msg,
"job_queue: node failed" "job_queue: node failed"
); );
coord let terminal = coord
.job_queue .job_queue
.complete_node(claim.dag_id, claim.node_id, Err(msg)); .complete_node(claim.dag_id, claim.node_id, Err(msg));
fire_terminal_hook(coord, terminal);
} }
} }
// The next loop iteration re-reconciles the transient pills against the // The next loop iteration re-reconciles the transient pills against the
@ -130,6 +132,19 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
} }
/// Fire a settled DAG's inline terminal hook (approval-resolve / rebuilt-emit /
/// intent-revert) off the container-terminal summary `complete_node` returned —
/// spawned so the async hook doesn't block the scheduler loop.
fn fire_terminal_hook(coord: &Arc<Coordinator>, terminal: Option<super::TerminalDag>) {
let Some(terminal) = terminal else {
return;
};
let coord = Arc::clone(coord);
tokio::spawn(async move {
exec::run_terminal_hook(&coord, &terminal).await;
});
}
/// Reconcile the transient-guard set against live lease ownership: drop pills /// Reconcile the transient-guard set against live lease ownership: drop pills
/// whose lease is no longer held, create one for each newly-held `(dag, agent)`. /// whose lease is no longer held, create one for each newly-held `(dag, agent)`.
fn reconcile_transients( fn reconcile_transients(