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

wip(#3001): test helpers off the container id

Browse source 
submit() -> insert() in tests, and the two shape walkers lose their dag
param: with no container there is no per-DAG root to filter on, nothing
to exclude (every node is real work now), and a group root genuinely has
parent = None, so the container-parent normalisation goes too. Each test
builds a fresh JobQueue, so "the DAG" is "the graph".

20 errors remain, all in tests.rs, and they are the point: changing the
helper's type from u64 to () made every site that consumed the container
id light up as `expected u64, found ()`. A type error is an exhaustive
grep -- ten helpers take a dag id, not the three I had measured.

state_of(q, dag_id) is not mechanical: it read the DAG's ROLLED-UP state,
which was the container node's own. That makes it the second consumer of
the container-as-roll-up-point, alongside hivectl's queued_dags poll.
Both want the same answer, so it waits on the same ruling.
This commit is contained in:
atlas 2026-08-04 11:53:02 +02:00 • committed by mara
commit be4763678b
1 changed files with 79 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -20,9 +20,15 @@ use super::*;
/// Submit a declared shape with the metadata every mechanics test uses.
/// `Source::Manual` because none of these exercise provenance — the tests that
/// do name their own source at the call site.
fn submit(q: &JobQueue, reason: &str, declare: impl FnOnce(&JobBuilder)) -> u64 {
q.submit(Source::Manual, reason.to_owned(), declare)
.expect("valid shape")
/// Insert a declared job, naming nothing — the shape assertions read the whole
/// graph. A test that needs a handle calls `q.insert` directly and names the
/// node it cares about.
fn insert(q: &JobQueue, declare: impl FnOnce(&JobBuilder)) {
q.insert(|b| {
declare(b);
Vec::new()
})
.expect("valid shape");
}
fn ident(s: &str) -> hive_types::Ident {
@ -104,25 +110,27 @@ fn when_tag(when: hive_jobq::DepWhen) -> String {
/// keeps the assertion on c0re's own product. Whether the scheduler then
/// *honours* those edges — runs a chain serially, holds a grant across a
/// subtree — is `hive_jobq`'s property and is tested in `hive_jobq`.
fn declared_shape(q: &JobQueue, dag: u64) -> Vec<Declared> {
declared_shape_filtered(q, dag, &|_| true)
fn declared_shape(q: &JobQueue) -> Vec<Declared> {
declared_shape_filtered(q, &|_| true)
}
fn declared_shape_filtered(
q: &JobQueue,
dag: u64,
keep: &dyn Fn(&NodeKind) -> bool,
) -> Vec<Declared> {
/// Every node in the graph, since each test inserts into a fresh [`JobQueue`].
///
/// This used to take a DAG id and filter by `root_of(n) == Some(container)`.
/// With no container node there is no per-DAG root to filter on — and nothing
/// to exclude either, because every node in the graph is now real work. Tests
/// that insert more than one job name a node per job and assert on the ids
/// [`JobQueue::insert`] hands back.
fn declared_shape_filtered(q: &JobQueue, keep: &dyn Fn(&NodeKind) -> bool) -> Vec<Declared> {
let sched = q.sched().lock().expect("job_queue mutex poisoned");
let graph = sched.graph();
let root = graph.resolve_id(dag).expect("dag id is a real node id");
let kind_of = |id: NodeId| graph.node(id).map(|n| n.payload.as_str());
graph
.nodes()
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && keep(&n.payload))
.filter(|n| keep(&n.payload))
.map(|n| Declared {
kind: n.payload.as_str(),
parent: n.parent.filter(|p| *p != root).and_then(kind_of),
parent: n.parent.and_then(kind_of),
after: n
.deps
.iter()
@ -140,19 +148,18 @@ fn declared_shape_filtered(
/// Panics unless there is exactly one — every caller is about a shape where the
/// kind is unique, so two would mean the assertion had quietly stopped being
/// about the node the test names.
fn node_of(q: &JobQueue, dag: u64, kind: &str) -> hive_jobq::NodeId {
fn node_of(q: &JobQueue, kind: &str) -> hive_jobq::NodeId {
let sched = q.sched().lock().expect("job_queue mutex poisoned");
let graph = sched.graph();
let root = graph.resolve_id(dag).expect("dag id is a real node id");
let mut found: Vec<_> = graph
.nodes()
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
.filter(|n| n.payload.as_str() == kind)
.map(|n| n.id)
.collect();
assert_eq!(
found.len(),
1,
"expected exactly one {kind} node in the dag"
"expected exactly one {kind} node in the graph"
);
found.pop().expect("checked above")
}
@ -312,8 +319,8 @@ fn dag_count(q: &JobQueue) -> usize {
#[test]
fn submit_assigns_distinct_ids() {
let q = JobQueue::new(1);
let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let second = submit(&q, "second", |builder| rebuild(builder, "agent-b"));
let first = insert(&q, |builder| rebuild(builder, "agent-a"));
let second = insert(&q, |builder| rebuild(builder, "agent-b"));
assert_ne!(first, second);
assert_eq!(dag_count(&q), 2);
}
@ -326,8 +333,8 @@ fn submit_assigns_distinct_ids() {
#[test]
fn identical_resubmit_is_a_distinct_dag() {
let q = JobQueue::new(1);
let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let resubmit = submit(&q, "again", |builder| rebuild(builder, "agent-a"));
let first = insert(&q, |builder| rebuild(builder, "agent-a"));
let resubmit = insert(&q, |builder| rebuild(builder, "agent-a"));
assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG");
assert_eq!(dag_count(&q), 2);
}
@ -335,9 +342,9 @@ fn identical_resubmit_is_a_distinct_dag() {
#[test]
fn distinct_submits_never_collapse() {
let q = JobQueue::new(1);
let rebuild_a = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let rebuild_b = submit(&q, "r", |builder| rebuild(builder, "agent-b"));
let restart_a = submit(&q, "r", |builder| {
let rebuild_a = insert(&q, |builder| rebuild(builder, "agent-a"));
let rebuild_b = insert(&q, |builder| rebuild(builder, "agent-b"));
let restart_a = insert(&q, |builder| {
restart_online(builder, &["agent-a"], false);
});
assert_ne!(rebuild_a, rebuild_b);
@ -357,8 +364,8 @@ fn resubmit_while_running_is_new_dag() {
// swallowed" is the scenario people worry about, and a reader looking for
// it should find it.
let q = JobQueue::new(1);
let a = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let again = submit(&q, "config bumped during build", |builder| {
let a = insert(&q, |builder| rebuild(builder, "agent-a"));
let again = insert(&q, |builder| {
rebuild(builder, "agent-a");
});
assert_ne!(a, again);
@ -401,9 +408,9 @@ fn rebuild_chain_is_declared_serial() {
// The non-graceful shape asserted here has no quiesce chain, so nothing here
// is concurrent — but the name would mislead about the graceful one.
let q = JobQueue::new(1);
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![
row("meta_sync", None, &[]),
// The brace: holds the lease + slot for everything nested below it.
@ -481,7 +488,7 @@ fn graceful_rebuild_chain_drains_before_stopping() {
// the quiesce chain runs beside the build or nested under it, so a
// kind-only assertion cannot see the bug this shape exists to fix.
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![
row("meta_sync", None, &[]),
row("agent_window", None, &[("meta_sync", "done")]),
@ -530,11 +537,11 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
// job keeps its nodes to itself and inserts them, so what it built is
// observable where it matters — in what the scheduler runs.
let q = JobQueue::new(1);
let id = submit(&q, "manual", |builder| {
let id = insert(&q, |builder| {
templates::rebuild_nodes(builder, "agent-a", true, None);
});
assert_eq!(
declared_shape(&q, id)
declared_shape(&q)
.iter()
.map(|d| d.kind)
.collect::<Vec<_>>(),
@ -567,8 +574,8 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
// `a_contended_resource_goes_to_the_oldest_waiter`. What is c0re's is
// *which* nodes contend in the first place — a declaration, asserted here.)
let q = JobQueue::new(1);
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let res = |kind: &str| declared_resources(&q, node_of(&q, id, kind));
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
let res = |kind: &str| declared_resources(&q, node_of(&q, kind));
let agent = || Resource::Agent("agent-a".to_owned());
assert_eq!(
@ -612,7 +619,7 @@ fn rebuild_chain_declares_its_resources_on_the_brace() {
#[test]
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = submit(&q, "hive-wide", |builder| {
let id = insert(&q, |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
});
// A hive-wide restart is ONE DAG, not one-per-agent.
@ -623,7 +630,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
// Those two declared facts are what "they run concurrently" *means* here —
// that a scheduler then does run independent, resource-disjoint roots at
// once is hive_jobq's property, tested there.
let heads: Vec<_> = declared_shape(&q, id)
let heads: Vec<_> = declared_shape(&q)
.into_iter()
.filter(|d| d.kind == "stop_for_update")
.collect();
@ -648,7 +655,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
#[test]
fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = submit(&q, "hive-wide stop", |builder| {
let id = insert(&q, |builder| {
stop_online(builder, &["agent-a", "agent-b"], false);
});
// A hive-wide stop is ONE DAG, not one-per-agent.
@ -657,7 +664,7 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
// is a group root with no node-deps, holding only its own agent's lease.
// Independent roots on disjoint resources is what "concurrently" means at
// this layer — the running of them is hive_jobq's.
let heads: Vec<_> = declared_shape(&q, id)
let heads: Vec<_> = declared_shape(&q)
.into_iter()
.filter(|d| d.kind == "set_wanted")
.collect();
@ -681,7 +688,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
let q = JobQueue::new(4);
// fresh: offline + not stale → SetWanted → Reconcile.
// stale: offline + stale → SetWanted → «rebuild subgraph».
let id = submit(&q, "hive-wide start", |builder| {
let id = insert(&q, |builder| {
power::start_nodes(
builder,
&[
@ -741,13 +748,13 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
// read and node exec.
let q = JobQueue::new(4);
// Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
let stop = submit(&q, "stop down", |builder| {
let stop = insert(&q, |builder| {
power::stop_nodes(builder, &[("down".to_owned(), false)], true);
});
// Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate):
// nothing to bounce, and restart never rewrites intent, so the tail
// Reconcile converges the down agent to its existing `wanted`.
let restart = submit(&q, "restart down", |builder| {
let restart = insert(&q, |builder| {
power::restart_nodes(builder, &[("down2".to_owned(), false)], true);
});
let shape = |id: u64| -> Vec<String> {
@ -797,7 +804,7 @@ fn boot_sweep_nodes_declare_their_own_resources() {
)
.expect("valid shape");
let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock"));
let mut lock = declared_resources(&q, node_of(&q, "meta_lock"));
lock.sort_by_key(|r| format!("{r:?}"));
assert_eq!(
lock,
@ -806,7 +813,7 @@ fn boot_sweep_nodes_declare_their_own_resources() {
);
assert_eq!(
declared_resources(&q, node_of(&q, id, "reconcile")),
declared_resources(&q, node_of(&q, "reconcile")),
vec![Resource::Agent("drifted-agent".to_owned())],
"a boot Reconcile touches the container, so it holds that agent's lease"
);
@ -901,8 +908,8 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
// **did** flatten this chain, and the guarantee survives only because the
// roll-up point moved with it.
let q = JobQueue::new(1);
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let shape = declared_shape(&q, id);
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
let shape = declared_shape(&q);
let parent_of = |kind: &str| {
shape
.iter()
@ -973,7 +980,7 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
#[test]
fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
let q = JobQueue::new(4);
let id = submit(&q, "fan-out", |builder| {
let id = insert(&q, |builder| {
templates::fanned_out_mechanical(
builder,
NodeKind::Start {
@ -981,9 +988,9 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
},
);
});
assert_eq!(declared_shape(&q, id), vec![row("start", None, &[])]);
assert_eq!(declared_shape(&q), vec![row("start", None, &[])]);
assert_eq!(
declared_resources(&q, node_of(&q, id, "start")),
declared_resources(&q, node_of(&q, "start")),
vec![Resource::Agent("agent-a".to_owned())],
"the fanned-out node carries the lease itself, rather than relying on \
whoever happened to fan it out"
@ -1020,7 +1027,7 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
// One chain per agent, each an independent group root — so the two rebuild
// concurrently, each on its own lease.
let shape = declared_shape(&q, id);
let shape = declared_shape(&q);
let heads: Vec<_> = shape
.iter()
.filter(|d| d.kind == "meta_sync")
@ -1046,7 +1053,7 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
#[test]
fn cancel_clears_queued_dag() {
let q = JobQueue::new(1);
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let id = insert(&q, |builder| rebuild(builder, "agent-a"));
assert!(q.cancel(id), "fully-queued dag cancels");
// The operator sees `Cancelled` the moment the cancel returns — the spared
// tail is still `Pending`, and a DAG must not read `Queued` back to the
@ -1075,7 +1082,7 @@ fn cancel_clears_queued_dag() {
#[test]
fn cancel_drops_one_agents_branch_leaving_the_rest() {
let q = JobQueue::new(2);
let id = submit(&q, "r", |builder| {
let id = insert(&q, |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
});
// Per-agent subgraphs hang directly off the container, one per agent.
@ -1152,19 +1159,19 @@ fn cancelled_power_op_runs_no_compensating_node() {
let case = format!("graceful={graceful} running={running}");
let q = JobQueue::new(1);
let id = submit(&q, "bounce", |builder| {
let id = insert(&q, |builder| {
power::restart_nodes(builder, &targets, graceful);
});
assert_cancels_clean(&q, id, false, &format!("restart {case}"));
let q = JobQueue::new(1);
let id = submit(&q, "stop", |builder| {
let id = insert(&q, |builder| {
power::stop_nodes(builder, &targets, graceful);
});
assert_cancels_clean(&q, id, true, &format!("stop {case}"));
let q = JobQueue::new(1);
let id = submit(&q, "start", |builder| {
let id = insert(&q, |builder| {
power::start_nodes(builder, &[("agent-a".to_owned(), running, false)]);
});
assert_cancels_clean(&q, id, true, &format!("start {case}"));
@ -1185,7 +1192,7 @@ fn cancelled_power_op_runs_no_compensating_node() {
#[test]
fn cancelled_dag_still_runs_its_approval_tail() {
let q = JobQueue::new(1);
let id = submit(&q, "approval #7", |builder| {
let id = insert(&q, |builder| {
templates::approval_deploy(builder, "agent-a", 7);
});
assert!(q.cancel(id), "fully-queued dag cancels");
@ -1215,7 +1222,7 @@ fn cancelled_dag_still_runs_its_approval_tail() {
assert_eq!(state_of(&q, id), State::Finishing);
// An unrelated group landing in the same graph doesn't disturb this one's
// state — a root rolls up its own subtree, not the graph.
let _other = submit(&q, "r", |builder| rebuild(builder, "agent-b"));
let _other = insert(&q, |builder| rebuild(builder, "agent-b"));
assert_eq!(state_of(&q, id), State::Finishing);
}
@ -1229,12 +1236,12 @@ fn cancelled_dag_still_runs_its_approval_tail() {
#[test]
fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() {
let q = JobQueue::new(1);
let id = submit(&q, "approval #7", |builder| {
let id = insert(&q, |builder| {
templates::approval_deploy(builder, "agent-a", 7);
});
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![
// The window is the group root and holds the meta window for the
// whole subtree; the three phases are its sub-nodes.
@ -1287,12 +1294,12 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
// children run (`a_completing_node_grows_the_work_it_declared`,
// `parent_parks_in_finishing_until_children_roll_up`).
let q = JobQueue::new(1);
let id = submit(&q, "deploy graft", |builder| {
let id = insert(&q, |builder| {
templates::deploy_rebuild_nodes(builder, "agent-a", 11);
});
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![
row("meta_sync", None, &[]),
row("agent_window", None, &[("meta_sync", "done")]),
@ -1467,11 +1474,11 @@ fn error_truncation_cuts_on_a_char_boundary() {
#[test]
fn graceful_stop_shape_signal_drain_reconcile() {
let q = JobQueue::new(1);
let id = submit(&q, "graceful", |builder| {
let id = insert(&q, |builder| {
stop_online(builder, &["agent-a"], true);
});
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![
// The whole stop hangs under `set_wanted`: the durable intent is
// written first, and the mechanical steps are its sub-nodes.
@ -1494,8 +1501,8 @@ fn graceful_stop_shape_signal_drain_reconcile() {
// `exec.rs`.
assert_eq!(
[
declared_resources(&q, node_of(&q, id, "signal")),
declared_resources(&q, node_of(&q, id, "drain")),
declared_resources(&q, node_of(&q, "signal")),
declared_resources(&q, node_of(&q, "drain")),
],
[vec![], vec![]],
"the quiesce pair borrows the brace's lease and declares nothing"
@ -1505,11 +1512,11 @@ fn graceful_stop_shape_signal_drain_reconcile() {
#[test]
fn spawn_shape_provision_create_dropin_reconcile() {
let q = JobQueue::new(1);
let id = submit(&q, "approval #7 spawn", |builder| {
let id = insert(&q, |builder| {
templates::spawn(builder, "newbie", 7);
});
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![
row("provision", None, &[]),
row("create", Some("provision"), &[]),
@ -1530,7 +1537,7 @@ fn spawn_shape_provision_create_dropin_reconcile() {
#[test]
fn perm_change_shape_prefixes_rebuild_chain() {
let q = JobQueue::new(1);
let id = submit(&q, "perm", |builder| {
let id = insert(&q, |builder| {
templates::perm_change(
builder,
"agent-a",
@ -1541,7 +1548,7 @@ fn perm_change_shape_prefixes_rebuild_chain() {
);
});
assert_eq!(
declared_shape(&q, id)
declared_shape(&q)
.iter()
.map(|d| d.kind)
.collect::<Vec<_>>(),
@ -1569,15 +1576,15 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() {
// `MetaLock`, and it must declare the meta window — a topology commit
// must not land inside another node's staged deploy window.
let q = JobQueue::new(1);
let id = submit(&q, "set-parent", |builder| {
let id = insert(&q, |builder| {
templates::reparent(builder, vec![(ident("alice"), Some(ident("bob")))]);
});
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![row("reparent", None, &[])],
"one node, no rebuild subgraph"
);
let node = node_of(&q, id, "reparent");
let node = node_of(&q, "reparent");
assert_eq!(
declared_resources(&q, node),
vec![Resource::MetaWindow],
@ -1593,11 +1600,11 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() {
// request is the reason a single node was chosen in the first place.
let moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)];
let q = JobQueue::new(1);
let id = submit(&q, "set-parent-bulk", |builder| {
let id = insert(&q, |builder| {
templates::reparent(builder, moves.clone());
});
assert_eq!(
declared_shape(&q, id),
declared_shape(&q),
vec![row("reparent", None, &[])],
"one node for the whole request, not one per move"
);