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

c0re: the per-agent template tests read declared shape

Browse source 
`multi_agent_stop` claimed both heads to show they start together, and
`multi_agent_start` completed both heads to show the stale agent rebuilds
first. Neither needs the scheduler: what makes the subgraphs concurrent is
that each head is a group root with no node-deps holding only its own
agent's lease, and the stale fold is a longer chain declared at submit.
Both are readable the moment submit returns.

`declared_shape_for` slices the shape by the agent a payload names — a
hive-wide DAG interleaves one subgraph per agent and the kinds alone can't
tell two `set_wanted` rows apart. `declared_resources_of_kind` does the
same for the whole family of one kind, replacing the hand-rolled lease
extraction in the restart test.

`boot_sweep_nodes_declare_their_own_resources` only claimed to get at two
node ids; `node_of` gets them without running anything.
This commit is contained in:
atlas 2026-08-02 21:13:46 +02:00 • committed by mara
commit a039a10e40
1 changed files with 93 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -161,13 +161,21 @@ fn when_tag(when: hive_jobq::DepWhen) -> String {
/// *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_filtered(
q: &JobQueue,
dag: u64,
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))
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && keep(&n.payload))
.map(|n| Declared {
kind: n.payload.as_str(),
parent: n.parent.filter(|p| *p != root).and_then(kind_of),
@ -304,6 +312,44 @@ fn declared_resources(q: &JobQueue, node_id: hive_jobq::NodeId) -> Vec<Resource>
.collect()
}
/// The resources declared by **every** node of `kind` under `dag`, one row per
/// 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
/// 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
/// [`declared_resources`]'s single node.
fn declared_resources_of_kind(q: &JobQueue, dag: u64, kind: &str) -> Vec<Vec<Resource>> {
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 rows: Vec<Vec<Resource>> = graph
.nodes()
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
.map(|n| {
n.deps
.iter()
.filter_map(|dep| match dep {
hive_jobq::Dep::Resource { name, .. } => Some(name.clone()),
hive_jobq::Dep::Node { .. } => None,
})
.collect()
})
.collect();
rows.sort_by_key(|r| format!("{r:?}"));
rows
}
/// [`declared_shape`] restricted to the nodes whose payload names `agent`.
///
/// A hive-wide DAG interleaves one subgraph per agent, and the kinds alone
/// 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
/// stale one rebuilds first" expressible as a declared shape.
fn declared_shape_for(q: &JobQueue, dag: u64, agent: &str) -> Vec<Declared> {
declared_shape_filtered(q, dag, &|kind: &NodeKind| kind.agent() == agent)
}
fn state_of(q: &JobQueue, dag_id: u64) -> State {
// A DAG whose nodes have all settled `Done` or `Skipped` drops out of the
// snapshot — absence is the completion signal, so map it to `Done`.
@ -628,25 +674,11 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
],
"both per-agent heads are independent group roots"
);
let sched = q.sched().lock().expect("job_queue mutex poisoned");
let graph = sched.graph();
let root = graph.resolve_id(id).expect("dag id");
let mut leases: Vec<String> = graph
.nodes()
.filter(|n| graph.root_of(n.id) == Some(root) && n.payload.as_str() == "stop_for_update")
.flat_map(|n| {
n.deps.iter().filter_map(|dep| match dep {
hive_jobq::Dep::Resource { name, .. } => Some(format!("{name:?}")),
hive_jobq::Dep::Node { .. } => None,
})
})
.collect();
leases.sort();
assert_eq!(
leases,
declared_resources_of_kind(&q, id, "stop_for_update"),
vec![
format!("{:?}", Resource::Agent("agent-a".to_owned())),
format!("{:?}", Resource::Agent("agent-b".to_owned())),
vec![Resource::Agent("agent-a".to_owned())],
vec![Resource::Agent("agent-b".to_owned())],
],
"each head declares only its own agent's lease — disjoint, so no contention"
);
@ -661,17 +693,26 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
);
// A hive-wide stop is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1);
let claims = q.claim_ready();
assert!(claims.iter().all(|c| c.dag_id == id));
let mut heads: Vec<(&str, &str)> = claims
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str()))
// Same declared story as the restart case above: each agent's subgraph head
// 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)
.into_iter()
.filter(|d| d.kind == "set_wanted")
.collect();
heads.sort_unstable();
assert_eq!(
heads,
vec![("agent-a", "set_wanted"), ("agent-b", "set_wanted")],
"both per-agent stop subgraphs start concurrently, each on its own lease"
vec![row("set_wanted", None, &[]), row("set_wanted", None, &[])],
"both per-agent stop subgraph heads are independent group roots"
);
assert_eq!(
declared_resources_of_kind(&q, id, "set_wanted"),
vec![
vec![Resource::Agent("agent-a".to_owned())],
vec![Resource::Agent("agent-b".to_owned())],
],
"each head declares only its own agent's lease"
);
}
@ -693,32 +734,31 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
);
// One DAG spanning both agents.
assert_eq!(q.snapshot().len(), 1);
// Both subgraph heads (SetWanted(Up)) are roots — claimable at once,
// each acquiring its own agent lease.
let heads = q.claim_ready();
assert!(
heads
.iter()
.all(|c| c.dag_id == id && c.kind.as_str() == "set_wanted")
);
let mut head_agents: Vec<&str> = heads.iter().map(|c| c.agent.as_str()).collect();
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 (meta_sync first).
for c in &heads {
q.complete_node(c.node_id, Ok(()));
}
let next = q.claim_ready();
let mut kinds: Vec<(&str, &str)> = next
.iter()
.map(|c| (c.agent.as_str(), c.kind.as_str()))
.collect();
kinds.sort_unstable();
// The fold is a *declared* difference, readable the moment submit returns:
// both agents get a `SetWanted(Up)` group root, but the fresh agent's
// subgraph ends at the Reconcile behind it while the stale agent's carries
// the whole rebuild chain in between.
assert_eq!(
kinds,
vec![("fresh", "reconcile"), ("stale", "meta_sync")],
"fresh agent starts directly; stale agent rebuilds first, all in one DAG"
declared_shape_for(&q, id, "fresh"),
vec![
row("set_wanted", None, &[]),
row("reconcile", Some("set_wanted"), &[]),
],
"a fresh agent is intent + convergence, nothing in between"
);
assert_eq!(
declared_shape_for(&q, id, "stale"),
vec![
row("set_wanted", None, &[]),
row("meta_sync", None, &[("set_wanted", "done")]),
row("prebuild", None, &[("meta_sync", "done")]),
row("stop_for_update", Some("prebuild"), &[]),
row("swap", Some("stop_for_update"), &[]),
row("post_swap", Some("stop_for_update"), &[("swap", "done")]),
row("reconcile", None, &[("prebuild", "done|failed|skipped")]),
],
"a stale agent gets the whole rebuild chain wedged between intent and \
convergence — same DAG, same head kind, more in the middle"
);
}
@ -784,7 +824,7 @@ fn boot_sweep_nodes_declare_their_own_resources() {
// meta commit inside another node's staged deploy window. Nothing failed to
// compile; only an exhaustive caller list would have caught it.
let q = JobQueue::new(4);
let _id = submit(
let id = submit(
&q,
DagSpec {
source: Source::AutoUpdate,
@ -800,16 +840,7 @@ fn boot_sweep_nodes_declare_their_own_resources() {
},
);
// Both are independent roots on disjoint resources, so both start at once.
let claims = q.claim_ready();
let by_kind = |kind: &str| {
claims
.iter()
.find(|c| c.kind.as_str() == kind)
.unwrap_or_else(|| panic!("no {kind} claim in {claims:?}"))
};
let mut lock = declared_resources(&q, by_kind("meta_lock").node_id);
let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock"));
lock.sort_by_key(|r| format!("{r:?}"));
assert_eq!(
lock,
@ -818,7 +849,7 @@ fn boot_sweep_nodes_declare_their_own_resources() {
);
assert_eq!(
declared_resources(&q, by_kind("reconcile").node_id),
declared_resources(&q, node_of(&q, id, "reconcile")),
vec![Resource::Agent("drifted-agent".to_owned())],
"a boot Reconcile touches the container, so it holds that agent's lease"
);