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: split RebuildOpts into two rebuild entry points

Browse source 
RebuildOpts held one real parameter (relock) and one single-call-site
flag (graceful). The struct justified itself as swap-protection for two
positional bools; with graceful out of the signature there is nothing
left to swap.

graceful stays an internal switch rather than moving to the caller: it
re-parents the stop root (StopForUpdate goes from part_of(prebuild) to
part_of(signal)) rather than prepending nodes, so a caller could only
declare it by being handed the subtree's internals — and that nesting
keeps the agent lease continuous across the whole stop.

run_meta_lock no longer returns options: both fields were a pure
function of the sweep flag its caller had just passed in.

315 tests pass unchanged.
This commit is contained in:
atlas 2026-08-03 02:00:16 +02:00
commit 05f84191fb
4 changed files with 80 additions and 110 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -72,7 +72,18 @@ pub(super) async fn run_node(
inputs,
} => run_meta_lock(coord, *sweep, fanout.clone(), inputs)
.await
.map(|(agents, opts)| super::templates::grown_rebuilds(&job, &agents, opts)),
.map(|agents| {
// `sweep` is the whole difference: it relocks per-agent like a
// manual rebuild, and it drains agents that were mid-turn when
// the host came up. A cascade does neither. Decided here rather
// than returned, since `run_meta_lock` would only be deriving
// it from the `sweep` this call site already holds.
if *sweep {
super::templates::grown_graceful_rebuilds(&job, &agents, true);
} else {
super::templates::grown_rebuilds(&job, &agents, false);
}
}),
NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
if let Some(kind) = sub {
super::templates::fanned_out_mechanical(&job, kind);
@ -313,35 +324,29 @@ async fn run_create(name: &str) -> Result<()> {
/// the current lock, exactly like today's sweep); the meta-update
/// flavour propagates errors, and a failed bump fans out nothing.
/// Returns the agents whose rebuild subgraphs the caller should grow into this
/// node, and the options to build them with — rather than declaring them here.
/// The declaration has to happen outside any `.await` (see [`run_node`]).
/// node, rather than declaring them here — the declaration has to happen outside
/// any `.await` (see [`run_node`]).
///
/// Only the agent list: *which* rebuild flavour to grow is a pure function of
/// `sweep`, which the caller passed in, so returning it too would be a round
/// trip rather than a decision.
async fn run_meta_lock(
coord: &Arc<Coordinator>,
sweep: bool,
fanout: Option<Vec<String>>,
inputs: &[String],
) -> Result<(Vec<String>, super::templates::RebuildOpts)> {
) -> Result<Vec<String>> {
if sweep {
if let Err(e) = crate::meta::lock_update_hyperhive().await {
tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
}
// Grow one rebuild subgraph per stale agent into *this* boot DAG
// (rooted on this `MetaLock`, so they build against the post-bump
// lock), rather than fanning out child DAGs. `relock = true` — a
// boot sweep relocks per-agent like a manual rebuild.
//
// `graceful = true` here and nowhere else: a boot sweep stops agents
// that were already mid-turn when the host came up, so they get their
// drain window rather than being cut off. The per-agent drains overlap,
// so the sweep's cost ceiling is one `GRACEFUL_STOP_TIMEOUT` in total,
// not one per agent.
return Ok((
fanout.unwrap_or_default(),
super::templates::RebuildOpts {
relock: true,
graceful: true,
},
));
// lock), rather than fanning out child DAGs. The caller grows them
// with the graceful flavour: the per-agent drains overlap, so the
// sweep's cost ceiling is one `GRACEFUL_STOP_TIMEOUT` in total, not
// one per agent.
return Ok(fanout.unwrap_or_default());
}
let _progress = coord.meta_update_guard();
crate::meta::lock_update(inputs).await?;
@ -357,13 +362,7 @@ async fn run_meta_lock(
// cascade children must NOT re-lock, which would revert the bump this
// node just committed (the property the old `fanout_specs` meta-update
// branch encoded).
Ok((
cascade,
super::templates::RebuildOpts {
relock: false,
graceful: false,
},
))
Ok(cascade)
}
/// Idempotent power-converge *planner*: compare `wanted` (durable

12
hive-c0re/src/job_queue/submit.rs
View file

@ -26,7 +26,7 @@ use std::sync::Arc;
use super::model::NodeKind;
use super::resource::Resource;
use super::templates::{RebuildOpts, rebuild_nodes};
use super::templates::rebuild_nodes;
use super::{Job, Source, templates};
use crate::coordinator::Coordinator;
use crate::lifecycle;
@ -118,15 +118,7 @@ fn start_chain(b: &Job, agent: &str, running: bool, stale: bool) {
// Rebuild subtree chained behind the `SetWanted` head. `MetaSync`,
// `Prebuild` + `Reconcile` are their own group roots (top-level, per
// `rebuild_nodes`).
rebuild_nodes(
b,
agent,
RebuildOpts {
relock: true,
graceful: false,
},
Some(wanted),
);
rebuild_nodes(b, agent, true, Some(wanted));
} else {
let _ = b
.node(NodeKind::Reconcile {

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

@ -90,9 +90,18 @@ fn resolve_approval_tails(b: &Job, approval_id: i64, root: Handle<'_>) {
///
/// 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) {
pub(crate) fn grown_rebuilds(b: &Job, agents: &[String], relock: bool) {
for agent in agents {
rebuild_nodes(b, agent, opts, None);
rebuild_nodes(b, agent, relock, None);
}
}
/// As [`grown_rebuilds`], but each agent gets its `Signal` → `Drain` window
/// before being stopped. The boot sweep's flavour: it stops agents that were
/// mid-turn when the host came up, so they drain rather than being cut off.
pub(crate) fn grown_graceful_rebuilds(b: &Job, agents: &[String], relock: bool) {
for agent in agents {
graceful_rebuild_nodes(b, agent, relock, None);
}
}
@ -113,19 +122,6 @@ pub(crate) fn fanned_out_mechanical(b: &Job, kind: NodeKind) {
let _ = b.node(kind).needs(lease);
}
/// Knobs for [`rebuild_nodes`]. A struct rather than two positional `bool`s so
/// a call site cannot silently swap them.
#[derive(Debug, Clone, Copy)]
pub(crate) struct RebuildOpts {
/// Re-lock the meta flake inside `MetaSync`.
pub relock: bool,
/// Give the agent its `Signal` → `Drain` window to finish the turn in
/// flight before the container is stopped, instead of stopping it
/// outright. Costs up to one `GRACEFUL_STOP_TIMEOUT` per subgraph, and
/// those overlap across agents.
pub graceful: bool,
}
/// The group-roots a [`rebuild_nodes`] subgraph exposes to its caller: what a
/// tail node edges onto, and what a follow-up node waits for.
///
@ -179,14 +175,14 @@ impl<'a> RebuildRoots<'a> {
/// cancel-cascades `Prebuild`, i.e. terminal, so the tail still runs). It
/// takes a fresh lease; the tiny gap is harmless — `Reconcile` converges to
/// the persisted `wanted` idempotently.
pub(crate) fn rebuild_nodes<'a>(
fn rebuild_subtree<'a>(
b: &'a Job,
agent: &str,
opts: RebuildOpts,
relock: bool,
graceful: bool,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
let a = || agent.to_owned();
let RebuildOpts { relock, graceful } = opts;
let mut meta_sync = b
.node(NodeKind::MetaSync { agent: a(), relock })
@ -249,6 +245,35 @@ pub(crate) fn rebuild_nodes<'a>(
}
}
/// The rebuild subtree, stopping the agent outright — the shape five of the six
/// call sites want. `relock` re-locks the meta flake inside `MetaSync`; `after`,
/// when given, is the node this subgraph chains behind. See
/// [`rebuild_subtree`] for the structure.
pub(crate) fn rebuild_nodes<'a>(
b: &'a Job,
agent: &str,
relock: bool,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
rebuild_subtree(b, agent, relock, false, after)
}
/// As [`rebuild_nodes`], but the agent gets a `Signal` → `Drain` window to
/// finish the turn in flight before it is stopped. Costs up to one
/// `GRACEFUL_STOP_TIMEOUT` per subgraph, and those overlap across agents.
///
/// A separate entry point rather than a flag because `graceful` does not
/// *prepend* nodes — it **re-parents** the stop root, so a caller cannot
/// declare it without being handed the internals. Only the boot sweep wants it.
pub(crate) fn graceful_rebuild_nodes<'a>(
b: &'a Job,
agent: &str,
relock: bool,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
rebuild_subtree(b, agent, relock, true, after)
}
/// The rebuild subgraph a [`NodeKind::DeployApply`] grows into its own DAG once
/// the merge has landed and `prepare_deploy` has staged the lock, plus the
/// [`NodeKind::FinalizeDeploy`] that closes the window behind it.
@ -274,15 +299,7 @@ pub(crate) fn rebuild_nodes<'a>(
/// and `FinalizeDeploy` declare is re-entered from the ancestor already holding
/// it rather than deadlocking against it.
pub(crate) fn deploy_rebuild_nodes(b: &Job, agent: &str, approval_id: i64) {
let roots = rebuild_nodes(
b,
agent,
RebuildOpts {
relock: false,
graceful: false,
},
None,
);
let roots = rebuild_nodes(b, agent, false, None);
let _finalize = b
.node(NodeKind::FinalizeDeploy {
agent: agent.to_owned(),
@ -305,15 +322,7 @@ pub(crate) fn deploy_rebuild_nodes(b: &Job, agent: &str, approval_id: i64) {
/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so
/// it reaches `Done` even after a failed swap and the tail would report success.
pub fn rebuild(b: &Job, agent: &str, relock: bool) {
let roots = rebuild_nodes(
b,
agent,
RebuildOpts {
relock,
graceful: false,
},
None,
);
let roots = rebuild_nodes(b, agent, relock, None);
emit_rebuilt_tails(b, agent, &roots.all());
}
@ -428,15 +437,7 @@ pub fn perm_change(b: &Job, agent: &str, payload: PermPayload) {
payload,
})
.needs(Resource::MetaWindow);
let roots = rebuild_nodes(
b,
agent,
RebuildOpts {
relock: true,
graceful: false,
},
Some(write),
);
let roots = rebuild_nodes(b, agent, true, Some(write));
emit_rebuilt_tails(
b,
agent,

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

@ -427,15 +427,7 @@ fn graceful_rebuild_chain_drains_before_stopping() {
let q = JobQueue::new(1);
let id = q
.submit(Source::AutoUpdate, "sweep".to_owned(), |b: &Job| {
templates::rebuild_nodes(
b,
"agent-a",
templates::RebuildOpts {
relock: true,
graceful: true,
},
None,
);
templates::graceful_rebuild_nodes(b, "agent-a", true, None);
})
.expect("valid shape");
assert_eq!(
@ -469,15 +461,7 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
// 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",
templates::RebuildOpts {
relock: true,
graceful: false,
},
None,
);
templates::rebuild_nodes(b, "agent-a", true, None);
});
assert_eq!(
declared_shape(&q, id)
@ -957,9 +941,10 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
/// 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.
/// `meta_update_grows_cascade_in_dag`, which differed only in which rebuild
/// flavour they grew and each minted a builder by hand to simulate the graft.
/// What they were checking is the `grown_*_rebuilds` templates, so this calls
/// one — the boot sweep's, since that is the caller that grows a graceful one.
///
/// That the grafted work lands under the emitter, and that the emitter parks in
/// `Finishing` until it settles, is `hive_jobq`'s
@ -970,14 +955,7 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
let agents = vec!["alice".to_owned(), "bob".to_owned()];
let id = q
.submit(Source::AutoUpdate, "sweep".to_owned(), |b: &Job| {
templates::grown_rebuilds(
b,
&agents,
templates::RebuildOpts {
relock: true,
graceful: true,
},
);
templates::grown_graceful_rebuilds(b, &agents, true);
})
.expect("valid shape");