diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 52eba21b..0c7d9d08 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -100,76 +100,69 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::WritePermFile => run_write_perm_file(coord, claim, &ctx).await, NodeKind::ApprovalDeploy => run_approval_deploy(coord, claim).await, 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 // `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()), } } -/// Terminal hook (approval DAGs — spawn / opaque deploy): resolve the DAG's -/// approval row from its rolled-up outcome. Its own graph node, weak-dep on the -/// DAG tails, so it runs once everything has settled (any outcome, incl. a -/// cancel before starting — the fallback that resolves a queued-then-cancelled -/// approval whose node never ran). Always succeeds — a hook failure is logged -/// inside, not surfaced as a node failure. -async fn run_resolve_approval(coord: &Arc, claim: &Claim) -> Result { - if let Some(terminal) = coord.job_queue.terminal_summary(claim.dag_id) { - crate::actions::resolve_approval_dag(coord, &terminal).await; +/// Run a settled DAG's inline terminal hook, dispatched off its rolled-up +/// summary — the container-terminal replacement for the old per-DAG hook node. +/// Always best-effort: a hook failure is logged inside, never surfaced. +pub(super) async fn run_terminal_hook(coord: &Arc, terminal: &super::TerminalDag) { + match super::terminal_hook(terminal.template, terminal.approval_id) { + Some(super::HookKind::ResolveApproval) => { + 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 -/// per targeted agent — `ok` on `Done`, `!ok` on `Failed`, none on cancel. -fn run_emit_rebuilt(coord: &Arc, claim: &Claim) -> NodeOutput { - if let Some(terminal) = coord.job_queue.terminal_summary(claim.dag_id) { - for agent in &terminal.agents { - match terminal.state { - State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: agent.clone(), - ok: true, - note: None, - sha: None, - tag: None, - }), - State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: agent.clone(), - ok: false, - note: terminal.error.clone(), - sha: None, - tag: None, - }), - _ => {} - } +/// Rebuild / perm-change hook: emit one `Rebuilt` manager event per targeted +/// agent — `ok` on `Done`, `!ok` on `Failed`, none on cancel. +fn emit_rebuilt(coord: &Arc, terminal: &super::TerminalDag) { + for agent in &terminal.agents { + match terminal.state { + State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent: agent.clone(), + ok: true, + note: None, + sha: None, + tag: None, + }), + State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent: agent.clone(), + ok: false, + note: terminal.error.clone(), + sha: None, + tag: None, + }), + _ => {} } } - NodeOutput::default() } -/// Terminal hook (power-op DAGs): on a *cancelled* DAG, revert each targeted -/// agent's `wanted` intent to its observed state — the operator's cancel means -/// "don't do it", so the intent snaps back instead of the flip executing as a -/// surprise side effect of some later reconcile. Noop on any non-cancelled -/// outcome. Always succeeds — a revert failure is logged, not surfaced. -async fn run_revert_intent(coord: &Arc, claim: &Claim) -> Result { - if let Some(terminal) = coord.job_queue.terminal_summary(claim.dag_id) - && terminal.state == State::Cancelled - { - for agent in &terminal.agents { - let running = crate::lifecycle::is_running(agent).await; - if let Err(e) = coord - .power - .set(agent, crate::power::Wanted::from_running(running)) - { - tracing::warn!(%agent, error = ?e, "agent_power: cancel revert failed"); - } +/// Power-op hook: on a *cancelled* DAG, revert each targeted agent's `wanted` +/// intent to its observed state — the operator's cancel means "don't do it", so +/// the intent snaps back instead of the flip executing as a surprise side effect +/// of some later reconcile. Noop on any non-cancelled outcome. +async fn revert_intent(coord: &Arc, terminal: &super::TerminalDag) { + if terminal.state != State::Cancelled { + return; + } + for agent in &terminal.agents { + let running = crate::lifecycle::is_running(agent).await; + if let Err(e) = coord + .power + .set(agent, crate::power::Wanted::from_running(running)) + { + 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 diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 9a23eb80..1d15acdd 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -97,19 +97,6 @@ pub struct TerminalDag { pub error: Option, } -/// 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, - inputs: Vec, - perm_payload: Option, - transient: Option, - created_at: i64, -} - /// Per-node runtime metadata the crate graph doesn't carry (kind + agent live /// in the node payload; state lives in the node). #[derive(Debug, Default, Clone)] @@ -121,23 +108,31 @@ struct NodeRuntime { error: Option, } +/// 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, + approval_id: Option, + inputs: Vec, + perm_payload: Option, + created_at: i64, +} + /// 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 -/// runtime metadata). One shared crate [`Graph`] holds every DAG's nodes. +/// per-node runtime metadata the graph can't carry. A **DAG is a single +/// 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 { sched: Scheduler, - next_dag: u64, - /// Per-DAG metadata, keyed by DAG id. - dag_meta: HashMap, - /// Work nodes of each DAG, in insert order (drives rollup + the view). - /// Excludes the internal terminal node. - dag_nodes: HashMap>, - /// 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, - /// Reverse lookup: crate node id → its owning DAG id. - node_dag: HashMap, - /// Per-node runtime metadata. + /// Per-node runtime metadata (build-log id, step, timestamps, error) — + /// mutable after insert, so it can't ride the immutable node payload. node_rt: HashMap, } @@ -169,22 +164,32 @@ fn to_crate_when(when: DepWhen) -> JobDepWhen { } } -/// The terminal-hook node kind a DAG needs, from its template + approval id — -/// or `None` for a DAG with no terminal side effect (meta-update, boot, bare -/// reconcile). Each concern is its own focused node rather than one node that -/// branches on metadata: an approval DAG resolves its approval, a rebuild / -/// perm-change emits `Rebuilt`, a power-op reverts its `wanted` intent on cancel. -fn terminal_kind(template: Template, approval_id: Option) -> Option { +/// The inline terminal-hook a settled DAG fires — dispatched off its container's +/// template + approval id when the container rolls up terminal (no hook node). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HookKind { + /// Approval-driven DAG (spawn / opaque deploy): resolve the approval row. + 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) -> Option { if approval_id.is_some() { - return Some(NodeKind::ResolveApproval); + return Some(HookKind::ResolveApproval); } match template { - Template::Rebuild | Template::PermChange => Some(NodeKind::EmitRebuilt), + Template::Rebuild | Template::PermChange => Some(HookKind::EmitRebuilt), Template::Start | Template::Stop | Template::GracefulStop | Template::Restart - | Template::GracefulRestart => Some(NodeKind::RevertIntent), + | Template::GracefulRestart => Some(HookKind::RevertIntent), _ => None, } } @@ -212,22 +217,20 @@ fn to_wire_state(state: JobState) -> State { /// keeps a resource continuous across a subtree (a root owns it, descendants /// 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 -/// on its own lease. Records per-node bookkeeping (`node_dag`, `node_rt`); the -/// caller owns `dag_nodes`. Returns the inserted ids (index-aligned with `nodes`) -/// and the group roots (the `parent = None` nodes). A node's `parent` / dep -/// targets must precede it in `nodes` (submit-time `validate` enforces density + -/// acyclicity). +/// on its own lease. Records per-node `node_rt`. Returns the inserted ids +/// (index-aligned with `nodes`). A node with `parent = None` is re-parented to +/// `group_parent` (the DAG container for a template, or the emitting node for a +/// runtime-appended subgraph); a node's `parent` / dep targets must precede it +/// in `nodes` (submit-time `validate` enforces density + acyclicity). /// /// # Errors /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). fn insert_group( inner: &mut QueueInner, - dag_id: u64, nodes: &[NodeSpec], group_parent: Option, -) -> anyhow::Result<(Vec, Vec)> { +) -> anyhow::Result> { let mut ids: Vec = Vec::with_capacity(nodes.len()); - let mut roots: Vec = Vec::new(); for ns in nodes { let payload = JobPayload { kind: ns.kind.clone(), @@ -248,14 +251,10 @@ fn insert_group( .sched .append(payload, deps, parent) .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; - if ns.parent.is_none() { - roots.push(id); - } ids.push(id); - inner.node_dag.insert(id, dag_id); inner.node_rt.insert(id, NodeRuntime::default()); } - Ok((ids, roots)) + Ok(ids) } impl JobQueue { @@ -269,11 +268,6 @@ impl JobQueue { Self { inner: Mutex::new(QueueInner { 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(), }), notify: Notify::new(), @@ -284,69 +278,43 @@ impl JobQueue { self.inner.lock().expect("job_queue mutex poisoned") } - /// Submit a DAG. Validates the spec (cycle rejection), inserts its nodes - /// into the shared graph (chain deps + resource edges), appends the per-DAG - /// terminal node (weak-depending on the DAG's tail nodes), and returns the - /// new DAG id. + /// Submit a DAG. Validates the spec, inserts a [`NodeKind::Dag`] **container + /// node** carrying the group's metadata, then inserts the template's nodes as + /// its subtree (their roots re-parented to the container). Returns the + /// 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 - /// Propagates the spec-validation error (empty / cyclic) or a graph-insert - /// error (a spec whose dependencies aren't dependency-topological). + /// Propagates the spec-validation error (empty / cyclic / bad parent) or a + /// graph-insert error (dependencies that aren't dependency-topological). pub fn submit(&self, spec: DagSpec) -> anyhow::Result { templates::validate(&spec)?; let mut inner = self.lock(); - inner.next_dag += 1; - let dag_id = inner.next_dag; - - let (work, roots) = insert_group(&mut inner, dag_id, &spec.nodes, None)?; - - // Append the DAG's terminal-hook node — but only if it needs one. It - // weak-depends (`AfterAny`) on every group **root**, each of which the - // crate rolls up terminal only once its whole subtree — the entire op, - // including any runtime-appended subgraphs (children of nodes inside the - // group) — has settled. So the hook runs exactly when the DAG is done, - // with no `add_dep` wiring. Hook-less DAGs (meta-update, boot, reconcile) - // get none. - if let Some(kind) = terminal_kind(spec.template, spec.approval_id) { - let term = inner - .sched - .append( - JobPayload { - kind, - agent: String::new(), + let container = inner + .sched + .append( + JobPayload { + kind: NodeKind::Dag { + template: spec.template, + source: spec.source, + reason: spec.reason, + transient: spec.transient, + approval_id: spec.approval_id, + inputs: spec.inputs, + perm_payload: spec.perm_payload, + created_at: now_unix(), }, - roots - .iter() - .map(|&id| Dep::Node { - id, - when: JobDepWhen::AfterAny, - }) - .collect(), - None, - ) - .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(), - }, - ); + agent: String::new(), + }, + Vec::new(), + None, + ) + .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; + inner.node_rt.insert(container, NodeRuntime::default()); + insert_group(&mut inner, &spec.nodes, Some(container))?; drop(inner); self.notify.notify_one(); - Ok(dag_id) + Ok(container.get()) } /// Append a whole *subgraph* into a live DAG at runtime — the single @@ -364,16 +332,17 @@ impl JobQueue { return Vec::new(); } let mut inner = self.lock(); - if !inner.dag_meta.contains_key(&dag_id) { + if inner.container(dag_id).is_none() { return Vec::new(); } // 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 // that root. No terminal-node wiring — roll-up carries terminality: 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. - let ids = match insert_group(&mut inner, dag_id, nodes, Some(dep_on)) { - Ok((ids, _roots)) => ids, + // container node rolls up terminal only once its whole subtree (incl. this + // appended work) has settled, so the DAG hook waits for free. + let ids = match insert_group(&mut inner, nodes, Some(dep_on)) { + Ok(ids) => ids, Err(e) => { tracing::error!( dag = dag_id, @@ -383,9 +352,6 @@ impl JobQueue { return Vec::new(); } }; - if let Some(list) = inner.dag_nodes.get_mut(&dag_id) { - list.extend(ids.iter().copied()); - } drop(inner); self.notify.notify_one(); ids @@ -393,9 +359,9 @@ impl JobQueue { /// Claim every currently-runnable node, acquiring its resources, and mark it /// `Running`. Delegates readiness + resource acquisition to the crate's - /// settle loop; builds a [`Claim`] per started node from its payload + the - /// DAG's metadata. The internal terminal node is claimed like any other - /// (its executor runs the terminal hooks). + /// settle loop; builds a [`Claim`] per started node from its payload + its + /// DAG container's metadata. The container node itself is claimed like any + /// other (its executor is an instant no-op that lets its subtree start). pub fn claim_ready(&self) -> Vec { let mut inner = self.lock(); let inner = &mut *inner; @@ -408,21 +374,21 @@ impl JobQueue { }; let kind = node.payload.kind.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; }; - let Some(meta) = inner.dag_meta.get(&dag_id) else { + let Some(meta) = inner.dag_meta(container) else { continue; }; claims.push(Claim { - dag_id, + dag_id: container.get(), node_id: id, kind, agent, template: meta.template, approval_id: meta.approval_id, - inputs: meta.inputs.clone(), - perm_payload: meta.perm_payload.clone(), + inputs: meta.inputs, + perm_payload: meta.perm_payload, transient: meta.transient, }); 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. /// The crate releases the node's build slot immediately and cascades the - /// `AfterOk` failure cancellation + subtree lease release; terminal hooks - /// run later as the terminal node (no drain). - pub fn complete_node(&self, _dag_id: u64, node_id: NodeId, result: Result<(), String>) { + /// `AfterOk` failure cancellation + subtree lease release. Returns the DAG's + /// terminal summary **iff** this completion rolled its container terminal — + /// 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 { let mut inner = self.lock(); let now = now_unix(); let (error, outcome) = match result { @@ -450,24 +422,33 @@ impl JobQueue { rt.error = Some(e); } } + let container = inner.dag_of(node_id); 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); self.notify.notify_one(); + terminal } - /// Cancel a DAG that hasn't started yet: every work node is still `Queued`, - /// so each is cancelled. No-op (`false`) once any node is running or - /// terminal — an in-flight nix build isn't interruptible. The terminal node - /// (a weak edge) still runs afterwards, so the cancel's terminal hooks - /// (approval resolution, power-intent revert) fire. + /// Cancel a DAG that hasn't started yet: the container + every work node is + /// still `Pending`, so each is cancelled. No-op (`false`) once any node is + /// running or terminal — an in-flight nix build isn't interruptible. The + /// cancel rolls the container up to `Cancelled`, so its inline hook (approval + /// resolution, power-intent revert) still fires. pub fn cancel(&self, dag_id: u64) -> bool { let mut inner = self.lock(); let inner = &mut *inner; - let Some(nodes) = inner.dag_nodes.get(&dag_id) else { + let Some(container) = inner.container(dag_id) else { return false; }; - let all_pending = nodes.iter().all(|&id| { + let ids: Vec = std::iter::once(container) + .chain(inner.subtree(container)) + .collect(); + let all_pending = ids.iter().all(|&id| { inner .sched .graph() @@ -477,7 +458,6 @@ impl JobQueue { if !all_pending { return false; } - let ids: Vec = nodes.clone(); for id in ids { inner.sched.cancel_node(id); } @@ -488,7 +468,7 @@ impl JobQueue { /// 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 { 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; } 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. pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool { 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; } inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id); @@ -535,20 +515,15 @@ impl JobQueue { true } - /// A DAG's terminal roll-up summary, computed on demand — a terminal hook - /// node's executor calls this to run its side effect. `None` if the DAG - /// is unknown (already history-trimmed). + /// 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 + /// `complete_node` returns when the container rolls up terminal. + #[cfg(test)] #[must_use] - pub fn terminal_summary(&self, dag_id: u64) -> Option { + pub(crate) fn terminal_summary(&self, dag_id: u64) -> Option { let inner = self.lock(); - let meta = inner.dag_meta.get(&dag_id)?; - Some(TerminalDag { - 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), - }) + let container = inner.container(dag_id)?; + inner.terminal_dag(container) } /// The `(dag_id, agent, kind)` triples for every per-agent lease currently @@ -566,9 +541,9 @@ impl JobQueue { let Resource::Agent(agent) = res else { return None; }; - let dag = *inner.node_dag.get(&holder)?; - let kind = inner.dag_meta.get(&dag)?.transient?; - Some((dag, agent, kind)) + let container = inner.dag_of(holder)?; + let kind = inner.dag_meta(container)?.transient?; + Some((container.get(), agent, kind)) }) .collect() } @@ -576,10 +551,16 @@ impl JobQueue { /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`. #[must_use] pub fn snapshot(&self) -> Vec { + 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 { let inner = self.lock(); - let mut ids: Vec = inner.dag_meta.keys().copied().collect(); - ids.sort_unstable(); - ids.into_iter().filter_map(|d| inner.dag_view(d)).collect() + let mut ids = inner.visible_dags(grace_cutoff); + ids.sort_unstable_by_key(|c| c.get()); + ids.into_iter().filter_map(|c| inner.dag_view(c)).collect() } /// Number of live (non-terminal) DAGs — tests + diagnostics. @@ -588,17 +569,18 @@ impl JobQueue { pub fn live_count(&self) -> usize { let inner = self.lock(); inner - .dag_meta - .keys() - .filter(|&&d| !inner.dag_is_terminal(d)) + .containers() + .into_iter() + .filter(|&c| !inner.dag_is_terminal(c)) .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)] - pub(crate) fn trim_ignoring_grace(&self) { - let mut inner = self.lock(); - inner.trim_history(i64::MAX); + #[must_use] + pub(crate) fn snapshot_no_grace(&self) -> Vec { + self.snapshot_capped(i64::MAX) } } @@ -611,27 +593,88 @@ impl QueueInner { .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 { + 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 { + 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 { + 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 { + 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 /// pipeline's single-node DAGs make this exact). fn running_node_of(&self, dag_id: u64) -> Option { - self.dag_nodes - .get(&dag_id)? - .iter() - .copied() + let container = self.container(dag_id)?; + self.subtree(container) + .into_iter() .find(|&id| self.node_running(id)) } /// 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` - /// if any cancelled; else `Done`. - fn dag_rollup(&self, dag_id: u64) -> State { - let Some(nodes) = self.dag_nodes.get(&dag_id) else { - return State::Done; - }; + /// if any cancelled; else `Done`. (Kept eager over the subtree — a failed + /// child shows `Failed` immediately, before the container finishes rolling + /// up — matching the pre-container behaviour.) + fn dag_rollup(&self, container: NodeId) -> State { let mut any_running = false; let mut any_queued = 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) { Some(JobState::Failed) => return State::Failed, Some(JobState::Running | JobState::Finishing) => any_running = true, @@ -651,28 +694,23 @@ impl QueueInner { } } - /// True when every work node of the DAG is terminal. - fn dag_is_terminal(&self, dag_id: u64) -> bool { - self.dag_nodes.get(&dag_id).is_some_and(|nodes| { - nodes.iter().all(|&id| { - self.sched - .graph() - .node(id) - .is_some_and(|n| n.state.is_terminal()) - }) - }) + /// True when the DAG has settled — its container has rolled up terminal + /// (equivalent to every work node being terminal). + fn dag_is_terminal(&self, container: NodeId) -> bool { + self.sched + .graph() + .node(container) + .is_some_and(|n| n.state.is_terminal()) } /// Distinct agents a DAG's work nodes target, in first-seen order. - fn dag_agents(&self, dag_id: u64) -> Vec { + fn dag_agents(&self, container: NodeId) -> Vec { let mut seen: Vec = Vec::new(); - if let Some(nodes) = self.dag_nodes.get(&dag_id) { - for &id in nodes { - if let Some(n) = self.sched.graph().node(id) { - let agent = &n.payload.agent; - if !agent.is_empty() && !seen.iter().any(|s| s == agent) { - seen.push(agent.clone()); - } + for id in self.subtree(container) { + if let Some(n) = self.sched.graph().node(id) { + let agent = &n.payload.agent; + if !agent.is_empty() && !seen.iter().any(|s| s == agent) { + seen.push(agent.clone()); } } } @@ -680,9 +718,8 @@ impl QueueInner { } /// First failed work node's stored error, for the roll-up `error` field. - fn dag_first_error(&self, dag_id: u64) -> Option { - let nodes = self.dag_nodes.get(&dag_id)?; - for &id in nodes { + fn dag_first_error(&self, container: NodeId) -> Option { + for id in self.subtree(container) { if self .sched .graph() @@ -696,15 +733,27 @@ impl QueueInner { None } - /// Rebuild the wire [`DagView`] for a DAG from its metadata + work nodes + - /// per-node runtime. The internal terminal node is excluded. - fn dag_view(&self, dag_id: u64) -> Option { - let meta = self.dag_meta.get(&dag_id)?; - let node_ids = self.dag_nodes.get(&dag_id)?; + /// A DAG's terminal roll-up summary — the input to its inline hook. + fn terminal_dag(&self, container: NodeId) -> Option { + let meta = self.dag_meta(container)?; + Some(TerminalDag { + 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 { + 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 { + for &id in &node_ids { let Some(node) = self.sched.graph().node(id) else { continue; }; @@ -736,11 +785,11 @@ impl QueueInner { 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 { - id: dag_id, + id: container.get(), kind: meta.template, - state: self.dag_rollup(dag_id), + state: self.dag_rollup(container), source: meta.source, reason: meta.reason.clone(), enqueued_at: meta.created_at, @@ -757,56 +806,55 @@ impl QueueInner { }) } - /// Keep only the newest [`MAX_HISTORY_PER_TEMPLATE`] terminal DAGs per - /// template, evicting older ones' side-tables (their now-terminal crate - /// nodes linger harmlessly in the graph — a bounded-prune primitive is a - /// tracked follow-up). Terminal DAGs finished after `grace_cutoff` are - /// exempt (never counted). - fn trim_history(&mut self, grace_cutoff: i64) { - let mut terminal: Vec<(u64, Template, i64)> = self - .dag_meta - .keys() - .copied() - .filter(|&d| self.dag_is_terminal(d)) - .map(|d| { - let finished = self - .dag_nodes - .get(&d) - .into_iter() - .flatten() - .filter_map(|id| self.node_rt.get(id).and_then(|r| r.finished_at)) - .max() - .unwrap_or(0); - (d, self.dag_meta[&d].template, finished) - }) - .collect(); - // Newest first, so the cap keeps the most recent per template. - terminal.sort_by(|a, b| b.2.cmp(&a.2).then(b.0.cmp(&a.0))); - let mut counts: HashMap = HashMap::new(); - let mut evict: Vec = Vec::new(); - for (d, template, finished) in terminal { - if finished > grace_cutoff { - continue; - } - let c = counts.entry(template).or_insert(0); - *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); + /// When a DAG's work node finishes on `finished_at` — the max over its + /// subtree, 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)) + .max() + .unwrap_or(0) + } + + /// Every DAG container node id in the graph. + fn containers(&self) -> Vec { + self.sched + .graph() + .nodes() + .filter(|n| n.parent.is_none() && matches!(n.payload.kind, NodeKind::Dag { .. })) + .map(|n| n.id) + .collect() + } + + /// The **visible** DAG set for the snapshot: every live (non-terminal) DAG, + /// plus the newest [`MAX_HISTORY_PER_TEMPLATE`] terminal DAGs per template + /// (terminal DAGs finished after `grace_cutoff` are always kept). Crate nodes + /// for evicted DAGs linger in the graph (bounded-prune is a Stage-C + /// follow-up); this filter is what bounds what the dashboard sees. + fn visible_dags(&self, grace_cutoff: i64) -> Vec { + let mut live: Vec = Vec::new(); + let mut terminal: Vec<(NodeId, Template, i64)> = Vec::new(); + for c in self.containers() { + if self.dag_is_terminal(c) { + if let Some(meta) = self.dag_meta(c) { + terminal.push((c, meta.template, self.dag_finished_at(c))); } + } 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 = 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 } } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index aa85eaf9..7365ce89 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -141,21 +141,6 @@ pub enum NodeKind { /// `Prebuild`, but that's a no-op there — the agent is down, so prebuild /// is skipped.) 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 /// 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 @@ -196,9 +181,6 @@ impl NodeKind { NodeKind::WritePermFile => "write_perm_file", NodeKind::ApprovalDeploy => "approval_deploy", NodeKind::SetWanted { .. } => "set_wanted", - NodeKind::ResolveApproval => "resolve_approval", - NodeKind::EmitRebuilt => "emit_rebuilt", - NodeKind::RevertIntent => "revert_intent", NodeKind::Dag { .. } => "dag", } } diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 1b9a843d..1c2022f9 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -106,9 +106,10 @@ fn handle_completion(coord: &Arc, done: NodeDone) { .job_queue .append_subgraph(claim.dag_id, subgraph, claim.node_id); } - coord + let terminal = coord .job_queue .complete_node(claim.dag_id, claim.node_id, Ok(())); + fire_terminal_hook(coord, terminal); } Err(e) => { let msg = format!("{e:#}"); @@ -120,9 +121,10 @@ fn handle_completion(coord: &Arc, done: NodeDone) { error = %msg, "job_queue: node failed" ); - coord + let terminal = coord .job_queue .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 @@ -130,6 +132,19 @@ fn handle_completion(coord: &Arc, done: NodeDone) { 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, terminal: Option) { + 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 /// whose lease is no longer held, create one for each newly-held `(dag, agent)`. fn reconcile_transients(