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

jobq: rename the builder parameters too, not just the alias

Browse source 
Renaming `pub type Job` fixed the definition and left every use site
reading `b` and `job` — including `job: super::JobBuilder`, where the
parameter still asserted it was a job while its type said otherwise.
The propagation is what the issue was about, so the parameters are the
half that matters at a call site.

Two spots deliberately untouched: `auto_update`'s `sort_by(|a, b| …)`
comparator, and the prose that means the job *queue* (main.rs's
"Job-queue scheduler", scheduler.rs's "not this module's job any more",
the "grown job rejected" log).

315 tests pass unchanged.
This commit is contained in:
atlas 2026-08-03 13:09:24 +02:00 • committed by mara
commit 684f6da78e
6 changed files with 203 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -29,24 +29,24 @@ fn ident(s: &str) -> hive_types::Ident {
hive_types::Ident::parse(s).expect("valid test ident")
}
fn rebuild(b: &JobBuilder, agent: &str) {
templates::rebuild(b, agent, true);
fn rebuild(builder: &JobBuilder, agent: &str) {
templates::rebuild(builder, agent, true);
}
/// Restart shape with every agent treated as **running** — the online
/// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head)
/// most queue-mechanics tests assume. Mirrors the pre-dynamic
/// `templates::restart` (which is now the state-aware `submit::restart_nodes`).
fn restart_online(b: &JobBuilder, agents: &[&str], graceful: bool) {
fn restart_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
submit::restart_nodes(b, &targets, graceful);
submit::restart_nodes(builder, &targets, graceful);
}
/// Stop shape with every agent treated as **running** — the online shape
/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
fn stop_online(b: &JobBuilder, agents: &[&str], graceful: bool) {
fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
submit::stop_nodes(b, &targets, graceful);
submit::stop_nodes(builder, &targets, graceful);
}
// `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type
@ -299,8 +299,8 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State {
#[test]
fn submit_assigns_distinct_ids() {
let q = JobQueue::new(1);
let first = submit(&q, "first", |b| rebuild(b, "agent-a"));
let second = submit(&q, "second", |b| rebuild(b, "agent-b"));
let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let second = submit(&q, "second", |builder| rebuild(builder, "agent-b"));
assert_ne!(first, second);
assert_eq!(q.snapshot().len(), 2);
}
@ -313,8 +313,8 @@ fn submit_assigns_distinct_ids() {
#[test]
fn identical_resubmit_is_a_distinct_dag() {
let q = JobQueue::new(1);
let first = submit(&q, "first", |b| rebuild(b, "agent-a"));
let resubmit = submit(&q, "again", |b| rebuild(b, "agent-a"));
let first = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let resubmit = submit(&q, "again", |builder| rebuild(builder, "agent-a"));
assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG");
assert_eq!(q.snapshot().len(), 2);
}
@ -322,9 +322,11 @@ fn identical_resubmit_is_a_distinct_dag() {
#[test]
fn distinct_submits_never_collapse() {
let q = JobQueue::new(1);
let rebuild_a = submit(&q, "r", |b| rebuild(b, "agent-a"));
let rebuild_b = submit(&q, "r", |b| rebuild(b, "agent-b"));
let restart_a = submit(&q, "r", |b| restart_online(b, &["agent-a"], false));
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| {
restart_online(builder, &["agent-a"], false);
});
assert_ne!(rebuild_a, rebuild_b);
assert_ne!(rebuild_a, restart_a);
assert_eq!(q.snapshot().len(), 3);
@ -342,8 +344,10 @@ 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", |b| rebuild(b, "agent-a"));
let again = submit(&q, "config bumped during build", |b| rebuild(b, "agent-a"));
let a = submit(&q, "first", |builder| rebuild(builder, "agent-a"));
let again = submit(&q, "config bumped during build", |builder| {
rebuild(builder, "agent-a");
});
assert_ne!(a, again);
assert_eq!(q.snapshot().len(), 2);
}
@ -379,7 +383,7 @@ fn rebuild_chain_is_declared_serial() {
// logic"). Both axes are asserted below because a template can break either
// one independently.
let q = JobQueue::new(1);
let id = submit(&q, "r", |b| rebuild(b, "agent-a"));
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
assert_eq!(
declared_shape(&q, id),
vec![
@ -426,9 +430,13 @@ fn rebuild_chain_is_declared_serial() {
fn graceful_rebuild_chain_drains_before_stopping() {
let q = JobQueue::new(1);
let id = q
.submit(Source::AutoUpdate, "sweep".to_owned(), |b: &JobBuilder| {
templates::graceful_rebuild_nodes(b, "agent-a", true, None);
})
.submit(
Source::AutoUpdate,
"sweep".to_owned(),
|builder: &JobBuilder| {
templates::graceful_rebuild_nodes(builder, "agent-a", true, None);
},
)
.expect("valid shape");
assert_eq!(
declared_shape(&q, id)
@ -460,8 +468,8 @@ 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", |b| {
templates::rebuild_nodes(b, "agent-a", true, None);
let id = submit(&q, "manual", |builder| {
templates::rebuild_nodes(builder, "agent-a", true, None);
});
assert_eq!(
declared_shape(&q, id)
@ -541,7 +549,7 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
// a resource unit is held for the acquirer's whole subtree, so the slot
// `Prebuild` takes covers `StopForUpdate` → `Swap` → `PostSwap` beneath it.
let q = JobQueue::new(1);
let id = submit(&q, "r", |b| rebuild(b, "agent-a"));
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let res = |kind: &str| declared_resources(&q, node_of(&q, id, kind));
let agent = || Resource::Agent("agent-a".to_owned());
@ -578,8 +586,8 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
#[test]
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4);
let id = submit(&q, "hive-wide", |b| {
restart_online(b, &["agent-a", "agent-b"], false);
let id = submit(&q, "hive-wide", |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
});
// A hive-wide restart is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
@ -614,8 +622,8 @@ 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", |b| {
stop_online(b, &["agent-a", "agent-b"], false);
let id = submit(&q, "hive-wide stop", |builder| {
stop_online(builder, &["agent-a", "agent-b"], false);
});
// A hive-wide stop is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
@ -647,9 +655,9 @@ 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", |b| {
let id = submit(&q, "hive-wide start", |builder| {
submit::start_nodes(
b,
builder,
&[
("fresh".to_owned(), false, false),
("stale".to_owned(), false, true),
@ -694,14 +702,14 @@ 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", |b| {
submit::stop_nodes(b, &[("down".to_owned(), false)], true);
let stop = submit(&q, "stop down", |builder| {
submit::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", |b| {
submit::restart_nodes(b, &[("down2".to_owned(), false)], true);
let restart = submit(&q, "restart down", |builder| {
submit::restart_nodes(builder, &[("down2".to_owned(), false)], true);
});
let shape = |id: u64| -> Vec<String> {
q.snapshot()
@ -737,14 +745,18 @@ fn boot_sweep_nodes_declare_their_own_resources() {
// compile; only an exhaustive caller list would have caught it.
let q = JobQueue::new(4);
let id = q
.submit(Source::AutoUpdate, "boot".to_owned(), |b: &JobBuilder| {
crate::workers::auto_update::boot_nodes(
b,
true,
vec!["stale-agent".to_owned()],
vec!["drifted-agent".to_owned()],
);
})
.submit(
Source::AutoUpdate,
"boot".to_owned(),
|builder: &JobBuilder| {
crate::workers::auto_update::boot_nodes(
builder,
true,
vec!["stale-agent".to_owned()],
vec!["drifted-agent".to_owned()],
);
},
)
.expect("valid shape");
let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock"));
@ -846,7 +858,7 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
// easy thing to break — someone flattening the chain would keep every edge
// and still lose the guarantee.
let q = JobQueue::new(1);
let id = submit(&q, "r", |b| rebuild(b, "agent-a"));
let id = submit(&q, "r", |builder| rebuild(builder, "agent-a"));
let shape = declared_shape(&q, id);
let parent_of = |kind: &str| {
shape
@ -918,9 +930,9 @@ 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", |b| {
let id = submit(&q, "fan-out", |builder| {
templates::fanned_out_mechanical(
b,
builder,
NodeKind::Start {
agent: "agent-a".to_owned(),
},
@ -954,9 +966,13 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
let q = JobQueue::new(4);
let agents = vec!["alice".to_owned(), "bob".to_owned()];
let id = q
.submit(Source::AutoUpdate, "sweep".to_owned(), |b: &JobBuilder| {
templates::grown_graceful_rebuilds(b, &agents, true);
})
.submit(
Source::AutoUpdate,
"sweep".to_owned(),
|builder: &JobBuilder| {
templates::grown_graceful_rebuilds(builder, &agents, true);
},
)
.expect("valid shape");
// One chain per agent, each an independent group root — so the two rebuild
@ -987,7 +1003,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", |b| rebuild(b, "agent-a"));
let id = submit(&q, "r", |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
@ -1016,8 +1032,8 @@ 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", |b| {
restart_online(b, &["agent-a", "agent-b"], false);
let id = submit(&q, "r", |builder| {
restart_online(builder, &["agent-a", "agent-b"], false);
});
// Per-agent subgraphs are independent roots; find agent-a's.
let snap = q.snapshot();
@ -1085,20 +1101,20 @@ 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", |b| {
submit::restart_nodes(b, &targets, graceful);
let id = submit(&q, "bounce", |builder| {
submit::restart_nodes(builder, &targets, graceful);
});
assert_cancels_clean(&q, id, false, &format!("restart {case}"));
let q = JobQueue::new(1);
let id = submit(&q, "stop", |b| {
submit::stop_nodes(b, &targets, graceful);
let id = submit(&q, "stop", |builder| {
submit::stop_nodes(builder, &targets, graceful);
});
assert_cancels_clean(&q, id, true, &format!("stop {case}"));
let q = JobQueue::new(1);
let id = submit(&q, "start", |b| {
submit::start_nodes(b, &[("agent-a".to_owned(), running, false)]);
let id = submit(&q, "start", |builder| {
submit::start_nodes(builder, &[("agent-a".to_owned(), running, false)]);
});
assert_cancels_clean(&q, id, true, &format!("start {case}"));
}
@ -1118,8 +1134,8 @@ 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", |b| {
templates::approval_deploy(b, "agent-a", 7);
let id = submit(&q, "approval #7", |builder| {
templates::approval_deploy(builder, "agent-a", 7);
});
assert!(q.cancel(id), "fully-queued dag cancels");
// The `Cancelled` tail is the only node whose edge accepts a dropped
@ -1141,7 +1157,7 @@ fn cancelled_dag_still_runs_its_approval_tail() {
assert_eq!(state_of(&q, id), State::Cancelled);
// An unrelated DAG landing in the same graph doesn't disturb this one's
// roll-up — the snapshot is per-DAG, not a global state machine.
let _other = submit(&q, "r", |b| rebuild(b, "agent-b"));
let _other = submit(&q, "r", |builder| rebuild(builder, "agent-b"));
assert_eq!(state_of(&q, id), State::Cancelled);
}
@ -1155,8 +1171,8 @@ 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", |b| {
templates::approval_deploy(b, "agent-a", 7);
let id = submit(&q, "approval #7", |builder| {
templates::approval_deploy(builder, "agent-a", 7);
});
assert_eq!(
@ -1213,8 +1229,8 @@ 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", |b| {
templates::deploy_rebuild_nodes(b, "agent-a", 11);
let id = submit(&q, "deploy graft", |builder| {
templates::deploy_rebuild_nodes(builder, "agent-a", 11);
});
assert_eq!(
@ -1380,7 +1396,9 @@ 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", |b| stop_online(b, &["agent-a"], true));
let id = submit(&q, "graceful", |builder| {
stop_online(builder, &["agent-a"], true);
});
assert_eq!(
declared_shape(&q, id),
vec![
@ -1410,8 +1428,8 @@ 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", |b| {
templates::spawn(b, "newbie", 7);
let id = submit(&q, "approval #7 spawn", |builder| {
templates::spawn(builder, "newbie", 7);
});
assert_eq!(
declared_shape(&q, id),
@ -1435,9 +1453,9 @@ 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", |b| {
let id = submit(&q, "perm", |builder| {
templates::perm_change(
b,
builder,
"agent-a",
PermPayload::Combined {
groups: Some(vec![]),
@ -1473,8 +1491,8 @@ 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", |b| {
templates::reparent(b, vec![(ident("alice"), Some(ident("bob")))]);
let id = submit(&q, "set-parent", |builder| {
templates::reparent(builder, vec![(ident("alice"), Some(ident("bob")))]);
});
assert_eq!(
declared_shape(&q, id),
@ -1497,8 +1515,8 @@ 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", |b| {
templates::reparent(b, moves.clone());
let id = submit(&q, "set-parent-bulk", |builder| {
templates::reparent(builder, moves.clone());
});
assert_eq!(
declared_shape(&q, id),