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(job-queue): promote the meta-repo deploy window to a queue resource

Browse source 
The two-phase approval deploy keeps a bumped `flake.lock` staged
uncommitted for the whole container build, so no other meta mutation may
land inside that span — until now enforced by a process-global
`meta::exclusive()` mutex held inside each executor fn.

A `MutexGuard` cannot outlive the fn that takes it, which is what blocks
decomposing the opaque `ApprovalDeploy` node into scheduler-visible
sub-nodes: the window has to span them. Replace the mutex with
`Resource::MetaWindow`, a global capacity-1 queue resource declared by
every meta-mutating node kind (`NodeKind::needs_meta_window`). Resources
are held by a subtree root across its whole subtree, so a later increment
can hang the deploy's phases under one window-holding parent.

Same global serialisation as before, and the scheduler now blocks a node
from being claimed rather than parking a worker on a mutex.

Split the rebuild's meta preamble out of `Prebuild` into a new `MetaSync`
node. `Prebuild` must NOT hold the window: the old mutex was deliberately
scoped to drop before the multi-minute toplevel build, which only reads
the store, and a cap-1 global held across it would serialise every
agent's rebuild behind every other's. `MetaSync` is a sibling root that
`Prebuild` deps `AfterOk` on — not its parent, since a parent's resource
covers its whole subtree and would reintroduce exactly that problem.

