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(job-queue): build DAGs by naming nodes, not counting them

Browse source 
Every template built a `Vec<NodeSpec>` whose edges and parents were
positional indices into that vector, so a shape was expressed as
arithmetic: `base + 1`, `stop_root + 2`, `sfu + 1`, and a
`reconcile_index()` helper that read the emitted vector's length to find
out where its own last node had landed. `concat_subgraphs` existed
solely to rebase one per-agent subgraph's indices onto another's.

Templates now declare into a `hive_jobq::JobBuilder` and hold the
handles they get back, so an edge names the node it waits on. The
arithmetic is gone, and with it:

- `NodeSpec` and the job-queue's own index-based `Dep`.
- `insert_group`'s index resolution — it wraps `Scheduler::insert_job`.
- `concat_subgraphs` — per-agent chains share one builder and each keeps
  its own root, so independence is structural rather than computed.
- `reconcile_index` and `dep_index`.
- `templates::validate` and its petgraph toposort. It rejected dangling
  deps and cycles; both are now unrepresentable, since a handle only
  exists for an already-declared node and every edge therefore points
  backwards. (petgraph stays in the tree for `agent_config::topology`.)

`NodeOutput.append_subgraph` becomes `Vec<Job>`: an executor cannot
reach the queue, so it hands back declarations and the scheduler inserts
them under its own lock. That is what the in-DAG growth path always
wanted — a transferable declaration, not a vector of specs.

Resource declaration is unchanged in behaviour: the `templates::node`
helper applies `NodeKind::resource_deps()` at the construction site, so
every node still declares what its kind needs. Moving that declaration
to the call sites is #2818's job; this leaves it one place to delete.

