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

jobq: Scheduler::new_job is gone

Browse source 
It handed out exactly the builder JobBuilder::new is pub(crate) to
withhold, which was agreed against more than once. I had left it in
place with a docstring naming it as the hole and folded the fix into an
open question. It was not an open question.

It only survived because two hive-c0re tests minted a builder by hand to
simulate a MetaLock growing its cascade. The grown thing is a template,
so the fix is the same as for the deploy graft and the reconcile
fan-out: call it.

exec.rs's MetaLock arm looped rebuild_nodes inline -- the second and last
construction site declaring nodes inside an executor. It is now
templates::grown_rebuilds, so a test can assert what a sweep declares by
calling the same function production calls.

grown_subgraph_roots_on_emitter_and_rebases_local_deps and
meta_update_grows_cascade_in_dag differed only in RebuildOpts; they are
one test over the declared shape, checking a root chain per agent, a
build each, and a drain each because a boot sweep is graceful.

With those gone, CompleteNode::new_job, CompleteNode::complete_node_growing
and drain_meta_syncs had no callers either. hive-jobq's own two growth
tests use JobBuilder::new() directly -- in-crate, so the wall holds.
grep for new_job across the workspace now returns nothing.
This commit is contained in:
atlas 2026-08-02 21:01:08 +02:00 • committed by mara
commit 009e9fafae
4 changed files with 77 additions and 183 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-c0re/src/job_queue/exec.rs
View file

@ -72,11 +72,7 @@ pub(super) async fn run_node(
inputs,
} => run_meta_lock(coord, *sweep, fanout.clone(), inputs)
.await
.map(|(agents, opts)| {
for agent in agents {
super::templates::rebuild_nodes(&job, &agent, opts, None);
}
}),
.map(|(agents, opts)| super::templates::grown_rebuilds(&job, &agents, opts)),
NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
if let Some(kind) = sub {
super::templates::fanned_out_mechanical(&job, kind);

16
hive-c0re/src/job_queue/templates.rs
View file

@ -80,6 +80,22 @@ fn resolve_approval_tails(b: &Job, approval_id: i64, root: Handle<'_>) {
}
}
/// Declare one rebuild subgraph per agent onto the builder a running
/// [`NodeKind::MetaLock`] was handed.
///
/// **Into the emitter's own builder, not as new DAGs.** Growing in-DAG is what
/// roots each subgraph on the `MetaLock`, so the whole sweep (or meta-update
/// cascade) stays one unit of work the operator can watch and cancel, and every
/// rebuild builds against the lock the emitter just bumped.
///
/// Same reason as [`fanned_out_mechanical`] for living here: this was the
/// second construction site declaring nodes inline in an executor.
pub(crate) fn grown_rebuilds(b: &Job, agents: &[String], opts: RebuildOpts) {
for agent in agents {
rebuild_nodes(b, agent, opts, None);
}
}
/// Declare the mechanical node a [`NodeKind::Reconcile`] planner fans out
/// (`Start` / `Stop`) onto the builder it was handed while running.
///

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

