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

refactor(#2441): move agent field from DAG onto Node; drop dedup

Browse source 
Agent was a single field on Dag/DagSpec, making a DAG structurally
one-agent — a multi-agent op could only ever be N separate DAGs. Move it
onto Node/NodeSpec (and the NodeView wire type), drop it from Dag/DagSpec
(and DagView): a DAG can now span agents.

- lifecycle lease keys on the node's agent, still globally exclusive per
  agent across all DAGs (Inner.leases unchanged in shape). A DAG holds one
  lease per distinct agent it touches; settle() frees each at DAG-terminal
  (per-agent-subgraph early release is a follow-up, only observable with
  multi-agent DAGs).
- transient guard keyed (dag_id, agent); cancel-revert + Rebuilt events
  walk TerminalDag.agents.
- submit-time dedup removed (a multi-agent DAG has no single agent to key
  on); every submit enqueues a fresh DAG. Whether dedup needs reintroducing
  is tracked in a follow-up sub-issue.
- templates gain a node(agent, kind, deps) helper stamping the agent onto
  every node; meta templates stamp "hyperhive".

Templates stay single-agent in this PR — behaviour is unchanged, only the
representation + wire shape. Multi-agent DAG emission (restart/restart-all/
broad stop+start as one DAG) and the SetWanted-as-a-node change are
follow-ups off #2439.
This commit is contained in:
atlas 2026-07-14 21:42:26 +02:00 • committed by mara
commit 2a59f2f5fc
7 changed files with 223 additions and 356 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -7,10 +7,12 @@
//! Concurrency is gated by two resource classes:
//! 1. **Build slots** — N permits (`services.hyperhive.c0re.buildSlots`,
//! default 1) held by nix-heavy nodes for the node's duration.
//! 2. **Per-agent lifecycle lease** — DAG-scoped: acquired before the
//! DAG's first container-affecting node runs, held until the DAG is
//! terminal, so two lifecycle DAGs for one agent never interleave
//! their container ops.
//! 2. **Per-agent lifecycle lease** — keyed on the *node's* agent and
//! globally exclusive per agent across all DAGs: acquired before a
//! container-affecting node runs, held (by the owning DAG) until no
//! live node of that DAG still targets the agent, so two DAGs never
//! interleave container ops on the same agent. A DAG spanning multiple
//! agents holds one lease per agent it touches.
//!
//! The meta *repo* is serialized by `meta::META_LOCK` inside the
//! executors themselves. Per-agent power *intent* (`wanted`) lives in
@ -57,15 +59,18 @@ pub struct Claim {
pub dag_id: u64,
pub node_id: NodeId,
pub kind: NodeKind,
/// The agent this node targets (its own, not a DAG-level field). The
/// executor operates on this agent's container; the lease is keyed on
/// it.
pub agent: String,
pub template: Template,
pub source: Source,
pub approval_id: Option<i64>,
pub inputs: Vec<String>,
pub perm_payload: Option<PermPayload>,
/// True when claiming this node acquired the DAG's agent lease —
/// the scheduler creates the DAG-scoped transient guard on this
/// edge.
/// True when claiming this node newly acquired its agent's lease —
/// the scheduler creates the per-`(dag, agent)` transient guard on
/// this edge.
pub lease_acquired: bool,
/// Transient pill kind for the lease window (from the spec).
pub transient: Option<crate::coordinator::TransientKind>,
@ -78,7 +83,10 @@ pub struct Claim {
pub struct TerminalDag {
pub dag_id: u64,
pub template: Template,
pub agent: String,
/// Distinct agents this DAG's nodes targeted (one for a single-agent
/// DAG). The cancel-revert hook walks these to snap each agent's
/// power intent back on a cancelled power-op DAG.
pub agents: Vec<String>,
pub approval_id: Option<i64>,
pub state: State,
/// First failed node's error when `state == Failed`.
@ -128,27 +136,17 @@ impl JobQueue {
}
}
/// Submit a DAG. Validates the spec (cycle rejection) and dedups
/// against non-started DAGs; returns the DAG id (newly-allocated,
/// or the existing DAG's id with the new reason appended).
/// Submit a DAG. Validates the spec (cycle rejection) and returns the
/// newly-allocated DAG id.
///
/// Dedup: a DAG whose roll-up is still `Queued` (no node started)
/// with the same `(template, agent, parent_id, approval_id)` — plus
/// `inputs` for `MetaUpdate` and the perm-type discriminant for
/// `PermChange` — swallows the repeat. `parent_id` is part of the
/// key so a meta-update cascade rebuild never collapses into a
/// standalone or sweep rebuild. Running / terminal DAGs never
/// dedup — operators are free to re-queue.
/// Submit-time dedup was removed with the agent-per-node refactor
/// (a multi-agent DAG has no single agent to key a dedup on) — every
/// submit now enqueues a fresh DAG. Whether any dedup needs
/// reintroducing (and in what form) is tracked as a follow-up; see the
/// dedup re-evaluation issue.
pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> {
templates::validate(&spec)?;
let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
if let Some(existing) = Self::dedup_target(&mut inner, &spec) {
if !existing.reason.contains(&spec.reason) {
use std::fmt::Write as _;
let _ = write!(existing.reason, "\nalso requested by: {}", spec.reason);
}
return Ok(existing.id);
}
let id = Self::push_dag(&mut inner, spec);
drop(inner);
self.notify.notify_one();
@ -156,8 +154,8 @@ impl JobQueue {
}
/// Append fan-out children under a parent DAG (meta-update / sweep
/// cascade). Applies the same dedup as [`Self::submit`]; returns
/// the child ids actually created or coalesced into.
/// cascade); returns the child ids created. No dedup (see
/// [`Self::submit`]).
pub fn append_children(&self, specs: Vec<DagSpec>) -> Vec<u64> {
let mut ids = Vec::with_capacity(specs.len());
for spec in specs {
@ -183,9 +181,14 @@ impl JobQueue {
pub fn append_node(&self, dag_id: u64, kind: NodeKind, dep_on: NodeId) -> Option<NodeId> {
let mut inner = self.inner.lock().expect("job_queue mutex poisoned");
let dag = inner.dags.iter_mut().find(|d| d.id == dag_id)?;
// The appended sub-step targets the same agent as the node that
// emitted it (a `Reconcile` fanning out its `Start`/`Stop` acts on
// the same container), so inherit `dep_on`'s agent.
let agent = dag.node(dep_on)?.agent.clone();
let new_id: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX);
dag.nodes.push(Node {
id: new_id,
agent,
kind,
deps: vec![model::Dep {
on: dep_on,
@ -203,21 +206,6 @@ impl JobQueue {
Some(new_id)
}
fn dedup_target<'a>(inner: &'a mut Inner, spec: &DagSpec) -> Option<&'a mut Dag> {
inner.dags.iter_mut().find(|d| {
d.rollup() == State::Queued
&& d.template == spec.template
&& d.agent == spec.agent
&& d.parent_id == spec.parent_id
&& d.approval_id == spec.approval_id
&& (d.template != Template::MetaUpdate || d.inputs == spec.inputs)
&& model::perm_payload_same_type(
d.perm_payload.as_ref(),
spec.perm_payload.as_ref(),
)
})
}
fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 {
inner.next_id += 1;
let id = inner.next_id;
@ -227,6 +215,7 @@ impl JobQueue {
.enumerate()
.map(|(i, n)| Node {
id: u32::try_from(i).unwrap_or(u32::MAX),
agent: n.agent,
kind: n.kind,
deps: n.deps,
state: State::Queued,
@ -240,7 +229,6 @@ impl JobQueue {
inner.dags.push_back(Dag {
id,
template: spec.template,
agent: spec.agent,
source: spec.source,
reason: spec.reason,
parent_id: spec.parent_id,
@ -282,16 +270,23 @@ impl JobQueue {
let dag = &inner.dags[di];
let node = dag.node(node_id).expect("node id from same dag");
let needs_slot = node.kind.needs_build_slot();
let needs_lease = node.kind.needs_lease();
// The lifecycle lease is keyed on the *node's* agent, held
// by this DAG (dag_id) — still globally exclusive per agent
// across all DAGs. A multi-agent DAG acquires one lease per
// agent it touches; each is released in `settle` once no
// live node of this DAG still targets that agent.
let node_agent = node.agent.clone();
if needs_slot && inner.slots_used >= inner.build_slots {
continue;
}
let mut lease_acquired = false;
if node.kind.needs_lease() {
match inner.leases.get(dag.agent.as_str()) {
if needs_lease {
match inner.leases.get(node_agent.as_str()) {
Some(&holder) if holder != dag_id => continue,
Some(_) => {}
None => {
inner.leases.insert(dag.agent.clone(), dag_id);
inner.leases.insert(node_agent.clone(), dag_id);
lease_acquired = true;
}
}
@ -304,7 +299,7 @@ impl JobQueue {
dag_id,
node_id,
kind: dag.node(node_id).expect("node").kind.clone(),
agent: dag.agent.clone(),
agent: node_agent,
template: dag.template,
source: dag.source,
approval_id: dag.approval_id,
@ -425,13 +420,23 @@ impl JobQueue {
continue;
}
dag.terminal_reported = true;
if inner.leases.get(dag.agent.as_str()) == Some(&dag.id) {
freed.push(dag.agent.clone());
// Free every agent-lease this DAG holds (one per distinct
// agent it touched). Single-agent DAGs release their one lease;
// a future multi-agent DAG releases all of them at terminal.
// (Per-agent early release — freeing an agent's lease the moment
// that agent's subgraph is terminal rather than at whole-DAG
// terminal — is a refinement for the multi-agent-emission
// follow-up, where it actually matters.)
let agents = dag.agents();
for agent in &agents {
if inner.leases.get(agent.as_str()) == Some(&dag.id) {
freed.push(agent.clone());
}
}
reports.push(TerminalDag {
dag_id: dag.id,
template: dag.template,
agent: dag.agent.clone(),
agents,
approval_id: dag.approval_id,
state: dag.rollup(),
error: dag.first_error().map(str::to_owned),