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

job_queue: drop Claim, claim_ready and the completion wrappers

Browse source 
c0re's run loop now goes through hive_jobq's claim_next seam, so the
host layer no longer needs its own claim/complete vocabulary.

exec::run_node takes (NodeId, &NodeKind) instead of a &Claim snapshot.
The agent already rides the payload, and the DAG id is a derived read
(JobQueue::dag_of) that only three arms want, so it is taken per-arm
rather than eagerly for every node. Two arms (WritePermFile, Reparent)
re-matched the kind behind a bail! that could never fire; the match arm
already destructures the payload, so they take it directly now.

Deleted from the c0re layer:
  - struct Claim
  - JobQueue::claim_ready
  - JobQueue::complete_node / complete_node_growing
  - scheduler::NodeDone / handle_completion

Completion happens inside the future claim_next hands back, so "ran the
node but forgot to complete it" is not expressible on the production
path any more. The node done / node failed logging moved with it -- it
lived in handle_completion but is not dead code.

claim_ready and the completion wrappers were left with no non-test
callers, so the tests carry them as ClaimReady / CompleteNode extension
traits over the crate primitives. JobQueue::new_job stays: run_worker
still mints an empty builder on a failed outcome.
This commit is contained in:
atlas 2026-08-02 19:01:53 +02:00 • committed by mara
commit ab53f6710d
4 changed files with 263 additions and 255 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -10,8 +10,7 @@ use std::sync::Arc;
use anyhow::{Context as _, Result};
use super::Claim;
use hive_jobq::TerminalState;
use hive_jobq::{NodeId, TerminalState};
use super::model::NodeKind;
use super::resource::Resource;
@ -43,22 +42,31 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
/// future non-`Send`, which the scheduler's `tokio::spawn` rejects. So the
/// growth executors below return *what to grow* and the declaration happens
/// here, synchronously, between awaits.
///
/// The node is identified by its own id + payload rather than by a `Claim`
/// side-struct: `kind` already carries the agent, and the DAG id is a
/// derived read (`JobQueue::dag_of`) the three arms that need it take
/// themselves. Nothing here needs a claim to exist as a type.
pub(super) async fn run_node(
coord: &Arc<Coordinator>,
job: super::Job,
claim: &Claim,
id: NodeId,
kind: &NodeKind,
) -> (super::Job, Result<()>) {
// The agent this node targets rides the payload — empty for the agentless
// container kinds (`MetaLock`, `Dag`), which never read it.
let agent = kind.agent();
// Every arm is `Result<()>`; the three that grow work declare into `job`
// *synchronously*, after their own awaits have finished. Borrowing `&job`
// inside an `.await` would make this future non-`Send` (see above), so the
// growth executors return what to grow rather than taking the builder.
let result = match &claim.kind {
NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, claim, *relock).await,
NodeKind::Prebuild { .. } => run_prebuild(claim).await,
NodeKind::Swap { .. } => run_swap(coord, claim).await,
NodeKind::PostSwap { .. } => run_post_swap(coord, claim).await,
NodeKind::Provision { .. } => run_provision(coord, claim).await,
NodeKind::Create { .. } => run_create(claim).await,
let result = match kind {
NodeKind::MetaSync { relock, .. } => run_meta_sync(coord, agent, *relock).await,
NodeKind::Prebuild { .. } => run_prebuild(agent, id).await,
NodeKind::Swap { .. } => run_swap(coord, agent, id).await,
NodeKind::PostSwap { .. } => run_post_swap(coord, agent).await,
NodeKind::Provision { .. } => run_provision(coord, agent).await,
NodeKind::Create { .. } => run_create(agent).await,
NodeKind::MetaLock {
sweep,
fanout,
@ -70,7 +78,7 @@ pub(super) async fn run_node(
super::templates::rebuild_nodes(&job, &agent, opts, None);
}
}),
NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await.map(|sub| {
NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| {
if let Some(kind) = sub {
// `Start` / `Stop` declare the lease they run under. This node
// is their parent and holds it, so the declaration is a
@ -81,38 +89,41 @@ pub(super) async fn run_node(
let _ = job.node(kind).needs(lease);
}
}),
NodeKind::Start { .. } => run_start(coord, claim).await,
NodeKind::Stop { .. } => run_stop(coord, claim).await,
NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, claim).await,
NodeKind::Start { .. } => run_start(coord, agent).await,
NodeKind::Stop { .. } => run_stop(coord, agent).await,
NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, agent).await,
NodeKind::Signal { .. } => {
run_signal(coord, claim);
run_signal(coord, agent);
Ok(())
}
NodeKind::Drain { .. } => run_drain(coord, claim).await,
NodeKind::WriteDropin { .. } => run_write_dropin(coord, claim).await,
NodeKind::WritePermFile { .. } => run_write_perm_file(coord, claim).await,
NodeKind::Reparent { .. } => run_reparent(coord, claim).await,
NodeKind::Drain { .. } => run_drain(coord, agent).await,
NodeKind::WriteDropin { .. } => run_write_dropin(coord, agent).await,
// The payload rides the node and is destructured here, so the executor
// takes it directly instead of re-matching the kind behind a `bail!`
// that could never fire.
NodeKind::WritePermFile { payload, .. } => run_write_perm_file(coord, agent, payload).await,
NodeKind::Reparent { moves } => run_reparent(coord, moves).await,
NodeKind::MergeVerify { approval_id, .. } => run_merge_verify(coord, *approval_id).await,
NodeKind::DeployApply { approval_id, .. } => {
run_deploy_apply(coord, *approval_id).await.map(|()| {
super::templates::deploy_rebuild_nodes(&job, claim.kind.agent(), *approval_id);
super::templates::deploy_rebuild_nodes(&job, agent, *approval_id);
})
}
NodeKind::FinalizeDeploy { approval_id, .. } => {
run_finalize_deploy(coord, *approval_id).await
}
NodeKind::DeployTail { approval_id, .. } => {
run_deploy_tail(coord, claim, *approval_id).await
run_deploy_tail(coord, coord.job_queue.dag_of(id), agent, *approval_id).await
}
NodeKind::ResolveApproval {
approval_id,
outcome,
} => run_resolve_approval(coord, claim, *approval_id, *outcome).await,
} => run_resolve_approval(coord, coord.job_queue.dag_of(id), *approval_id, *outcome).await,
NodeKind::EmitRebuilt { ok, .. } => {
run_emit_rebuilt(coord, claim, *ok);
run_emit_rebuilt(coord, agent, coord.job_queue.dag_of(id), *ok);
Ok(())
}
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up),
NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up),
// The two nodes that carry no work of their own; completing either
// lets it reach `Finishing` so the nodes under it start.
// - `Dag`: pure grouping container. The DAG's terminal side effect, if
@ -133,12 +144,12 @@ pub(super) async fn run_node(
/// since the work already happened and failing the tail would only misreport it.
async fn run_resolve_approval(
coord: &Arc<Coordinator>,
claim: &Claim,
dag_id: Option<u64>,
approval_id: i64,
outcome: TerminalState,
) -> Result<()> {
let reason = (outcome == TerminalState::Failed)
.then(|| coord.job_queue.first_error(claim.dag_id))
.then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
.flatten();
crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await;
Ok(())
@ -147,12 +158,12 @@ async fn run_resolve_approval(
/// Emit this agent's `Rebuilt` manager event. `ok` is not computed — it is which
/// of the tail pair the graph let run. The failure note comes from the DAG's
/// first failing node, since the branch knows *that* it failed but not *why*.
fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) {
fn run_emit_rebuilt(coord: &Arc<Coordinator>, agent: &str, dag_id: Option<u64>, ok: bool) {
coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
agent: claim.agent.clone(),
agent: agent.to_owned(),
ok,
note: (!ok)
.then(|| coord.job_queue.first_error(claim.dag_id))
.then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
.flatten(),
sha: None,
tag: None,
@ -167,7 +178,7 @@ fn run_emit_rebuilt(coord: &Arc<Coordinator>, claim: &Claim, ok: bool) {
/// warn-and-continue write, a failed write fails the node (cancel-downstream
/// cancels the `Reconcile`) rather than letting it converge to a stale
/// intent — that atomicity is the point of moving it into the DAG.
fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<()> {
fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()> {
let wanted = if up {
crate::power::Wanted::Up
} else {
@ -175,8 +186,8 @@ fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<(
};
coord
.power
.set(&claim.agent, wanted)
.with_context(|| format!("set wanted={} for agent {}", wanted.as_str(), claim.agent))?;
.set(agent, wanted)
.with_context(|| format!("set wanted={} for agent {agent}", wanted.as_str()))?;
Ok(())
}
@ -189,8 +200,7 @@ fn run_set_wanted(coord: &Arc<Coordinator>, claim: &Claim, up: bool) -> Result<(
/// Deliberately a separate node from the [`run_prebuild`] it feeds: that
/// build takes minutes and only *reads* the store, so keeping the global
/// window off it is what lets rebuilds of different agents overlap.
async fn run_meta_sync(coord: &Arc<Coordinator>, claim: &Claim, relock: bool) -> Result<()> {
let name = &claim.agent;
async fn run_meta_sync(coord: &Arc<Coordinator>, name: &str, relock: bool) -> Result<()> {
// Runs while the agent is still up — the runtime dir and MCP listener
// already exist. Use the pure path accessor; no need to re-register the
// listener (event-driven: registered at start/create).
@ -217,15 +227,14 @@ async fn run_meta_sync(coord: &Arc<Coordinator>, claim: &Claim, relock: bool) ->
/// container is already down: its only purpose is to shrink the swap's
/// downtime window, so a stopped agent (no uptime to preserve) doesn't
/// pay the double eval — `Swap` builds inline instead.
async fn run_prebuild(claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_prebuild(name: &str, id: NodeId) -> Result<()> {
// Warm the toplevel build only when the container is up — the whole
// point of prebuild is to shrink the swap's downtime window. A
// stopped agent has no uptime to preserve, so skip the (expensive)
// eval and let the downstream `Swap` build inline.
if crate::lifecycle::is_running(name).await {
let flake_ref = format!("{}#{name}", crate::paths::meta_root().display());
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(claim.node_id.get())).await?;
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(id.get())).await?;
}
Ok(())
}
@ -235,15 +244,13 @@ async fn run_prebuild(claim: &Claim) -> Result<()> {
/// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan).
/// The recovery-start on failure is NOT here — the DAG's tail
/// `Reconcile` runs after this node terminal ok *or* fail.
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_swap(coord: &Arc<Coordinator>, name: &str, id: NodeId) -> Result<()> {
// Swap runs on an already-existing (stopped) container — runtime dir
// and listener were created earlier. Pure path accessor suffices.
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result =
crate::lifecycle::swap_update(name, &hive, &paths, Some(claim.node_id.get())).await;
let result = crate::lifecycle::swap_update(name, &hive, &paths, Some(id.get())).await;
// On success the Ok-only bookkeeping tail (rev marker, forge/matrix
// sync, kick, rescan, snapshot) runs in the sibling `PostSwap` node,
// which deps `AfterOk(Swap)`. On failure `PostSwap` is cancel-cascaded
@ -262,8 +269,7 @@ async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
/// means the profile swap succeeded. Store/forge/matrix work only — no nix
/// build (build-slot-exempt); the agent lease taken at `Swap` is still held
/// (the whole chain up to `Reconcile` is one agent's subgraph).
async fn run_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_post_swap(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
&& let Err(e) = std::fs::write(crate::paths::applied_rev_marker(name), rev)
{
@ -290,8 +296,7 @@ async fn run_post_swap(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
/// subvolume, and the meta `sync_agents` registration. Runs under the
/// deploy window (it declares `Resource::MetaWindow`) so its commit can't
/// land inside another node's staged deploy window.
async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_provision(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
@ -305,8 +310,8 @@ async fn run_provision(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
/// dir creation and MCP listener registration are deferred to the tail
/// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
/// node stays purely "create", not "create + start".
async fn run_create(claim: &Claim) -> Result<()> {
crate::lifecycle::create_only(&claim.agent).await?;
async fn run_create(name: &str) -> Result<()> {
crate::lifecycle::create_only(name).await?;
Ok(())
}
@ -377,18 +382,17 @@ async fn run_meta_lock(
/// guard) rides across it.
/// Returns the mechanical node to fan out (`None` on a noop) rather than
/// declaring it — the declaration has to happen outside any `.await`, see
/// [`run_node`]. `NodeKind` carries the agent it targets, so `claim.agent` is
/// stamped into the kind here.
async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Option<NodeKind>> {
let name = &claim.agent;
/// [`run_node`]. `NodeKind` carries the agent it targets, so this node's agent
/// is stamped into the fanned-out kind here.
async fn run_reconcile(coord: &Arc<Coordinator>, name: &str) -> Result<Option<NodeKind>> {
let running = crate::lifecycle::is_running(name).await;
let wanted = coord.power.get_or_seed(name, running)?;
Ok(match reconcile_action(wanted, running) {
ReconcileAction::Start => Some(NodeKind::Start {
agent: name.clone(),
agent: name.to_owned(),
}),
ReconcileAction::Stop => Some(NodeKind::Stop {
agent: name.clone(),
agent: name.to_owned(),
}),
ReconcileAction::Noop => {
tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop");
@ -399,8 +403,7 @@ async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<Option
/// Mechanical container start — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Up` and the container down.
async fn run_start(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_start(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
// No node-local transient guard: the pill is derived from the running node
// set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This
// used to take one "only when the DAG holds none", which was a second
@ -428,14 +431,13 @@ async fn run_start(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
/// Mechanical container stop — the sub-step a `Reconcile` planner fans
/// out when it observes `wanted = Offline` and the container up.
async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_stop(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
// See `run_start`: no node-local guard — `Stop` reports `Stopping` from its
// own kind now.
crate::lifecycle::kill(name).await?;
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.clone(),
agent: name.to_owned(),
});
coord.rescan_containers_and_emit().await;
Ok(())
@ -443,8 +445,7 @@ async fn run_stop(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
/// Mechanical stop for the profile swap. Never *changes* `wanted`;
/// noop when already stopped.
async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_stop_for_update(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
if crate::lifecycle::is_running(name).await {
// Seed a missing agent_power row from the PRE-stop observation
// — the DAG's tail `Reconcile` observes only the mechanically
@ -468,19 +469,18 @@ async fn run_stop_for_update(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
/// `GRACEFUL_STOP_TIMEOUT`. Safe because the harness tests the marker at
/// the top of its loop — a paused agent has no turn in flight, so there
/// is nothing to checkpoint.
fn run_signal(coord: &Arc<Coordinator>, claim: &Claim) {
if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) {
fn run_signal(coord: &Arc<Coordinator>, name: &str) {
if hive_types::Ident::parse(name).is_ok_and(|a| Coordinator::is_paused(&a)) {
return;
}
coord.mark_graceful_stop(&claim.agent);
coord.kick_agent(&claim.agent, "graceful stop requested");
coord.mark_graceful_stop(name);
coord.kick_agent(name, "graceful stop requested");
}
/// Await the harness clearing the fence (`GracefulStopComplete`) or
/// the timeout — either way the downstream `Reconcile` proceeds with
/// the actual stop.
async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_drain(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
while coord.is_graceful_stop_pending(name) {
if std::time::Instant::now() >= deadline {
@ -494,8 +494,7 @@ async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
}
/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let name = &claim.agent;
async fn run_write_dropin(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
// write_dropins only needs the path value to build AgentPaths; the
// dir doesn't need to exist at this point (created by ensure_agent_runtime_dir
// on the upstream Prebuild/Start node).
@ -509,13 +508,12 @@ async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()>
/// Write + commit the perm file(s) (fused under `META_LOCK` so the
/// working tree is never left dirty), then emit the P3RM1SS10NS-tab
/// snapshots so the dashboard reflects the new assignment.
async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
async fn run_write_perm_file(
coord: &Arc<Coordinator>,
name: &str,
payload: &super::model::PermPayload,
) -> Result<()> {
use super::model::PermPayload;
let name = &claim.agent;
// The perm file payload rides the node itself (the only consumer).
let NodeKind::WritePermFile { payload, .. } = &claim.kind else {
anyhow::bail!("run_write_perm_file on a non-WritePermFile node");
};
// Runs under the deploy window (it declares `Resource::MetaWindow`): a
// perm commit landing inside another node's staged prepare→finalize
// window would sweep the staged deploy lock into its commit (the
@ -555,10 +553,10 @@ async fn run_write_perm_file(coord: &Arc<Coordinator>, claim: &Claim) -> Result<
/// the deploy window (it declares `Resource::MetaWindow`), same reasoning as
/// `run_write_perm_file`: a topology commit landing inside another node's
/// staged deploy window would sweep the staged lock into its commit.
async fn run_reparent(coord: &Arc<Coordinator>, claim: &Claim) -> Result<()> {
let NodeKind::Reparent { moves } = &claim.kind else {
anyhow::bail!("run_reparent on a non-Reparent node");
};
async fn run_reparent(
coord: &Arc<Coordinator>,
moves: &[(hive_types::Ident, Option<hive_types::Ident>)],
) -> Result<()> {
let refs: Vec<(&str, Option<&str>)> = moves
.iter()
.map(|(child, parent)| {
@ -609,9 +607,13 @@ async fn run_finalize_deploy(coord: &Arc<Coordinator>, approval_id: i64) -> Resu
///
/// Takes the agent from the node payload so the tail can still compensate when
/// the approval row is gone (deny race, purge).
async fn run_deploy_tail(coord: &Arc<Coordinator>, claim: &Claim, approval_id: i64) -> Result<()> {
crate::actions::run_deploy_tail(coord, Some(claim.dag_id), claim.kind.agent(), approval_id)
.await;
async fn run_deploy_tail(
coord: &Arc<Coordinator>,
dag_id: Option<u64>,
agent: &str,
approval_id: i64,
) -> Result<()> {
crate::actions::run_deploy_tail(coord, dag_id, agent, approval_id).await;
Ok(())
}

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

@ -88,18 +88,6 @@ pub struct RunningTransient {
pub since: DateTime<Utc>,
}
/// A node claimed for execution — everything the executor needs, snapshotted at
/// claim time.
#[derive(Debug, Clone)]
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). Empty for
/// the agentless [`NodeKind::MetaLock`] + [`NodeKind::Dag`] container nodes.
pub agent: String,
}
/// An owned read-view of a DAG container's carried metadata ([`NodeKind::Dag`]).
/// Derived on read from the container node — the data has a single home (the
/// node payload); this is not a stored side-table.
@ -245,88 +233,41 @@ impl JobQueue {
Ok(container.get())
}
/// Claim every currently-runnable node, acquiring its resources, and mark it
/// `Running`. Delegates readiness + resource acquisition to the crate's
/// settle loop; builds a [`Claim`] per started node from its payload + its
/// DAG container's metadata. The container node itself is claimed like any
/// other (its executor is an instant no-op that lets its subtree start).
pub fn claim_ready(&self) -> Vec<Claim> {
let mut inner = self.lock();
let inner = &mut *inner;
let started = inner.settle();
let mut claims = Vec::with_capacity(started.len());
for id in started {
let Some(node) = inner.graph().node(id) else {
continue;
};
let kind = node.payload.clone();
let agent = node.payload.agent().to_owned();
let Some(container) = inner.graph().root_of(id) else {
continue;
};
claims.push(Claim {
dag_id: container.get(),
node_id: id,
kind,
agent,
});
// `started_at` is stamped on the graph `Node` by the scheduler's
// transition to `Running` — no host-side copy needed.
}
claims
/// The scheduler itself, for `hive_jobq`'s run-loop seam
/// (`Scheduler::claim_next`), which takes exactly this type.
///
/// Handing out the `Arc` rather than wrapping each crate call keeps the
/// host from growing a parallel API: the run loop uses `hive_jobq`'s
/// functions directly, and this module stays the thin glue it is being
/// reduced to.
pub(crate) fn sched(&self) -> &Arc<Mutex<Sched>> {
&self.sched
}
/// Mark a claimed node terminal, recording its outcome + (truncated) error.
/// The crate releases the node's build slot immediately and cascades the
/// `AfterOk` failure cancellation + subtree lease release.
///
/// Nothing is returned: a DAG's terminal side effects are its own tail nodes
/// ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]), which the
/// scheduler claims and runs like any other node.
pub fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
// Deliberately not `complete_node_growing(.., self.new_job())`: that
// would take the lock twice (once to mint an empty builder, once to
// complete) to express "grew nothing". The shared part is the outcome
// mapping, and that's a free fn.
let mut inner = self.lock();
inner.complete(node_id, outcome_of(result));
drop(inner);
self.notify.notify_one();
/// The DAG container id owning `node`, for log lines and the dashboard.
/// Derived from the graph rather than carried alongside the node — the
/// parent axis already knows it.
#[must_use]
pub fn dag_of(&self, node: NodeId) -> Option<u64> {
self.lock().graph().root_of(node).map(NodeId::get)
}
/// A builder for a node to declare more work into while it runs.
///
/// Handed to [`exec::run_node`] and returned to
/// [`JobQueue::complete_node_growing`]. Only `hive_jobq` can construct one,
/// which is why this goes through the scheduler rather than
/// `Job::default()`.
/// Handed to [`exec::run_node`] and returned to the crate's completion.
/// Only `hive_jobq` can construct one, which is why this goes through the
/// scheduler rather than `Job::default()`.
///
/// The completion wrappers that used to live beside this (`complete_node`,
/// `complete_node_growing`) are **gone**: a node is completed inside the
/// future [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so
/// this layer has nothing left to wrap. The tests keep their own extension
/// trait for driving completions by hand.
#[must_use]
pub fn new_job(&self) -> Job {
self.lock().new_job()
}
/// [`JobQueue::complete_node`] plus the work the node declared while it ran.
///
/// `grown` is inserted **under `node_id`** before the completion, so the DAG
/// cannot roll terminal with the appended work still pending — the property
/// the old two-call `append_subgraph` + `complete_node` sequence had to
/// arrange by hand at every call site.
pub fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
let mut inner = self.lock();
// A rejected grown job is logged, not propagated: the node's own work
// already ran, and refusing to complete it here would both misreport
// that and wedge the DAG on a node stuck `Running`.
if let Err(e) = inner.complete_growing(node_id, outcome_of(result), grown) {
tracing::error!(
node = node_id.get(),
error = %e,
"job_queue: work grown by a completing node was rejected"
);
}
drop(inner);
self.notify.notify_one();
}
/// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
/// so each is cancelled. `false` once any work node is running or terminal —
/// an in-flight nix build isn't interruptible.

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

@ -17,23 +17,17 @@
//!
//! In-DAG growth (a `MetaLock` growing rebuild subgraphs, a `Reconcile` fanning
//! its `Start`/`Stop`) is declared onto the builder each node is handed, and
//! inserted as part of completing that node — see `handle_completion`.
//! inserted as part of completing that node. Completion itself is not this
//! module's job any more: it happens *inside* the future
//! [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so a node that
//! ran but was never completed is not an expressible state here.
use std::collections::HashMap;
use std::sync::Arc;
use super::exec;
use super::{Claim, Job};
use crate::coordinator::Coordinator;
struct NodeDone {
claim: Claim,
/// Whatever the node declared into its builder while running — usually
/// nothing. Inserted under the node as part of completing it.
grown: Job,
result: anyhow::Result<()>,
}
/// Scheduler loop. Spawned once at hive-c0re startup from `main.rs`.
///
/// Shutdown semantics: subscribes to `coord.shutdown_rx()`. On a true signal
@ -56,7 +50,6 @@ struct NodeDone {
/// reconverging silently.
pub async fn run_worker(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<NodeDone>();
// Last derived pill set we published, keyed by agent (its lease is cap-1,
// so one pill each). Purely the previous value of a *derived* quantity —
// it exists to spot transitions, since the dashboard wants edges
@ -73,33 +66,69 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
return;
}
reconcile_transients(&coord, &mut transients);
let claims = coord.job_queue.claim_ready();
if !claims.is_empty() {
for claim in claims {
tracing::info!(
dag = claim.dag_id,
node = claim.node_id.get(),
kind = claim.kind.as_str(),
agent = %claim.agent,
"job_queue: node running"
);
let coord = Arc::clone(&coord);
let tx = tx.clone();
tokio::spawn(async move {
// The node's growth channel. Local state, so it costs
// nothing to carry and holds no lock while the node runs.
// The builder is passed by value and handed back: owned it
// is `Send`, a `&Job` held across an await is not.
let job = coord.job_queue.new_job();
let (grown, result) = exec::run_node(&coord, job, &claim).await;
// Send failure = scheduler gone (shutdown); drop.
let _ = tx.send(NodeDone {
claim,
grown,
result,
});
});
}
// Claim exactly one node and get back the work that runs it. `Some`
// means something started, so there may be more runnable right now —
// loop again immediately. `None` means nothing is runnable and the
// loop parks below. That decision is the whole reason the crate hands
// back a task rather than an id.
let runner = {
// Two handles, deliberately: `sched` is the scheduler the crate
// locks, `node_coord` is what the node's own future captures. One
// binding can't do both — passing `coord.job_queue.sched()` borrows
// `coord` for the whole call while the `move` closure wants to take
// it.
let sched = Arc::clone(coord.job_queue.sched());
let node_coord = Arc::clone(&coord);
hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, job| {
let coord = node_coord;
async move {
tracing::info!(
dag = coord.job_queue.dag_of(id).unwrap_or_default(),
node = id.get(),
kind = kind.as_str(),
agent = %kind.agent(),
"job_queue: node running"
);
let (grown, result) = exec::run_node(&coord, job, id, &kind).await;
match &result {
Ok(()) => tracing::info!(node = id.get(), "job_queue: node done"),
Err(e) => tracing::warn!(
node = id.get(),
kind = kind.as_str(),
agent = %kind.agent(),
error = %format!("{e:#}"),
grown_nodes = !grown.is_empty(),
"job_queue: node failed"
),
}
let outcome = super::outcome_of(result.map_err(|e| format!("{e:#}")));
// Growth is dropped on failure: a node that declared
// follow-up work and *then* failed does not want it run —
// failure cancel-cascades, so inserting it would only add
// nodes to immediately cancel.
let grown = if matches!(outcome, hive_jobq::scheduler::Outcome::Failed(_)) {
coord.job_queue.new_job()
} else {
grown
};
(grown, outcome)
}
})
};
if let Some(runner) = runner {
let done_coord = Arc::clone(&coord);
tokio::spawn(async move {
// Completion happens inside `runner` — it cannot be forgotten
// here, which is why there is no completion channel any more.
let (id, grew) = runner.await;
if let Err(e) = grew {
tracing::warn!(node = id.get(), error = %e, "job_queue: grown job rejected");
}
done_coord.emit_rebuild_queue_snapshot();
// Wake the loop: this node's completion may have unblocked
// dependents. Previously the completion channel did this.
done_coord.job_queue.notify.notify_one();
});
// Newly-started owner nodes now hold their leases — surface the pills.
reconcile_transients(&coord, &mut transients);
coord.emit_rebuild_queue_snapshot();
@ -113,63 +142,11 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
return;
}
}
Some(done) = rx.recv() => {
handle_completion(&coord, done);
}
() = coord.job_queue.notify.notified() => {}
}
}
}
fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
let NodeDone {
claim,
grown,
result,
} = done;
match result {
Ok(()) => {
tracing::info!(
dag = claim.dag_id,
node = claim.node_id.get(),
"job_queue: node done"
);
// Whatever the node declared goes in under it as part of this
// completion, so the DAG cannot roll terminal while the appended
// work is still pending. Covers the multi-node case (a `MetaLock`
// growing per-agent rebuild subgraphs) and the single-node case (a
// `Reconcile` planner's `Start` / `Stop`) identically.
coord
.job_queue
.complete_node_growing(claim.node_id, Ok(()), grown);
}
Err(e) => {
let msg = format!("{e:#}");
tracing::warn!(
dag = claim.dag_id,
node = claim.node_id.get(),
kind = claim.kind.as_str(),
agent = %claim.agent,
error = %msg,
grown_nodes = !grown.is_empty(),
"job_queue: node failed"
);
// `grown` is deliberately dropped on failure. A node that declared
// follow-up work and *then* failed does not want that work run —
// failure cancel-cascades downstream, so inserting it would only
// add nodes to immediately cancel. This preserves the old shape,
// where growth could only be expressed on the success path at all;
// the difference is that it is now possible to declare and then
// fail, so the drop has to be a decision rather than an accident.
drop(grown);
coord.job_queue.complete_node(claim.node_id, Err(msg));
}
}
// The next loop iteration re-reconciles the transient pills against the
// post-completion lease state (a settled subgraph drops its pill).
coord.emit_rebuild_queue_snapshot();
}
/// Publish the transitions between the previously-derived pill set and the
/// current one. `prev` is last loop's derived value, keyed by agent (an agent's
/// lease is cap-1, so at most one pill each).

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

@ -65,8 +65,96 @@ fn stop_online(
submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
}
/// What a test observes about a node the settle loop just started: its id, its
/// DAG, and its payload.
///
/// **Test-only, and deliberately not a production type.** `exec::run_node`
/// takes `(NodeId, &NodeKind)` and derives the DAG id on the two arms that
/// actually want it — nothing in production needs a claim snapshot to exist.
/// The assertions here are about *which* node the graph let run, which does
/// need the payload next to the id.
#[derive(Debug, Clone)]
struct Claimed {
dag_id: u64,
node_id: NodeId,
kind: NodeKind,
agent: String,
}
/// Drive one settle wave and report every node that started.
///
/// An **extension trait rather than a method on [`JobQueue`]**: production
/// claims one node at a time ([`hive_jobq::scheduler::Scheduler::claim_next`])
/// and has no use for a whole wave, so this must not be reachable from
/// non-test code. `settle()` is that same claim primitive in a loop, so a test
/// driving it here exercises the production path.
trait ClaimReady {
fn claim_ready(&self) -> Vec<Claimed>;
}
impl ClaimReady for JobQueue {
fn claim_ready(&self) -> Vec<Claimed> {
let mut sched = self.sched().lock().expect("job_queue mutex poisoned");
let started = sched.settle();
started
.into_iter()
.filter_map(|node_id| {
let kind = sched.graph().node(node_id)?.payload.clone();
Some(Claimed {
dag_id: sched.graph().root_of(node_id)?.get(),
node_id,
agent: kind.agent().to_owned(),
kind,
})
})
.collect()
}
}
/// Drive a node terminal by hand.
///
/// Also an extension trait, for the same reason as [`ClaimReady`]: production
/// completes a node **inside** the future
/// [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so "run the node,
/// then remember to complete it" is not an expressible sequence there — which
/// was the whole point of the seam. These tests need to express it, because
/// they exercise the graph without running any executor.
trait CompleteNode {
fn complete_node(&self, node_id: NodeId, result: Result<(), String>);
fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job);
}
impl CompleteNode for JobQueue {
fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
self.sched()
.lock()
.expect("job_queue mutex poisoned")
.complete(node_id, outcome_of(result));
self.notify.notify_one();
}
fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
// A rejected grown job is logged, not propagated: the node's own work
// already ran, and refusing to complete it here would both misreport
// that and wedge the DAG on a node stuck `Running`.
if let Err(e) = self
.sched()
.lock()
.expect("job_queue mutex poisoned")
.complete_growing(node_id, outcome_of(result), grown)
{
tracing::error!(
node = node_id.get(),
error = %e,
"job_queue: work grown by a completing node was rejected"
);
}
self.notify.notify_one();
}
}
/// Claim helper asserting exactly one node comes back.
fn claim_one(q: &JobQueue) -> Claim {
fn claim_one(q: &JobQueue) -> Claimed {
let mut claims = q.claim_ready();
assert_eq!(
claims.len(),