vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor(#2756): replace the DAG terminal hook with real tail nodes

Browse source 
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<DepOutcome>`, 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.
This commit is contained in:
atlas 2026-07-27 14:25:03 +02:00 • committed by mara
commit e8e6998ac5
11 changed files with 521 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -16,10 +16,12 @@
//! `NodeId` is the DAG id, its rolled-up state is the DAG state, and membership
//! is a graph walk — there are no host grouping side-tables. The lease is owned
//! by a subtree root and borrowed by its descendants (continuity);
//! - per-DAG terminal work runs **inline** ([`exec::run_terminal_hook`]) when the
//! container rolls up terminal, off the [`HookKind`] the *builder* stated on
//! the spec: approval-resolve or `Rebuilt`-emit. No terminal-hook node, no
//! drained event stream.
//! - per-DAG terminal work is an ordinary **tail node**
//! ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
//! appends in [`templates`], weak-edged (`AfterAny`) onto the DAG's other group
//! roots so it runs on success, failure and cancel alike. It reads how the work
//! went off its own [`Claim::deps`] — no inline hook fired from outside the
//! graph, no drained event stream.
//!
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
//! state is re-derived by the reconcile sweep. A single scheduler task
@ -47,9 +49,7 @@ use hive_sh4re::wire_time::now_unix;
use tokio::sync::Notify;
use crate::coordinator::TransientKind;
pub use model::{
DagSpec, DagView, DepWhen, HookKind, NodeKind, NodeSpec, PermPayload, Source, State,
};
pub use model::{DagSpec, DagView, DepWhen, NodeKind, NodeSpec, PermPayload, Source, State};
use resource::Resource;
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
@ -60,6 +60,31 @@ const MAX_HISTORY_DAGS: usize = 50;
/// Cap on stored node error strings.
const MAX_ERROR_LEN: usize = 2_000;
/// How one of a claimed node's dependencies finished, snapshotted at claim time.
///
/// A node only starts once its edges are satisfied, so every dep named here is
/// already terminal: an `AfterOk` edge means [`State::Done`], an `AfterAny` edge
/// means any of `Done` / `Failed` / `Cancelled`.
///
/// This is what lets a tail node be an ordinary node. A tail that must report
/// how the work below it went — "emit `Rebuilt { ok }`", "resolve this approval
/// with the failure note" — reads its deps' outcomes off its own claim, rather
/// than re-deriving them from the world after the fact. The alternative in this
/// codebase is `DeployTail`, which infers success by re-reading *git state*; that
/// works only because a deploy happens to write its result somewhere durable, and
/// it is not a pattern to copy.
///
/// Carries no node id: a tail acts on *how* its dependencies ended, never on
/// which one it was, so an id here would be a field with no reader.
#[derive(Debug, Clone)]
pub struct DepOutcome {
/// Its terminal state, in wire terms.
pub state: State,
/// Its failure reason, when it failed with one of its own. A node that
/// rolled up `Failed` from a child, or was cancelled, carries no error.
pub error: Option<String>,
}
/// 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<TransientKind>,
/// 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<DepOutcome>,
}
/// 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<HookKind>,
/// Distinct agents this DAG's nodes targeted (one for a single-agent DAG).
pub agents: Vec<String>,
pub approval_id: Option<i64>,
pub state: State,
/// First failed node's error when `state == Failed`.
pub error: Option<String>,
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<HookKind>,
source: Source,
reason: String,
transient: Option<TransientKind>,
@ -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<NodeId> = node
.deps
.iter()
.filter_map(|d| match d {
Dep::Node { id, .. } => Some(*id),
Dep::Resource { .. } => None,
})
.collect();
let deps: Vec<DepOutcome> = 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<TerminalDag> {
/// `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<TerminalDag> {
/// 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<TerminalDag> {
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<DagMeta> {
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<String> {
let mut seen: Vec<String> = 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<String> {
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<TerminalDag> {
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