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

wip(#3001): 9 of 10 test helpers off the container id; sweep stale prose

Browse source 
The helpers keep shrinking the same way: resolve_id goes, 'n.id != root'
goes (the container was the only non-work node), root_of goes, and the
container-parent normalisation goes because a group root now genuinely
has parent = None. pending_kinds_filtered drops from a four-clause
multi-line filter to one line.

Also removed a doc block my earlier edit had orphaned above the renamed
helper, and swept 'under `dag`' / '`submit` returns' out of the prose.

state_of stays untouched: it reads a roll-up, which is the same question
as hivectl's queued_dags.
This commit is contained in:
atlas 2026-08-04 12:23:24 +02:00 • committed by mara
commit 102ebdc03d
1 changed files with 40 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -4,7 +4,7 @@
//! that graph (wire projection, history retention, error truncation). //! that graph (wire projection, history retention, error truncation).
//! //!
//! **Nothing here runs a node.** Everything a template declares is in the graph //! **Nothing here runs a node.** Everything a template declares is in the graph
//! the moment `submit` returns, so the assertions read it there. Whether the //! the moment `insert` returns, so the assertions read it there. Whether the
//! scheduler then honours those declarations — cascade, roll-up, grant //! scheduler then honours those declarations — cascade, roll-up, grant
//! borrow/release, fairness, the `Finishing` gate — is `hive_jobq`'s property //! borrow/release, fairness, the `Finishing` gate — is `hive_jobq`'s property
//! and is tested in `hive_jobq`, against its own primitives rather than through //! and is tested in `hive_jobq`, against its own primitives rather than through
@ -17,9 +17,6 @@
use super::model::NodeKind; use super::model::NodeKind;
use super::*; 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.
/// Insert a declared job, naming nothing — the shape assertions read the whole /// 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 /// graph. A test that needs a handle calls `q.insert` directly and names the
/// node it cares about. /// node it cares about.
@ -68,8 +65,8 @@ fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) {
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
struct Declared { struct Declared {
kind: &'static str, kind: &'static str,
/// Parent kind, or `None` when the node hangs directly under the DAG /// Parent kind, or `None` when the node is a group root — which is now a
/// container (i.e. it is a group root). /// genuine `parent = None`, not "hangs under the container".
parent: Option<&'static str>, parent: Option<&'static str>,
/// Kinds this node declared a node-dep on, in declaration order, each with /// Kinds this node declared a node-dep on, in declaration order, each with
/// the outcome set that satisfies it. /// the outcome set that satisfies it.
@ -102,10 +99,10 @@ fn when_tag(when: hive_jobq::DepWhen) -> String {
.join("|") .join("|")
} }
/// Every work node under `dag`, in insertion order, as its declared shape. /// Every work node in the graph, in insertion order, as its declared shape.
/// ///
/// **This is what the template tests are actually about.** A template's output /// **This is what the template tests are actually about.** A template's output
/// is fully determined the moment `submit` returns: the kinds, the parent /// is fully determined the moment `insert` returns: the kinds, the parent
/// nesting and the dep edges are all sitting in the graph. Reading them here /// nesting and the dep edges are all sitting in the graph. Reading them here
/// keeps the assertion on c0re's own product. Whether the scheduler then /// keeps the assertion on c0re's own product. Whether the scheduler then
/// *honours* those edges — runs a chain serially, holds a grant across a /// *honours* those edges — runs a chain serially, holds a grant across a
@ -143,7 +140,7 @@ fn declared_shape_filtered(q: &JobQueue, keep: &dyn Fn(&NodeKind) -> bool) -> Ve
.collect() .collect()
} }
/// The id of the one node of `kind` under `dag`, for the resource assertions. /// The id of the one node of `kind`, for the resource assertions.
/// ///
/// Panics unless there is exactly one — every caller is about a shape where the /// 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 /// kind is unique, so two would mean the assertion had quietly stopped being
@ -164,56 +161,45 @@ fn node_of(q: &JobQueue, kind: &str) -> hive_jobq::NodeId {
found.pop().expect("checked above") found.pop().expect("checked above")
} }
/// The payload of the one node of `kind` under `dag`, for assertions about what /// The payload of the one node of `kind`, for assertions about what a node
/// a node *carries* rather than how it is wired. /// *carries* rather than how it is wired.
fn payload_of(q: &JobQueue, dag: u64, kind: &str) -> NodeKind { fn payload_of(q: &JobQueue, kind: &str) -> NodeKind {
let id = node_of(q, dag, kind); let id = node_of(q, kind);
let sched = q.sched().lock().expect("job_queue mutex poisoned"); let sched = q.sched().lock().expect("job_queue mutex poisoned");
sched.graph().node(id).expect("node exists").payload.clone() sched.graph().node(id).expect("node exists").payload.clone()
} }
/// Kinds of every node under `dag` still `Pending` — the nodes that could yet /// Kinds of every node still `Pending` — the nodes that could yet run.
/// run. Stronger than asking the scheduler what is *ready right now*: a node /// Stronger than asking the scheduler what is *ready right now*: a node
/// blocked on a dep is not ready but is very much still alive. /// blocked on a dep is not ready but is very much still alive.
fn pending_kinds(q: &JobQueue, dag: u64) -> Vec<&'static str> { fn pending_kinds(q: &JobQueue) -> Vec<&'static str> {
pending_kinds_filtered(q, dag, &|_| true) pending_kinds_filtered(q, &|_| true)
} }
/// [`pending_kinds`] restricted to the nodes whose payload names `agent`. /// [`pending_kinds`] restricted to the nodes whose payload names `agent`.
fn pending_kinds_for(q: &JobQueue, dag: u64, agent: &str) -> Vec<&'static str> { fn pending_kinds_for(q: &JobQueue, agent: &str) -> Vec<&'static str> {
pending_kinds_filtered(q, dag, &|kind: &NodeKind| kind.agent() == agent) pending_kinds_filtered(q, &|kind: &NodeKind| kind.agent() == agent)
} }
/// The payloads of every node under `dag` still `Pending`, for the cases where /// The payloads of every node still `Pending`, for the cases where *which* of a
/// *which* of a family of same-kind nodes survived is the assertion — a /// family of same-kind nodes survived is the assertion — a template emits one
/// template emits one tail per outcome and they differ only in what they carry. /// tail per outcome and they differ only in what they carry.
fn pending_payloads(q: &JobQueue, dag: u64) -> Vec<NodeKind> { fn pending_payloads(q: &JobQueue) -> Vec<NodeKind> {
let sched = q.sched().lock().expect("job_queue mutex poisoned"); let sched = q.sched().lock().expect("job_queue mutex poisoned");
let graph = sched.graph(); let graph = sched.graph();
let root = graph.resolve_id(dag).expect("dag id is a real node id");
graph graph
.nodes() .nodes()
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.state == State::Pending) .filter(|n| n.state == State::Pending)
.map(|n| n.payload.clone()) .map(|n| n.payload.clone())
.collect() .collect()
} }
fn pending_kinds_filtered( fn pending_kinds_filtered(q: &JobQueue, keep: &dyn Fn(&NodeKind) -> bool) -> Vec<&'static str> {
q: &JobQueue,
dag: u64,
keep: &dyn Fn(&NodeKind) -> bool,
) -> Vec<&'static str> {
let sched = q.sched().lock().expect("job_queue mutex poisoned"); let sched = q.sched().lock().expect("job_queue mutex poisoned");
let graph = sched.graph(); let graph = sched.graph();
let root = graph.resolve_id(dag).expect("dag id is a real node id");
graph graph
.nodes() .nodes()
.filter(|n| { .filter(|n| n.state == State::Pending && keep(&n.payload))
n.id != root
&& graph.root_of(n.id) == Some(root)
&& n.state == State::Pending
&& keep(&n.payload)
})
.map(|n| n.payload.as_str()) .map(|n| n.payload.as_str())
.collect() .collect()
} }
@ -252,20 +238,19 @@ fn declared_resources(q: &JobQueue, node_id: hive_jobq::NodeId) -> Vec<Resource>
.collect() .collect()
} }
/// The resources declared by **every** node of `kind` under `dag`, one row per /// The resources declared by **every** node of `kind`, one row per
/// node, sorted so the rows read as a set rather than an insertion order. /// node, sorted so the rows read as a set rather than an insertion order.
/// ///
/// The per-agent templates emit several nodes of one kind — one per agent — and /// The per-agent templates emit several nodes of one kind — one per agent — and
/// what makes them concurrent is that each holds only its *own* agent's lease. /// what makes them concurrent is that each holds only its *own* agent's lease.
/// That is a statement about the whole family, so it needs all the rows, not /// That is a statement about the whole family, so it needs all the rows, not
/// [`declared_resources`]'s single node. /// [`declared_resources`]'s single node.
fn declared_resources_of_kind(q: &JobQueue, dag: u64, kind: &str) -> Vec<Vec<Resource>> { fn declared_resources_of_kind(q: &JobQueue, kind: &str) -> Vec<Vec<Resource>> {
let sched = q.sched().lock().expect("job_queue mutex poisoned"); let sched = q.sched().lock().expect("job_queue mutex poisoned");
let graph = sched.graph(); let graph = sched.graph();
let root = graph.resolve_id(dag).expect("dag id is a real node id");
let mut rows: Vec<Vec<Resource>> = graph let mut rows: Vec<Vec<Resource>> = graph
.nodes() .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| { .map(|n| {
n.deps n.deps
.iter() .iter()
@ -286,8 +271,8 @@ fn declared_resources_of_kind(q: &JobQueue, dag: u64, kind: &str) -> Vec<Vec<Res
/// cannot tell them apart — two `set_wanted` rows look identical. Slicing by /// cannot tell them apart — two `set_wanted` rows look identical. Slicing by
/// agent is what makes "the fresh agent goes straight to reconcile while the /// agent is what makes "the fresh agent goes straight to reconcile while the
/// stale one rebuilds first" expressible as a declared shape. /// stale one rebuilds first" expressible as a declared shape.
fn declared_shape_for(q: &JobQueue, dag: u64, agent: &str) -> Vec<Declared> { fn declared_shape_for(q: &JobQueue, agent: &str) -> Vec<Declared> {
declared_shape_filtered(q, dag, &|kind: &NodeKind| kind.agent() == agent) declared_shape_filtered(q, &|kind: &NodeKind| kind.agent() == agent)
} }
fn state_of(q: &JobQueue, dag_id: u64) -> State { fn state_of(q: &JobQueue, dag_id: u64) -> State {
@ -394,7 +379,7 @@ fn resubmit_while_running_is_new_dag() {
#[test] #[test]
fn rebuild_chain_is_declared_serial() { fn rebuild_chain_is_declared_serial() {
// Was `rebuild_chain_claims_in_dep_order`, which drove the whole DAG to // Was `rebuild_chain_claims_in_dep_order`, which drove the whole DAG to
// observe an order that is fully declared the moment `submit` returns. // observe an order that is fully declared the moment `insert` returns.
// //
// ⚠️ The old name was also wrong about the mechanism, and reading it rather // ⚠️ The old name was also wrong about the mechanism, and reading it rather
// than the graph is how you'd stay wrong: **only part of this chain is dep // than the graph is how you'd stay wrong: **only part of this chain is dep
@ -643,7 +628,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
"both per-agent heads are independent group roots" "both per-agent heads are independent group roots"
); );
assert_eq!( assert_eq!(
declared_resources_of_kind(&q, id, "stop_for_update"), declared_resources_of_kind(&q, "stop_for_update"),
vec![ vec![
vec![Resource::Agent("agent-a".to_owned())], vec![Resource::Agent("agent-a".to_owned())],
vec![Resource::Agent("agent-b".to_owned())], vec![Resource::Agent("agent-b".to_owned())],
@ -674,7 +659,7 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
"both per-agent stop subgraph heads are independent group roots" "both per-agent stop subgraph heads are independent group roots"
); );
assert_eq!( assert_eq!(
declared_resources_of_kind(&q, id, "set_wanted"), declared_resources_of_kind(&q, "set_wanted"),
vec![ vec![
vec![Resource::Agent("agent-a".to_owned())], vec![Resource::Agent("agent-a".to_owned())],
vec![Resource::Agent("agent-b".to_owned())], vec![Resource::Agent("agent-b".to_owned())],
@ -704,7 +689,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
// subgraph ends at the Reconcile behind it while the stale agent's carries // subgraph ends at the Reconcile behind it while the stale agent's carries
// the whole rebuild chain in between. // the whole rebuild chain in between.
assert_eq!( assert_eq!(
declared_shape_for(&q, id, "fresh"), declared_shape_for(&q, "fresh"),
vec![ vec![
row("set_wanted", None, &[]), row("set_wanted", None, &[]),
row("reconcile", Some("set_wanted"), &[]), row("reconcile", Some("set_wanted"), &[]),
@ -712,7 +697,7 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
"a fresh agent is intent + convergence, nothing in between" "a fresh agent is intent + convergence, nothing in between"
); );
assert_eq!( assert_eq!(
declared_shape_for(&q, id, "stale"), declared_shape_for(&q, "stale"),
vec![ vec![
row("set_wanted", None, &[]), row("set_wanted", None, &[]),
row("meta_sync", None, &[("set_wanted", "done")]), row("meta_sync", None, &[("set_wanted", "done")]),
@ -1065,9 +1050,9 @@ fn cancel_clears_queued_dag() {
// with the work and **nothing is left that could still run**: no node is // with the work and **nothing is left that could still run**: no node is
// spared, so a rebuild that never ran emits nothing. // spared, so a rebuild that never ran emits nothing.
assert!( assert!(
pending_kinds(&q, id).is_empty(), pending_kinds(&q).is_empty(),
"a dropped rebuild leaves nothing alive, got {:?}", "a dropped rebuild leaves nothing alive, got {:?}",
pending_kinds(&q, id) pending_kinds(&q)
); );
} }
@ -1106,12 +1091,12 @@ fn cancel_drops_one_agents_branch_leaving_the_rest() {
// agent-a's subgraph is gone; agent-b's is untouched and still alive. // agent-a's subgraph is gone; agent-b's is untouched and still alive.
assert!( assert!(
pending_kinds_for(&q, id, "agent-a").is_empty(), pending_kinds_for(&q, "agent-a").is_empty(),
"agent-a's branch was dropped whole, got {:?}", "agent-a's branch was dropped whole, got {:?}",
pending_kinds_for(&q, id, "agent-a") pending_kinds_for(&q, "agent-a")
); );
assert!( assert!(
!pending_kinds_for(&q, id, "agent-b").is_empty(), !pending_kinds_for(&q, "agent-b").is_empty(),
"agent-b's branch survives its sibling's cancel" "agent-b's branch survives its sibling's cancel"
); );
} }
@ -1201,7 +1186,7 @@ fn cancelled_dag_still_runs_its_approval_tail() {
// survives is the whole assertion: the template emits one per outcome and // survives is the whole assertion: the template emits one per outcome and
// the spared one names how the approval row is about to be resolved. // the spared one names how the approval row is about to be resolved.
// Nothing computes it, so reading the survivor is reading the answer. // Nothing computes it, so reading the survivor is reading the answer.
let spared = pending_payloads(&q, id); let spared = pending_payloads(&q);
assert!( assert!(
matches!( matches!(
spared.as_slice(), spared.as_slice(),
@ -1608,7 +1593,7 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() {
vec![row("reparent", None, &[])], vec![row("reparent", None, &[])],
"one node for the whole request, not one per move" "one node for the whole request, not one per move"
); );
let NodeKind::Reparent { moves: got } = payload_of(&q, id, "reparent") else { let NodeKind::Reparent { moves: got } = payload_of(&q, "reparent") else {
panic!("expected a Reparent node"); panic!("expected a Reparent node");
}; };
assert_eq!(got, moves, "every move rides the single node"); assert_eq!(got, moves, "every move rides the single node");