Queue tests: shape assertions gain the extra node, which is the point of
the change (phases become nodes). The concurrency invariants are intact
but observed one step later — the `MetaSync` heads take turns on the
window, exactly as the runtime mutex made them, so those tests now
complete the heads before asserting that the prebuilds overlap.
This commit is contained in:
atlas 2026-07-25 20:00:29 +02:00 • committed by mara
commit dfadacd45f
8 changed files with 311 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -176,6 +176,7 @@ fn rebuild_chain_claims_in_dep_order() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
for expected in [
"meta_sync",
"prebuild",
"stop_for_update",
"swap",
@ -201,15 +202,29 @@ fn build_slot_serializes_nix_heavy_nodes() {
let q = JobQueue::new(1);
let a = submit(&q, rebuild("agent-a", "r"));
let b = submit(&q, rebuild("agent-b", "r"));
let first = claim_one(&q); // a's Prebuild takes the only slot
assert_eq!(first.dag_id, a);
assert_eq!(first.kind.as_str(), "prebuild");
q.complete_node(a, first.node_id, Ok(()));
// The rebuild heads are `MetaSync` (slot-free, but serialized on the
// global meta window), so drive each chain's head out of the way first.
let head_a = claim_one(&q);
assert_eq!(head_a.dag_id, a);
assert_eq!(head_a.kind.as_str(), "meta_sync");
q.complete_node(a, head_a.node_id, Ok(()));
// a's Prebuild takes the only slot; b's MetaSync is free to run beside it
// (different resources), but b's Prebuild is not.
let claims = q.claim_ready();
let mut kinds: Vec<(u64, &str)> = claims.iter().map(|c| (c.dag_id, c.kind.as_str())).collect();
kinds.sort_unstable();
assert_eq!(kinds, vec![(a, "prebuild"), (b, "meta_sync")]);
for c in &claims {
q.complete_node(c.dag_id, c.node_id, Ok(()));
}
// 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();
let kinds: Vec<(u64, &str)> = q
.claim_ready()
.iter()
.map(|c| (c.dag_id, c.kind.as_str()))
.collect();
assert_eq!(kinds, vec![(a, "stop_for_update")]);
assert!(
!kinds.iter().any(|&(d, _)| d == b),
@ -222,9 +237,23 @@ fn two_build_slots_run_two_prebuilds() {
let q = JobQueue::new(2);
submit(&q, rebuild("agent-a", "r"));
submit(&q, rebuild("agent-b", "r"));
let claims = q.claim_ready();
assert_eq!(claims.len(), 2, "two slots → two concurrent prebuilds");
assert!(claims.iter().all(|c| c.kind.as_str() == "prebuild"));
// Each rebuild's head `MetaSync` holds the cap-1 global meta window, so the
// two heads take turns — exactly the serialization the old runtime
// `meta::exclusive()` mutex imposed inside the prebuild executor. What must
// NOT serialize is the build itself: complete only the meta heads and watch
// both prebuilds end up in flight together, neither of them completed.
let mut prebuilds = Vec::new();
for _ in 0..3 {
for c in q.claim_ready() {
if c.kind.as_str() == "meta_sync" {
q.complete_node(c.dag_id, c.node_id, Ok(()));
} else {
prebuilds.push(c);
}
}
}
assert_eq!(prebuilds.len(), 2, "two slots → two concurrent prebuilds");
assert!(prebuilds.iter().all(|c| c.kind.as_str() == "prebuild"));
}
#[test]
@ -312,12 +341,23 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
None,
),
);
// Prebuild is lease-exempt: the stop's Reconcile takes the lease
// Both DAGs' heads are lease-independent of each other: the rebuild's
// MetaSync (meta window) and the stop's Reconcile (agent lease).
let heads = q.claim_ready();
let head_kinds: Vec<&str> = heads.iter().map(|c| c.kind.as_str()).collect();
assert!(head_kinds.contains(&"meta_sync"));
assert!(head_kinds.contains(&"reconcile"));
let meta_sync = heads
.iter()
.find(|c| c.kind.as_str() == "meta_sync")
.expect("meta_sync claim")
.clone();
q.complete_node(meta_sync.dag_id, meta_sync.node_id, Ok(()));
// Prebuild is lease-exempt: the stop's Reconcile keeps the lease
// and runs concurrently with the rebuild's out-of-band nix build.
let claims = q.claim_ready();
let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect();
assert!(kinds.contains(&"prebuild"));
assert!(kinds.contains(&"reconcile"));
// But the rebuild's StopForUpdate must then wait for the stop DAG
// to finish (lease).
let prebuild = claims
@ -330,7 +370,7 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
q.claim_ready().is_empty(),
"StopForUpdate blocked while stop DAG holds the lease"
);
let reconcile = claims
let reconcile = heads
.iter()
.find(|c| c.kind.as_str() == "reconcile")
.expect("reconcile claim")
@ -484,7 +524,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
head_agents.sort_unstable();
assert_eq!(head_agents, vec!["fresh", "stale"]);
// Complete both heads; the fresh agent then reconciles directly while
// the stale agent's subgraph is the rebuild chain (prebuild first).
// the stale agent's subgraph is the rebuild chain (meta_sync first).
for c in &heads {
q.complete_node(id, c.node_id, Ok(()));
}
@ -496,7 +536,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
kinds.sort_unstable();
assert_eq!(
kinds,
vec![("fresh", "reconcile"), ("stale", "prebuild")],
vec![("fresh", "reconcile"), ("stale", "meta_sync")],
"fresh agent starts directly; stale agent rebuilds first, all in one DAG"
);
}
@ -578,30 +618,48 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
let emitter = claim_one(&q);
assert_eq!(emitter.kind.as_str(), "meta_lock");
// Two independent per-agent subgraphs — the REAL production shape the
// sweep MetaLock grows (`rebuild_nodes(_, true, 0)`: root Prebuild →
// StopForUpdate → Swap → Reconcile, local 0-based deps), so this test
// tracks any drift in that builder's root-first (`base = 0`) shape.
// sweep MetaLock grows (`rebuild_nodes(_, true, 0)`: root MetaSync → root
// Prebuild → StopForUpdate → Swap → Reconcile, local 0-based deps), so this
// test 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.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.
// Done (rooted on it), each on its own agent lease. Their `MetaSync` heads
// take turns on the cap-1 global meta window, so drain those first — what
// must be concurrent is the builds.
assert_eq!(q.snapshot().len(), 1);
let next = q.claim_ready();
let mut kinds: Vec<(&str, &str)> = next
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str()))
.collect();
let mut kinds = drain_meta_syncs(&q, id);
kinds.sort_unstable();
assert_eq!(
kinds,
vec![("a", "prebuild"), ("b", "prebuild")],
vec![
("a".to_owned(), "prebuild".to_owned()),
("b".to_owned(), "prebuild".to_owned())
],
"both rebuild subgraphs root on the emitter and run concurrently in one DAG"
);
}
/// Complete every `MetaSync` head the queue offers (they take turns on the
/// cap-1 global meta window) and return whatever else got claimed alongside
/// them, as `(agent, kind)` pairs left in flight.
fn drain_meta_syncs(q: &JobQueue, dag: u64) -> Vec<(String, String)> {
let mut rest = Vec::new();
for _ in 0..3 {
for c in q.claim_ready() {
if c.kind.as_str() == "meta_sync" {
q.complete_node(dag, c.node_id, Ok(()));
} else {
rest.push((c.agent.clone(), c.kind.as_str().to_owned()));
}
}
}
rest
}
#[test]
fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() {
// The meta-update `MetaLock` grows one rebuild subgraph per affected
@ -637,17 +695,18 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() {
}
q.complete_node(id, meta_lock.node_id, Ok(()));
// Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root
// on the MetaLock, each on its own agent lease.
// on the MetaLock, each on its own agent lease. The per-agent `MetaSync`
// heads serialize on the global meta window (they commit to the meta repo);
// the builds behind them do not.
assert_eq!(q.snapshot().len(), 1);
let next = q.claim_ready();
let mut kinds: Vec<(&str, &str)> = next
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str()))
.collect();
let mut kinds = drain_meta_syncs(&q, id);
kinds.sort_unstable();
assert_eq!(
kinds,
vec![("alice", "prebuild"), ("bob", "prebuild")],
vec![
("alice".to_owned(), "prebuild".to_owned()),
("bob".to_owned(), "prebuild".to_owned())
],
"cascade rebuilds grow in the meta-update DAG, concurrent per agent"
);
}
@ -658,7 +717,11 @@ fn meta_update_carries_rebuilding_transient_and_grows_cascade_in_dag() {
fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
let meta_sync = claim_one(&q);
assert_eq!(meta_sync.kind.as_str(), "meta_sync");
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()));
// StopForUpdate + Swap are cancelled (AfterOk on a failed chain);
// the AfterAny Reconcile still runs once Swap is terminal.
@ -702,7 +765,8 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
fn swap_failure_still_runs_reconcile() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
for _ in 0..2 {
// meta_sync + prebuild + stop_for_update
for _ in 0..3 {
let c = claim_one(&q);
q.complete_node(id, c.node_id, Ok(()));
}
@ -735,8 +799,8 @@ fn swap_failure_still_runs_reconcile() {
fn swap_ok_runs_post_swap_before_reconcile() {
let q = JobQueue::new(1);
let id = submit(&q, rebuild("agent-a", "r"));
// prebuild + stop_for_update
for _ in 0..2 {
// meta_sync + prebuild + stop_for_update
for _ in 0..3 {
let c = claim_one(&q);
q.complete_node(id, c.node_id, Ok(()));
}
@ -979,13 +1043,25 @@ fn graceful_signal_and_drain_hold_no_build_slot() {
submit(&q, rebuild("builder", "slot hog"));
submit(&q, stop_online(&["agent-a"], true, "g"));
submit(&q, stop_online(&["agent-b"], true, "g"));
let claims = q.claim_ready();
let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect();
// All three DAG heads are build-slot-exempt, so they run at once.
let heads = q.claim_ready();
let kinds: Vec<&str> = heads.iter().map(|c| c.kind.as_str()).collect();
assert_eq!(kinds, vec!["meta_sync", "set_wanted", "set_wanted"]);
for c in &heads {
q.complete_node(c.dag_id, c.node_id, Ok(()));
}
// Now the rebuild's Prebuild holds the single slot — and both graceful
// stops still proceed to their Signal beside it.
let kinds: Vec<&str> = q
.claim_ready()
.iter()
.map(|c| c.kind.as_str())
.collect::<Vec<_>>();
assert_eq!(
kinds,
vec!["prebuild", "set_wanted", "set_wanted"],
"both agents' graceful-stop heads (SetWanted, build-slot-exempt) run \
while the slot is held; their signals follow"
vec!["prebuild", "signal", "signal"],
"both agents' graceful-stop signals (build-slot-exempt) run while the \
rebuild holds the slot"
);
}
@ -1023,6 +1099,7 @@ fn perm_change_shape_prefixes_rebuild_chain() {
);
for expected in [
"write_perm_file",
"meta_sync",
"prebuild",
"stop_for_update",
"swap",