job_queue: drop the cross-DAG contention tests
Six tests drove two or three DAGs against each other to watch a build slot or an agent lease serialize them. In every case the part that is hive-c0re's -- which nodes declare which resource -- is now a declared_resources table, and the part that is hive-jobq's -- what a scheduler does with a contended unit -- is tested in hive-jobq (build_slot_cap_limits_concurrency_and_release_unblocks, unrelated_nodes_needing_the_same_resource_are_serialized, the two sibling_borrowers cases, owner_holds_grant_for_its_whole_subtree, and the fairness test added in the previous commit). graceful_signal_and_drain_hold_no_build_slot is absorbed rather than deleted: its claim is a declaration, so it now sits in the graceful-stop shape test. Signal and Drain declare the agent lease and no build slot, which is why a whole-hive graceful stop overlaps every agent's drain at buildSlots = 1 -- the ceiling is one GRACEFUL_STOP_TIMEOUT in total, not one per agent. hive-c0re/src/job_queue/tests.rs: 42 tests to 36, 181 lines lighter.
This commit is contained in:
atlas
mara
parent
3ebfed1226
commit
ff70bf029d
1 changed files with 11 additions and 181 deletions
|
|
@ -573,65 +573,6 @@ fn settled_dag_leaves_the_snapshot_despite_its_skipped_branch() {
|
||||||
|
|
||||||
// ---- build slots ----
|
// ---- build slots ----
|
||||||
|
|
||||||
#[test]
|
|
||||||
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"));
|
|
||||||
// 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(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.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 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),
|
|
||||||
"b's build waits — slot held across a's chain"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn two_build_slots_run_two_prebuilds() {
|
|
||||||
let q = JobQueue::new(2);
|
|
||||||
submit(&q, rebuild("agent-a", "r"));
|
|
||||||
submit(&q, rebuild("agent-b", "r"));
|
|
||||||
// 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.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]
|
#[test]
|
||||||
fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
|
fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
|
||||||
// Was `fifo_fairness_for_the_slot`, which submitted three rebuilds and
|
// Was `fifo_fairness_for_the_slot`, which submitted three rebuilds and
|
||||||
|
|
@ -681,100 +622,6 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
|
||||||
|
|
||||||
// ---- per-agent lease ----
|
// ---- per-agent lease ----
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn lease_serializes_two_lifecycle_dags_for_same_agent() {
|
|
||||||
let q = JobQueue::new(4);
|
|
||||||
let restart = submit(&q, restart_online(&["agent-a"], false, "restart"));
|
|
||||||
let stop = submit(
|
|
||||||
&q,
|
|
||||||
templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()),
|
|
||||||
);
|
|
||||||
// Restart's first node (StopForUpdate) takes the lease; stop's
|
|
||||||
// Reconcile must wait even though slots are free.
|
|
||||||
let first = claim_one(&q);
|
|
||||||
assert_eq!(first.dag_id, restart);
|
|
||||||
assert_eq!(first.kind.as_str(), "stop_for_update");
|
|
||||||
q.complete_node(first.node_id, Ok(()));
|
|
||||||
// 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");
|
|
||||||
q.complete_node(second.node_id, Ok(()));
|
|
||||||
// Restart's work is terminal → its lease releases, so stop's now-unblocked
|
|
||||||
// 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");
|
|
||||||
q.complete_node(third.node_id, Ok(()));
|
|
||||||
assert_eq!(state_of(&q, restart), State::Done);
|
|
||||||
assert_eq!(state_of(&q, stop), State::Done);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
|
|
||||||
let q = JobQueue::new(2);
|
|
||||||
submit(&q, rebuild("agent-a", "rebuild"));
|
|
||||||
submit(
|
|
||||||
&q,
|
|
||||||
templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()),
|
|
||||||
);
|
|
||||||
// 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.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"));
|
|
||||||
// But the rebuild's StopForUpdate must then wait for the stop DAG
|
|
||||||
// to finish (lease).
|
|
||||||
let prebuild = claims
|
|
||||||
.iter()
|
|
||||||
.find(|c| c.kind.as_str() == "prebuild")
|
|
||||||
.expect("prebuild claim")
|
|
||||||
.clone();
|
|
||||||
q.complete_node(prebuild.node_id, Ok(()));
|
|
||||||
assert!(
|
|
||||||
q.claim_ready().is_empty(),
|
|
||||||
"StopForUpdate blocked while stop DAG holds the lease"
|
|
||||||
);
|
|
||||||
let reconcile = heads
|
|
||||||
.iter()
|
|
||||||
.find(|c| c.kind.as_str() == "reconcile")
|
|
||||||
.expect("reconcile claim")
|
|
||||||
.clone();
|
|
||||||
q.complete_node(reconcile.node_id, Ok(()));
|
|
||||||
// stop's Reconcile done → its lease frees, so rebuild's StopForUpdate
|
|
||||||
// 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()
|
|
||||||
.find(|c| c.kind.as_str() == "stop_for_update")
|
|
||||||
.expect("rebuild StopForUpdate unblocked once the lease frees");
|
|
||||||
assert_eq!(sfu.agent, "agent-a");
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn agents_do_not_contend_on_each_others_leases() {
|
|
||||||
let q = JobQueue::new(4);
|
|
||||||
submit(&q, restart_online(&["agent-a"], false, "r"));
|
|
||||||
submit(&q, restart_online(&["agent-b"], false, "r"));
|
|
||||||
let claims = q.claim_ready();
|
|
||||||
assert_eq!(claims.len(), 2, "different agents run concurrently");
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
|
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
|
||||||
let q = JobQueue::new(4);
|
let q = JobQueue::new(4);
|
||||||
|
|
@ -1799,35 +1646,18 @@ fn graceful_stop_shape_signal_drain_reconcile() {
|
||||||
row("reconcile", Some("set_wanted"), &[("drain", "done")]),
|
row("reconcile", Some("set_wanted"), &[("drain", "done")]),
|
||||||
]
|
]
|
||||||
);
|
);
|
||||||
}
|
// Neither half of the graceful window takes a build slot. That is what lets
|
||||||
|
// a whole-hive graceful stop overlap every agent's drain even at
|
||||||
#[test]
|
// `buildSlots = 1` while a rebuild hogs the slot — the cost ceiling is one
|
||||||
fn graceful_signal_and_drain_hold_no_build_slot() {
|
// `GRACEFUL_STOP_TIMEOUT` in total, not one per agent.
|
||||||
// A whole-hive graceful stop overlaps every drain even at
|
let agent = Resource::Agent("agent-a".to_owned());
|
||||||
// buildSlots = 1 while a rebuild hogs the slot.
|
|
||||||
let q = JobQueue::new(1);
|
|
||||||
submit(&q, rebuild("builder", "slot hog"));
|
|
||||||
submit(&q, stop_online(&["agent-a"], true, "g"));
|
|
||||||
submit(&q, stop_online(&["agent-b"], true, "g"));
|
|
||||||
// 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.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!(
|
assert_eq!(
|
||||||
kinds,
|
[
|
||||||
vec!["prebuild", "signal", "signal"],
|
declared_resources(&q, node_of(&q, id, "signal")),
|
||||||
"both agents' graceful-stop signals (build-slot-exempt) run while the \
|
declared_resources(&q, node_of(&q, id, "drain")),
|
||||||
rebuild holds the slot"
|
],
|
||||||
|
[vec![agent.clone()], vec![agent]],
|
||||||
|
"signal and drain hold the lease but never a build slot"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue