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

refactor(#2897): carry the meta-update inputs on the MetaLock node

Browse source 
Second of the `Dag` field removals, and the same shape as the first:
`DagSpec`/`NodeKind::Dag` carried an `inputs: Vec<String>` that exactly
one node ever read. Both reads live inside `run_meta_lock` — the
`meta::lock_update` call and the `meta_update_cascade_agents` fan-out —
so the list now rides `NodeKind::MetaLock` itself.

The executor stops touching `Claim` for this node entirely: its dispatch
arm already destructured `MetaLock { sweep, fanout }`, so `inputs` joins
them and the `claim` parameter, which had no other use, is gone.

Falls out of that:
- `Claim::inputs` and `DagMeta::inputs` delete.
- `dag_view`'s DAG-level projection onto the `MetaLock` node reads the
  payload instead. The wire `NodeView::inputs` is unchanged: still
  populated on the `meta_lock` node alone.
- the boot sweep names no inputs (it bumps `hyperhive` alone via
  `lock_update_hyperhive`), which the construction site now says out loud
  rather than leaving implicit in an empty DAG-level field.

Checked with clippy (`--all-targets -D warnings`), `cargo test -p
hive-c0re` (320 passed) and `nix fmt`. No option surface is touched, so
no nix-eval gate.
This commit is contained in:
atlas 2026-08-01 13:23:19 +02:00
commit af2b1ce0e2
7 changed files with 22 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -80,9 +80,11 @@ pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await,
NodeKind::Provision { .. } => run_provision(coord, claim).await,
NodeKind::Create { .. } => run_create(claim).await,
NodeKind::MetaLock { sweep, fanout } => {
run_meta_lock(coord, claim, *sweep, fanout.clone()).await
}
NodeKind::MetaLock {
sweep,
fanout,
inputs,
} => run_meta_lock(coord, *sweep, fanout.clone(), inputs).await,
NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await,
NodeKind::Start { .. } => run_start(coord, claim).await,
NodeKind::Stop { .. } => run_stop(coord, claim).await,
@ -318,9 +320,9 @@ async fn run_create(claim: &Claim) -> Result<NodeOutput> {
/// flavour propagates errors, and a failed bump fans out nothing.
async fn run_meta_lock(
coord: &Arc<Coordinator>,
claim: &Claim,
sweep: bool,
fanout: Option<Vec<String>>,
inputs: &[String],
) -> Result<NodeOutput> {
if sweep {
if let Err(e) = crate::meta::lock_update_hyperhive().await {
@ -353,12 +355,12 @@ async fn run_meta_lock(
return Ok(NodeOutput { append_subgraph });
}
let _progress = coord.meta_update_guard();
crate::meta::lock_update(&claim.inputs).await?;
crate::meta::lock_update(inputs).await?;
// Lock file changed — meta-inputs panel re-renders.
crate::dashboard::emit_meta_inputs_snapshot(coord);
let cascade = match fanout {
Some(list) => list,
None => meta_update_cascade_agents(&claim.inputs).await,
None => meta_update_cascade_agents(inputs).await,
};
// Grow one rebuild subgraph per affected agent into *this* meta-update
// DAG (rooted on this `MetaLock`, so they build against the post-bump

13
hive-c0re/src/job_queue/mod.rs
View file

@ -70,7 +70,6 @@ pub struct Claim {
/// The agent this node targets (its own, not a DAG-level field). Empty for
/// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes.
pub agent: String,
pub inputs: Vec<String>,
/// Transient pill kind for the lease window (from the spec). Whether the
/// pill is currently shown is derived from live lease ownership
/// ([`JobQueue::held_transients`]), not a per-claim edge.
@ -93,7 +92,6 @@ struct DagMeta {
source: Source,
reason: String,
transient: Option<TransientKind>,
inputs: Vec<String>,
created_at: i64,
}
@ -218,7 +216,6 @@ impl JobQueue {
source: spec.source,
reason: spec.reason,
transient: spec.transient,
inputs: spec.inputs,
created_at: now_unix(),
},
Vec::new(),
@ -304,7 +301,6 @@ impl JobQueue {
node_id: id,
kind,
agent,
inputs: meta.inputs,
transient: meta.transient,
});
// `started_at` is stamped on the graph `Node` by the scheduler's
@ -481,7 +477,6 @@ impl QueueInner {
source,
reason,
transient,
inputs,
created_at,
} = &self.sched.graph().node(container)?.payload
else {
@ -491,7 +486,6 @@ impl QueueInner {
source: *source,
reason: reason.clone(),
transient: *transient,
inputs: inputs.clone(),
created_at: *created_at,
})
}
@ -549,10 +543,9 @@ impl QueueInner {
NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id),
_ => None,
};
let inputs = if matches!(node.payload, NodeKind::MetaLock { .. }) {
meta.inputs.clone()
} else {
Vec::new()
let inputs = match &node.payload {
NodeKind::MetaLock { inputs, .. } => inputs.clone(),
_ => Vec::new(),
};
let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id);
// `node.parent` is the structural jobq parent. Top-level nodes

10
hive-c0re/src/job_queue/model.rs
View file

@ -84,15 +84,20 @@ pub enum NodeKind {
/// upstream `Provision` node already registered the agent in meta.
Create { agent: String },
/// Meta flake lock bump. `sweep = false`: `meta::lock_update`
/// (commit fused, under `META_LOCK`) with the DAG's `inputs`;
/// (commit fused, under `META_LOCK`) with this node's own `inputs`;
/// `sweep = true`: `meta::lock_update_hyperhive`, *non-fatal* (a
/// failed boot-time bump must not cancel the fan-out rebuilds).
/// On success the scheduler appends child `Rebuild` DAGs: the
/// precomputed `fanout` list when present (boot sweep), else the
/// post-bump affected set (`meta_update_cascade_agents`).
///
/// `inputs` are the flake inputs to bump — empty means "all", and the
/// boot sweep leaves them empty since it bumps `hyperhive` alone. They
/// ride this node because it is the only thing that reads them.
MetaLock {
sweep: bool,
fanout: Option<Vec<String>>,
inputs: Vec<String>,
},
/// Idempotent power converge *planner*: read `wanted` + observed
/// state and decide the action (start if `Up` & down, stop if
@ -281,7 +286,6 @@ pub enum NodeKind {
source: Source,
reason: String,
transient: Option<TransientKind>,
inputs: Vec<String>,
created_at: i64,
},
}
@ -459,8 +463,6 @@ pub struct DagSpec {
pub source: Source,
/// Free-form "why".
pub reason: String,
/// Meta-update only: the inputs to bump. Display copy lives on the DAG.
pub inputs: Vec<String>,
/// Dashboard transient pill (and crash-watch suppression) held for
/// the lease window — from lease acquisition to DAG terminal.
pub transient: Option<crate::coordinator::TransientKind>,

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

@ -207,7 +207,6 @@ fn power_dag(
DagSpec {
source,
reason,
inputs: Vec::new(),
transient: Some(transient),
nodes,
}

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

@ -367,7 +367,6 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag
DagSpec {
source,
reason,
inputs: Vec::new(),
transient: Some(TransientKind::Rebuilding),
nodes,
}
@ -403,7 +402,6 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec
DagSpec {
source: Source::Approval,
reason,
inputs: Vec::new(),
transient: Some(TransientKind::Rebuilding),
nodes: vec![
node(
@ -462,7 +460,6 @@ pub fn reconcile_only(
DagSpec {
source,
reason,
inputs: Vec::new(),
transient,
nodes: vec![node(
NodeKind::Reconcile {
@ -488,7 +485,6 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec {
DagSpec {
source: Source::Approval,
reason,
inputs: Vec::new(),
transient: Some(TransientKind::Spawning),
nodes: {
let a = || agent.to_owned();
@ -533,7 +529,6 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay
DagSpec {
source,
reason,
inputs: Vec::new(),
transient: Some(TransientKind::Rebuilding),
nodes,
}
@ -559,6 +554,7 @@ pub fn meta_update(
NodeKind::MetaLock {
sweep: false,
fanout: None,
inputs,
},
Vec::new(),
)];
@ -572,7 +568,6 @@ pub fn meta_update(
DagSpec {
source,
reason,
inputs,
transient: Some(TransientKind::Rebuilding),
nodes,
}
@ -595,7 +590,6 @@ pub fn reparent(
DagSpec {
source,
reason,
inputs: Vec::new(),
transient: None,
nodes: vec![node(NodeKind::Reparent { moves }, Vec::new())],
}

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

@ -246,7 +246,6 @@ fn graceful_rebuild_chain_drains_before_stopping() {
let spec = DagSpec {
source: Source::AutoUpdate,
reason: "sweep".to_owned(),
inputs: Vec::new(),
transient: None,
nodes: templates::rebuild_nodes(
"agent-a",
@ -724,12 +723,12 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
let spec = DagSpec {
source: Source::AutoUpdate,
reason: "sweep".to_owned(),
inputs: Vec::new(),
transient: None,
nodes: vec![NodeSpec {
kind: NodeKind::MetaLock {
sweep: true,
fanout: None,
inputs: Vec::new(),
},
deps: Vec::new(),
parent: None,

4
hive-c0re/src/workers/auto_update.rs
View file

@ -328,6 +328,9 @@ fn submit_boot_tree(
kind: NodeKind::MetaLock {
sweep: true,
fanout: Some(fanout),
// A sweep bumps `hyperhive` alone (`lock_update_hyperhive`),
// so it names no inputs.
inputs: Vec::new(),
},
deps: Vec::new(),
parent: None,
@ -347,7 +350,6 @@ fn submit_boot_tree(
// land; the boot DAG as a whole has no terminal side effect, so no tail.
source: Source::AutoUpdate,
reason,
inputs: Vec::new(),
// Rebuilding when the sweep will grow rebuild subgraphs (per-agent
// crash-watch suppression during their Swap, applied at claim time);
// a reconcile-only boot needs no transient.