From e8e6998ac5c3e34c5ea78a7327895e1192d576b1 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 14:25:03 +0200 Subject: [PATCH] refactor(#2756): replace the DAG terminal hook with real tail nodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The queue carried a per-DAG `HookKind` that fired an inline side effect from outside the graph when a container rolled up terminal. mara asked three times why this could not be an ordinary node; the answer in the code was a doc-comment claiming a node could not work, and it was wrong. `DepWhen::AfterAny` already existed with two live users, and a weak edge is satisfied by a `Cancelled` dep, so a tail node runs on success, failure and cancel alike. What was genuinely missing was smaller than a hook: a node had no way to learn how the work it followed ended. So: `Claim` now carries `deps: Vec`, snapshotted at claim time from the graph the scheduler already holds (no `hive-jobq` change). `Claim::deps_state()` / `deps_error()` roll that up, and two new kinds consume it — `ResolveApproval { approval_id }` and `EmitRebuilt { agent }`. Templates append one as a group-root with `AfterAny` edges onto the DAG's other group roots; a root's state is its subtree's roll-up, so that covers every node without fanning out to each of them. Deleted: `HookKind`, `DagSpec.hook`, `NodeKind::Dag.hook`, `DagMeta.hook`, `TerminalDag`, `terminal_dag()`, `terminal_summary()`, `dag_agents()`, `dag_rollup()`, `fire_terminal_hook()`, `run_terminal_hook()`, `emit_rebuilt()`. `complete_node` returns `()`. Load-bearing details: - `JobQueue::cancel` spares tail nodes instead of cancelling the whole subtree, and returns `bool`. Without this a cancelled approval DAG would dangle its approval forever — the hazard `tests.rs` already named. The spared tail's deps are `Cancelled`, which satisfies its weak edge, so the scheduler claims it and it resolves the row as cancelled. `hive-jobq` anticipated exactly this: `cancel_node`'s doc already says to settle afterwards so "a weak-edge terminal node observing the cancellation" can advance. - The existing `complete(container)` call after cancelling is kept and is deliberately a no-op when a tail was spared (a non-terminal child parks the container back in `Finishing`), so power ops still settle synchronously with no branch. - `DeployTail` is NOT `is_tail()`: it does real compensating work, and a cancelled DAG has nothing to compensate. - `exec::failure_reason` falls back to `first_error(dag_id)` because a group root that rolled up `Failed` from a child carries no error of its own — without it every tail-reported failure would lose its reason. - `EmitRebuilt` is per agent, so a multi-agent DAG reports each agent's own outcome rather than painting all of them with the DAG roll-up. - `ResolveApproval` is agentless: the approval row already names its agent, and that is also what lets one tail close a multi-agent DAG. Transients-derived-from-running-nodes and the frontend's node-kind strings stay out of this change; they touch iris's slice and review better next to their own diff. --- docs/coordinator.md | 2 +- hive-c0re/src/actions.rs | 38 ++-- hive-c0re/src/dashboard/schedules.rs | 7 +- hive-c0re/src/job_queue/exec.rs | 99 ++++++----- hive-c0re/src/job_queue/mod.rs | 248 +++++++++++++-------------- hive-c0re/src/job_queue/model.rs | 87 ++++++---- hive-c0re/src/job_queue/scheduler.rs | 19 +- hive-c0re/src/job_queue/submit.rs | 7 +- hive-c0re/src/job_queue/templates.rs | 99 ++++++++--- hive-c0re/src/job_queue/tests.rs | 236 ++++++++++++++++++------- hive-c0re/src/workers/auto_update.rs | 3 +- 11 files changed, 521 insertions(+), 324 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index c8df2b12..af66f329 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -49,7 +49,7 @@ Cheap — no build slot: | `MetaSync` | the rebuild's meta preamble — rebuild-dir prep, idempotent meta `sync_agents`, optional per-agent relock. Holds the `MetaWindow` resource (below); deliberately its own node so the window never covers `Prebuild`'s multi-minute build | | `Reconcile` | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop | | `StopForUpdate` | mechanical `nixos-container stop` for the profile swap; never touches `wanted`; noop if already stopped | -| `PostSwap` | the swap's Ok-only bookkeeping tail — rev marker, forge/matrix sync, manager kick, rescan, meta-inputs snapshot; `AfterOk(Swap)` so it runs only on a successful swap (the `Rebuilt` manager event still fires once per DAG from the terminal hook, not here) | +| `PostSwap` | the swap's Ok-only bookkeeping tail — rev marker, forge/matrix sync, manager kick, rescan, meta-inputs snapshot; `AfterOk(Swap)` so it runs only on a successful swap (the `Rebuilt` manager event is emitted by the DAG's `EmitRebuilt` tail node, not here) | | `Signal` | set the graceful fence + kick, so the harness runs one stop-checkpoint turn | | `Drain` | await the harness clearing the fence, bounded by the 3-min graceful-stop timeout; resolves ok either way | | `WriteDropin` | `set_nspawn_flags` + `set_resource_limits` + daemon-reload | diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index a84fdde6..d00d9227 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -501,19 +501,26 @@ async fn run_approval_schedule_prompt( finish_approval(coord, &approval, result, None) } -/// Terminal hook for approval-carrying DAGs — the job queue's scheduler calls -/// this exactly once when such a DAG settles terminal. Every approval-carrying -/// template resolves here, deploys included: the deploy pipeline is ordinary -/// queue nodes now, so the DAG's own terminal state is the authoritative -/// outcome and there's no in-node resolution to skip around. +/// Resolve an approval row from how its DAG's work ended — the body of the +/// [`NodeKind::ResolveApproval`] tail node. Every approval-carrying template +/// resolves here, deploys included: the deploy pipeline is ordinary queue nodes, +/// so the work's terminal state is the authoritative outcome and there's no +/// in-node resolution to skip around. +/// +/// `state` / `error` come from the tail node's own dependency roll-up +/// ([`Claim::deps_state`] / [`Claim::deps_error`]), so this runs on the success, +/// failure **and cancel** paths alike. +/// +/// [`NodeKind::ResolveApproval`]: crate::job_queue::NodeKind::ResolveApproval +/// [`Claim::deps_state`]: crate::job_queue::Claim::deps_state +/// [`Claim::deps_error`]: crate::job_queue::Claim::deps_error pub(crate) async fn resolve_approval_dag( coord: &Arc, - terminal: &crate::job_queue::TerminalDag, + approval_id: i64, + state: crate::job_queue::State, + error: Option<&str>, ) { use crate::job_queue::State; - let Some(approval_id) = terminal.approval_id else { - return; - }; let approval = match coord.approvals.get(approval_id) { Ok(Some(a)) => a, Ok(None) => { @@ -525,16 +532,10 @@ pub(crate) async fn resolve_approval_dag( return; } }; - let result: Result<()> = match terminal.state { + let result: Result<()> = match state { State::Done => Ok(()), State::Cancelled => Err(anyhow::anyhow!("cancelled before completion")), - _ => Err(anyhow::anyhow!( - "{}", - terminal - .error - .clone() - .unwrap_or_else(|| "job dag failed".to_owned()) - )), + _ => Err(anyhow::anyhow!("{}", error.unwrap_or("job dag failed"))), }; let mut terminal_tag = None; match approval.kind { @@ -550,8 +551,7 @@ pub(crate) async fn resolve_approval_dag( } } ApprovalKind::MergeConfigPr => { - terminal_tag = - deploy_terminal_tag(approval.agent.as_str(), approval_id, terminal.state).await; + terminal_tag = deploy_terminal_tag(approval.agent.as_str(), approval_id, state).await; // On a failed deploy, surface the failing build log back onto the // PR so the manager sees why it was rejected without leaving the // forge. Posted here rather than inside a node because this is the diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs index 6b990f93..2cfd002f 100644 --- a/hive-c0re/src/dashboard/schedules.rs +++ b/hive-c0re/src/dashboard/schedules.rs @@ -127,10 +127,9 @@ pub(super) async fn post_rebuild_queue_cancel( State(state): State, AxumPath(id): AxumPath, ) -> Response { - if let Some(terminal) = state.coord.job_queue.cancel(id) { - // Fire the DAG's inline terminal hook (power-op intent revert / approval - // resolution) off the cancel roll-up, then surface the flip live. - crate::job_queue::exec::run_terminal_hook(&state.coord, &terminal).await; + if state.coord.job_queue.cancel(id) { + // Any terminal side effect is the DAG's own spared tail node, which the + // scheduler picks up on its next pass — nothing to fire from here. state.coord.emit_rebuild_queue_snapshot(); axum::Json(serde_json::json!({"cancelled": true})).into_response() } else { diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 178b55a2..21ea0728 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -95,50 +95,71 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< NodeKind::DeployApply { .. } => run_deploy_apply(coord, claim).await, NodeKind::FinalizeDeploy { .. } => run_finalize_deploy(coord, claim).await, NodeKind::DeployTail { .. } => run_deploy_tail(coord, claim).await, + NodeKind::ResolveApproval { approval_id, .. } => { + run_resolve_approval(coord, claim, *approval_id).await + } + NodeKind::EmitRebuilt { .. } => Ok(run_emit_rebuilt(coord, claim)), NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up), // Pure grouping container — no work; completing it lets it reach - // `Finishing` so its child work nodes start. The DAG's terminal - // hook fires (inline, via `run_terminal_hook`) when the container itself - // rolls up terminal — not as a scheduled node. + // `Finishing` so its child work nodes start. The DAG's terminal side + // effect, if any, is its own tail node in the graph. NodeKind::Dag { .. } => Ok(NodeOutput::default()), } } -/// Run a settled DAG's inline terminal hook — the container-terminal -/// replacement for the old per-DAG hook node. Always best-effort: a hook -/// failure is logged inside, never surfaced. -pub(crate) async fn run_terminal_hook(coord: &Arc, terminal: &super::TerminalDag) { - match terminal.hook { - Some(super::HookKind::ResolveApproval) => { - crate::actions::resolve_approval_dag(coord, terminal).await; - } - Some(super::HookKind::EmitRebuilt) => emit_rebuilt(coord, terminal), - None => {} - } +/// Resolve the DAG's approval row from how the work it follows ended. The +/// outcome comes off this node's own dependency roll-up, not from re-reading +/// the world. Best-effort: a resolution failure is logged inside +/// [`crate::actions::resolve_approval_dag`], never surfaced as a node failure — +/// the work already happened, and failing the tail would only misreport it. +async fn run_resolve_approval( + coord: &Arc, + claim: &Claim, + approval_id: i64, +) -> Result { + let reason = failure_reason(coord, claim); + crate::actions::resolve_approval_dag(coord, approval_id, claim.deps_state(), reason.as_deref()) + .await; + Ok(NodeOutput::default()) } -/// 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, - }), - _ => {} - } +/// Why the work a tail node follows failed, as a human-readable string. +/// +/// Prefers the tail's own dependency error, but a dep that is a **group root** +/// rolled up `Failed` from a child carries no error of its own (the reason lives +/// on the leaf that actually failed) — and a grafted subgraph's nodes can't be +/// edged statically anyway. So fall back to the DAG's first failing node. Still +/// the queue's own graph, not the outside world. +fn failure_reason(coord: &Arc, claim: &Claim) -> Option { + claim + .deps_error() + .map(str::to_owned) + .or_else(|| coord.job_queue.first_error(claim.dag_id)) +} + +/// Emit this agent's `Rebuilt` manager event — `ok` when the work it follows is +/// `Done`, `!ok` with the failure note when it `Failed`, and nothing at all when +/// it `Cancelled` (nothing ran, so there is no rebuild to report). +fn run_emit_rebuilt(coord: &Arc, claim: &Claim) -> NodeOutput { + let agent = claim.agent.clone(); + match claim.deps_state() { + State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent, + ok: true, + note: None, + sha: None, + tag: None, + }), + State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { + agent, + ok: false, + note: failure_reason(coord, claim), + sha: None, + tag: None, + }), + _ => {} } + NodeOutput::default() } /// Write the agent's durable power intent — the DAG-node form of the old @@ -238,8 +259,8 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res // which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded // and the tail `Reconcile` (`AfterAny(PostSwap)`) handles recovery; here // we only refresh the observed state so dashboards reflect the failed - // swap immediately. The `Rebuilt { ok: false }` manager event fires once - // per DAG from the terminal hook (any node may be the one that failed). + // swap immediately. The `Rebuilt { ok: false }` manager event is emitted by + // the DAG's `EmitRebuilt` tail (any node may be the one that failed). if result.is_err() { coord.rescan_containers_and_emit().await; } @@ -258,8 +279,8 @@ async fn run_post_swap(coord: &Arc, claim: &Claim) -> Result, +} + /// A node claimed for execution — everything the executor needs, snapshotted at /// claim time. #[derive(Debug, Clone)] @@ -76,21 +101,38 @@ pub struct Claim { /// pill is currently shown is derived from live lease ownership /// ([`JobQueue::held_transients`]), not a per-claim edge. pub transient: Option, + /// How each node this one depends on finished. Empty for a head node. + /// See [`DepOutcome`] — this is how a tail node learns the outcome of the + /// work it follows without going back to the world to ask. + pub deps: Vec, } -/// Summary of a DAG's terminal roll-up — the input to the terminal node's -/// executor (approval resolution, `Rebuilt` emission, cancelled-power-op intent -/// revert). Computed on demand from live graph state, not drained. -#[derive(Debug, Clone)] -pub struct TerminalDag { - /// The side effect to fire, carried from the DAG container's payload. - pub hook: Option, - /// Distinct agents this DAG's nodes targeted (one for a single-agent DAG). - pub agents: Vec, - pub approval_id: Option, - pub state: State, - /// First failed node's error when `state == Failed`. - pub error: Option, +impl Claim { + /// Roll this claim's dependencies up into one outcome — what a weak-edged + /// tail node acts on. `Done` only when every dep succeeded; `Failed` when any + /// failed; otherwise `Cancelled` (all terminal, none failed, so the work was + /// dropped before it ran). + /// + /// A head node has no deps and rolls up `Done` — vacuously true, and never + /// reached in practice since only tail kinds consult this. + pub fn deps_state(&self) -> State { + if self.deps.iter().all(|d| d.state == State::Done) { + State::Done + } else if self.deps.iter().any(|d| d.state == State::Failed) { + State::Failed + } else { + State::Cancelled + } + } + + /// The first failed dependency's error, for reporting *why* the work ended + /// badly. `None` when nothing failed. + pub fn deps_error(&self) -> Option<&str> { + self.deps + .iter() + .find(|d| d.state == State::Failed) + .and_then(|d| d.error.as_deref()) + } } /// Per-node runtime metadata the crate graph doesn't carry. Lifecycle @@ -106,7 +148,6 @@ struct NodeRuntime { /// 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 { - hook: Option, source: Source, reason: String, transient: Option, @@ -241,8 +282,7 @@ impl JobQueue { /// 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. + /// container's id as the DAG id — its rolled-up state is the DAG state. /// /// # Errors /// Propagates the spec-validation error (empty / cyclic / bad parent) or a @@ -254,7 +294,6 @@ impl JobQueue { .sched .append( NodeKind::Dag { - hook: spec.hook, source: spec.source, reason: spec.reason, transient: spec.transient, @@ -334,6 +373,24 @@ impl JobQueue { }; let kind = node.payload.clone(); let agent = node.payload.agent().to_owned(); + let dep_ids: Vec = node + .deps + .iter() + .filter_map(|d| match d { + Dep::Node { id, .. } => Some(*id), + Dep::Resource { .. } => None, + }) + .collect(); + let deps: Vec = dep_ids + .into_iter() + .filter_map(|dep| { + let n = inner.sched.graph().node(dep)?; + Some(DepOutcome { + state: to_wire_state(n.state), + error: n.error.clone(), + }) + }) + .collect(); let Some(container) = inner.dag_of(id) else { continue; }; @@ -348,6 +405,7 @@ impl JobQueue { approval_id: meta.approval_id, inputs: meta.inputs, transient: meta.transient, + deps, }); // `started_at` is stamped on the graph `Node` by the scheduler's // transition to `Running` — no host-side copy needed. @@ -357,15 +415,12 @@ 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. 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 { + /// `AfterOk` failure cancellation + subtree lease release. + /// + /// Nothing is returned: a DAG's terminal side effects are its own tail nodes + /// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the + /// scheduler claims and runs like any other node. + pub fn complete_node(&self, _dag_id: u64, node_id: NodeId, result: Result<(), String>) { let mut inner = self.lock(); // The failure reason + `finished_at` are stamped onto the graph `Node` // by the scheduler (the reason rides `Outcome::Failed`); no host-side @@ -374,27 +429,31 @@ impl JobQueue { Ok(()) => Outcome::Done, Err(e) => Outcome::Failed(truncate_error(&e)), }; - let container = inner.dag_of(node_id); inner.sched.complete(node_id, outcome); - // 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 `Pending`, - /// so each is cancelled. `None` once any work node is running or terminal — - /// an in-flight nix build isn't interruptible. Otherwise the container is - /// rolled up so the DAG settles (wire state `Cancelled`) and its terminal - /// summary is returned — the caller fires the inline hook (power-intent - /// revert / approval resolution) off it. - pub fn cancel(&self, dag_id: u64) -> Option { + /// so each is cancelled. `false` once any work node is running or terminal — + /// an in-flight nix build isn't interruptible. + /// + /// **Tail nodes are spared** ([`NodeKind::is_tail`]). They are weak-edged + /// (`AfterAny`), and a `Cancelled` dep satisfies a weak edge, so sparing one + /// leaves it *ready* rather than stranded: the scheduler claims it on the next + /// pass, its [`Claim::deps_state`] reads `Cancelled`, and it resolves the + /// approval as "cancelled before completion". That is what stops a queued + /// approval DAG the operator cancelled from dangling its approval forever — + /// the job the inline hook used to do from outside the graph. + /// + /// Cancelling the tail too would be the bug: `cancel_node` only cascades along + /// `AfterOk` edges and parent links, so nothing else would reach it, and the + /// approval row would simply never be touched. + pub fn cancel(&self, dag_id: u64) -> bool { let mut inner = self.lock(); - let container = inner.container(dag_id)?; + let Some(container) = inner.container(dag_id) else { + return false; + }; let work = inner.subtree(container); let all_pending = work.iter().all(|&id| { inner @@ -404,22 +463,29 @@ impl JobQueue { .is_some_and(|n| n.state == JobState::Pending) }); if !all_pending { - return None; + return false; } for id in work { + if inner + .sched + .graph() + .node(id) + .is_some_and(|n| n.payload.is_tail()) + { + continue; + } inner.sched.cancel_node(id); } - // The container was settled to `Finishing` at submit; completing it again - // now re-runs the roll-up with its children all `Cancelled`, driving it to - // a terminal state synchronously within this lock — so the caller reads - // the terminal summary immediately instead of waiting for the scheduler - // loop to observe the cancellation. `dag_rollup` reports `Cancelled` to - // the wire (a container whose children all cancelled). + // Re-run the container's roll-up now that its children are `Cancelled`. + // With a spared tail still `Pending` this is a deliberate no-op — the + // container has a non-terminal child, so `settle_terminal` parks it back + // in `Finishing` and it rolls up for real once the tail finishes. With no + // tail (a power op) every child *is* terminal, so it settles synchronously + // here exactly as before. inner.sched.complete(container, Outcome::Done); - let terminal = inner.terminal_dag(container); drop(inner); self.notify.notify_one(); - terminal + true } /// Link a `build_logs` row to a specific `Running` node. @@ -463,17 +529,6 @@ impl JobQueue { inner.dag_first_error(container) } - /// A DAG's terminal roll-up summary, computed on demand from its container. - /// `None` if the DAG id is unknown. Test-only — production reads the summary - /// `complete_node` returns when the container rolls up terminal. - #[cfg(test)] - #[must_use] - pub(crate) fn terminal_summary(&self, dag_id: u64) -> Option { - let inner = self.lock(); - let container = inner.container(dag_id)?; - inner.terminal_dag(container) - } - /// The `(dag_id, agent, kind)` triples for every per-agent lease currently /// held by a DAG that carries a transient pill — the live transient-pill /// set, a pull query over crate resource ownership (replaces the old @@ -566,7 +621,6 @@ impl QueueInner { /// not a stored side-table. fn dag_meta(&self, container: NodeId) -> Option { let NodeKind::Dag { - hook, source, reason, transient, @@ -578,7 +632,6 @@ impl QueueInner { return None; }; Some(DagMeta { - hook: *hook, source: *source, reason: reason.clone(), transient: *transient, @@ -588,35 +641,6 @@ impl QueueInner { }) } - /// 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`. (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 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, - Some(JobState::Pending) => any_queued = true, - Some(JobState::Cancelled) => any_cancelled = true, - Some(JobState::Done) | None => {} - } - } - if any_running { - State::Running - } else if any_queued { - State::Queued - } else if any_cancelled { - State::Cancelled - } else { - State::Done - } - } - /// 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 { @@ -626,22 +650,8 @@ impl QueueInner { .is_some_and(|n| n.state.is_terminal()) } - /// Distinct agents a DAG's work nodes target, in first-seen order. - fn dag_agents(&self, container: NodeId) -> Vec { - let mut seen: Vec = Vec::new(); - 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.to_owned()); - } - } - } - seen - } - /// First failed work node's error (read off the graph `Node`), for the - /// terminal roll-up summary the inline hook consumes. + /// dashboard's DAG-level error line. fn dag_first_error(&self, container: NodeId) -> Option { for id in self.subtree(container) { if let Some(n) = self.sched.graph().node(id) @@ -654,18 +664,6 @@ impl QueueInner { None } - /// 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 { - hook: meta.hook, - agents: self.dag_agents(container), - approval_id: meta.approval_id, - state: self.dag_rollup(container), - error: self.dag_first_error(container), - }) - } - /// 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 diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 03ea621f..0d321a36 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -17,27 +17,6 @@ use serde::Serialize; use crate::coordinator::TransientKind; -/// The inline side effect a settled DAG fires when its container node rolls -/// up terminal (there is no hook *node*). Stated explicitly by the builder in -/// `templates.rs` / `submit.rs` rather than inferred from a DAG-level enum: -/// only the builder knows why it assembled the DAG, so only the builder can -/// say what should happen at the end of it. -/// -/// A cancelled DAG deliberately gets **no** compensating hook. [`super::JobQueue::cancel`] -/// refuses unless every work node is still `Pending`, and a cancel *cascade* -/// rolls up `Failed` (see `dag_rollup`), never `Cancelled` — so on a -/// `Cancelled` DAG no node ever executed and there is nothing to undo. A power -/// op's `SetWanted` head provably never ran, so its intent is still whatever -/// the operator last set it to. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] -#[serde(rename_all = "snake_case")] -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, -} - /// When a dependency edge is considered satisfied. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] @@ -248,6 +227,31 @@ pub enum NodeKind { /// means it survives a `hive-c0re` restart mid-deploy, which an in-memory /// queue does not. DeployTail { agent: String }, + /// Tail node of an approval-carrying DAG (spawn / opaque deploy / config-PR + /// merge): resolve the approval row from how the work actually ended. + /// + /// Weak-edged (`DepWhen::AfterAny`) like [`NodeKind::DeployTail`], so it runs on + /// success, failure **and cancel** alike and decides internally. It reads its + /// dependencies' terminal states off its own [`Claim::deps`] rather than + /// re-deriving them from the world the way `DeployTail` reads git: a node is + /// *told* how the work it follows ended, it does not go back out and ask. + /// + /// Agentless on purpose: the approval row already names its agent, so + /// carrying one here would be a second copy free to drift. Like + /// [`NodeKind::MetaLock`] it reports `""` from [`NodeKind::agent`] and takes + /// no lease — which is also what lets one close a multi-agent DAG. + /// + /// [`Claim::deps`]: super::Claim::deps + ResolveApproval { approval_id: i64 }, + /// Tail node of a rebuild / perm-change: emit this agent's `Rebuilt` manager + /// event — `ok` when its deps are `Done`, `!ok` carrying the failure note when + /// they `Failed`, and **nothing at all** when they `Cancelled` (a cancelled DAG + /// never ran, so there is no rebuild to report). + /// + /// One node **per agent**, unlike the DAG-wide hook it replaces: a multi-agent + /// DAG now reports each agent's own outcome instead of painting every agent with + /// the whole DAG's roll-up. + EmitRebuilt { agent: String }, /// Write the agent's durable power intent (`wanted = Up` when `up`, else /// `Offline`) as a first-class DAG node, at the head of a power-op /// template so the downstream `Reconcile` reads it. Replaces the old @@ -264,15 +268,11 @@ pub enum NodeKind { SetWanted { agent: String, up: bool }, /// The **DAG container** node: one per submitted DAG, carrying the group's /// domain metadata. Every node hangs *under* it (its subtree), so - /// the container's `NodeId` **is** the DAG id, its rolled-up state **is** the - /// DAG state, and it reaching terminal **is** the completion signal that - /// fires the DAG's inline `hook`. Pure grouping — lease- and + /// the container's `NodeId` **is** the DAG id and its rolled-up state **is** + /// the DAG state. Pure grouping — lease- and /// build-slot-exempt; the executor instant-completes it (`Done`) so it /// reaches `Finishing` and its children start. Dag { - /// The side effect to run when this DAG settles, or `None` for a DAG - /// with none (power op, meta-update, boot). - hook: Option, source: Source, reason: String, transient: Option, @@ -307,6 +307,8 @@ impl NodeKind { NodeKind::DeployApply { .. } => "deploy_apply", NodeKind::FinalizeDeploy { .. } => "finalize_deploy", NodeKind::DeployTail { .. } => "deploy_tail", + NodeKind::ResolveApproval { .. } => "resolve_approval", + NodeKind::EmitRebuilt { .. } => "emit_rebuilt", NodeKind::SetWanted { .. } => "set_wanted", NodeKind::Dag { .. } => "dag", } @@ -338,11 +340,29 @@ impl NodeKind { | NodeKind::DeployApply { agent } | NodeKind::FinalizeDeploy { agent } | NodeKind::DeployTail { agent } + | NodeKind::EmitRebuilt { agent } | NodeKind::SetWanted { agent, .. } => agent, - NodeKind::MetaLock { .. } | NodeKind::Reparent { .. } | NodeKind::Dag { .. } => "", + NodeKind::MetaLock { .. } + | NodeKind::Reparent { .. } + | NodeKind::ResolveApproval { .. } + | NodeKind::Dag { .. } => "", } } + /// Whether this is a DAG's **tail** — a node that reports how the rest of the + /// DAG ended rather than doing work of its own. + /// + /// The one place this matters is [`super::JobQueue::cancel`], which spares + /// tails so they still run (and report `Cancelled`) on a cancelled DAG. Note + /// [`NodeKind::DeployTail`] is *not* one: despite the name it does real + /// compensating work, and on a cancelled DAG there is nothing to compensate. + pub fn is_tail(&self) -> bool { + matches!( + self, + NodeKind::ResolveApproval { .. } | NodeKind::EmitRebuilt { .. } + ) + } + /// Nix-heavy kinds hold one of the `buildSlots` semaphore permits /// for the node's duration. pub fn needs_build_slot(&self) -> bool { @@ -446,14 +466,15 @@ pub struct NodeSpec { /// ([`NodeKind::WritePermFile`]), not this generic spec. #[derive(Debug, Clone)] pub struct DagSpec { - /// The inline side effect to fire when this DAG settles. Explicit — the - /// builder assembling the DAG is the only thing that knows its intent. - pub hook: Option, pub source: Source, /// Free-form "why". pub reason: String, - /// The approval row [`HookKind::ResolveApproval`] resolves. Set together - /// with that hook; carried separately because the hook needs the id. + /// The approval row this DAG belongs to, for display + the `approval_id` on + /// every [`Claim`]. The *resolving* of it rides the + /// [`NodeKind::ResolveApproval`] tail node instead — this field does not + /// drive it. + /// + /// [`Claim`]: super::Claim pub approval_id: Option, /// Meta-update only: the inputs to bump. Display copy lives on the DAG. pub inputs: Vec, diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 87536ef9..6191ab6b 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -105,10 +105,9 @@ fn handle_completion(coord: &Arc, done: NodeDone) { .job_queue .append_subgraph(claim.dag_id, subgraph, claim.node_id); } - let terminal = coord + coord .job_queue .complete_node(claim.dag_id, claim.node_id, Ok(())); - fire_terminal_hook(coord, terminal); } Err(e) => { let msg = format!("{e:#}"); @@ -120,10 +119,9 @@ fn handle_completion(coord: &Arc, done: NodeDone) { error = %msg, "job_queue: node failed" ); - let terminal = coord + 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 @@ -131,19 +129,6 @@ 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( diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 3584793c..e07475d9 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -188,9 +188,9 @@ fn concat_subgraphs(chains: Vec>) -> Vec { out } -/// Wrap assembled power-op `nodes` in a `DagSpec`. No terminal hook: a power -/// op's effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to -/// do once they settle. +/// Wrap assembled power-op `nodes` in a `DagSpec`. No tail node: a power op's +/// effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to do once +/// they settle. fn power_dag( transient: TransientKind, source: Source, @@ -198,7 +198,6 @@ fn power_dag( nodes: Vec, ) -> DagSpec { DagSpec { - hook: None, source, reason, approval_id: None, diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 1b85f338..d47be49d 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -30,7 +30,7 @@ use anyhow::{Result, bail}; -use super::model::{DagSpec, Dep, DepWhen, HookKind, NodeKind, NodeSpec, PermPayload, Source}; +use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source}; use crate::coordinator::TransientKind; /// After-ok edge on the previous node — the common chain link. Shared with @@ -43,6 +43,23 @@ pub(crate) fn after_ok(on: u64) -> Vec { }] } +/// Weak edges onto every one of a DAG's other **group-roots** — how a tail node +/// (`ResolveApproval` / `EmitRebuilt`) sees the whole DAG's outcome. +/// +/// Group-roots are the right granularity, not "every node": a root's state *is* +/// its subtree's roll-up, so edging the roots covers every descendant while +/// keeping the tail's dep list small and stable as subtrees grow. `AfterAny` +/// throughout, so the tail runs on success, failure and cancel alike and decides +/// from [`super::Claim::deps_state`]. +pub(crate) fn after_any_all(ons: &[u64]) -> Vec { + ons.iter() + .map(|&on| Dep { + on, + when: DepWhen::AfterAny, + }) + .collect() +} + /// Build one **top-level (group-root)** node — `parent = None`. `kind` carries /// the agent it targets ([`NodeKind`] is the payload directly). Shared with /// `submit.rs`'s dynamic power-op builders. A root owns whatever resource it @@ -173,15 +190,27 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str) -> Vec { /// when `wanted = Offline` (a rebuild of a deliberately-stopped agent /// leaves it stopped). `relock = false` only for meta-update cascade /// children. +/// +/// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots +/// (`MetaSync`, `Prebuild`, `Reconcile`) — `Prebuild`'s roll-up carries the +/// whole `StopForUpdate`→`Swap`→`PostSwap` subtree, so those three cover every +/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so +/// it reaches `Done` even after a failed swap and the tail would report success. pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { + let mut nodes = rebuild_nodes(agent, relock, 0); + nodes.push(node( + NodeKind::EmitRebuilt { + agent: agent.to_owned(), + }, + after_any_all(&[0, 1, 5]), + )); DagSpec { - hook: Some(HookKind::EmitRebuilt), source, reason, approval_id: None, inputs: Vec::new(), transient: Some(TransientKind::Rebuilding), - nodes: rebuild_nodes(agent, relock, 0), + nodes, } } @@ -201,12 +230,18 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag /// bookkeeping tail — rollback when a merge landed unfinalized, forge tag /// mirror, PR failure comment (see [`NodeKind::DeployTail`]). /// +/// - `ResolveApproval` (4, **root**, `AfterAny` `DeployWindow`): resolves the +/// approval row. A root rather than another child, so it isn't inside the +/// window's resource subtree — it runs once the window has released the meta +/// window, lease and build slot. One edge suffices here: `DeployWindow` is the +/// DAG's only other group-root, so its roll-up already *is* the whole +/// pipeline's outcome. +/// /// The window still spans the container build, as it must: `prepare_deploy` /// leaves `flake.lock` staged-uncommitted for the build's whole duration. pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { let a = || agent.to_owned(); DagSpec { - hook: Some(HookKind::ResolveApproval), source: Source::Approval, reason, approval_id: Some(approval_id), @@ -224,6 +259,10 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec when: DepWhen::AfterAny, }], ), + node( + NodeKind::ResolveApproval { approval_id }, + after_any_all(&[0]), + ), ], } } @@ -241,7 +280,6 @@ pub fn reconcile_only( transient: Option, ) -> DagSpec { DagSpec { - hook: None, source, reason, approval_id: None, @@ -264,10 +302,11 @@ pub fn reconcile_only( /// `Create` (child) owns the agent lease; `WriteDropin` + `Reconcile` /// (children of `Create`) borrow it. A failure cancel-cascades the rest — /// unlike rebuild there's no recovery-reconcile (nothing to converge if the -/// container was never created). +/// container was never created). Closed by a `ResolveApproval` tail root edged +/// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up +/// already carries the whole cascade. pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { DagSpec { - hook: Some(HookKind::ResolveApproval), source: Source::Approval, reason, approval_id: Some(approval_id), @@ -280,6 +319,10 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { child(0, NodeKind::Create { agent: a() }, Vec::new()), child(1, NodeKind::WriteDropin { agent: a() }, Vec::new()), child(1, NodeKind::Reconcile { agent: a() }, after_ok(2)), + node( + NodeKind::ResolveApproval { approval_id }, + after_any_all(&[0]), + ), ] }, } @@ -287,7 +330,9 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { /// Perm change: commit the JSON file(s), then the rebuild subgraph so /// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes -/// effect in the container. +/// effect in the container. Group-roots are `WritePermFile`(0) plus the rebuild +/// subgraph's `MetaSync`(1) / `Prebuild`(2) / `Reconcile`(6), so the +/// `EmitRebuilt` tail edges all four. pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { let mut nodes = vec![node( NodeKind::WritePermFile { @@ -297,8 +342,13 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay Vec::new(), )]; nodes.extend(rebuild_nodes(agent, true, 1)); + nodes.push(node( + NodeKind::EmitRebuilt { + agent: agent.to_owned(), + }, + after_any_all(&[0, 1, 2, 6]), + )); DagSpec { - hook: Some(HookKind::EmitRebuilt), source, reason, approval_id: None, @@ -324,22 +374,30 @@ pub fn meta_update( reason: String, approval_id: Option, ) -> DagSpec { + let mut nodes = vec![node( + NodeKind::MetaLock { + sweep: false, + fanout: None, + }, + Vec::new(), + )]; + // The bump itself has no side effect, so an operator-driven one ends at the + // `MetaLock`; an approval-driven one still has its row to resolve and gets a + // tail edged onto that single group-root — whose roll-up covers the rebuild + // subgraphs `MetaLock` grows into itself. + if let Some(approval_id) = approval_id { + nodes.push(node( + NodeKind::ResolveApproval { approval_id }, + after_any_all(&[0]), + )); + } DagSpec { - // The bump itself has no side effect; an approval-driven one still has - // its row to resolve. - hook: approval_id.map(|_| HookKind::ResolveApproval), source, reason, approval_id, inputs, transient: Some(TransientKind::Rebuilding), - nodes: vec![node( - NodeKind::MetaLock { - sweep: false, - fanout: None, - }, - Vec::new(), - )], + nodes, } } @@ -351,14 +409,13 @@ pub fn meta_update( /// (dashboard tree, ``/`` sentinel routing, permission /// checks), so a parent move needs no container rebuild to take effect. /// No transient pill either — the node is agentless (no lease to hang one -/// off of) and near-instant. No terminal hook: the write is the whole effect. +/// off of) and near-instant. No tail node: the write is the whole effect. pub fn reparent( moves: Vec<(hive_types::Ident, Option)>, source: Source, reason: String, ) -> DagSpec { DagSpec { - hook: None, source, reason, approval_id: None, diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index d3b7b5ce..e1a5e9fc 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -48,6 +48,39 @@ fn claim_one(q: &JobQueue) -> Claim { claims.pop().expect("one claim") } +/// Claim an approval DAG's `ResolveApproval` tail, assert which approval it +/// carries and what it will report to that row, then complete it. Replaces the +/// old `terminal_summary()` assertions: the outcome is no longer a struct handed +/// to a hook, it's what this node reads off its own deps. +fn settle_approval_tail(q: &JobQueue, dag_id: u64, approval_id: i64, expect: State) { + let tail = claim_one(q); + assert!( + matches!(tail.kind, NodeKind::ResolveApproval { approval_id: got } if got == approval_id), + "expected the ResolveApproval tail for #{approval_id}, got {:?}", + tail.kind + ); + assert_eq!( + tail.deps_state(), + expect, + "outcome the tail reports to approval #{approval_id}" + ); + q.complete_node(dag_id, tail.node_id, Ok(())); +} + +/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim a rebuild / +/// perm-change DAG's tail, assert the `Rebuilt` event it will emit, complete it. +fn settle_rebuild_tail(q: &JobQueue, dag_id: u64, agent: &str, expect: State) { + let tail = claim_one(q); + assert_eq!(tail.kind.as_str(), "emit_rebuilt"); + assert_eq!(tail.agent, agent, "`Rebuilt` is emitted per agent"); + assert_eq!( + tail.deps_state(), + expect, + "ok-ness of the emitted `Rebuilt`" + ); + q.complete_node(dag_id, tail.node_id, Ok(())); +} + fn state_of(q: &JobQueue, dag_id: u64) -> State { // A fully-`Done` DAG drops out of the snapshot (its nodes are all // excluded) — absence is the completion signal, so map it to `Done`. @@ -196,6 +229,7 @@ fn rebuild_chain_claims_in_dep_order() { ); q.complete_node(id, c.node_id, Ok(())); } + settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -315,8 +349,8 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() { assert_eq!(second.kind.as_str(), "reconcile"); q.complete_node(restart, second.node_id, Ok(())); // Restart's work is terminal → its lease releases, so stop's now-unblocked - // Reconcile becomes ready (restart's inline hook fired off the returned - // summary — no terminal-hook node). + // Reconcile becomes ready (a power op has no tail node, so nothing of + // restart's remains claimable). let third = claim_one(&q); assert_eq!(third.dag_id, stop); assert_eq!(third.kind.as_str(), "reconcile"); @@ -369,8 +403,8 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { .clone(); q.complete_node(stop, reconcile.node_id, Ok(())); // stop's Reconcile done → its lease frees, so rebuild's StopForUpdate - // unblocks. (stop's DAG rolls up terminal; its inline hook fires off the - // returned summary — no terminal-hook node in the claim set.) + // unblocks. (stop's DAG rolls up terminal; a power op has no tail node, so + // nothing of stop's is left in the claim set.) let after = q.claim_ready(); let sfu = after .iter() @@ -591,7 +625,6 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // the emitter and its LOCAL 0-based deps are rebased onto the DAG. let q = JobQueue::new(4); let spec = DagSpec { - hook: None, source: Source::AutoUpdate, reason: "sweep".to_owned(), approval_id: None, @@ -703,6 +736,73 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() { ); } +// ---- dep outcomes on the claim ---- + +/// Every claimed node reports how each node it depends on finished, and those +/// deps are always already terminal — that is what a node's edges being +/// satisfied *means*. A tail node reads its `deps` instead of going back to the +/// world to find out how the work below it went. +#[test] +fn claim_carries_terminal_dep_outcomes() { + let q = JobQueue::new(1); + let id = submit(&q, rebuild("agent-a", "r")); + let mut saw_a_dep = false; + loop { + let mut claims = q.claim_ready(); + let Some(claim) = claims.pop() else { break }; + assert!(claims.is_empty(), "one build slot ⇒ one claim at a time"); + for dep in &claim.deps { + saw_a_dep = true; + assert!( + dep.state.is_terminal(), + "{} was claimed with a non-terminal dep ({:?}) — a node's edges \ + being satisfied is exactly the claim that its deps have finished", + claim.kind.as_str(), + dep.state + ); + assert_eq!( + dep.state, + State::Done, + "on the happy path every dep of {} finished Done", + claim.kind.as_str() + ); + assert_eq!(dep.error, None, "a Done dep carries no error"); + } + q.complete_node(id, claim.node_id, Ok(())); + } + assert!(saw_a_dep, "the rebuild DAG has at least one dependent node"); + assert_eq!(state_of(&q, id), State::Done); +} + +/// The failure direction, which is the whole point of carrying outcomes at all: +/// `Reconcile` hangs off `Prebuild` with `AfterAny`, so a failed prebuild +/// cancel-cascades `StopForUpdate`/`Swap`/`PostSwap` and `Reconcile` still runs +/// — and its claim hands it the failure, including the reason, rather than +/// leaving the executor to go and re-derive it from the world. +#[test] +fn claim_dep_outcome_reports_a_failed_dep_with_its_error() { + let q = JobQueue::new(1); + let id = submit(&q, rebuild("agent-a", "r")); + let meta_sync = claim_one(&q); + q.complete_node(id, meta_sync.node_id, Ok(())); + let prebuild = claim_one(&q); + assert_eq!(prebuild.kind.as_str(), "prebuild"); + q.complete_node(id, prebuild.node_id, Err("nix build exploded".to_owned())); + let reconcile = claim_one(&q); + assert_eq!(reconcile.kind.as_str(), "reconcile"); + assert_eq!( + reconcile + .deps + .iter() + .map(|d| (d.state, d.error.as_deref())) + .collect::>(), + vec![(State::Failed, Some("nix build exploded"))], + "reconcile's claim carries the failed prebuild and its reason" + ); + assert_eq!(reconcile.deps_state(), State::Failed); + assert_eq!(reconcile.deps_error(), Some("nix build exploded")); +} + // ---- failure: cancel-downstream + AfterAny ---- #[test] @@ -811,6 +911,7 @@ fn swap_ok_runs_post_swap_before_reconcile() { let reconcile = claim_one(&q); assert_eq!(reconcile.kind.as_str(), "reconcile"); q.complete_node(id, reconcile.node_id, Ok(())); + settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } @@ -832,10 +933,17 @@ fn failed_reconcile_marks_dag_failed() { fn cancel_clears_queued_dag() { let q = JobQueue::new(1); let id = submit(&q, rebuild("agent-a", "r")); - // Cancel returns the terminal summary (state `Cancelled`) — the inline hook - // fires off it at the caller; there's no terminal-hook node to claim. - let terminal = q.cancel(id).expect("cancelled"); - assert_eq!(terminal.state, State::Cancelled); + assert!(q.cancel(id), "fully-queued dag cancels"); + // Every work node is `Cancelled`, but the tail is spared so it can still + // report the cancellation — so it is the one thing left to claim. + let tail = claim_one(&q); + assert_eq!(tail.kind.as_str(), "emit_rebuilt"); + assert_eq!( + tail.deps_state(), + State::Cancelled, + "the spared tail sees its deps cancelled, so it emits no `Rebuilt`" + ); + q.complete_node(id, tail.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Cancelled); assert!(q.claim_ready().is_empty()); } @@ -845,23 +953,24 @@ fn cancel_refuses_running_dag() { let q = JobQueue::new(1); let id = submit(&q, rebuild("agent-a", "r")); let _ = claim_one(&q); - assert!(q.cancel(id).is_none()); + assert!(!q.cancel(id)); assert_eq!(state_of(&q, id), State::Running); } -/// A cancelled power op must fire **no** compensating hook — not even one that +/// A cancelled power op must run **no** compensating node — not even one that /// carries a `SetWanted` head. /// -/// `cancel` refuses unless every work node is still `Pending` -/// (`cancel_refuses_running_dag`) and a cancel *cascade* rolls up `Failed` -/// rather than `Cancelled`, so a `Cancelled` DAG provably never executed a -/// node: its `SetWanted` never ran and the agent's intent still reads whatever +/// Now structural rather than a property of a hook enum: a power op emits no +/// tail node at all, so once its work nodes cancel there is simply nothing left +/// to claim. `cancel` also refuses unless every work node is still `Pending` +/// (`cancel_refuses_running_dag`), so a `Cancelled` DAG provably never executed +/// a node: its `SetWanted` never ran and the agent's intent still reads whatever /// the operator last set. A "revert" instead writes the agent's *observed* /// state, which for a down-but-`wanted = Up` agent (crashed, or caught /// mid-bounce) flips the intent to `Offline` and leaves it /// deliberately-stopped as far as reconcile and crash-watch are concerned. #[test] -fn cancelled_power_op_fires_no_hook() { +fn cancelled_power_op_runs_no_compensating_node() { for graceful in [false, true] { for running in [false, true] { let targets = vec![("agent-a".to_owned(), running)]; @@ -896,12 +1005,12 @@ fn cancelled_power_op_fires_no_hook() { ); let q = JobQueue::new(1); let id = submit(&q, spec); - let summary = q.cancel(id).expect("cancelled while queued"); - assert_eq!(summary.state, State::Cancelled); - assert_eq!( - summary.hook, None, + assert!(q.cancel(id), "cancelled while queued"); + assert_eq!(state_of(&q, id), State::Cancelled); + assert!( + q.claim_ready().is_empty(), "cancelled {name} (graceful={graceful}, running={running}) must \ - fire no hook — no node of it ever ran" + leave nothing to run — a power op emits no tail node" ); } } @@ -920,13 +1029,10 @@ fn dag_settles_terminal_and_releases_lease_after_work() { q.complete_node(id, stop.node_id, Ok(())); let rec = claim_one(&q); assert_eq!(rec.kind.as_str(), "reconcile"); - // Completing the last work node rolls the container up terminal and returns - // the summary the inline hook consumes — there is no terminal-hook node. - let summary = q - .complete_node(id, rec.node_id, Ok(())) - .expect("terminal summary"); - assert_eq!(summary.state, State::Done); - assert!(q.claim_ready().is_empty(), "no terminal-hook node to claim"); + // Completing the last work node rolls the container up terminal. A power op + // has no tail node, so nothing is left to claim. + q.complete_node(id, rec.node_id, Ok(())); + assert!(q.claim_ready().is_empty(), "no tail node to claim"); assert_eq!(state_of(&q, id), State::Done); // Lease released when the work chain settled: a new DAG for the agent claims // immediately. @@ -938,31 +1044,44 @@ fn dag_settles_terminal_and_releases_lease_after_work() { assert_eq!(c.dag_id, next); } -/// A DAG cancelled while fully queued must still surface a terminal -/// roll-up for the scheduler's hooks — otherwise a queued approval -/// DAG cancelled by the operator would dangle its approval forever. +/// A DAG cancelled while fully queued must still **run its tail**, or a queued +/// approval DAG cancelled by the operator would dangle its approval forever. +/// +/// This is the load-bearing case for sparing tails in [`JobQueue::cancel`]: the +/// work nodes all cancel, but `ResolveApproval` is weak-edged, so a `Cancelled` +/// dep satisfies its edge and it becomes claimable instead of being cancelled +/// along with everything else. It reads `Cancelled` off its own deps and resolves +/// the approval as "cancelled before completion". #[test] -fn cancelled_dag_finalizes_with_terminal_rollup() { +fn cancelled_dag_still_runs_its_approval_tail() { let q = JobQueue::new(1); let id = submit( &q, templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), ); - // Cancel rolls the DAG up terminal and returns its summary — the inline hook - // (approval resolution) runs off it at the caller. Cancelled + approval id 7. - let summary = q.cancel(id).expect("cancelled"); - assert_eq!(summary.state, State::Cancelled); - assert_eq!(summary.approval_id, Some(7)); - // The cancelled DAG's summary stays available (until history-trimmed) and - // unrelated later activity doesn't disturb it. + assert!(q.cancel(id), "fully-queued dag cancels"); + let tail = claim_one(&q); + assert_eq!(tail.dag_id, id); + assert_eq!(tail.kind.as_str(), "resolve_approval"); + assert!( + matches!(tail.kind, NodeKind::ResolveApproval { approval_id: 7 }), + "the tail carries the approval to resolve, got {:?}", + tail.kind + ); + assert_eq!( + tail.deps_state(), + State::Cancelled, + "so the executor resolves the approval as cancelled, not as a failure" + ); + assert_eq!(tail.deps_error(), None, "a cancelled dep carries no error"); + q.complete_node(id, tail.node_id, Ok(())); + assert_eq!(state_of(&q, id), State::Cancelled); + // Unrelated later activity doesn't disturb the settled DAG. let other = submit(&q, rebuild("agent-b", "r")); let c = claim_one(&q); assert_eq!(c.dag_id, other); q.complete_node(other, c.node_id, Err("boom".to_owned())); - assert_eq!( - q.terminal_summary(id).map(|t| t.state), - Some(State::Cancelled) - ); + assert_eq!(state_of(&q, id), State::Cancelled); } // ---- approval deploy subtree ---- @@ -1006,13 +1125,12 @@ fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { ); q.complete_node(id, tail.node_id, Ok(())); - let summary = q.terminal_summary(id).expect("dag terminal"); + settle_approval_tail(&q, id, 7, State::Failed); assert_eq!( - summary.state, + state_of(&q, id), State::Failed, "an Ok tail must not launder a failed deploy into a success" ); - assert_eq!(summary.approval_id, Some(7)); } /// The deploy's happy path: `DeployApply` does not build. It grows the ordinary @@ -1080,9 +1198,8 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { assert!(matches!(tail.kind, NodeKind::DeployTail { .. })); q.complete_node(id, tail.node_id, Ok(())); - let summary = q.terminal_summary(id).expect("dag terminal"); - assert_eq!(summary.state, State::Done); - assert_eq!(summary.approval_id, Some(11)); + settle_approval_tail(&q, id, 11, State::Done); + assert_eq!(state_of(&q, id), State::Done); } /// A failure *inside* the grafted rebuild is the failure mode the subgraph @@ -1132,12 +1249,14 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() { ); q.complete_node(id, tail.node_id, Ok(())); - let summary = q.terminal_summary(id).expect("dag terminal"); - assert_eq!(summary.state, State::Failed); + settle_approval_tail(&q, id, 13, State::Failed); + assert_eq!(state_of(&q, id), State::Failed); assert_eq!( q.first_error(id).as_deref(), Some("profile swap failed"), - "the tail annotates failed/ with this" + "the tail annotates failed/ with this — and it is also what + `exec::failure_reason` falls back to, since the tail's own dep is a + group root that rolled up Failed and so carries no error itself" ); } @@ -1168,9 +1287,8 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { ); q.complete_node(id, tail.node_id, Ok(())); - let summary = q.terminal_summary(id).expect("dag terminal"); - assert_eq!(summary.state, State::Failed); - assert_eq!(summary.approval_id, Some(9)); + settle_approval_tail(&q, id, 9, State::Failed); + assert_eq!(state_of(&q, id), State::Failed); } // ---- build logs, history ---- @@ -1219,8 +1337,7 @@ fn history_evicts_oldest_terminals_past_flat_cap() { // Fail the single work node so the DAG *lingers*: a fully-`Done` DAG // drops off the wire entirely, but a `Failed` one is retained (+ // history-capped) so the operator can still triage it. Completing the - // node rolls the container up terminal (its inline hook fires off the - // returned summary — no terminal-hook node). + // node rolls the container up terminal. q.complete_node(id, c.node_id, Err("boom".to_owned())); ids.push(id); } @@ -1310,8 +1427,8 @@ fn spawn_shape_provision_create_dropin_reconcile() { assert_eq!(c.approval_id, Some(7)); q.complete_node(id, c.node_id, Ok(())); } - let report_terminal = state_of(&q, id); - assert_eq!(report_terminal, State::Done); + settle_approval_tail(&q, id, 7, State::Done); + assert_eq!(state_of(&q, id), State::Done); } #[test] @@ -1342,6 +1459,7 @@ fn perm_change_shape_prefixes_rebuild_chain() { assert_eq!(c.kind.as_str(), expected); q.complete_node(id, c.node_id, Ok(())); } + settle_rebuild_tail(&q, id, "agent-a", State::Done); assert_eq!(state_of(&q, id), State::Done); } diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 0ba9460a..90d3e555 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -344,8 +344,7 @@ fn submit_boot_tree( let spec = DagSpec { // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they - // land; the boot DAG as a whole has no terminal side effect. - hook: None, + // land; the boot DAG as a whole has no terminal side effect, so no tail. source: Source::AutoUpdate, reason, approval_id: None,