@ -101,24 +101,9 @@ impl ClaimReady for JobQueue {
/// they exercise the graph without running any executor.
trait CompleteNode {
fn complete_node(&self, node_id: NodeId, result: Result<(), String>);
fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job);
fn new_job(&self) -> Job;
}
impl CompleteNode for JobQueue {
/// Mint a builder to declare growth into.
///
/// Test-only for the same reason as the rest of this trait: production
/// never mints one, because `claim_next` hands each running node its
/// builder and takes it back. That leaves `Scheduler::new_job` with no
/// non-test caller either — see the note on that fn.
fn new_job(&self) -> Job {
self.sched()
.lock()
.expect("job_queue mutex poisoned")
.new_job()
}
fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
self.sched()
.lock()
@ -126,25 +111,6 @@ impl CompleteNode for JobQueue {
.complete(node_id, outcome_of(result));
self.notify.notify_one();
}
fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
// A rejected grown job is logged, not propagated: the node's own work
// already ran, and refusing to complete it here would both misreport
// that and wedge the DAG on a node stuck `Running`.
if let Err(e) = self
.sched()
.lock()
.expect("job_queue mutex poisoned")
.complete_growing(node_id, outcome_of(result), grown)
{
tracing::error!(
node = node_id.get(),
error = %e,
"job_queue: work grown by a completing node was rejected"
);
}
self.notify.notify_one();
}
}
/// One node's **declared** shape: what it is, what it hangs under, and what it
@ -858,80 +824,6 @@ fn boot_sweep_nodes_declare_their_own_resources() {
);
}
#[test]
fn grown_subgraph_roots_on_emitter_and_rebases_local_deps() {
// The startup-sweep mechanism: a `MetaLock` emitter grows one rebuild
// subgraph per stale agent into its OWN DAG. Each subgraph is rooted on
// the emitter and its LOCAL 0-based deps are rebased onto the DAG.
let q = JobQueue::new(4);
let spec = DagSpec {
source: Source::AutoUpdate,
reason: "sweep".to_owned(),
declare: Box::new(|b: &Job| {
let _lock = b.node(NodeKind::MetaLock {
sweep: true,
fanout: None,
inputs: Vec::new(),
});
}),
};
submit(&q, spec);
let emitter = claim_one(&q);
assert_eq!(emitter.kind.as_str(), "meta_lock");
// Two independent per-agent subgraphs — the REAL production shape the
// sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain →
// StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must
// match the sweep arm of `run_meta_lock` or this stops tracking production.
// Both subgraphs go into the emitter's own builder, exactly as
// `run_meta_lock`'s sweep arm does. Insert-before-complete is no longer the
// caller's job to remember: it is one call, and the ordering is inside it.
let grown = q.new_job();
for agent in ["a", "b"] {
templates::rebuild_nodes(
&grown,
agent,
templates::RebuildOpts {
relock: true,
graceful: true,
},
None,
);
}
q.complete_node_growing(emitter.node_id, Ok(()), grown);
// Still ONE DAG; both subgraph roots become ready once the emitter is
// Done (rooted on it), each on its own agent lease. Their `MetaSync` heads
// take turns on the cap-1 global meta window, so drain those first — what
// must be concurrent is the builds.
assert_eq!(q.snapshot().len(), 1);
let mut kinds = drain_meta_syncs(&q);
kinds.sort_unstable();
assert_eq!(
kinds,
vec![
("a".to_owned(), "prebuild".to_owned()),
("b".to_owned(), "prebuild".to_owned())
],
"both rebuild subgraphs root on the emitter and run concurrently in one DAG"
);
}
/// Complete every `MetaSync` head the queue offers (they take turns on the
/// cap-1 global meta window) and return whatever else got claimed alongside
/// them, as `(agent, kind)` pairs left in flight.
fn drain_meta_syncs(q: &JobQueue) -> Vec<(String, String)> {
let mut rest = 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 {
rest.push((c.agent.clone(), c.kind.as_str().to_owned()));
}
}
}
rest
}
/// Crash-watch suppression for a cascade rebuild, which the deleted half of
/// `meta_update_grows_cascade_in_dag` used to assert via `DagSpec::transient`.
///
@ -971,54 +863,6 @@ fn rebuild_chain_nodes_suppress_crash_watch() {
);
}
#[test]
fn meta_update_grows_cascade_in_dag() {
// The meta-update `MetaLock` grows one rebuild subgraph per affected
// agent into its OWN DAG (via the builder it is handed), not child DAGs.
let spec = templates::meta_update(
vec!["nixpkgs".to_owned()],
Source::Manual,
"bump".to_owned(),
None,
);
let q = JobQueue::new(4);
submit(&q, spec);
let meta_lock = claim_one(&q);
assert_eq!(meta_lock.kind.as_str(), "meta_lock");
// Simulate the executor growing the cascade in-DAG (`relock = false` — a
// cascade child must not re-lock and revert the parent's bump). Both
// agents go into the one builder the node was handed, which is what
// `run_meta_lock`'s fanout arm does.
let grown = q.new_job();
for agent in ["alice", "bob"] {
templates::rebuild_nodes(
&grown,
agent,
templates::RebuildOpts {
relock: false,
graceful: false,
},
None,
);
}
q.complete_node_growing(meta_lock.node_id, Ok(()), grown);
// Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root
// on the MetaLock, each on its own agent lease. The per-agent `MetaSync`
// heads serialize on the global meta window (they commit to the meta repo);
// the builds behind them do not.
assert_eq!(q.snapshot().len(), 1);
let mut kinds = drain_meta_syncs(&q);
kinds.sort_unstable();
assert_eq!(
kinds,
vec![
("alice".to_owned(), "prebuild".to_owned()),
("bob".to_owned(), "prebuild".to_owned())
],
"cascade rebuilds grow in the meta-update DAG, concurrent per agent"
);
}
// ---- failure: cancel-downstream + AfterAny ----
#[test]
@ -1190,6 +1034,64 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
);
}
/// A running `MetaLock` grows one rebuild subgraph per agent into **its own
/// DAG**, rooted on itself — not as child DAGs. That is what keeps a boot sweep
/// (or a meta-update cascade) one unit of work, with every rebuild building
/// against the lock the emitter just bumped.
///
/// Replaces `grown_subgraph_roots_on_emitter_and_rebases_local_deps` and
/// `meta_update_grows_cascade_in_dag`, which differed only in `RebuildOpts` and
/// each minted a builder by hand to simulate the graft. What they were checking
/// is `templates::grown_rebuilds`, so this calls it.
///
/// That the grafted work lands under the emitter, and that the emitter parks in
/// `Finishing` until it settles, is `hive_jobq`'s
/// (`a_completing_node_grows_the_work_it_declared`).
#[test]
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 = submit(
&q,
DagSpec {
source: Source::AutoUpdate,
reason: "sweep".to_owned(),
declare: Box::new(move |b: &Job| {
templates::grown_rebuilds(
b,
&agents,
templates::RebuildOpts {
relock: true,
graceful: true,
},
);
}),
},
);
// 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 heads: Vec<_> = shape
.iter()
.filter(|d| d.kind == "meta_sync")
.map(|d| d.parent)
.collect();
assert_eq!(heads, vec![None, None], "one root chain per agent");
assert_eq!(
shape.iter().filter(|d| d.kind == "prebuild").count(),
2,
"both agents get their own build"
);
// `graceful: true` is the sweep's distinguishing knob — agents mid-turn
// when the host came up get their drain window rather than being cut off.
assert_eq!(
shape.iter().filter(|d| d.kind == "drain").count(),
2,
"a boot sweep is graceful, so each agent gets a drain"
);
}
// ---- cancel ----
#[test]