Three tests went with the guard they covered — they hand-built malformed
specs out of indices, which is the representation that made those shapes
possible. Two more now read a DAG's shape off the queue rather than out
of a spec vector, which is where it is observable. The remaining 45
job-queue tests are unchanged and still pass: lease serialization,
roll-up, cancel-cascade, in-DAG growth and per-agent concurrency all
behave as before.
This commit is contained in:
atlas 2026-08-02 13:05:17 +02:00 • committed by mara
commit e7c3cf5a3d
9 changed files with 528 additions and 696 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -10,10 +10,10 @@ use std::sync::Arc;
use anyhow::{Context as _, Result}; use anyhow::{Context as _, Result};
use super::Claim; use super::{Claim, Job};
use hive_jobq::TerminalState; use hive_jobq::TerminalState;
use super::model::{NodeKind, NodeSpec}; use super::model::NodeKind;
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
use crate::power::{ReconcileAction, reconcile_action}; use crate::power::{ReconcileAction, reconcile_action};
@ -30,19 +30,19 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
#[derive(Debug, Default)] #[derive(Debug, Default)]
pub struct NodeOutput { pub struct NodeOutput {
/// Whole per-agent *subgraphs* to append into *this same* DAG at /// Whole per-agent *subgraphs* to append into *this same* DAG at
/// runtime — the single in-DAG-growth channel. Each inner /// runtime — the single in-DAG-growth channel. Each [`Job`] is one
/// `Vec<NodeSpec>` is one independent subgraph whose `deps` are local /// independent subgraph, declared but not yet inserted: an executor cannot
/// (0-based within that subgraph); the scheduler appends each via /// reach the queue, so it hands the declaration back and the scheduler
/// [`super::JobQueue::append_subgraph`], which rebases the deps onto the DAG's /// inserts it via [`super::JobQueue::append_subgraph`] under its own lock,
/// node-id space and roots the subgraph on the emitting node. Used both /// rooted on the emitting node. Used both for the multi-node case
/// for the multi-node case (`MetaLock` growing one rebuild subgraph per /// (`MetaLock` growing one rebuild subgraph per agent — the startup
/// agent — the startup sweep's stale agents, the meta-update cascade's /// sweep's stale agents, the meta-update cascade's affected agents) and
/// affected agents) and the single-node case (a `Reconcile` planner /// the single-node case (a `Reconcile` planner emitting its mechanical
/// emitting its mechanical `Start` / `Stop` as a one-node subgraph). The /// `Start` / `Stop` as a one-node subgraph). The scheduler applies these
/// scheduler applies these *before* the emitting node's completion so the /// *before* the emitting node's completion so the DAG never rolls terminal
/// DAG never rolls terminal with the appended work still pending — keeping /// with the appended work still pending — keeping the lease-window
/// the lease-window transient held across the sub-step. /// transient held across the sub-step.
pub append_subgraph: Vec<Vec<NodeSpec>>, pub append_subgraph: Vec<Job>,
} }
/// Build-log sink for one claimed node. /// Build-log sink for one claimed node.
@ -342,14 +342,17 @@ async fn run_meta_lock(
.unwrap_or_default() .unwrap_or_default()
.iter() .iter()
.map(|agent| { .map(|agent| {
let job = Job::new();
super::templates::rebuild_nodes( super::templates::rebuild_nodes(
&job,
agent, agent,
super::templates::RebuildOpts { super::templates::RebuildOpts {
relock: true, relock: true,
graceful: true, graceful: true,
}, },
0, None,
) );
job
}) })
.collect(); .collect();
return Ok(NodeOutput { append_subgraph }); return Ok(NodeOutput { append_subgraph });
@ -371,14 +374,17 @@ async fn run_meta_lock(
let append_subgraph = cascade let append_subgraph = cascade
.iter() .iter()
.map(|agent| { .map(|agent| {
let job = Job::new();
super::templates::rebuild_nodes( super::templates::rebuild_nodes(
&job,
agent, agent,
super::templates::RebuildOpts { super::templates::RebuildOpts {
relock: false, relock: false,
graceful: false, graceful: false,
}, },
0, None,
) );
job
}) })
.collect(); .collect();
Ok(NodeOutput { append_subgraph }) Ok(NodeOutput { append_subgraph })
@ -398,7 +404,11 @@ async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOu
// One node targeting this agent, rooted on this reconcile node. `NodeKind` // One node targeting this agent, rooted on this reconcile node. `NodeKind`
// carries the agent it targets, so stamp `claim.agent` into the fanned-out // carries the agent it targets, so stamp `claim.agent` into the fanned-out
// Start/Stop kind (one in-DAG-growth channel). // Start/Stop kind (one in-DAG-growth channel).
let sub = |kind| vec![vec![super::templates::node(kind, Vec::new())]]; let sub = |kind| {
let job = Job::new();
let _ = super::templates::node(&job, kind);
vec![job]
};
let append_subgraph = match reconcile_action(wanted, running) { let append_subgraph = match reconcile_action(wanted, running) {
ReconcileAction::Start => sub(NodeKind::Start { ReconcileAction::Start => sub(NodeKind::Start {
agent: name.clone(), agent: name.clone(),

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

@ -47,9 +47,19 @@ use hive_jobq::{Dep, Graph, NodeId};
use tokio::sync::Notify; use tokio::sync::Notify;
pub use hive_jobq::TerminalState; pub use hive_jobq::TerminalState;
pub use model::{DagSpec, DagView, NodeKind, NodeSpec, PermPayload, Source, State}; pub use model::{DagSpec, DagView, NodeKind, PermPayload, Source, State};
use resource::Resource; use resource::Resource;
/// A job under construction: `hive_jobq`'s builder over this queue's payload
/// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into
/// one of these; [`JobQueue::submit`] inserts it.
pub type Job = hive_jobq::JobBuilder<NodeKind, Resource>;
/// A handle to one node a template declared — where its edges, grouping and
/// resources are declared. `Copy`; naming a node as a dependency does not
/// consume the ability to name it again.
pub type Handle<'a> = hive_jobq::NodeRef<'a, NodeKind, Resource>;
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
/// retains, newest first. A flat cap over the whole sorted list: the /// retains, newest first. A flat cap over the whole sorted list: the
/// dashboard renders one recent-builds list, so one number bounds it. /// dashboard renders one recent-builds list, so one number bounds it.
@ -143,56 +153,33 @@ impl Default for JobQueue {
} }
} }
/// Insert `nodes` into the shared graph, honouring the spec's explicit **parent /// Insert a declared `job` into the shared graph and record its per-node
/// axis**: a node with `parent = None` is a top-level group root (re-parented to /// `node_rt`, returning the inserted ids.
/// `group_parent`, which is `None` for `submit` and the emitting node for ///
/// `append_subgraph`); a node with `parent = Some(idx)` becomes a child of the /// A node that declared no parent hangs under `group_parent` — the DAG
/// already-inserted node at spec index `idx`. `deps` are translated to crate /// container for a template, the emitting node for a runtime-appended
/// `Dep::Node` edges verbatim — templates declare the parent axis + sibling /// subgraph. Templates declare the parent axis + sibling ordering directly, so
/// ordering directly, so there is no dep-on-root to drop and no lease to hoist: /// there is no dep-on-root to drop and no lease to hoist: each node declares
/// each node declares its own `Dep::Resource`, and the crate's borrow model /// its own resources, and the crate's borrow model keeps a resource continuous
/// keeps a resource continuous across a subtree (a root owns it, descendants /// across a subtree (a root owns it, descendants borrow it). Independent group
/// borrow it). Independent group roots (multiple `parent = None` nodes) carry no /// roots carry no cross-links, so a multi-agent DAG's per-agent subgraphs run
/// cross-links, so a multi-agent DAG's per-agent subgraphs run concurrently, each /// concurrently, each on its own lease.
/// on its own lease. Records per-node `node_rt`. Returns the inserted ids
/// (index-aligned with `nodes`). A node with `parent = None` is re-parented to
/// `group_parent` (the DAG container for a template, or the emitting node for a
/// runtime-appended subgraph); a node's `parent` / dep targets must precede it
/// in `nodes` (submit-time `validate` enforces density + acyclicity).
/// ///
/// # Errors /// # Errors
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group( fn insert_group(
inner: &mut QueueInner, inner: &mut QueueInner,
nodes: &[NodeSpec], job: Job,
group_parent: Option<NodeId>, group_parent: Option<NodeId>,
) -> anyhow::Result<Vec<NodeId>> { ) -> anyhow::Result<Vec<NodeId>> {
let mut ids: Vec<NodeId> = Vec::with_capacity(nodes.len()); let ids = inner
for ns in nodes { .sched
let payload = ns.kind.clone(); .insert_job(job, group_parent)
let mut deps: Vec<Dep<Resource>> = payload .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
.resource_deps() for &id in ids.values() {
.into_iter()
.map(|(name, count)| Dep::Resource { name, count })
.collect();
for d in &ns.deps {
deps.push(Dep::Node {
id: ids[dep_index(d.on)],
when: d.when,
});
}
let parent = match ns.parent {
Some(idx) => Some(ids[dep_index(idx)]),
None => group_parent,
};
let id = inner
.sched
.append(payload, deps, parent)
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
ids.push(id);
inner.node_rt.insert(id, NodeRuntime::default()); inner.node_rt.insert(id, NodeRuntime::default());
} }
Ok(ids) Ok(ids.into_values().collect())
} }
impl JobQueue { impl JobQueue {
@ -225,7 +212,6 @@ impl JobQueue {
/// Propagates the spec-validation error (empty / cyclic / bad parent) or a /// Propagates the spec-validation error (empty / cyclic / bad parent) or a
/// graph-insert error (dependencies that aren't dependency-topological). /// graph-insert error (dependencies that aren't dependency-topological).
pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> { pub fn submit(&self, spec: DagSpec) -> anyhow::Result<u64> {
templates::validate(&spec)?;
let mut inner = self.lock(); let mut inner = self.lock();
let container = inner let container = inner
.sched .sched
@ -240,7 +226,7 @@ impl JobQueue {
) )
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
inner.node_rt.insert(container, NodeRuntime::default()); inner.node_rt.insert(container, NodeRuntime::default());
insert_group(&mut inner, &spec.nodes, Some(container))?; insert_group(&mut inner, spec.job, Some(container))?;
// Settle the container's own (no-op) logic immediately so it parks in // Settle the container's own (no-op) logic immediately so it parks in
// `Finishing` and its children become runnable — it never needs claiming // `Finishing` and its children become runnable — it never needs claiming
// or executing, and stays out of `claim_ready`. It rolls up terminal when // or executing, and stays out of `claim_ready`. It rolls up terminal when
@ -261,8 +247,8 @@ impl JobQueue {
/// the DAG's terminal node deps on the top root, roll-up keeps the DAG from /// the DAG's terminal node deps on the top root, roll-up keeps the DAG from
/// settling early with no explicit wiring. Returns the new node ids; empty if /// settling early with no explicit wiring. Returns the new node ids; empty if
/// the DAG is gone or `nodes` is empty. /// the DAG is gone or `nodes` is empty.
pub fn append_subgraph(&self, dag_id: u64, nodes: &[NodeSpec], dep_on: NodeId) -> Vec<NodeId> { pub fn append_subgraph(&self, dag_id: u64, job: Job, dep_on: NodeId) -> Vec<NodeId> {
if nodes.is_empty() { if job.is_empty() {
return Vec::new(); return Vec::new();
} }
let mut inner = self.lock(); let mut inner = self.lock();
@ -275,7 +261,7 @@ impl JobQueue {
// emitter stays `Finishing` until this appended subtree settles, and the // emitter stays `Finishing` until this appended subtree settles, and the
// container node rolls up terminal only once its whole subtree (incl. this // container node rolls up terminal only once its whole subtree (incl. this
// appended work) has settled, so the DAG hook waits for free. // appended work) has settled, so the DAG hook waits for free.
let ids = match insert_group(&mut inner, nodes, Some(dep_on)) { let ids = match insert_group(&mut inner, job, Some(dep_on)) {
Ok(ids) => ids, Ok(ids) => ids,
Err(e) => { Err(e) => {
tracing::error!( tracing::error!(
@ -679,12 +665,6 @@ impl QueueInner {
} }
} }
/// A spec dependency index (`Dep.on`, a wire `u64`) as a `usize` for indexing
/// into the node/id vectors. `templates::validate` guarantees it's in range.
fn dep_index(on: u64) -> usize {
usize::try_from(on).unwrap_or(usize::MAX)
}
/// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`. /// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`.
fn truncate_error(e: &str) -> String { fn truncate_error(e: &str) -> String {
if e.len() <= MAX_ERROR_LEN { if e.len() <= MAX_ERROR_LEN {

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

@ -13,18 +13,10 @@
//! full design. //! full design.
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
pub use hive_host_sock::jobs::{DagView, NodeId, PermPayload, Source, State}; pub use hive_host_sock::jobs::{DagView, PermPayload, Source, State};
use serde::Serialize; use serde::Serialize;
use hive_jobq::{DepWhen, TerminalState}; use hive_jobq::TerminalState;
/// A dependency edge (intra-DAG only — cross-DAG ordering comes from
/// the per-agent lease + dedup, never from edges between DAGs).
#[derive(Debug, Clone, Copy, Serialize)]
pub struct Dep {
pub on: NodeId,
pub when: DepWhen,
}
/// The primitive operations — each kind maps to one executor fn in /// The primitive operations — each kind maps to one executor fn in
/// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta.rs` /// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta.rs`
@ -469,35 +461,24 @@ impl NodeKind {
} }
} }
/// Submit-time spec for one node. /// Submit-time spec for a whole DAG: the group's metadata plus the declared —
#[derive(Debug, Clone)] /// not yet inserted — nodes. Built by `templates.rs`, inserted by
pub struct NodeSpec { /// `JobQueue::submit`.
/// The node's payload — [`NodeKind`] is the queue's payload type directly, ///
/// and each variant carries the agent it targets (a DAG can span agents; /// No DAG-level `agent` — every node carries its own (a DAG can span agents),
/// the queue derives per-agent leasing from [`NodeKind::agent`]). /// and the queue derives per-agent leasing from [`NodeKind::agent`].
pub kind: NodeKind, /// Type-specific payloads (`PermChange`'s file payload) ride the node that
pub deps: Vec<Dep>, /// consumes them ([`NodeKind::WritePermFile`]), not this generic spec.
/// The **structural parent** axis — the spec-local index of this node's ///
/// group parent, or `None` for a top-level (group-root) node. Independent /// There is no separate per-node spec type: the nodes live in the builder,
/// of `deps`: `deps` order execution, `parent` groups nodes into a subtree /// which inserts them itself. A shape that has been declared is therefore
/// whose resource the whole subtree borrows (the agent lease is owned by a /// always insertable — a dangling edge or a cycle cannot be expressed, so
/// group root and re-entered by its descendants for continuity). A child /// there is nothing left for a submit-time validation pass to reject.
/// runs once its parent reaches `Finishing` (the parent gate), so a child #[derive(Debug)]
/// never `deps` on its own parent (that would deadlock — dep-scope
/// validation rejects it).
pub parent: Option<u64>,
}
/// Submit-time spec for a whole DAG. Built by `templates.rs`; validated
/// (cycle rejection) by `JobQueue::submit`. No DAG-level `agent` — every
/// node carries its own (a DAG can span agents), and the queue derives
/// per-agent leasing from [`NodeKind::agent`]. Type-specific payloads
/// (`PermChange`'s file payload) ride the node that consumes them
/// ([`NodeKind::WritePermFile`]), not this generic spec.
#[derive(Debug, Clone)]
pub struct DagSpec { pub struct DagSpec {
pub source: Source, pub source: Source,
/// Free-form "why". /// Free-form "why".
pub reason: String, pub reason: String,
pub nodes: Vec<NodeSpec>, /// The DAG's declared nodes, with their edges, grouping and resources.
pub job: super::Job,
} }

2
hive-c0re/src/job_queue/scheduler.rs
View file

@ -125,7 +125,7 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
// `Done` just below — covers both the multi-node case (a `MetaLock` // `Done` just below — covers both the multi-node case (a `MetaLock`
// growing per-agent rebuild subgraphs) and the single-node case (a // growing per-agent rebuild subgraphs) and the single-node case (a
// `Reconcile` planner's `Start` / `Stop`). // `Reconcile` planner's `Start` / `Stop`).
for subgraph in &output.append_subgraph { for subgraph in output.append_subgraph {
coord coord
.job_queue .job_queue
.append_subgraph(claim.dag_id, subgraph, claim.node_id); .append_subgraph(claim.dag_id, subgraph, claim.node_id);

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

@ -6,8 +6,8 @@
//! state, which needs an async `lifecycle::is_running` read that a pure/sync //! state, which needs an async `lifecycle::is_running` read that a pure/sync
//! template can't do. So these fns are async — they read each agent's state, //! template can't do. So these fns are async — they read each agent's state,
//! assemble a per-agent subgraph out of the shared pure primitives //! assemble a per-agent subgraph out of the shared pure primitives
//! (`templates::{node, after_ok, rebuild_nodes}`), and concatenate them into //! (`templates::{node, rebuild_nodes}`), all declaring into ONE job
//! ONE DAG (independent per-agent roots, concurrent on their own leases). //! (independent per-agent roots, concurrent on their own leases).
//! //!
//! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable //! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable
//! intent write) — `restart` does NOT (it bounces the container but leaves //! intent write) — `restart` does NOT (it bounces the container but leaves
@ -24,9 +24,9 @@
use std::sync::Arc; use std::sync::Arc;
use super::model::{DagSpec, Dep, NodeKind, NodeSpec}; use super::model::{DagSpec, NodeKind};
use super::templates::{RebuildOpts, after_ok, child, node, rebuild_nodes}; use super::templates::{RebuildOpts, node, rebuild_nodes};
use super::{Source, templates}; use super::{Job, Source, templates};
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
use crate::lifecycle; use crate::lifecycle;
@ -51,71 +51,75 @@ pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: St
// The pure per-agent chain builders below take `running` (and `stale`) // The pure per-agent chain builders below take `running` (and `stale`)
// explicitly so they stay pure + unit-testable without a live container; // explicitly so they stay pure + unit-testable without a live container;
// the async `*_many` fns read the real state via `lifecycle::is_running` // the async `*_many` fns read the real state via `lifecycle::is_running`
// then hand it in. Each chain uses LOCAL (0-based) deps; `concat_subgraphs` // then hand it in. Each chain declares into the shared job it is handed, and
// rebases them into one DAG. // names the nodes it depends on — so there is nothing to rebase.
/// One agent's **stop** subgraph. `SetWanted(Off)` head + `Reconcile` tail /// One agent's **stop** subgraph. `SetWanted(Off)` head + `Reconcile` tail
/// always; the graceful `Signal → Drain` quiesce only when the agent is /// always; the graceful `Signal → Drain` quiesce only when the agent is
/// actually running (nothing to drain on a down container). The `Reconcile` /// actually running (nothing to drain on a down container). The `Reconcile`
/// stays even for a down agent so a race-up between the state read and exec /// stays even for a down agent so a race-up between the state read and exec
/// is still stopped in-DAG. /// is still stopped in-DAG.
fn stop_chain(agent: &str, graceful: bool, running: bool) -> Vec<NodeSpec> { fn stop_chain(b: &Job, agent: &str, graceful: bool, running: bool) {
// `SetWanted` is the group root and owns the agent lease; the mechanical // `SetWanted` is the group root and owns the agent lease; the mechanical
// steps are its children (borrow the lease, run once it reaches `Finishing`, // steps are its children (borrow the lease, run once it reaches `Finishing`,
// dep-ordered among themselves). // dep-ordered among themselves).
let a = || agent.to_owned(); let a = || agent.to_owned();
let mut n = vec![node( let wanted = node(
b,
NodeKind::SetWanted { NodeKind::SetWanted {
agent: a(), agent: a(),
up: false, up: false,
}, },
Vec::new(), );
)]; // Declaration order is dependency order: the quiesce steps come first so
// the `Reconcile` that waits on them can name them.
if graceful && running { if graceful && running {
n.push(child(0, NodeKind::Signal { agent: a() }, Vec::new())); let signal = node(b, NodeKind::Signal { agent: a() }).part_of(wanted);
n.push(child(0, NodeKind::Drain { agent: a() }, after_ok(1))); let drain = node(b, NodeKind::Drain { agent: a() })
n.push(child(0, NodeKind::Reconcile { agent: a() }, after_ok(2))); .part_of(wanted)
.after_ok(signal);
let _ = node(b, NodeKind::Reconcile { agent: a() })
.part_of(wanted)
.after_ok(drain);
} else { } else {
n.push(child(0, NodeKind::Reconcile { agent: a() }, Vec::new())); let _ = node(b, NodeKind::Reconcile { agent: a() }).part_of(wanted);
} }
n
} }
/// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev /// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev
/// agent gets the rebuild subgraph (its tail `Reconcile` starts it on /// agent gets the rebuild subgraph (its tail `Reconcile` starts it on
/// current derivations), otherwise a plain `Reconcile` (which starts a down /// current derivations), otherwise a plain `Reconcile` (which starts a down
/// agent and noops an already-running one). /// agent and noops an already-running one).
fn start_chain(agent: &str, running: bool, stale: bool) -> Vec<NodeSpec> { fn start_chain(b: &Job, agent: &str, running: bool, stale: bool) {
let mut n = vec![node( let wanted = node(
b,
NodeKind::SetWanted { NodeKind::SetWanted {
agent: agent.to_owned(), agent: agent.to_owned(),
up: true, up: true,
}, },
Vec::new(), );
)];
if !running && stale { if !running && stale {
// Rebuild subtree after the SetWanted head (base = 1, so the rebuild's // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`,
// `MetaSync` root deps `after_ok(0)` = the head). `MetaSync`,
// `Prebuild` + `Reconcile` are their own group roots (top-level, per // `Prebuild` + `Reconcile` are their own group roots (top-level, per
// `rebuild_nodes`). // `rebuild_nodes`).
n.extend(rebuild_nodes( rebuild_nodes(
b,
agent, agent,
RebuildOpts { RebuildOpts {
relock: true, relock: true,
graceful: false, graceful: false,
}, },
1, Some(wanted),
)); );
} else { } else {
n.push(child( let _ = node(
0, b,
NodeKind::Reconcile { NodeKind::Reconcile {
agent: agent.to_owned(), agent: agent.to_owned(),
}, },
Vec::new(), )
)); .part_of(wanted);
} }
n
} }
/// One agent's **restart** subgraph. Restart NEVER rewrites `wanted` /// One agent's **restart** subgraph. Restart NEVER rewrites `wanted`
@ -127,82 +131,49 @@ fn start_chain(agent: &str, running: bool, stale: bool) -> Vec<NodeSpec> {
/// before `Reconcile`; a down agent gets just `Reconcile`, which /// before `Reconcile`; a down agent gets just `Reconcile`, which
/// converges to intent — a stopped (`wanted = Off`) agent stays stopped, /// converges to intent — a stopped (`wanted = Off`) agent stays stopped,
/// a crashed (`wanted = Up`) agent comes back up. /// a crashed (`wanted = Up`) agent comes back up.
fn restart_chain(agent: &str, graceful: bool, running: bool) -> Vec<NodeSpec> { fn restart_chain(b: &Job, agent: &str, graceful: bool, running: bool) {
let a = || agent.to_owned(); let a = || agent.to_owned();
if !running { if !running {
// Nothing to bounce — a lone Reconcile converges to intent. // Nothing to bounce — a lone Reconcile converges to intent.
return vec![node(NodeKind::Reconcile { agent: a() }, Vec::new())]; let _ = node(b, NodeKind::Reconcile { agent: a() });
return;
} }
// Running: mechanical stop then Reconcile. The first stop node is the group // Running: mechanical stop then Reconcile. The first stop node is the group
// ROOT (no SetWanted head) and owns the agent lease; the rest are its // ROOT (no SetWanted head) and owns the agent lease; the rest are its
// children (borrow the lease, dep-ordered), so the bounce holds one // children (borrow the lease, dep-ordered), so the bounce holds one
// continuous lease and `Reconcile` cancel-cascades if a stop step fails. // continuous lease and `Reconcile` cancel-cascades if a stop step fails.
let mut n = vec![if graceful { //
node(NodeKind::Signal { agent: a() }, Vec::new()) // `Reconcile` gates on the last mechanical step. For a non-graceful bounce
} else { // that step *is* the root, and the parent gate already orders it — a child
node(NodeKind::StopForUpdate { agent: a() }, Vec::new()) // must NOT dep on its own parent (dep-scope), so it takes no sibling edge.
}];
if graceful { if graceful {
n.push(child(0, NodeKind::Drain { agent: a() }, Vec::new())); let signal = node(b, NodeKind::Signal { agent: a() });
n.push(child( let drain = node(b, NodeKind::Drain { agent: a() }).part_of(signal);
0, let stop = node(b, NodeKind::StopForUpdate { agent: a() })
NodeKind::StopForUpdate { agent: a() }, .part_of(signal)
after_ok(1), .after_ok(drain);
)); let _ = node(b, NodeKind::Reconcile { agent: a() })
} .part_of(signal)
// `Reconcile` gates on the last mechanical step. When the only step is the .after_ok(stop);
// root itself (non-graceful, `StopForUpdate` == index 0), the parent gate
// already orders `Reconcile` after it — a child must NOT dep on its own
// parent (dep-scope). So the sibling dep is added only for a graceful
// bounce, where the last step is a sibling child.
let deps = if n.len() > 1 {
after_ok(u64::try_from(n.len() - 1).unwrap_or(0))
} else { } else {
Vec::new() let stop = node(b, NodeKind::StopForUpdate { agent: a() });
}; let _ = node(b, NodeKind::Reconcile { agent: a() }).part_of(stop);
n.push(child(0, NodeKind::Reconcile { agent: a() }, deps));
n
}
/// Concatenate per-agent subgraphs (each with LOCAL 0-based deps) into one
/// node list, rebasing each subgraph's internal deps by its offset. A
/// subgraph root (empty deps — the `SetWanted` head) stays a root, so the
/// per-agent subgraphs are independent and run concurrently, each on its
/// own lease.
fn concat_subgraphs(chains: Vec<Vec<NodeSpec>>) -> Vec<NodeSpec> {
let mut out: Vec<NodeSpec> = Vec::new();
for chain in chains {
let base = u64::try_from(out.len()).unwrap_or(u64::MAX);
for spec in chain {
let deps = spec
.deps
.into_iter()
.map(|d| Dep {
on: base + d.on,
when: d.when,
})
.collect();
out.push(NodeSpec {
kind: spec.kind,
deps,
// Rebase the structural parent by the same offset (a subgraph
// root keeps `parent = None`, so the per-agent groups stay
// independent + concurrent).
parent: spec.parent.map(|p| base + p),
});
}
} }
out
} }
/// Wrap assembled power-op `nodes` in a `DagSpec`. No tail node: a power op's /// Wrap the per-agent subgraphs in a `DagSpec`. No tail node: a power op's
/// effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to do once /// effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to do once
/// they settle. /// they settle.
fn power_dag(source: Source, reason: String, nodes: Vec<NodeSpec>) -> DagSpec { ///
/// There is no concatenation step: every chain declares into the same builder
/// and each keeps its own root, so the per-agent subgraphs are independent and
/// run concurrently, each on its own lease. Rebasing one subgraph's indices
/// onto another's used to be a function.
fn power_dag(source: Source, reason: String, job: Job) -> DagSpec {
DagSpec { DagSpec {
source, source,
reason, reason,
nodes, job,
} }
} }
@ -218,11 +189,11 @@ pub(crate) fn stop_spec(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let chains = targets let job = Job::new();
.iter() for (agent, running) in targets {
.map(|(agent, running)| stop_chain(agent, graceful, *running)) stop_chain(&job, agent, graceful, *running);
.collect(); }
power_dag(source, reason, concat_subgraphs(chains)) power_dag(source, reason, job)
} }
/// Assemble the start DAG from explicit `(agent, running, stale)` targets. /// Assemble the start DAG from explicit `(agent, running, stale)` targets.
@ -236,11 +207,11 @@ pub(crate) fn start_spec(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let chains = targets let job = Job::new();
.iter() for (agent, running, stale) in targets {
.map(|(agent, running, stale)| start_chain(agent, *running, *stale)) start_chain(&job, agent, *running, *stale);
.collect(); }
power_dag(source, reason, concat_subgraphs(chains)) power_dag(source, reason, job)
} }
/// Assemble the restart DAG from explicit `(agent, running)` targets. /// Assemble the restart DAG from explicit `(agent, running)` targets.
@ -250,11 +221,11 @@ pub(crate) fn restart_spec(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let chains = targets let job = Job::new();
.iter() for (agent, running) in targets {
.map(|(agent, running)| restart_chain(agent, graceful, *running)) restart_chain(&job, agent, graceful, *running);
.collect(); }
power_dag(source, reason, concat_subgraphs(chains)) power_dag(source, reason, job)
} }
/// Restart a single agent. Thin wrapper over [`restart_many`]. /// Restart a single agent. Thin wrapper over [`restart_many`].

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

@ -1,21 +1,19 @@
//! DAG shape builders — every operation as a template over the shared //! DAG shape builders — every operation as a template over the shared
//! node primitives — plus submit-time cycle validation (petgraph is //! node primitives.
//! confined to this validation; the runtime store stays the plain
//! `Vec<Node>` + `deps`).
//! //!
//! Every node carries its own `agent` (there is no DAG-level agent) — the //! Every node carries its own `agent` (there is no DAG-level agent) — the
//! `node` helper stamps each node's agent. This module holds the *pure* //! [`node`] helper stamps each node's agent and declares the resources that
//! shape builders (no I/O). The hive-wide **power ops** (`stop` / `start` / //! node's kind needs. This module holds the *pure* shape builders (no I/O).
//! `restart`) are NOT here: their per-agent shape depends on each agent's //! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here:
//! live running state (an async `lifecycle::is_running` read), so they are //! their per-agent shape depends on each agent's live running state (an async
//! assembled dynamically in `submit.rs` out of the shared pure primitives //! `lifecycle::is_running` read), so they are assembled dynamically in
//! this module exports (`node`, `after_ok`, `rebuild_nodes`) — one //! `submit.rs` out of the shared pure primitives this module exports
//! independent per-agent subgraph each, concurrent on its own lease, ONE //! ([`node`], [`rebuild_nodes`]) — one independent per-agent subgraph each,
//! DAG for the whole hive-wide op. `stop`/`start` write the durable `wanted` //! concurrent on its own lease, ONE DAG for the whole hive-wide op.
//! intent via a head `SetWanted(w)` node (holding the agent lease, so //! `stop`/`start` write the durable `wanted` intent via a head `SetWanted(w)`
//! intent+reconcile is atomic per-agent); `restart` writes no intent — it //! node (holding the agent lease, so intent+reconcile is atomic per-agent);
//! bounces the container and lets the tail `Reconcile` converge to the //! `restart` writes no intent — it bounces the container and lets the tail
//! agent's existing `wanted`. //! `Reconcile` converge to the agent's existing `wanted`.
//! //!
//! ```text //! ```text
//! rebuild(a): MetaSync(a) → Prebuild(a) → StopForUpdate(a) → Swap(a) →(ok) PostSwap(a) →(any) Reconcile(a) //! rebuild(a): MetaSync(a) → Prebuild(a) → StopForUpdate(a) → Swap(a) →(ok) PostSwap(a) →(any) Reconcile(a)
@ -26,113 +24,77 @@
//! reparent(moves): Reparent(moves) [no rebuild — topology.json is read live] //! reparent(moves): Reparent(moves) [no rebuild — topology.json is read live]
//! ``` //! ```
//! //!
//! Nodes are **named, not counted**: a template declares a node and holds the
//! handle it gets back, so an edge says which node it waits on instead of
//! computing where that node landed. There is no submit-time cycle validation
//! left to do — `hive_jobq`'s builder inserts in declaration order and rejects
//! a reference to a node declared later, so every edge points backwards and a
//! cycle is unrepresentable rather than merely rejected.
//!
//! For the dynamic power-op shapes (`stop` / `start` / `restart`, built from //! For the dynamic power-op shapes (`stop` / `start` / `restart`, built from
//! live online/offline state), see `submit.rs`. //! live online/offline state), see `submit.rs`.
use anyhow::{Result, bail}; use hive_jobq::TerminalState;
use hive_jobq::{DepWhen, TerminalState}; use super::model::{DagSpec, NodeKind, PermPayload, Source};
use super::{Handle, Job};
use super::model::{DagSpec, Dep, NodeKind, NodeSpec, PermPayload, Source}; /// Declare one node carrying `kind`, with the resources that kind needs.
/// After-ok edge on the previous node — the common chain link. Shared with
/// the async power-op builders in `submit.rs` (which assemble per-agent
/// chains dynamically from live container state).
pub(crate) fn after_ok(on: u64) -> Vec<Dep> {
vec![Dep {
on,
when: DepWhen::AFTER_OK,
}]
}
/// `AfterOk` edges onto every one of a DAG's **group-roots** — the success
/// branch of a per-outcome tail pair, and the aggregator the failure branch
/// keys off.
/// ///
/// Group-roots are the right granularity, not "every node": a root's state *is* /// The resource declaration is [`NodeKind::resource_deps`] applied at the
/// its subtree's roll-up, so edging the roots covers every descendant while /// construction site — a build slot for nix-heavy kinds, the agent lease for
/// keeping the dep list small and stable as subtrees grow. Because every edge is /// container-affecting ones, the global meta window for meta-mutating ones. A
/// `AFTER_OK`, this node runs only if *all* of them succeeded — and is ruled out /// node that declares a resource an ancestor already holds re-enters that
/// ([`TerminalState::Skipped`]) the moment one doesn't, which is precisely the /// grant rather than taking a fresh unit, so declaring costs nothing.
/// signal [`on_elimination_of`] waits for.
pub(crate) fn after_ok_all(ons: &[u64]) -> Vec<Dep> {
ons.iter()
.map(|&on| Dep {
on,
when: DepWhen::AFTER_OK,
})
.collect()
}
/// `AFTER_ANY` edges onto every group-root — "wait for all of these to finish,
/// however they went". Ordering only; it accepts any outcome except the DAG
/// being dropped.
pub(crate) fn after_any_all(ons: &[u64]) -> Vec<Dep> {
ons.iter()
.map(|&on| Dep {
on,
when: DepWhen::AFTER_ANY,
})
.collect()
}
/// A single edge satisfied only when `on` was **ruled out** by its own edges.
/// ///
/// Dependency edges are conjunctive, so "any one of these several nodes failed" /// The returned handle is where edges and grouping are declared, and is `Copy`
/// cannot be written directly. This is the composition that expresses it: point /// — naming a node as a dependency does not consume the ability to name it
/// the success branch at every root with [`after_ok_all`], then hang the failure /// again.
/// branch off *that* node's elimination. Exactly one of the pair ever runs. pub(crate) fn node(b: &Job, kind: NodeKind) -> Handle<'_> {
/// // Read the resources off the kind before handing it over — the payload is
/// Note it accepts `Skipped` and **not** `Cancelled`: if the whole DAG was // moved into the node, not cloned for it.
/// dropped before it started, the success branch is marked `Cancelled` directly let resources = kind.resource_deps();
/// and this branch is ruled out too — a job nobody ran reports nothing. let mut handle = b.node(kind);
pub(crate) fn on_elimination_of(on: u64) -> Vec<Dep> { for (name, count) in resources {
vec![Dep { handle = handle.needs_units(name, count);
on, }
when: DepWhen::of(&[TerminalState::Skipped]), handle
}]
}
/// A single edge satisfied only by the listed outcomes of `on` — for the
/// one-tail-per-outcome shape an approval DAG uses.
pub(crate) fn on_outcome(on: u64, outcomes: &[TerminalState]) -> Vec<Dep> {
vec![Dep {
on,
when: DepWhen::of(outcomes),
}]
} }
/// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node /// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node
/// gated on every group-root in `roots`, and the failure node gated on *its* /// gated on every group-root in `roots`, and the failure node gated on *its*
/// elimination. `base` is the spec index the pair starts at. /// elimination.
/// ///
/// Exactly one runs on a DAG that executed, and neither runs on one the operator /// Exactly one runs on a DAG that executed, and neither runs on one the operator
/// dropped — see [`on_elimination_of`]. /// dropped — see [`hive_jobq::NodeRef::on_elimination_of`].
fn emit_rebuilt_tails(agent: &str, roots: &[u64], base: u64) -> Vec<NodeSpec> { fn emit_rebuilt_tails(b: &Job, agent: &str, roots: &[Handle<'_>]) {
let ok = roots.iter().fold(
node(
b,
NodeKind::EmitRebuilt {
agent: agent.to_owned(),
ok: true,
},
),
hive_jobq::NodeRef::after_ok,
);
// The failure branch needs *both*: the ok branch being ruled out (that is the // The failure branch needs *both*: the ok branch being ruled out (that is the
// "something went wrong" signal) **and** every root actually finished. The // "something went wrong" signal) **and** every root actually finished. The
// second half is easy to forget and gets the ordering wrong without it — a // second half is easy to forget and gets the ordering wrong without it — a
// failed `Prebuild` eliminates the ok branch immediately, while the recovery // failed `Prebuild` eliminates the ok branch immediately, while the recovery
// `Reconcile` is still bringing the container back up, so reporting straight // `Reconcile` is still bringing the container back up, so reporting straight
// off the elimination would announce the failure mid-recovery. // off the elimination would announce the failure mid-recovery.
let mut on_fail = after_any_all(roots); let _failed = roots.iter().fold(
on_fail.extend(on_elimination_of(base));
vec![
node(
NodeKind::EmitRebuilt {
agent: agent.to_owned(),
ok: true,
},
after_ok_all(roots),
),
node( node(
b,
NodeKind::EmitRebuilt { NodeKind::EmitRebuilt {
agent: agent.to_owned(), agent: agent.to_owned(),
ok: false, ok: false,
}, },
on_fail, )
), .on_elimination_of(ok),
] hive_jobq::NodeRef::after_any,
);
} }
/// The approval-resolving tails for an approval-carrying DAG: one per outcome of /// The approval-resolving tails for an approval-carrying DAG: one per outcome of
@ -140,48 +102,20 @@ fn emit_rebuilt_tails(agent: &str, roots: &[u64], base: u64) -> Vec<NodeSpec> {
/// ///
/// The `Cancelled` node is what keeps a dropped approval DAG from dangling its /// The `Cancelled` node is what keeps a dropped approval DAG from dangling its
/// row forever — its edge is the only one [`super::JobQueue::cancel`] spares. /// row forever — its edge is the only one [`super::JobQueue::cancel`] spares.
fn resolve_approval_tails(approval_id: i64, root: u64) -> Vec<NodeSpec> { fn resolve_approval_tails(b: &Job, approval_id: i64, root: Handle<'_>) {
[ for outcome in [
TerminalState::Done, TerminalState::Done,
TerminalState::Failed, TerminalState::Failed,
TerminalState::Cancelled, TerminalState::Cancelled,
] ] {
.into_iter() let _ = node(
.map(|outcome| { b,
node(
NodeKind::ResolveApproval { NodeKind::ResolveApproval {
approval_id, approval_id,
outcome, outcome,
}, },
on_outcome(root, &[outcome]),
) )
}) .on_outcome(root, &[outcome]);
.collect()
}
/// Build one **top-level (group-root)** node — `parent = None`. `kind` carries
/// the agent it targets ([`NodeKind`] is the payload directly). Shared with
/// `submit.rs`'s dynamic power-op builders. A root owns whatever resource it
/// declares for its whole subtree; its descendants borrow it (agent-lease /
/// build-slot continuity). Ordering vs other nodes is `deps`; grouping is
/// `parent`.
pub(crate) fn node(kind: NodeKind, deps: Vec<Dep>) -> NodeSpec {
NodeSpec {
kind,
deps,
parent: None,
}
}
/// Build a **child** node whose structural parent is spec-index `parent`. The
/// child runs once its parent reaches `Finishing` (the parent gate), so it must
/// NOT `deps` on `parent` (dep-scope validation rejects a dep on one's own
/// parent). `deps` here order the child against its *siblings* only.
pub(crate) fn child(parent: u64, kind: NodeKind, deps: Vec<Dep>) -> NodeSpec {
NodeSpec {
kind,
deps,
parent: Some(parent),
} }
} }
@ -198,28 +132,51 @@ pub(crate) struct RebuildOpts {
pub graceful: bool, pub graceful: bool,
} }
/// The rebuild node subtree (nested, three group roots). `base` is the spec /// The group-roots a [`rebuild_nodes`] subgraph exposes to its caller: what a
/// index of the first node (`MetaSync`). Structure: /// tail node edges onto, and what a follow-up node waits for.
/// - `MetaSync` (base+0, **root**): the meta-repo preamble (dir prep, agent ///
/// sync, optional relock). Owns the global `MetaWindow` — and *only* for its /// Only the roots — a root's state *is* its subtree's roll-up, so these three
/// own short duration, which is why it is a sibling root rather than /// cover every node in the subgraph without the caller knowing its shape.
/// `Prebuild`'s parent: a resource is held across the holder's whole subtree, #[derive(Debug, Clone, Copy)]
/// so parenting the build under it would extend a hive-global window over pub(crate) struct RebuildRoots<'a> {
/// every rebuild's nix build. /// The meta-repo preamble.
/// - `Prebuild` (base+1, **root**): `AfterOk` `MetaSync`. Owns the build slot pub meta_sync: Handle<'a>,
/// for the whole mechanical subtree below it. Lease-exempt — the nix build /// The build root — its roll-up carries the whole
/// overlaps other DAGs on the same agent. /// `StopForUpdate` → `Swap` → `PostSwap` subtree.
/// - the **stop root** (base+2, child of `Prebuild`): owns the agent lease and pub prebuild: Handle<'a>,
/// runs once `Prebuild` reaches `Finishing` (parent gate). Non-graceful that /// The recovery/convergence tail root.
/// is `StopForUpdate` itself; graceful it is `Signal`, with `Drain` and then pub reconcile: Handle<'a>,
}
impl<'a> RebuildRoots<'a> {
/// The three roots as a slice, for edging a tail onto all of them.
fn all(self) -> [Handle<'a>; 3] {
[self.meta_sync, self.prebuild, self.reconcile]
}
}
/// The rebuild node subtree (nested, three group roots). `after`, when given, is
/// the node this subgraph chains behind. Structure:
/// - `MetaSync` (**root**): the meta-repo preamble (dir prep, agent sync,
/// optional relock). Owns the global `MetaWindow` — and *only* for its own
/// short duration, which is why it is a sibling root rather than `Prebuild`'s
/// parent: a resource is held across the holder's whole subtree, so parenting
/// the build under it would extend a hive-global window over every rebuild's
/// nix build.
/// - `Prebuild` (**root**): `AfterOk` `MetaSync`. Owns the build slot for the
/// whole mechanical subtree below it. Lease-exempt — the nix build overlaps
/// other DAGs on the same agent.
/// - the **stop root** (child of `Prebuild`): owns the agent lease and runs once
/// `Prebuild` reaches `Finishing` (parent gate). Non-graceful that is
/// `StopForUpdate` itself; graceful it is `Signal`, with `Drain` and then
/// `StopForUpdate` as its children so the lease stays continuous across the /// `StopForUpdate` as its children so the lease stays continuous across the
/// whole stop — siblings would each take the lease separately and leave a gap /// whole stop — siblings would each take the lease separately and leave a gap
/// another DAG could claim the agent in, mid-bounce. /// another DAG could claim the agent in, mid-bounce.
/// - `Swap` (child of `StopForUpdate`): borrows the agent lease from its /// - `Swap` (child of `StopForUpdate`): borrows the agent lease from its
/// ancestors and the build slot from `Prebuild` — both continuous. /// ancestors and the build slot from `Prebuild` — both continuous.
/// - `PostSwap` (child of `StopForUpdate`): the swap's Ok-only /// - `PostSwap` (child of `StopForUpdate`): the swap's Ok-only bookkeeping tail
/// bookkeeping tail (rev marker, forge/matrix sync, kick, rescan), `AfterOk` /// (rev marker, forge/matrix sync, kick, rescan), `AfterOk` its sibling
/// its sibling `Swap`. /// `Swap`.
/// - `Reconcile` (**last, root**): `AfterAny` `Prebuild`, which rolls up /// - `Reconcile` (**last, root**): `AfterAny` `Prebuild`, which rolls up
/// terminal only once its whole mechanical subtree (SFU→Swap→PostSwap) has /// terminal only once its whole mechanical subtree (SFU→Swap→PostSwap) has
/// settled — so `Reconcile` runs after the swap regardless of outcome, and as /// settled — so `Reconcile` runs after the swap regardless of outcome, and as
@ -228,56 +185,47 @@ pub(crate) struct RebuildOpts {
/// cancel-cascades `Prebuild`, i.e. terminal, so the tail still runs). It /// cancel-cascades `Prebuild`, i.e. terminal, so the tail still runs). It
/// takes a fresh lease; the tiny gap is harmless — `Reconcile` converges to /// takes a fresh lease; the tiny gap is harmless — `Reconcile` converges to
/// the persisted `wanted` idempotently. /// the persisted `wanted` idempotently.
pub(crate) fn rebuild_nodes(agent: &str, opts: RebuildOpts, base: u64) -> Vec<NodeSpec> { pub(crate) fn rebuild_nodes<'a>(
b: &'a Job,
agent: &str,
opts: RebuildOpts,
after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
let a = || agent.to_owned(); let a = || agent.to_owned();
let RebuildOpts { relock, graceful } = opts; let RebuildOpts { relock, graceful } = opts;
let mut nodes = vec![
node( let mut meta_sync = node(b, NodeKind::MetaSync { agent: a(), relock });
NodeKind::MetaSync { agent: a(), relock }, if let Some(after) = after {
if base == 0 { meta_sync = meta_sync.after_ok(after);
Vec::new() }
} else { let prebuild = node(b, NodeKind::Prebuild { agent: a() }).after_ok(meta_sync);
after_ok(base - 1)
},
),
node(NodeKind::Prebuild { agent: a() }, after_ok(base)),
];
// The stop root hangs off `Prebuild` and owns the agent lease for // The stop root hangs off `Prebuild` and owns the agent lease for
// everything below it. // everything below it. `StopForUpdate` parents the swap pair either way.
let stop_root = base + 2; let stop_for_update = if graceful {
if graceful { let signal = node(b, NodeKind::Signal { agent: a() }).part_of(prebuild);
nodes.push(child(base + 1, NodeKind::Signal { agent: a() }, Vec::new()));
// `Drain` is a *child* of `Signal`, so the parent gate already orders // `Drain` is a *child* of `Signal`, so the parent gate already orders
// it — a child must not dep on its own parent (dep-scope). // it — a child must not dep on its own parent (dep-scope).
nodes.push(child(stop_root, NodeKind::Drain { agent: a() }, Vec::new())); let drain = node(b, NodeKind::Drain { agent: a() }).part_of(signal);
nodes.push(child( node(b, NodeKind::StopForUpdate { agent: a() })
stop_root, .part_of(signal)
NodeKind::StopForUpdate { agent: a() }, .after_ok(drain)
after_ok(stop_root + 1),
));
} else { } else {
nodes.push(child( node(b, NodeKind::StopForUpdate { agent: a() }).part_of(prebuild)
base + 1, };
NodeKind::StopForUpdate { agent: a() },
Vec::new(), let swap = node(b, NodeKind::Swap { agent: a() }).part_of(stop_for_update);
)); let _post_swap = node(b, NodeKind::PostSwap { agent: a() })
.part_of(stop_for_update)
.after_ok(swap);
let reconcile = node(b, NodeKind::Reconcile { agent: a() }).after_any(prebuild);
RebuildRoots {
meta_sync,
prebuild,
reconcile,
} }
// Index of `StopForUpdate`, which parents the swap pair either way.
let sfu = if graceful { stop_root + 2 } else { stop_root };
nodes.push(child(sfu, NodeKind::Swap { agent: a() }, Vec::new()));
nodes.push(child(
sfu,
NodeKind::PostSwap { agent: a() },
after_ok(sfu + 1),
));
nodes.push(node(
NodeKind::Reconcile { agent: a() },
vec![Dep {
on: base + 1,
when: DepWhen::AFTER_ANY,
}],
));
nodes
} }
/// The rebuild subgraph a [`NodeKind::DeployApply`] grows into its own DAG once /// The rebuild subgraph a [`NodeKind::DeployApply`] grows into its own DAG once
@ -288,7 +236,7 @@ pub(crate) fn rebuild_nodes(agent: &str, opts: RebuildOpts, base: u64) -> Vec<No
/// relocked and staged `flake.lock`, so the appended `MetaSync` must do the dir /// relocked and staged `flake.lock`, so the appended `MetaSync` must do the dir
/// prep + `sync_agents` *without* re-locking over it. /// prep + `sync_agents` *without* re-locking over it.
/// ///
/// `FinalizeDeploy` waits on **two** siblings, which together reproduce the gate /// `FinalizeDeploy` waits on **two** roots, which together reproduce the gate
/// the old fused node had around its inline `rebuild_no_meta` call: /// the old fused node had around its inline `rebuild_no_meta` call:
/// - `AfterOk` `Prebuild` — a parent's state is its roll-up, so this is `Done` /// - `AfterOk` `Prebuild` — a parent's state is its roll-up, so this is `Done`
/// only once `StopForUpdate` → `Swap` → `PostSwap` all are (a failed *or* /// only once `StopForUpdate` → `Swap` → `PostSwap` all are (a failed *or*
@ -304,40 +252,27 @@ pub(crate) fn rebuild_nodes(agent: &str, opts: RebuildOpts, base: u64) -> Vec<No
/// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's /// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's
/// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor /// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor
/// already holding it rather than deadlocking against it. /// already holding it rather than deadlocking against it.
pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Vec<NodeSpec> { pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Job {
let mut nodes = rebuild_nodes( let b = Job::new();
let roots = rebuild_nodes(
&b,
agent, agent,
RebuildOpts { RebuildOpts {
relock: false, relock: false,
graceful: false, graceful: false,
}, },
0, None,
); );
let reconcile = reconcile_index(&nodes, 0); let _finalize = node(
nodes.push(node( &b,
NodeKind::FinalizeDeploy { NodeKind::FinalizeDeploy {
agent: agent.to_owned(), agent: agent.to_owned(),
approval_id, approval_id,
}, },
vec![ )
Dep { .after_ok(roots.prebuild)
on: 1, .after_ok(roots.reconcile);
when: DepWhen::AFTER_OK, b
},
Dep {
on: reconcile,
when: DepWhen::AFTER_OK,
},
],
));
nodes
}
/// Spec index of the `Reconcile` root a [`rebuild_nodes`] subgraph ends on,
/// for callers that gate a tail on it. Read off the emitted list rather than
/// hard-coded, because the subgraph's length depends on [`RebuildOpts`].
fn reconcile_index(rebuild: &[NodeSpec], base: u64) -> u64 {
base + u64::try_from(rebuild.len()).unwrap_or(0).saturating_sub(1)
} }
/// One uniform rebuild shape — no `was_running` branch. `StopForUpdate` /// One uniform rebuild shape — no `was_running` branch. `StopForUpdate`
@ -352,41 +287,41 @@ fn reconcile_index(rebuild: &[NodeSpec], base: u64) -> u64 {
/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so /// 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. /// it reaches `Done` even after a failed swap and the tail would report success.
pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec {
let mut nodes = rebuild_nodes( let job = Job::new();
let roots = rebuild_nodes(
&job,
agent, agent,
RebuildOpts { RebuildOpts {
relock, relock,
graceful: false, graceful: false,
}, },
0, None,
); );
let reconcile = reconcile_index(&nodes, 0); emit_rebuilt_tails(&job, agent, &roots.all());
let tail_base = u64::try_from(nodes.len()).unwrap_or(0);
nodes.extend(emit_rebuilt_tails(agent, &[0, 1, reconcile], tail_base));
DagSpec { DagSpec {
source, source,
reason, reason,
nodes, job,
} }
} }
/// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the /// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the
/// single opaque node it used to be. Structure: /// single opaque node it used to be. Structure:
/// - `DeployWindow` (0, **root**): the resource holder — global meta window, /// - `DeployWindow` (**root**): the resource holder — global meta window, agent
/// agent lease, build slot — held across every child below. No work of its /// lease, build slot — held across every child below. No work of its own; it
/// own; it reaches `Finishing` immediately and the children run inside it. /// reaches `Finishing` immediately and the children run inside it.
/// - `MergeVerify` (1, child): drift-gate + fetch + eval-verify. Mutates /// - `MergeVerify` (child): drift-gate + fetch + eval-verify. Mutates nothing,
/// nothing, so a failure here cancel-cascades its siblings with the forge and /// so a failure here cancel-cascades its siblings with the forge and the
/// the applied repo exactly as they were. /// applied repo exactly as they were.
/// - `DeployApply` (2, child, `AfterOk` `MergeVerify`): the irreversible half — /// - `DeployApply` (child, `AfterOk` `MergeVerify`): the irreversible half —
/// ff-merge + `prepare_deploy`. It doesn't rebuild inline; it grows /// ff-merge + `prepare_deploy`. It doesn't rebuild inline; it grows
/// [`deploy_rebuild_nodes`] into this DAG as its own children, so the build /// [`deploy_rebuild_nodes`] into this DAG as its own children, so the build
/// and the closing `FinalizeDeploy` are real nodes under the same window. /// and the closing `FinalizeDeploy` are real nodes under the same window.
/// - `DeployTail` (3, child, `AfterAny` `DeployApply`): the compensation + /// - `DeployTail` (child, `AfterAny` `DeployApply`): the compensation +
/// bookkeeping tail — rollback when a merge landed unfinalized, forge tag /// bookkeeping tail — rollback when a merge landed unfinalized, forge tag
/// mirror, PR failure comment (see [`NodeKind::DeployTail`]). /// mirror, PR failure comment (see [`NodeKind::DeployTail`]).
/// ///
/// - `ResolveApproval` (4, **root**, `AfterAny` `DeployWindow`): resolves the /// - `ResolveApproval` (**root**, `AfterAny` `DeployWindow`): resolves the
/// approval row. A root rather than another child, so it isn't inside the /// approval row. A root rather than another child, so it isn't inside the
/// window's resource subtree — it runs once the window has released the meta /// window's resource subtree — it runs once the window has released the meta
/// window, lease and build slot. One edge suffices here: `DeployWindow` is the /// window, lease and build slot. One edge suffices here: `DeployWindow` is the
@ -397,48 +332,47 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag
/// leaves `flake.lock` staged-uncommitted for the build's whole duration. /// leaves `flake.lock` staged-uncommitted for the build's whole duration.
pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec {
let a = || agent.to_owned(); let a = || agent.to_owned();
let job = Job::new();
let window = node(
&job,
NodeKind::DeployWindow {
agent: a(),
approval_id,
},
);
let verify = node(
&job,
NodeKind::MergeVerify {
agent: a(),
approval_id,
},
)
.part_of(window);
let apply = node(
&job,
NodeKind::DeployApply {
agent: a(),
approval_id,
},
)
.part_of(window)
.after_ok(verify);
let _tail = node(
&job,
NodeKind::DeployTail {
agent: a(),
approval_id,
},
)
.part_of(window)
.after_any(apply);
resolve_approval_tails(&job, approval_id, window);
DagSpec { DagSpec {
source: Source::Approval, source: Source::Approval,
reason, reason,
nodes: vec![ job,
node(
NodeKind::DeployWindow {
agent: a(),
approval_id,
},
Vec::new(),
),
child(
0,
NodeKind::MergeVerify {
agent: a(),
approval_id,
},
Vec::new(),
),
child(
0,
NodeKind::DeployApply {
agent: a(),
approval_id,
},
after_ok(1),
),
child(
0,
NodeKind::DeployTail {
agent: a(),
approval_id,
},
vec![Dep {
on: 2,
when: DepWhen::AFTER_ANY,
}],
),
]
.into_iter()
.chain(resolve_approval_tails(approval_id, 0))
.collect(),
} }
} }
@ -449,15 +383,17 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec
/// in the queue tests); production paths no longer emit a bare reconcile. /// in the queue tests); production paths no longer emit a bare reconcile.
#[cfg(test)] #[cfg(test)]
pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec {
let job = Job::new();
let _reconcile = node(
&job,
NodeKind::Reconcile {
agent: agent.to_owned(),
},
);
DagSpec { DagSpec {
source, source,
reason, reason,
nodes: vec![node( job,
NodeKind::Reconcile {
agent: agent.to_owned(),
},
Vec::new(),
)],
} }
} }
@ -473,53 +409,56 @@ pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec {
/// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up
/// already carries the whole cascade. /// already carries the whole cascade.
pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec {
let a = || agent.to_owned();
let job = Job::new();
let provision = node(&job, NodeKind::Provision { agent: a() });
let create = node(&job, NodeKind::Create { agent: a() }).part_of(provision);
let dropin = node(&job, NodeKind::WriteDropin { agent: a() }).part_of(create);
let _reconcile = node(&job, NodeKind::Reconcile { agent: a() })
.part_of(create)
.after_ok(dropin);
resolve_approval_tails(&job, approval_id, provision);
DagSpec { DagSpec {
source: Source::Approval, source: Source::Approval,
reason, reason,
nodes: { job,
let a = || agent.to_owned();
vec![
node(NodeKind::Provision { agent: a() }, Vec::new()),
child(0, NodeKind::Create { agent: a() }, Vec::new()),
child(1, NodeKind::WriteDropin { agent: a() }, Vec::new()),
child(1, NodeKind::Reconcile { agent: a() }, after_ok(2)),
]
.into_iter()
.chain(resolve_approval_tails(approval_id, 0))
.collect()
},
} }
} }
/// Perm change: commit the JSON file(s), then the rebuild subgraph so /// Perm change: commit the JSON file(s), then the rebuild subgraph so
/// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes /// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes
/// effect in the container. Group-roots are `WritePermFile`(0) plus the rebuild /// effect in the container. Group-roots are `WritePermFile` plus the rebuild
/// subgraph's `MetaSync`(1) / `Prebuild`(2) / `Reconcile`(6), so the /// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail
/// `EmitRebuilt` tail edges all four. /// edges all four.
pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec {
let mut nodes = vec![node( let job = Job::new();
let write = node(
&job,
NodeKind::WritePermFile { NodeKind::WritePermFile {
agent: agent.to_owned(), agent: agent.to_owned(),
payload, payload,
}, },
Vec::new(), );
)]; let roots = rebuild_nodes(
let rebuild = rebuild_nodes( &job,
agent, agent,
RebuildOpts { RebuildOpts {
relock: true, relock: true,
graceful: false, graceful: false,
}, },
1, Some(write),
);
emit_rebuilt_tails(
&job,
agent,
&[write, roots.meta_sync, roots.prebuild, roots.reconcile],
); );
let reconcile = reconcile_index(&rebuild, 1);
nodes.extend(rebuild);
let tail_base = u64::try_from(nodes.len()).unwrap_or(0);
nodes.extend(emit_rebuilt_tails(agent, &[0, 1, 2, reconcile], tail_base));
DagSpec { DagSpec {
source, source,
reason, reason,
nodes, job,
} }
} }
@ -539,25 +478,26 @@ pub fn meta_update(
reason: String, reason: String,
approval_id: Option<i64>, approval_id: Option<i64>,
) -> DagSpec { ) -> DagSpec {
let mut nodes = vec![node( let job = Job::new();
let lock = node(
&job,
NodeKind::MetaLock { NodeKind::MetaLock {
sweep: false, sweep: false,
fanout: None, fanout: None,
inputs, inputs,
}, },
Vec::new(), );
)];
// The bump itself has no side effect, so an operator-driven one ends at the // The bump itself has no side effect, so an operator-driven one ends at the
// `MetaLock`; an approval-driven one still has its row to resolve and gets the // `MetaLock`; an approval-driven one still has its row to resolve and gets the
// per-outcome tails edged onto that single group-root — whose roll-up covers // per-outcome tails edged onto that single group-root — whose roll-up covers
// the rebuild subgraphs `MetaLock` grows into itself. // the rebuild subgraphs `MetaLock` grows into itself.
if let Some(approval_id) = approval_id { if let Some(approval_id) = approval_id {
nodes.extend(resolve_approval_tails(approval_id, 0)); resolve_approval_tails(&job, approval_id, lock);
} }
DagSpec { DagSpec {
source, source,
reason, reason,
nodes, job,
} }
} }
@ -575,10 +515,12 @@ pub fn reparent(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let job = Job::new();
let _reparent = node(&job, NodeKind::Reparent { moves });
DagSpec { DagSpec {
source, source,
reason, reason,
nodes: vec![node(NodeKind::Reparent { moves }, Vec::new())], job,
} }
} }
@ -586,45 +528,3 @@ pub fn reparent(
// as ONE `Boot` DAG (a sweep `MetaLock` root that grows rebuild subgraphs // as ONE `Boot` DAG (a sweep `MetaLock` root that grows rebuild subgraphs
// in-DAG, plus a `Reconcile` root per drifted agent) — no anchor node and no // in-DAG, plus a `Reconcile` root per drifted agent) — no anchor node and no
// per-agent child DAGs. // per-agent child DAGs.
/// Validate a spec before it enters the queue: node ids are dense
/// (index = id), deps + parents reference existing *earlier* nodes, and the
/// dep graph is acyclic (petgraph `toposort`). Rejecting cycles here fixes the
/// old queue's documented "circular dep silently deadlocks forever" caveat.
pub fn validate(spec: &DagSpec) -> Result<()> {
if spec.nodes.is_empty() {
bail!("dag spec {:?} has no nodes", spec.reason);
}
let n = spec.nodes.len();
let mut graph = petgraph::graph::DiGraph::<u32, ()>::new();
let idx: Vec<_> = (0..n)
.map(|i| graph.add_node(u32::try_from(i).unwrap_or(u32::MAX)))
.collect();
for (i, node) in spec.nodes.iter().enumerate() {
// A `parent` must index an earlier node — `insert_group` resolves it to
// an already-inserted `NodeId`, so a forward/out-of-bounds parent would
// otherwise panic there.
if let Some(p) = node.parent
&& usize::try_from(p).is_ok_and(|p| p >= i)
{
bail!(
"dag spec {:?} node {i} has invalid parent {p} (must be an earlier node)",
spec.reason
);
}
for dep in &node.deps {
let Some(&dep_idx) = usize::try_from(dep.on).ok().and_then(|i| idx.get(i)) else {
bail!(
"dag spec {:?} node {i} depends on unknown node {}",
spec.reason,
dep.on
);
};
graph.add_edge(dep_idx, idx[i], ());
}
}
if petgraph::algo::toposort(&graph, None).is_err() {
bail!("dag spec {:?} contains a dependency cycle", spec.reason);
}
Ok(())
}

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

@ -6,9 +6,7 @@
//! scheduler's async loop is a thin claim/complete pump over the same //! scheduler's async loop is a thin claim/complete pump over the same
//! methods exercised here. //! methods exercised here.
use hive_jobq::DepWhen; use super::model::NodeKind;
use super::model::{Dep, NodeKind, NodeSpec};
use super::*; use super::*;
fn submit(q: &JobQueue, spec: DagSpec) -> u64 { fn submit(q: &JobQueue, spec: DagSpec) -> u64 {
@ -141,71 +139,22 @@ fn resubmit_while_running_is_new_dag() {
assert_eq!(q.snapshot().len(), 2); assert_eq!(q.snapshot().len(), 2);
} }
// ---- cycle rejection ---- // ---- malformed specs: no longer expressible ----
//
#[test] // Three tests lived here — a dependency cycle, a dependency on a node that
fn cyclic_dag_is_rejected_at_submit() { // does not exist, and an out-of-range parent index — each asserting that
let q = JobQueue::new(1); // `submit` refused the spec. All three built their spec by hand out of
let mut spec = rebuild("agent-a", "cyclic"); // positional indices, which is exactly the representation that made those
// 0 → 1 → 0 cycle. // shapes possible: an index can name a node that isn't there, or one that
spec.nodes = vec![ // comes later.
NodeSpec { //
kind: NodeKind::StopForUpdate { // A job is now declared against handles that only exist for nodes already
agent: "agent-a".to_owned(), // declared, so there is no index to put out of range, and every edge points
}, // backwards — a cycle needs a forward edge. The guard those tests covered was
deps: vec![Dep { // deleted along with the failure mode. What remains — a handle used against a
on: 1, // builder that never issued it — is `hive_jobq`'s to reject, and its builder
when: DepWhen::AFTER_OK, // tests cover it (`a_forward_edge_is_rejected_by_name`,
}], // `a_forward_parent_is_rejected_by_name`, `graph_rejection_surfaces_as_is`).
parent: None,
},
NodeSpec {
kind: NodeKind::Reconcile {
agent: "agent-a".to_owned(),
},
deps: vec![Dep {
on: 0,
when: DepWhen::AFTER_OK,
}],
parent: None,
},
];
assert!(q.submit(spec).is_err(), "cyclic spec must be refused");
assert!(q.snapshot().is_empty());
}
#[test]
fn unknown_dep_is_rejected_at_submit() {
let q = JobQueue::new(1);
let mut spec = rebuild("agent-a", "bad dep");
spec.nodes = vec![NodeSpec {
kind: NodeKind::Reconcile {
agent: "agent-a".to_owned(),
},
deps: vec![Dep {
on: 9,
when: DepWhen::AFTER_OK,
}],
parent: None,
}];
assert!(q.submit(spec).is_err());
}
#[test]
fn invalid_parent_is_rejected_at_submit() {
let q = JobQueue::new(1);
let mut spec = rebuild("agent-a", "bad parent");
// A forward/out-of-bounds parent index must be refused at validate, not
// panic in `insert_group`.
spec.nodes = vec![NodeSpec {
kind: NodeKind::Reconcile {
agent: "agent-a".to_owned(),
},
deps: Vec::new(),
parent: Some(3),
}];
assert!(q.submit(spec).is_err());
}
// ---- dependency order within a DAG ---- // ---- dependency order within a DAG ----
@ -243,20 +192,24 @@ fn rebuild_chain_claims_in_dep_order() {
#[test] #[test]
fn graceful_rebuild_chain_drains_before_stopping() { fn graceful_rebuild_chain_drains_before_stopping() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let spec = DagSpec { let job = Job::new();
source: Source::AutoUpdate, templates::rebuild_nodes(
reason: "sweep".to_owned(), &job,
"agent-a",
nodes: templates::rebuild_nodes( templates::RebuildOpts {
"agent-a", relock: true,
templates::RebuildOpts { graceful: true,
relock: true, },
graceful: true, None,
}, );
0, let id = submit(
), &q,
}; DagSpec {
let id = submit(&q, spec); source: Source::AutoUpdate,
reason: "sweep".to_owned(),
job,
},
);
for expected in [ for expected in [
"meta_sync", "meta_sync",
"prebuild", "prebuild",
@ -284,17 +237,34 @@ fn graceful_rebuild_chain_drains_before_stopping() {
/// drain window, so `StopForUpdate` still hangs straight off `Prebuild`. /// drain window, so `StopForUpdate` still hangs straight off `Prebuild`.
#[test] #[test]
fn non_graceful_rebuild_has_no_signal_or_drain() { fn non_graceful_rebuild_has_no_signal_or_drain() {
let kinds: Vec<String> = templates::rebuild_nodes( // Read the shape off the queue rather than out of a node list: a declared
// job keeps its nodes to itself and inserts them, so what it built is
// observable where it matters — in what the scheduler runs.
let q = JobQueue::new(1);
let job = Job::new();
templates::rebuild_nodes(
&job,
"agent-a", "agent-a",
templates::RebuildOpts { templates::RebuildOpts {
relock: true, relock: true,
graceful: false, graceful: false,
}, },
0, None,
) );
.iter() let id = submit(
.map(|n| n.kind.as_str().to_owned()) &q,
.collect(); DagSpec {
source: Source::Manual,
reason: "manual".to_owned(),
job,
},
);
let mut kinds = Vec::new();
for _ in 0..6 {
let c = claim_one(&q);
kinds.push(c.kind.as_str().to_owned());
q.complete_node(c.node_id, Ok(()));
}
assert_eq!( assert_eq!(
kinds, kinds,
vec![ vec![
@ -306,6 +276,8 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
"reconcile" "reconcile"
] ]
); );
// Settled after exactly those six — nothing else was declared.
assert_eq!(state_of(&q, id), State::Done);
} }
/// A cleanly-finished DAG leaves the snapshot even though its not-taken /// A cleanly-finished DAG leaves the snapshot even though its not-taken
@ -720,19 +692,19 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
// subgraph per stale agent into its OWN DAG. Each subgraph is rooted on // 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. // the emitter and its LOCAL 0-based deps are rebased onto the DAG.
let q = JobQueue::new(4); let q = JobQueue::new(4);
let job = Job::new();
let _lock = templates::node(
&job,
NodeKind::MetaLock {
sweep: true,
fanout: None,
inputs: Vec::new(),
},
);
let spec = DagSpec { let spec = DagSpec {
source: Source::AutoUpdate, source: Source::AutoUpdate,
reason: "sweep".to_owned(), reason: "sweep".to_owned(),
job,
nodes: vec![NodeSpec {
kind: NodeKind::MetaLock {
sweep: true,
fanout: None,
inputs: Vec::new(),
},
deps: Vec::new(),
parent: None,
}],
}; };
let id = submit(&q, spec); let id = submit(&q, spec);
let emitter = claim_one(&q); let emitter = claim_one(&q);
@ -742,18 +714,21 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
// StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must // StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must
// match the sweep arm of `run_meta_lock` or this stops tracking production. // match the sweep arm of `run_meta_lock` or this stops tracking production.
let subgraph = |agent: &str| { let subgraph = |agent: &str| {
let job = Job::new();
templates::rebuild_nodes( templates::rebuild_nodes(
&job,
agent, agent,
templates::RebuildOpts { templates::RebuildOpts {
relock: true, relock: true,
graceful: true, graceful: true,
}, },
0, None,
) );
job
}; };
// Must append BEFORE completing the emitter (the documented contract). // Must append BEFORE completing the emitter (the documented contract).
q.append_subgraph(id, &subgraph("a"), emitter.node_id); q.append_subgraph(id, subgraph("a"), emitter.node_id);
q.append_subgraph(id, &subgraph("b"), emitter.node_id); q.append_subgraph(id, subgraph("b"), emitter.node_id);
q.complete_node(emitter.node_id, Ok(())); q.complete_node(emitter.node_id, Ok(()));
// Still ONE DAG; both subgraph roots become ready once the emitter is // 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 // Done (rooted on it), each on its own agent lease. Their `MetaSync` heads
@ -845,18 +820,17 @@ fn meta_update_grows_cascade_in_dag() {
// Simulate the executor growing the cascade in-DAG (`relock = false` — a // Simulate the executor growing the cascade in-DAG (`relock = false` — a
// cascade child must not re-lock and revert the parent's bump). // cascade child must not re-lock and revert the parent's bump).
for agent in ["alice", "bob"] { for agent in ["alice", "bob"] {
q.append_subgraph( let job = Job::new();
id, templates::rebuild_nodes(
&templates::rebuild_nodes( &job,
agent, agent,
templates::RebuildOpts { templates::RebuildOpts {
relock: false, relock: false,
graceful: false, graceful: false,
}, },
0, None,
),
meta_lock.node_id,
); );
q.append_subgraph(id, job, meta_lock.node_id);
} }
q.complete_node(meta_lock.node_id, Ok(())); q.complete_node(meta_lock.node_id, Ok(()));
// Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root
@ -1119,15 +1093,21 @@ fn cancelled_power_op_runs_no_compensating_node() {
), ),
]; ];
for (name, writes_intent, spec) in cases { for (name, writes_intent, spec) in cases {
let q = JobQueue::new(1);
let id = submit(&q, spec);
// Read the intent head off the submitted DAG rather than out of
// the spec: a declared job holds its own nodes and inserts them.
assert_eq!( assert_eq!(
spec.nodes q.snapshot()
.iter() .iter()
.any(|n| matches!(n.kind, NodeKind::SetWanted { .. })), .find(|d| d.id == id)
.expect("submitted dag")
.nodes
.iter()
.any(|n| n.kind == "set_wanted"),
writes_intent, writes_intent,
"{name} intent head (graceful={graceful}, running={running})" "{name} intent head (graceful={graceful}, running={running})"
); );
let q = JobQueue::new(1);
let id = submit(&q, spec);
assert!(q.cancel(id), "cancelled while queued"); assert!(q.cancel(id), "cancelled while queued");
assert_eq!(state_of(&q, id), State::Cancelled); assert_eq!(state_of(&q, id), State::Cancelled);
assert!( assert!(
@ -1272,7 +1252,7 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
// gate immediately and letting the deploy "finish" before it had built. // gate immediately and letting the deploy "finish" before it had built.
let grown = q.append_subgraph( let grown = q.append_subgraph(
id, id,
&templates::deploy_rebuild_nodes("agent-a", 11), templates::deploy_rebuild_nodes("agent-a", 11),
apply.node_id, apply.node_id,
); );
assert!(!grown.is_empty(), "subgraph grafted onto the apply node"); assert!(!grown.is_empty(), "subgraph grafted onto the apply node");
@ -1330,7 +1310,7 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() {
let apply = claim_one(&q); let apply = claim_one(&q);
q.append_subgraph( q.append_subgraph(
id, id,
&templates::deploy_rebuild_nodes("agent-a", 13), templates::deploy_rebuild_nodes("agent-a", 13),
apply.node_id, apply.node_id,
); );
q.complete_node(apply.node_id, Ok(())); q.complete_node(apply.node_id, Ok(()));

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

@ -334,7 +334,7 @@ fn submit_boot_tree(
n_deferred: usize, n_deferred: usize,
n_skipped: usize, n_skipped: usize,
) { ) {
use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source}; use crate::job_queue::{DagSpec, Job, NodeKind, Source, templates};
// Fully-quiet boot (nothing stale, nothing drifted) submits nothing. // Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
if !any_stale && drifted.is_empty() { if !any_stale && drifted.is_empty() {
@ -348,32 +348,27 @@ fn submit_boot_tree(
n_skipped, n_skipped,
); );
let mut nodes: Vec<NodeSpec> = Vec::new(); let job = Job::new();
// Sweep whenever ANY marker is stale — even when every stale agent is // Sweep whenever ANY marker is stale — even when every stale agent is
// wanted-offline: the hyperhive lock bump must land now so their later // wanted-offline: the hyperhive lock bump must land now so their later
// start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock
// ⇒ no meta commit on a no-change boot. The `fanout` list rides the // ⇒ no meta commit on a no-change boot. The `fanout` list rides the
// MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs.
if any_stale { if any_stale {
nodes.push(NodeSpec { let _ = templates::node(
kind: NodeKind::MetaLock { &job,
NodeKind::MetaLock {
sweep: true, sweep: true,
fanout: Some(fanout), fanout: Some(fanout),
// A sweep bumps `hyperhive` alone (`lock_update_hyperhive`), // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`),
// so it names no inputs. // so it names no inputs.
inputs: Vec::new(), inputs: Vec::new(),
}, },
deps: Vec::new(), );
parent: None,
});
} }
// One boot Reconcile per drifted agent — independent roots. // One boot Reconcile per drifted agent — independent roots.
for name in drifted { for name in drifted {
nodes.push(NodeSpec { let _ = templates::node(&job, NodeKind::Reconcile { agent: name });
kind: NodeKind::Reconcile { agent: name },
deps: Vec::new(),
parent: None,
});
} }
let spec = DagSpec { let spec = DagSpec {
@ -384,7 +379,7 @@ fn submit_boot_tree(
// Rebuilding when the sweep will grow rebuild subgraphs (per-agent // Rebuilding when the sweep will grow rebuild subgraphs (per-agent
// crash-watch suppression during their Swap, applied at claim time); // crash-watch suppression during their Swap, applied at claim time);
// a reconcile-only boot needs no transient. // a reconcile-only boot needs no transient.
nodes, job,
}; };
if let Err(e) = coord.job_queue.submit(spec) { if let Err(e) = coord.job_queue.submit(spec) {
tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); tracing::warn!(error = ?e, "boot: sweep DAG submit failed");

15
hive-jobq/src/builder.rs
View file

@ -112,6 +112,13 @@ impl<N, R> JobBuilder<N, R> {
Self::default() Self::default()
} }
/// Whether nothing has been declared yet — for a caller deciding whether an
/// insertion is worth taking a lock for.
#[must_use]
pub fn is_empty(&self) -> bool {
self.nodes.borrow().is_empty()
}
/// Add a node carrying `payload`, with no edges, resources, or parent yet. /// Add a node carrying `payload`, with no edges, resources, or parent yet.
/// ///
/// The returned handle is where those are declared; it is [`Copy`], so it /// The returned handle is where those are declared; it is [`Copy`], so it
@ -275,6 +282,14 @@ impl<N, R> NodeRef<'_, N, R> {
self.edge(on.into(), DepWhen::AFTER_ANY) self.edge(on.into(), DepWhen::AFTER_ANY)
} }
/// Run only on the listed outcomes of `on` — the general form of
/// [`NodeRef::after_ok`] / [`NodeRef::after_any`], for the
/// one-node-per-outcome shape a job with a row to resolve uses.
#[must_use]
pub fn on_outcome(self, on: impl Into<NodeGuid>, outcomes: &[TerminalState]) -> Self {
self.edge(on.into(), DepWhen::of(outcomes))
}
/// Run only if `on` was **ruled out** — i.e. its own `after_ok` edges did /// Run only if `on` was **ruled out** — i.e. its own `after_ok` edges did
/// not hold. /// not hold.
/// ///