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

feat(#2591): port hive-c0re job_queue onto the hive-jobq crate

Browse source 
Replace the in-tree scheduler with the domain-agnostic hive-jobq crate
(merged in #2615): parent-axis grouping + borrow/subtree-reservation
resource model + roll-up completion (State::Finishing).

Host adaptation:
- NodeSpec gains an explicit `parent` axis; templates declare grouping +
  sibling ordering directly (deps order execution, parent groups a subtree
  whose resource the descendants borrow).
- Rebuild is a nested two-root subtree: Prebuild (root, owns the build slot
  for the whole subtree, lease-exempt) -> StopForUpdate (child, owns the
  agent lease) -> Swap/PostSwap (children, borrow both); Reconcile is a
  separate top-level root (AfterAny Prebuild) so it survives the cancel-
  cascade of any failed step (recovery-start invariant) and converges to
  the persisted `wanted` on a fresh lease. This is the multi-root
  correction to the single-root-chain sketch: node0=root broke lease-
  exemption (hoisting the lease onto Prebuild) and recovery-reconcile
  (root failure cancels all children).
- Spawn / perm-change / power-ops (stop/start/restart) group-rooted the
  same way; per-agent power-op subgraphs stay independent roots so a
  multi-agent DAG runs them concurrently, each on its own lease.
- insert_group honours the explicit parent axis (no lease hoisting); the
  DAG terminal node deps AfterAny on every group root and runs once the
  whole op rolls up. Drop the old Graph::add_dep terminal wiring.

36/36 job_queue tests, full hive-c0re suite green, clippy --all-targets.
This commit is contained in:
atlas 2026-07-20 21:46:08 +02:00 • committed by mara
commit a5c321a1a0
14 changed files with 1111 additions and 893 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -115,6 +115,7 @@ fn cyclic_dag_is_rejected_at_submit() {
on: 1,
when: DepWhen::AfterOk,
}],
parent: None,
},
NodeSpec {
agent: "agent-a".to_owned(),
@ -123,6 +124,7 @@ fn cyclic_dag_is_rejected_at_submit() {
on: 0,
when: DepWhen::AfterOk,
}],
parent: None,
},
];
assert!(q.submit(spec).is_err(), "cyclic spec must be refused");
@ -140,6 +142,7 @@ fn unknown_dep_is_rejected_at_submit() {
on: 9,
when: DepWhen::AfterOk,
}],
parent: None,
}];
assert!(q.submit(spec).is_err());
}
@ -180,13 +183,16 @@ fn build_slot_serializes_nix_heavy_nodes() {
assert_eq!(first.dag_id, a);
assert_eq!(first.kind.as_str(), "prebuild");
q.complete_node(a, first.node_id, Ok(()));
// With the slot free again, FIFO gives... a's StopForUpdate is
// slot-free (lease) and b's Prebuild takes the slot — both run.
// Uniform hold: agent-a keeps the build slot across its whole build chain
// (Swap re-enters it), so a's StopForUpdate (lease, slot-free) runs but b's
// Prebuild must wait for a's slot-needers (through Swap) to finish.
let claims = q.claim_ready();
let kinds: Vec<(u64, &str)> = claims.iter().map(|c| (c.dag_id, c.kind.as_str())).collect();
assert!(kinds.contains(&(a, "stop_for_update")));
assert!(kinds.contains(&(b, "prebuild")));
assert_eq!(claims.len(), 2);
assert_eq!(kinds, vec![(a, "stop_for_update")]);
assert!(
!kinds.iter().any(|&(d, _)| d == b),
"b's build waits — slot held across a's chain"
);
}
#[test]
@ -208,9 +214,27 @@ fn fifo_fairness_for_the_slot() {
let first = claim_one(&q);
assert_eq!(first.dag_id, a, "submit order wins the slot");
q.complete_node(a, first.node_id, Ok(()));
let next: Vec<u64> = q.claim_ready().iter().map(|cl| cl.dag_id).collect();
assert!(next.contains(&b), "b's prebuild before c's");
assert!(!next.contains(&c));
// Uniform hold: the slot stays with agent-a until its Swap (the last
// slot-needer) completes. Drive a's chain; the moment its slot frees,
// submit order (b before c) wins it.
let mut freed_to = None;
for _ in 0..6 {
let claims = q.claim_ready();
if let Some(nb) = claims.iter().find(|cl| cl.dag_id == b || cl.dag_id == c) {
freed_to = Some(nb.dag_id);
break;
}
for cl in claims {
if cl.dag_id == a {
q.complete_node(a, cl.node_id, Ok(()));
}
}
}
assert_eq!(
freed_to,
Some(b),
"b's prebuild wins the freed slot before c's"
);
}
// ---- per-agent lease ----
@ -234,18 +258,31 @@ fn lease_serializes_two_lifecycle_dags_for_same_agent() {
let first = claim_one(&q);
assert_eq!(first.dag_id, restart);
assert_eq!(first.kind.as_str(), "stop_for_update");
assert!(first.lease_acquired);
q.complete_node(restart, first.node_id, Ok(()));
// Same DAG keeps the lease through the tail Reconcile.
// Same DAG keeps the lease through the tail Reconcile (re-entered from the
// dep graph — no fresh acquire), since stop's Reconcile can't re-enter it.
let second = claim_one(&q);
assert_eq!(second.dag_id, restart);
assert_eq!(second.kind.as_str(), "reconcile");
assert!(!second.lease_acquired, "lease already held by this DAG");
q.complete_node(restart, second.node_id, Ok(()));
// Restart terminal → lease released → stop's Reconcile runs.
let third = claim_one(&q);
assert_eq!(third.dag_id, stop);
// 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");
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);
}
@ -288,8 +325,20 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
.expect("reconcile claim")
.clone();
q.complete_node(stop, reconcile.node_id, Ok(()));
let next = claim_one(&q);
assert_eq!(next.kind.as_str(), "stop_for_update");
// stop's Reconcile done → its lease frees (rebuild's StopForUpdate unblocks)
// and its terminal node becomes ready; both surface in the same pass.
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")
.expect("rebuild StopForUpdate unblocked once the lease frees");
assert_eq!(sfu.agent, "agent-a");
}
#[test]
@ -315,16 +364,16 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
// lease (no contention across distinct agents), all inside the single DAG.
let claims = q.claim_ready();
assert!(claims.iter().all(|c| c.dag_id == id));
let mut heads: Vec<(&str, &str, bool)> = claims
let mut heads: Vec<(&str, &str)> = claims
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str(), c.lease_acquired))
.map(|c| (c.agent.as_str(), c.kind.as_str()))
.collect();
heads.sort_unstable();
assert_eq!(
heads,
vec![
("agent-a", "stop_for_update", true),
("agent-b", "stop_for_update", true),
("agent-a", "stop_for_update"),
("agent-b", "stop_for_update"),
],
"both per-agent subgraphs start concurrently, each acquiring its own lease"
);
@ -387,17 +436,14 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
assert_eq!(q.snapshot().len(), 1);
let claims = q.claim_ready();
assert!(claims.iter().all(|c| c.dag_id == id));
let mut heads: Vec<(&str, &str, bool)> = claims
let mut heads: Vec<(&str, &str)> = claims
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str(), c.lease_acquired))
.map(|c| (c.agent.as_str(), c.kind.as_str()))
.collect();
heads.sort_unstable();
assert_eq!(
heads,
vec![
("agent-a", "set_wanted", true),
("agent-b", "set_wanted", true),
],
vec![("agent-a", "set_wanted"), ("agent-b", "set_wanted")],
"both per-agent stop subgraphs start concurrently, each on its own lease"
);
}
@ -521,6 +567,7 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
fanout: None,
},
deps: Vec::new(),
parent: None,
}],
};
let id = submit(&q, spec);
@ -532,8 +579,8 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
// tracks any drift in that builder's root-first (`base = 0`) shape.
let subgraph = |agent: &str| templates::rebuild_nodes(agent, true, 0);
// Must append BEFORE completing the emitter (the documented contract).
q.append_subgraph(id, subgraph("a"), emitter.node_id);
q.append_subgraph(id, subgraph("b"), emitter.node_id);
q.append_subgraph(id, &subgraph("a"), emitter.node_id);
q.append_subgraph(id, &subgraph("b"), emitter.node_id);
q.complete_node(id, emitter.node_id, Ok(()));
// Still ONE DAG; both subgraph roots become ready once the emitter is
// Done (rooted on it), each on its own agent lease.
@ -580,7 +627,7 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() {
for agent in ["alice", "bob"] {
q.append_subgraph(
id,
templates::rebuild_nodes(agent, false, 0),
&templates::rebuild_nodes(agent, false, 0),
meta_lock.node_id,
);
}
@ -728,6 +775,11 @@ fn cancel_clears_queued_dag() {
let id = submit(&q, rebuild("agent-a", "r"));
assert!(q.cancel(id));
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());
}
@ -743,22 +795,25 @@ fn cancel_refuses_running_dag() {
// ---- terminal reporting + lease release ----
#[test]
fn terminal_dag_reported_exactly_once_and_lease_released() {
fn terminal_node_runs_after_work_settles_and_lease_released() {
let q = JobQueue::new(1);
let id = submit(&q, restart_online(&["agent-a"], false, "r"));
// restart = StopForUpdate → Reconcile; not terminal until the last
// node completes.
// restart = StopForUpdate → Reconcile; the terminal node (which weak-deps on
// the chain tail) isn't runnable until the whole chain is terminal.
let stop = claim_one(&q);
q.complete_node(id, stop.node_id, Ok(()));
assert!(q.drain_terminal().is_empty(), "dag not terminal yet");
let rec = claim_one(&q);
assert_eq!(rec.kind.as_str(), "reconcile");
q.complete_node(id, rec.node_id, Ok(()));
let reports = q.drain_terminal();
assert_eq!(reports.len(), 1);
assert_eq!(reports[0].dag_id, id);
assert_eq!(reports[0].state, State::Done);
assert!(q.drain_terminal().is_empty(), "reported exactly once");
// Lease released: a new DAG for the agent can claim immediately.
// 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");
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.
let next = submit(
&q,
templates::reconcile_only(
@ -771,31 +826,38 @@ fn terminal_dag_reported_exactly_once_and_lease_released() {
);
let c = claim_one(&q);
assert_eq!(c.dag_id, next);
assert!(c.lease_acquired);
}
/// 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.
#[test]
fn cancelled_dag_reports_terminal_once() {
fn cancelled_dag_finalizes_with_terminal_rollup() {
let q = JobQueue::new(1);
let id = submit(
&q,
templates::approval_deploy("agent-a", 7, "approval #7".to_owned()),
);
assert!(q.cancel(id));
let reports = q.drain_terminal();
assert_eq!(reports.len(), 1);
assert_eq!(reports[0].dag_id, id);
assert_eq!(reports[0].state, State::Cancelled);
assert_eq!(reports[0].approval_id, Some(7));
// Never re-reported by later activity.
// 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");
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"));
let c = claim_one(&q);
assert_eq!(c.dag_id, other);
q.complete_node(other, c.node_id, Err("boom".to_owned()));
assert!(q.drain_terminal().iter().all(|t| t.dag_id != id));
assert_eq!(
q.terminal_summary(id).map(|t| t.state),
Some(State::Cancelled)
);
}
// ---- steps, build logs, history ----
@ -804,7 +866,10 @@ fn cancelled_dag_reports_terminal_once() {
fn set_step_only_on_running_and_signals_change() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
assert!(!q.set_step(id, 0, "too early"), "queued node refuses step");
assert!(
!q.set_step_running(id, "too early"),
"no running node yet → refused"
);
let c = claim_one(&q);
assert!(q.set_step(id, c.node_id, "nix build"));
assert!(
@ -823,7 +888,10 @@ fn set_step_only_on_running_and_signals_change() {
fn set_build_log_id_links_running_node() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
assert!(!q.set_build_log_id(id, 0, 41), "queued node refuses log id");
assert!(
!q.set_build_log_id_running(id, 41),
"no running node yet → refused"
);
let c = claim_one(&q);
assert!(q.set_build_log_id(id, c.node_id, 42));
assert!(q.set_build_log_id_running(id, 43));
@ -849,6 +917,11 @@ fn history_evicts_old_terminals_per_template() {
);
let c = claim_one(&q);
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