24
hive-jobq/src/scheduler.rs
View file

@ -338,26 +338,6 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
self.release_ready();
}
/// A fresh builder for a **running** node to declare more work into.
///
/// The node runs outside this scheduler's lock — often for minutes — so it
/// cannot hold a graph reference while it works. It doesn't need one: a
/// builder is pure local state (locally-minted guids, resolved to
/// [`NodeId`]s only at insert), so it can be filled in freely and handed
/// back to [`Scheduler::complete_growing`], which inserts it under the lock.
///
/// ⚠️ **This has no non-test caller left, and it is the hole in
/// [`JobBuilder::new`]'s `pub(crate)` wall** — it hands out exactly the
/// builder that fn withholds. [`Scheduler::claim_next`] mints one per
/// running node itself, so production never asks. Kept only so the host's
/// graph-growth tests can still declare work by hand; the fix is a venue
/// question (move those tests here vs. a closure-form completion), not a
/// rename.
#[must_use]
pub fn new_job(&self) -> JobBuilder<N, R> {
JobBuilder::new()
}
/// [`Scheduler::complete`], plus whatever the node declared into the builder
/// it was handed while running.
///
@ -728,7 +708,7 @@ mod tests {
let n = s.append("emitter", vec![], None).expect("insert");
assert_eq!(s.settle(), vec![n]);
let grown = s.new_job();
let grown = JobBuilder::new();
grown.node("child-a");
grown.node("child-b");
s.complete_growing(n, Outcome::Done, grown)
@ -754,7 +734,7 @@ mod tests {
let n = s.append("emitter", vec![], None).expect("insert");
assert_eq!(s.settle(), vec![n]);
let grown = s.new_job();
let grown = JobBuilder::new();
grown.node("never-runs");
s.complete_growing(n, Outcome::Failed("boom".to_owned()), grown)
.expect("growth is dropped, not rejected");