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

feat(#2591): auto-complete the DAG container + run terminal hooks inline

Browse source
This commit is contained in:
atlas 2026-07-20 22:44:19 +02:00 • committed by mara
commit 2294cd4516
4 changed files with 59 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-c0re/src/dashboard/schedules.rs
View file

@ -127,8 +127,10 @@ pub(super) async fn post_rebuild_queue_cancel(
State(state): State<AppState>,
AxumPath(id): AxumPath<u64>,
) -> Response {
let cancelled = state.coord.job_queue.cancel(id);
if cancelled {
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;
state.coord.emit_rebuild_queue_snapshot();
axum::Json(serde_json::json!({"cancelled": true})).into_response()
} else {

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

@ -111,7 +111,7 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
/// 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<Coordinator>, terminal: &super::TerminalDag) {
pub(crate) async fn run_terminal_hook(coord: &Arc<Coordinator>, terminal: &super::TerminalDag) {
match super::terminal_hook(terminal.template, terminal.approval_id) {
Some(super::HookKind::ResolveApproval) => {
crate::actions::resolve_approval_dag(coord, terminal).await;

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

@ -312,6 +312,11 @@ impl JobQueue {
.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))?;
// Settle the container's own (no-op) logic immediately so it parks in
// `Finishing` and its children become runnable — it never needs claiming
// or executing, and stays out of `claim_ready`. It rolls up terminal when
// its whole subtree settles (that's the DAG-done signal).
inner.sched.complete(container, Outcome::Done);
drop(inner);
self.notify.notify_one();
Ok(container.get())
@ -434,21 +439,17 @@ impl JobQueue {
terminal
}
/// 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 {
/// 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> {
let mut inner = self.lock();
let inner = &mut *inner;
let Some(container) = inner.container(dag_id) else {
return false;
};
let ids: Vec<NodeId> = std::iter::once(container)
.chain(inner.subtree(container))
.collect();
let all_pending = ids.iter().all(|&id| {
let container = inner.container(dag_id)?;
let work = inner.subtree(container);
let all_pending = work.iter().all(|&id| {
inner
.sched
.graph()
@ -456,13 +457,18 @@ impl JobQueue {
.is_some_and(|n| n.state == JobState::Pending)
});
if !all_pending {
return false;
return None;
}
for id in ids {
for id in work {
inner.sched.cancel_node(id);
}
// Roll the container up so the DAG reaches a terminal state (all children
// now `Cancelled`); `dag_rollup` reports `Cancelled` to the wire.
inner.sched.complete(container, Outcome::Done);
let terminal = inner.terminal_dag(container);
drop(inner);
self.notify.notify_one();
true
terminal
}
/// Set the step label on a `Running` node. Returns `true` when it changed.

89
hive-c0re/src/job_queue/tests.rs
View file

@ -280,24 +280,13 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() {
assert_eq!(second.dag_id, restart);
assert_eq!(second.kind.as_str(), "reconcile");
q.complete_node(restart, second.node_id, Ok(()));
// Restart's work is terminal → its lease releases. Its terminal node and
// stop's now-unblocked Reconcile both become ready in the same pass.
let ready = q.claim_ready();
let restart_fin = ready
.iter()
.find(|c| c.dag_id == restart && c.kind.as_str() == "revert_intent")
.expect("restart finalize ready");
q.complete_node(restart, restart_fin.node_id, Ok(()));
let third = ready
.iter()
.find(|c| c.dag_id == stop)
.expect("stop reconcile ready once the lease is freed");
// 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).
let third = claim_one(&q);
assert_eq!(third.dag_id, stop);
assert_eq!(third.kind.as_str(), "reconcile");
q.complete_node(stop, third.node_id, Ok(()));
// stop's terminal node (revert-intent) then runs.
let stop_fin = claim_one(&q);
assert_eq!(stop_fin.kind.as_str(), "revert_intent");
q.complete_node(stop, stop_fin.node_id, Ok(()));
assert_eq!(state_of(&q, restart), State::Done);
assert_eq!(state_of(&q, stop), State::Done);
}
@ -340,15 +329,10 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
.expect("reconcile claim")
.clone();
q.complete_node(stop, reconcile.node_id, Ok(()));
// stop's Reconcile done → its lease frees (rebuild's StopForUpdate unblocks)
// and its terminal node becomes ready; both surface in the same pass.
// 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.)
let after = q.claim_ready();
assert!(
after
.iter()
.any(|c| c.dag_id == stop && c.kind.as_str() == "revert_intent"),
"stop DAG's finalize runs once its work settles"
);
let sfu = after
.iter()
.find(|c| c.kind.as_str() == "stop_for_update")
@ -788,13 +772,11 @@ fn failed_reconcile_marks_dag_failed() {
fn cancel_clears_queued_dag() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
assert!(q.cancel(id));
// 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_eq!(state_of(&q, id), State::Cancelled);
// The terminal node's weak edges are satisfied by the cancelled (terminal)
// work nodes, so it still runs its hooks — it's the one thing left claimable.
let fin = claim_one(&q);
assert_eq!(fin.kind.as_str(), "emit_rebuilt");
q.complete_node(id, fin.node_id, Ok(()));
assert!(q.claim_ready().is_empty());
}
@ -803,32 +785,32 @@ 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));
assert!(q.cancel(id).is_none());
assert_eq!(state_of(&q, id), State::Running);
}
// ---- terminal reporting + lease release ----
#[test]
fn terminal_node_runs_after_work_settles_and_lease_released() {
fn dag_settles_terminal_and_releases_lease_after_work() {
let q = JobQueue::new(1);
let id = submit(&q, restart_online(&["agent-a"], false, "r"));
// restart = StopForUpdate → Reconcile; the terminal node (which weak-deps on
// the chain tail) isn't runnable until the whole chain is terminal.
// restart = StopForUpdate → Reconcile.
let stop = claim_one(&q);
assert_eq!(stop.kind.as_str(), "stop_for_update");
q.complete_node(id, stop.node_id, Ok(()));
let rec = claim_one(&q);
assert_eq!(rec.kind.as_str(), "reconcile");
q.complete_node(id, rec.node_id, Ok(()));
// Work settled → the terminal node is now the runnable one; it carries the
// terminal roll-up the hook consumes via `terminal_summary`.
let fin = claim_one(&q);
assert_eq!(fin.kind.as_str(), "revert_intent");
let summary = q.terminal_summary(id).expect("terminal summary");
// 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);
q.complete_node(id, fin.node_id, Ok(()));
// Lease released (freed when the work chain settled, ahead of finalize): a
// new DAG for the agent claims immediately.
assert!(q.claim_ready().is_empty(), "no terminal-hook 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.
let next = submit(
&q,
templates::reconcile_only(
@ -853,16 +835,11 @@ fn cancelled_dag_finalizes_with_terminal_rollup() {
&q,
templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
);
assert!(q.cancel(id));
// The terminal node's weak edges still fire on a fully-cancelled DAG, so its
// hook (approval resolution) runs — surfaced here as a claimable resolve-
// approval node whose `terminal_summary` is Cancelled + carries the approval id.
let fin = claim_one(&q);
assert_eq!(fin.kind.as_str(), "resolve_approval");
let summary = q.terminal_summary(id).expect("terminal summary");
// 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));
q.complete_node(id, fin.node_id, Ok(()));
// The cancelled DAG's summary stays available (until history-trimmed) and
// unrelated later activity doesn't disturb it.
let other = submit(&q, rebuild("agent-b", "r"));
@ -931,12 +908,9 @@ fn history_evicts_old_terminals_per_template() {
),
);
let c = claim_one(&q);
// Completing the single work node rolls the container up terminal (its
// inline hook fires off the returned summary — no terminal-hook node).
q.complete_node(id, c.node_id, Ok(()));
// Drain the DAG's terminal node too, so the next iteration's claim sees
// only its own work (the terminal node is excluded from the view + rollup).
let fin = claim_one(&q);
assert_eq!(fin.kind.as_str(), "revert_intent");
q.complete_node(id, fin.node_id, Ok(()));
}
// Fresh terminals are inside the grace window: nothing evicts yet,
// so a ~1s QueueDag poller can still observe every terminal state
@ -947,8 +921,7 @@ fn history_evicts_old_terminals_per_template() {
"grace window protects fresh terminals"
);
// Past the grace window the per-template cap applies.
q.trim_ignoring_grace();
assert_eq!(q.snapshot().len(), 5, "per-template history cap");
assert_eq!(q.snapshot_no_grace().len(), 5, "per-template history cap");
assert_eq!(q.live_count(), 0);
}