From 2454a1ea6a009d71d58a6af668fd6d6d3e9ddfc1 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 16:30:13 +0200 Subject: [PATCH 01/27] jobq: drop JobBuilder's Default impl so it is really unconstructible outside MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The builder's module doc claimed a builder "cannot be constructed, held or inserted from outside this crate". Two of those three were false: `new()` is `pub(crate)`, but a hand-written `impl Default for JobBuilder` is a trait impl on a `pub` type, so it is public regardless — `JobBuilder::default()` compiled downstream. Nothing was unsound (`insert_with` stayed `pub(crate)`, so an outside-built builder could not reach a graph), but the sentence claimed more than the visibility enforced, which is the bug this crate's docs have hit before. Delete the impl; `new()` constructs directly. The doc now says only what is enforced, and records why there is no `Default` — so the next person reaching for one finds the reason instead of adding it back. --- hive-jobq/src/builder.rs | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/hive-jobq/src/builder.rs b/hive-jobq/src/builder.rs index fa1e9807..a2e207e9 100644 --- a/hive-jobq/src/builder.rs +++ b/hive-jobq/src/builder.rs @@ -8,8 +8,10 @@ //! **An insertion API, not a spec factory.** A builder is only ever handed to a //! closure by the single insertion entry point //! ([`crate::scheduler::Scheduler::insert_job`]), which inserts the declared -//! nodes and returns the ids the job asked for. It cannot be constructed, held -//! or inserted from outside this crate, and there is no intermediate +//! nodes and returns the ids the job asked for. It cannot be constructed or +//! inserted from outside this crate — `new()` and `insert_with` are both +//! `pub(crate)`, and there is deliberately no `Default` impl, since a trait impl +//! on a `pub` type is public regardless. There is no intermediate //! node-description type to keep in sync with [`crate::Graph::insert`]'s signature — //! so a job has no representation that can be passed around instead of being //! inserted. @@ -242,16 +244,6 @@ pub struct JobBuilder { nodes: RefCell>>, } -// Hand-written rather than derived: `#[derive(Default)]` would demand -// `N: Default, R: Default`, which has nothing to do with an empty builder. -impl Default for JobBuilder { - fn default() -> Self { - Self { - nodes: RefCell::new(Vec::new()), - } - } -} - impl JobBuilder { /// A fresh, empty builder. /// @@ -261,8 +253,17 @@ impl JobBuilder { /// nodes and returns the ids. Nothing job-shaped is constructible or /// carryable outside this crate — otherwise it is a spec factory again, /// just with a builder's name on it. + /// + /// Deliberately **not** a `Default` impl. A trait impl on a `pub` type is + /// public no matter how private its inherent constructors are, so + /// `JobBuilder::default()` would hand every downstream crate the builder + /// this fn is `pub(crate)` to withhold. The body is what `#[derive(Default)]` + /// could not be anyway — deriving would demand `N: Default, R: Default`, + /// which has nothing to do with an empty builder. pub(crate) fn new() -> Self { - Self::default() + Self { + nodes: RefCell::new(Vec::new()), + } } /// Whether nothing has been declared yet — for a caller deciding whether an From 82ef06f445f64221a11346a7b582dbd11e01db25 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 17:20:43 +0200 Subject: [PATCH 02/27] =?UTF-8?q?refactor(#2949):=20kill=20Declare=20?= =?UTF-8?q?=E2=80=94=20a=20running=20node=20declares=20onto=20its=20own=20?= =?UTF-8?q?builder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A node no longer hands back a recipe for the scheduler to replay later. It declares straight onto a builder it was given, and that builder is inserted as part of completing the node. Deleted: `pub type Declare`, `struct NodeOutput` (+ its hand-written `Debug`), `JobQueue::append_subgraph`. Nothing added to `Dag` / `DagView`. jobq gains `Scheduler::new_job()` (the only way to obtain a `JobBuilder`) and `complete_growing(id, outcome, grown)`, which inserts under `id` and *then* completes it, so a DAG cannot roll terminal while grown work is still pending. `complete()` and `complete_growing()` share a private `finish()` rather than one redirecting through the other. The DAG-gone guard lives beside the graph now, where it cannot be skipped, instead of being a caller-side lookup. The growth executors return data (`run_meta_lock -> (Vec, RebuildOpts)`, `run_reconcile -> Option`) rather than taking the builder: a `&Job` parameter is live for the whole function body, and `&RefCell` is never `Send`, so an async fn taking one cannot be spawned. `run_node` threads the builder by value and hands it back. A node can now declare work and then fail, which was previously inexpressible. `grown` is dropped in that case — failure cancel-cascades downstream, so inserting it would only add nodes to immediately cancel — and the log line carries `grown_nodes` so the drop is visible. --- hive-c0re/src/job_queue/exec.rs | 297 ++++++++++++--------------- hive-c0re/src/job_queue/mod.rs | 105 +++++----- hive-c0re/src/job_queue/model.rs | 6 +- hive-c0re/src/job_queue/scheduler.rs | 62 ++++-- hive-c0re/src/job_queue/templates.rs | 55 +++-- hive-c0re/src/job_queue/tests.rs | 138 ++++++------- hive-c0re/src/workers/auto_update.rs | 3 +- hive-jobq/src/scheduler.rs | 66 ++++++ 8 files changed, 388 insertions(+), 344 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 95486951..2d69a6b1 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use anyhow::{Context as _, Result}; -use super::{Claim, Declare}; +use super::Claim; use hive_jobq::TerminalState; use super::model::NodeKind; @@ -26,36 +26,6 @@ use crate::power::{ReconcileAction, reconcile_action}; /// N × this timeout. pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3); -/// Extra signal an executor hands back to the scheduler alongside -/// success. -#[derive(Default)] -pub struct NodeOutput { - /// Whole per-agent *subgraphs* to append into *this same* DAG at - /// runtime — the single in-DAG-growth channel. Each [`Job`] is one - /// independent subgraph, declared but not yet inserted: an executor cannot - /// reach the queue, so it hands the declaration back and the scheduler - /// inserts it via [`super::JobQueue::append_subgraph`] under its own lock, - /// rooted on the emitting node. Used both for the multi-node case - /// (`MetaLock` growing one rebuild subgraph per agent — the startup - /// sweep's stale agents, the meta-update cascade's affected agents) and - /// the single-node case (a `Reconcile` planner emitting its mechanical - /// `Start` / `Stop` as a one-node subgraph). The scheduler applies these - /// *before* the emitting node's completion so the DAG never rolls terminal - /// with the appended work still pending — keeping the lease-window - /// transient held across the sub-step. - pub append_subgraph: Vec, -} - -impl std::fmt::Debug for NodeOutput { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - // The subgraphs are closures — how many were emitted is the only thing - // there is to say about them before the queue runs them. - f.debug_struct("NodeOutput") - .field("append_subgraph", &self.append_subgraph.len()) - .finish() - } -} - /// Build-log sink for one claimed node. struct Ctx<'a> { coord: &'a Arc, @@ -78,13 +48,35 @@ impl Ctx<'_> { /// Run one claimed node to completion. Called from a task the /// scheduler spawns per claim; the `Result` (stringified) becomes the /// node's terminal state. -pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result { +/// +/// `job` is the node's own growth channel: an executor that decides more work +/// is needed declares it here, and the scheduler inserts it under this node +/// when the node completes. Most executors never touch it. Nothing is inserted +/// while the node runs — the builder is local state, so this stays outside the +/// queue's lock for the whole (often multi-minute) execution. +/// +/// ⚠️ Taken **by value and handed back**, not by reference. A `JobBuilder` is +/// `RefCell`-backed: owned it is `Send`, but `&JobBuilder` is not (a shared ref +/// is `Send` only if the referent is `Sync`, and `RefCell` never is). A `&Job` +/// parameter would be live across every `.await` in this fn and make the whole +/// 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. +pub(super) async fn run_node( + coord: &Arc, + job: super::Job, + claim: &Claim, +) -> (super::Job, Result<()>) { let ctx = Ctx { coord, dag_id: claim.dag_id, node_id: claim.node_id, }; - match &claim.kind { + // 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, &ctx).await, NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).await, @@ -95,19 +87,40 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< sweep, fanout, inputs, - } => run_meta_lock(coord, *sweep, fanout.clone(), inputs).await, - NodeKind::Reconcile { .. } => run_reconcile(coord, claim).await, + } => run_meta_lock(coord, *sweep, fanout.clone(), inputs) + .await + .map(|(agents, opts)| { + for agent in agents { + super::templates::rebuild_nodes(&job, &agent, opts, None); + } + }), + NodeKind::Reconcile { .. } => run_reconcile(coord, claim).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 + // re-entrant borrow — no second unit, no deadlock. It exists so + // the requirement belongs to the node rather than to the fact + // that a `Reconcile` happens to fan it out. + let lease = Resource::Agent(kind.agent().to_owned()); + 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::Signal { .. } => Ok(run_signal(coord, claim)), + NodeKind::Signal { .. } => { + run_signal(coord, claim); + 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::MergeVerify { approval_id, .. } => run_merge_verify(coord, *approval_id).await, NodeKind::DeployApply { approval_id, .. } => { - run_deploy_apply(coord, claim, *approval_id).await + run_deploy_apply(coord, *approval_id).await.map(|()| { + super::templates::deploy_rebuild_nodes(&job, claim.kind.agent(), *approval_id); + }) } NodeKind::FinalizeDeploy { approval_id, .. } => { run_finalize_deploy(coord, *approval_id).await @@ -119,7 +132,10 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< approval_id, outcome, } => run_resolve_approval(coord, claim, *approval_id, *outcome).await, - NodeKind::EmitRebuilt { ok, .. } => Ok(run_emit_rebuilt(coord, claim, *ok)), + NodeKind::EmitRebuilt { ok, .. } => { + run_emit_rebuilt(coord, claim, *ok); + Ok(()) + } NodeKind::SetWanted { up, .. } => run_set_wanted(coord, claim, *up), // The two nodes that carry no work of their own; completing either // lets it reach `Finishing` so the nodes under it start. @@ -127,8 +143,9 @@ pub(super) async fn run_node(coord: &Arc, claim: &Claim) -> Result< // any, is its own tail node in the graph. // - `DeployWindow`: pure resource holder — the meta window, agent lease // and build slot it declares stay held until its subtree settles. - NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } => Ok(NodeOutput::default()), - } + NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } => Ok(()), + }; + (job, result) } /// Resolve the DAG's approval row the way this node's own `outcome` says. @@ -143,18 +160,18 @@ async fn run_resolve_approval( claim: &Claim, approval_id: i64, outcome: TerminalState, -) -> Result { +) -> Result<()> { let reason = (outcome == TerminalState::Failed) .then(|| coord.job_queue.first_error(claim.dag_id)) .flatten(); crate::actions::resolve_approval_dag(coord, approval_id, outcome, reason.as_deref()).await; - Ok(NodeOutput::default()) + Ok(()) } /// 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, claim: &Claim, ok: bool) -> NodeOutput { +fn run_emit_rebuilt(coord: &Arc, claim: &Claim, ok: bool) { coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { agent: claim.agent.clone(), ok, @@ -164,7 +181,6 @@ fn run_emit_rebuilt(coord: &Arc, claim: &Claim, ok: bool) -> NodeOu sha: None, tag: None, }); - NodeOutput::default() } /// Write the agent's durable power intent — the DAG-node form of the old @@ -175,7 +191,7 @@ fn run_emit_rebuilt(coord: &Arc, claim: &Claim, ok: bool) -> NodeOu /// 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, claim: &Claim, up: bool) -> Result { +fn run_set_wanted(coord: &Arc, claim: &Claim, up: bool) -> Result<()> { let wanted = if up { crate::power::Wanted::Up } else { @@ -185,7 +201,7 @@ fn run_set_wanted(coord: &Arc, claim: &Claim, up: bool) -> Result, claim: &Claim, up: bool) -> Result, - claim: &Claim, - relock: bool, -) -> Result { +async fn run_meta_sync(coord: &Arc, claim: &Claim, relock: bool) -> Result<()> { let name = &claim.agent; // 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 @@ -219,7 +231,7 @@ async fn run_meta_sync( if relock { crate::meta::lock_update_for_rebuild(name).await?; } - Ok(NodeOutput::default()) + Ok(()) } /// Out-of-band toplevel build while the container keeps serving: warm @@ -229,7 +241,7 @@ async fn run_meta_sync( /// 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, ctx: &Ctx<'_>) -> Result { +async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { let name = &claim.agent; // Warm the toplevel build only when the container is up — the whole // point of prebuild is to shrink the swap's downtime window. A @@ -240,7 +252,7 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result { crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id)) .await?; } - Ok(NodeOutput::default()) + Ok(()) } /// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb), @@ -248,7 +260,7 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> 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, claim: &Claim, ctx: &Ctx<'_>) -> Result { +async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { let name = &claim.agent; // Swap runs on an already-existing (stopped) container — runtime dir // and listener were created earlier. Pure path accessor suffices. @@ -269,7 +281,7 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res if result.is_err() { coord.rescan_containers_and_emit().await; } - result.map(|()| NodeOutput::default()) + result } /// The post-`Swap` bookkeeping tail, split into its own node for dashboard @@ -277,7 +289,7 @@ async fn run_swap(coord: &Arc, claim: &Claim, ctx: &Ctx<'_>) -> Res /// 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, claim: &Claim) -> Result { +async fn run_post_swap(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; 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) @@ -298,20 +310,20 @@ async fn run_post_swap(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { +async fn run_provision(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; let agent_dir = crate::paths::agent_runtime_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir); crate::lifecycle::provision_container(name, &hive, &paths).await?; - Ok(NodeOutput::default()) + Ok(()) } /// `nixos-container create` proper — the upstream `Provision` node @@ -320,21 +332,24 @@ async fn run_provision(coord: &Arc, claim: &Claim) -> Result Result { +async fn run_create(claim: &Claim) -> Result<()> { crate::lifecycle::create_only(&claim.agent).await?; - Ok(NodeOutput::default()) + Ok(()) } /// Meta flake lock bump. Boot-sweep flavour is non-fatal (a failed /// bump must not cancel the fan-out rebuilds — they proceed against /// the current lock, exactly like today's sweep); the meta-update /// flavour propagates errors, and a failed bump fans out nothing. +/// Returns the agents whose rebuild subgraphs the caller should grow into this +/// node, and the options to build them with — rather than declaring them here. +/// The declaration has to happen outside any `.await` (see [`run_node`]). async fn run_meta_lock( coord: &Arc, sweep: bool, fanout: Option>, inputs: &[String], -) -> Result { +) -> Result<(Vec, super::templates::RebuildOpts)> { if sweep { if let Err(e) = crate::meta::lock_update_hyperhive().await { tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed"); @@ -349,25 +364,13 @@ async fn run_meta_lock( // drain window rather than being cut off. The per-agent drains overlap, // so the sweep's cost ceiling is one `GRACEFUL_STOP_TIMEOUT` in total, // not one per agent. - let append_subgraph = fanout - .unwrap_or_default() - .iter() - .map(|agent| { - let agent = agent.clone(); - Box::new(move |b: &super::Job| { - super::templates::rebuild_nodes( - b, - &agent, - super::templates::RebuildOpts { - relock: true, - graceful: true, - }, - None, - ); - }) as Declare - }) - .collect(); - return Ok(NodeOutput { append_subgraph }); + return Ok(( + fanout.unwrap_or_default(), + super::templates::RebuildOpts { + relock: true, + graceful: true, + }, + )); } let _progress = coord.meta_update_guard(); crate::meta::lock_update(inputs).await?; @@ -383,68 +386,47 @@ async fn run_meta_lock( // cascade children must NOT re-lock, which would revert the bump this // node just committed (the property the old `fanout_specs` meta-update // branch encoded). - let append_subgraph = cascade - .iter() - .map(|agent| { - let agent = agent.clone(); - Box::new(move |b: &super::Job| { - super::templates::rebuild_nodes( - b, - &agent, - super::templates::RebuildOpts { - relock: false, - graceful: false, - }, - None, - ); - }) as Declare - }) - .collect(); - Ok(NodeOutput { append_subgraph }) + Ok(( + cascade, + super::templates::RebuildOpts { + relock: false, + graceful: false, + }, + )) } /// Idempotent power-converge *planner*: compare `wanted` (durable /// intent) against observed state and, when they diverge, fan the /// mechanical `Start` / `Stop` out as a first-class node appended to -/// *this* DAG (a single-node `NodeOutput::append_subgraph` rooted on -/// this node). Does no container work itself — the sub-step becomes -/// visible in the DAG and the lease-window transient (or the sub-step's -/// own node-local guard) rides across it. -async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result { +/// *this* DAG (a single node declared into `job`, rooted on this node). +/// Does no container work itself — the sub-step becomes visible in the +/// DAG and the lease-window transient (or the sub-step's own node-local +/// 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, claim: &Claim) -> Result> { let name = &claim.agent; let running = crate::lifecycle::is_running(name).await; let wanted = coord.power.get_or_seed(name, running)?; - // One node targeting this agent, rooted on this reconcile node. `NodeKind` - // carries the agent it targets, so stamp `claim.agent` into the fanned-out - // Start/Stop kind (one in-DAG-growth channel). - let sub = |kind: NodeKind| { - // `Start` / `Stop` declare the lease they run under. This node is their - // parent and holds it, so the declaration is a re-entrant borrow — no - // second unit, no deadlock. It exists so the requirement belongs to the - // node rather than to the fact that a `Reconcile` happens to fan it out. - let lease = Resource::Agent(kind.agent().to_owned()); - vec![Box::new(move |b: &super::Job| { - let _ = b.node(kind).needs(lease); - }) as Declare] - }; - let append_subgraph = match reconcile_action(wanted, running) { - ReconcileAction::Start => sub(NodeKind::Start { + Ok(match reconcile_action(wanted, running) { + ReconcileAction::Start => Some(NodeKind::Start { agent: name.clone(), }), - ReconcileAction::Stop => sub(NodeKind::Stop { + ReconcileAction::Stop => Some(NodeKind::Stop { agent: name.clone(), }), ReconcileAction::Noop => { tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop"); - Vec::new() + None } - }; - Ok(NodeOutput { append_subgraph }) + }) } /// 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, claim: &Claim) -> Result { +async fn run_start(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; // No node-local transient guard: the pill is derived from the running node // set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This @@ -468,12 +450,12 @@ async fn run_start(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { +async fn run_stop(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; // See `run_start`: no node-local guard — `Stop` reports `Stopping` from its // own kind now. @@ -483,12 +465,12 @@ async fn run_stop(coord: &Arc, claim: &Claim) -> Result agent: name.clone(), }); coord.rescan_containers_and_emit().await; - Ok(NodeOutput::default()) + Ok(()) } /// Mechanical stop for the profile swap. Never *changes* `wanted`; /// noop when already stopped. -async fn run_stop_for_update(coord: &Arc, claim: &Claim) -> Result { +async fn run_stop_for_update(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; if crate::lifecycle::is_running(name).await { // Seed a missing agent_power row from the PRE-stop observation @@ -501,7 +483,7 @@ async fn run_stop_for_update(coord: &Arc, claim: &Claim) -> Result< crate::lifecycle::kill(name).await?; coord.rescan_containers_and_emit().await; } - Ok(NodeOutput::default()) + Ok(()) } /// Set the graceful fence + kick so the harness sees it promptly and @@ -513,19 +495,18 @@ async fn run_stop_for_update(coord: &Arc, 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, claim: &Claim) -> NodeOutput { +fn run_signal(coord: &Arc, claim: &Claim) { if hive_types::Ident::parse(&claim.agent).is_ok_and(|a| Coordinator::is_paused(&a)) { - return NodeOutput::default(); + return; } coord.mark_graceful_stop(&claim.agent); coord.kick_agent(&claim.agent, "graceful stop requested"); - NodeOutput::default() } /// 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, claim: &Claim) -> Result { +async fn run_drain(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT; while coord.is_graceful_stop_pending(name) { @@ -536,11 +517,11 @@ async fn run_drain(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { +async fn run_write_dropin(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; // 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 @@ -549,13 +530,13 @@ async fn run_write_dropin(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result { +async fn run_write_perm_file(coord: &Arc, claim: &Claim) -> Result<()> { use super::model::PermPayload; let name = &claim.agent; // The perm file payload rides the node itself (the only consumer). @@ -591,7 +572,7 @@ async fn run_write_perm_file(coord: &Arc, claim: &Claim) -> Result< } } } - Ok(NodeOutput::default()) + Ok(()) } /// Apply the node's `(child, new_parent)` moves as one `META_LOCK`-fused @@ -601,7 +582,7 @@ async fn run_write_perm_file(coord: &Arc, 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, claim: &Claim) -> Result { +async fn run_reparent(coord: &Arc, claim: &Claim) -> Result<()> { let NodeKind::Reparent { moves } = &claim.kind else { anyhow::bail!("run_reparent on a non-Reparent node"); }; @@ -618,16 +599,14 @@ async fn run_reparent(coord: &Arc, claim: &Claim) -> Result, approval_id: i64) -> Result { - crate::actions::run_deploy_merge_verify(coord, approval_id) - .await - .map(|()| NodeOutput::default()) +async fn run_merge_verify(coord: &Arc, approval_id: i64) -> Result<()> { + crate::actions::run_deploy_merge_verify(coord, approval_id).await } /// Deploy phase 2 — the irreversible half: ff-merge, then phase 1 of the @@ -639,27 +618,15 @@ async fn run_merge_verify(coord: &Arc, approval_id: i64) -> Result< /// their `MetaSync` declares is re-entered rather than deadlocked against the /// ancestor already holding it. On failure nothing is appended and the tail /// compensates, exactly as before. -async fn run_deploy_apply( - coord: &Arc, - claim: &Claim, - approval_id: i64, -) -> Result { - crate::actions::run_deploy_apply(coord, approval_id).await?; - Ok(NodeOutput { - append_subgraph: vec![super::templates::deploy_rebuild_nodes( - claim.kind.agent(), - approval_id, - )], - }) +async fn run_deploy_apply(coord: &Arc, approval_id: i64) -> Result<()> { + crate::actions::run_deploy_apply(coord, approval_id).await } /// Deploy phase 3 — close the staged-lock window once the appended rebuild has /// come up clean: drop the rollback ref, plant the `deployed/` tag, commit /// the staged lock. -async fn run_finalize_deploy(coord: &Arc, approval_id: i64) -> Result { - crate::actions::run_finalize_deploy(coord, approval_id) - .await - .map(|()| NodeOutput::default()) +async fn run_finalize_deploy(coord: &Arc, approval_id: i64) -> Result<()> { + crate::actions::run_finalize_deploy(coord, approval_id).await } /// Deploy compensation + bookkeeping tail. `AfterAny` the apply node, so it @@ -669,14 +636,10 @@ async fn run_finalize_deploy(coord: &Arc, 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, - claim: &Claim, - approval_id: i64, -) -> Result { +async fn run_deploy_tail(coord: &Arc, claim: &Claim, approval_id: i64) -> Result<()> { crate::actions::run_deploy_tail(coord, Some(claim.dag_id), claim.kind.agent(), approval_id) .await; - Ok(NodeOutput::default()) + Ok(()) } /// Compute which agents a `nix flake update ` on the meta diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 3db89559..3e351673 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -55,16 +55,6 @@ use resource::Resource; /// borrowed one; only `hive_jobq` can make or insert it. pub type Job = hive_jobq::JobBuilder; -/// A job's shape as a **recipe**: given a builder, declare the nodes. -/// -/// What a template returns and what an executor hands back, because neither -/// can build a job itself — `hive_jobq` creates the builder inside its own -/// insertion call and never lets one out. So the transferable thing is the -/// declaring closure, and the queue runs it at the moment it inserts. -/// -/// `Send` because an executor's output crosses the scheduler's task boundary. -pub type Declare = Box; - /// 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. @@ -163,6 +153,18 @@ impl Default for JobQueue { } } +/// A node runner's `Result` as the scheduler's [`Outcome`]. +/// +/// The failure reason + `finished_at` are stamped onto the graph `Node` by the +/// scheduler (the reason rides `Outcome::Failed`); there is no host-side copy, +/// so nothing needs clearing on success. +fn outcome_of(result: Result<(), String>) -> Outcome { + match result { + Ok(()) => Outcome::Done, + Err(e) => Outcome::Failed(truncate_error(&e)), + } +} + /// Insert a declared `job` into the shared graph and record its per-node /// `node_rt`, returning the inserted ids. /// @@ -221,7 +223,7 @@ impl JobQueue { /// roots re-parented to the container). Returns the container's id as the /// DAG id — its rolled-up state is the DAG state. /// - /// Takes the spec's recipe by generic, not as a boxed [`Declare`]: a spec + /// Takes the spec's recipe by generic, not as a boxed closure: a spec /// travels from the template that built it directly into this call, so /// there is nothing to allocate for. /// @@ -254,38 +256,6 @@ impl JobQueue { Ok(container.get()) } - /// Append a whole *subgraph* into a live DAG at runtime — the single - /// in-DAG-growth primitive. The subgraph is inserted as a [`insert_group`] - /// rooted under `dep_on` (the emitting node): the subgraph's own root becomes - /// a *child* of `dep_on`, its steps children of that root, and the group's - /// agent lease is hoisted onto that root. Ordering root→`dep_on` is the parent - /// gate — the children run once `dep_on` reaches `Finishing`. Because the - /// emitting node stays `Finishing` until this appended subtree is terminal and - /// the DAG's terminal node deps on the top root, roll-up keeps the DAG from - /// settling early with no explicit wiring. A no-op if the DAG is gone. - pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) { - let mut inner = self.lock(); - if inner.container(dag_id).is_none() { - return; - } - // Insert the subgraph as a group rooted under the emitting node: the - // subgraph's own root becomes a child of `dep_on`, its steps children of - // that root. No terminal-node wiring — roll-up carries terminality: the - // emitter stays `Finishing` until this appended subtree settles, and the - // container node rolls up terminal only once its whole subtree (incl. this - // appended work) has settled, so the DAG hook waits for free. - if let Err(e) = insert_group(&mut inner, declare, Some(dep_on)) { - tracing::error!( - dag = dag_id, - error = %e, - "job_queue: append_subgraph insert failed" - ); - return; - } - drop(inner); - self.notify.notify_one(); - } - /// 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 @@ -325,15 +295,48 @@ impl JobQueue { /// ([`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(); - // The failure reason + `finished_at` are stamped onto the graph `Node` - // by the scheduler (the reason rides `Outcome::Failed`); no host-side - // copy, so there is nothing to clear here. - let outcome = match result { - Ok(()) => Outcome::Done, - Err(e) => Outcome::Failed(truncate_error(&e)), - }; - inner.sched.complete(node_id, outcome); + inner.sched.complete(node_id, outcome_of(result)); + drop(inner); + self.notify.notify_one(); + } + + /// 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()`. + #[must_use] + pub fn new_job(&self) -> Job { + self.lock().sched.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 + .sched + .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(); } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index b6a9fe8d..b95f89e8 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -407,9 +407,9 @@ impl NodeKind { /// Generic over the recipe rather than boxing it: a spec goes from the template /// that returns it straight to the `submit` that consumes it, so the closure's /// concrete type is known the whole way and needs neither an allocation nor a -/// `Send` bound. (The executor's `append_subgraph` is the case that *does* need -/// a boxed [`super::Declare`] — its recipes are collected into a `Vec` and -/// applied later, across a task boundary.) +/// `Send` bound. Nothing boxes a recipe any more — a running node grows its DAG +/// by declaring straight onto the builder it was handed, so there is no recipe +/// to store and replay across a task boundary. pub struct DagSpec { pub source: Source, /// Free-form "why". diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 3efdf913..d1aec330 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -16,19 +16,22 @@ //! the DAG settles. //! //! In-DAG growth (a `MetaLock` growing rebuild subgraphs, a `Reconcile` fanning -//! its `Start`/`Stop`) flows through `NodeOutput.append_subgraph`, applied -//! before the emitting node completes — see `handle_completion`. +//! its `Start`/`Stop`) is declared onto the builder each node is handed, and +//! inserted as part of completing that node — see `handle_completion`. use std::collections::HashMap; use std::sync::Arc; -use super::Claim; -use super::exec::{self, NodeOutput}; +use super::exec; +use super::{Claim, Job}; use crate::coordinator::Coordinator; struct NodeDone { claim: Claim, - result: anyhow::Result, + /// 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`. @@ -83,9 +86,18 @@ pub async fn run_worker(coord: Arc) { let coord = Arc::clone(&coord); let tx = tx.clone(); tokio::spawn(async move { - let result = exec::run_node(&coord, &claim).await; + // 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, result }); + let _ = tx.send(NodeDone { + claim, + grown, + result, + }); }); } // Newly-started owner nodes now hold their leases — surface the pills. @@ -110,27 +122,26 @@ pub async fn run_worker(coord: Arc) { } fn handle_completion(coord: &Arc, done: NodeDone) { - let NodeDone { claim, result } = done; + let NodeDone { + claim, + grown, + result, + } = done; match result { - Ok(output) => { + Ok(()) => { tracing::info!( dag = claim.dag_id, node = claim.node_id.get(), "job_queue: node done" ); - // Append any in-DAG subgraphs BEFORE completing this node, so - // completing it doesn't roll the DAG terminal while the appended - // work is still pending. Each subgraph roots on this node - // (`AfterOk`), so it becomes ready the instant this one settles - // `Done` just below — covers both the multi-node case (a `MetaLock` + // 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`). - for subgraph in output.append_subgraph { - coord - .job_queue - .append_subgraph(claim.dag_id, subgraph, claim.node_id); - } - coord.job_queue.complete_node(claim.node_id, Ok(())); + // `Reconcile` planner's `Start` / `Stop`) identically. + coord + .job_queue + .complete_node_growing(claim.node_id, Ok(()), grown); } Err(e) => { let msg = format!("{e:#}"); @@ -140,8 +151,17 @@ fn handle_completion(coord: &Arc, done: NodeDone) { 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)); } } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 8607958c..c0016c54 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -28,7 +28,7 @@ use hive_jobq::TerminalState; use super::model::{DagSpec, NodeKind, PermPayload, Source}; use super::resource::Resource; -use super::{Declare, Handle, Job}; +use super::{Handle, Job}; /// 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* @@ -235,32 +235,29 @@ pub(crate) fn rebuild_nodes<'a>( /// finalized. `Reconcile` alone would not do, being `AfterAny` — it reaches /// `Done` even after a failed `Swap`. /// -/// Appended, not submitted: the roots below become children of the emitting -/// `DeployApply` (see [`super::JobQueue::append_subgraph`]), which puts them -/// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's -/// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor -/// already holding it rather than deadlocking against it. -pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { - let agent = agent.to_owned(); - Box::new(move |b: &Job| { - let roots = rebuild_nodes( - b, - &agent, - RebuildOpts { - relock: false, - graceful: false, - }, - None, - ); - let _finalize = b - .node(NodeKind::FinalizeDeploy { - agent: agent.clone(), - approval_id, - }) - .needs(Resource::MetaWindow) - .after_ok(roots.prebuild) - .after_ok(roots.reconcile); - }) +/// Declared into a **running** `DeployApply`'s own builder, not submitted: the +/// roots below become children of that node, which puts them inside the +/// `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's `MetaSync` +/// and `FinalizeDeploy` declare is re-entered from the ancestor already holding +/// it rather than deadlocking against it. +pub(crate) fn deploy_rebuild_nodes(b: &Job, agent: &str, approval_id: i64) { + let roots = rebuild_nodes( + b, + agent, + RebuildOpts { + relock: false, + graceful: false, + }, + None, + ); + let _finalize = b + .node(NodeKind::FinalizeDeploy { + agent: agent.to_owned(), + approval_id, + }) + .needs(Resource::MetaWindow) + .after_ok(roots.prebuild) + .after_ok(roots.reconcile); } /// One uniform rebuild shape — no `was_running` branch. `StopForUpdate` @@ -480,8 +477,8 @@ pub fn perm_change( } /// Meta-input lock bump. The `MetaLock` executor grows one rebuild subgraph -/// per affected agent into *this same* DAG on completion (via -/// `append_subgraph`) — appended *after* the bump lands so their prebuilds +/// per affected agent into *this same* DAG on completion (declared onto the +/// builder it was handed) — appended *after* the bump lands so their prebuilds /// run against the post-bump lock, and a failed bump appends nothing /// (replacing the old fan-out-child-DAGs dance). /// `transient = Rebuilding` because those appended subgraphs are rebuilds: diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 13de8d5d..f460092a 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -13,13 +13,19 @@ fn submit(q: &JobQueue, spec: DagSpec) -> u64 { q.submit(spec).expect("valid spec") } -/// Erase a spec's recipe to the boxed [`Declare`] so specs of *different* -/// shapes can share one type — e.g. a table of `(name, spec)` cases. +/// A spec recipe with its concrete closure type erased. **Test-only** — the +/// module used to export this alias for the executor's growth path too, which +/// is exactly what a node declaring onto its own builder removed: nothing in +/// production stores a recipe to replay later, so nothing needs to box one. +type ErasedRecipe = Box; + +/// Erase a spec's recipe so specs of *different* shapes can share one type — +/// e.g. a table of `(name, spec)` cases. /// /// Production never needs this: each submit path builds one spec and hands it /// straight to `submit`, so the concrete closure type is known end to end. A -/// test table is the case where several shapes must be one type. -fn erase(spec: DagSpec) -> DagSpec { +/// test table is the one case where several shapes must be one type. +fn erase(spec: DagSpec) -> DagSpec { DagSpec { source: spec.source, reason: spec.reason, @@ -758,19 +764,16 @@ fn a_fanned_out_start_declares_the_lease_and_re_enters_its_reconciles_grant() { assert_eq!(reconcile.dag_id, id); assert_eq!(reconcile.kind.as_str(), "reconcile"); - // What `run_reconcile` does on observing a down container with wanted=Up. - q.append_subgraph( - id, - Box::new(|b: &Job| { - let kind = NodeKind::Start { - agent: "agent-a".to_owned(), - }; - let lease = Resource::Agent(kind.agent().to_owned()); - let _ = b.node(kind).needs(lease); - }), - reconcile.node_id, - ); - q.complete_node(reconcile.node_id, Ok(())); + // What `run_reconcile` does on observing a down container with wanted=Up: + // declare into the builder it was handed, then hand it back with the + // completion. Same two calls the scheduler makes, in the same order. + let grown = q.new_job(); + let kind = NodeKind::Start { + agent: "agent-a".to_owned(), + }; + let lease = Resource::Agent(kind.agent().to_owned()); + let _ = grown.node(kind).needs(lease); + q.complete_node_growing(reconcile.node_id, Ok(()), grown); // (a) + (b): the child runs, under the parent that parked in `Finishing`. let start = claim_one(&q); @@ -850,7 +853,7 @@ fn boot_sweep_nodes_declare_their_own_resources() { } #[test] -fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { +fn grown_subgraph_roots_on_emitter_and_rebases_local_deps() { // The startup-sweep mechanism: a `MetaLock` emitter grows one rebuild // 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. @@ -866,31 +869,29 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { }); }), }; - let id = submit(&q, spec); + submit(&q, spec); let emitter = claim_one(&q); assert_eq!(emitter.kind.as_str(), "meta_lock"); // Two independent per-agent subgraphs — the REAL production shape the // sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain → // StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must // match the sweep arm of `run_meta_lock` or this stops tracking production. - let subgraph = |agent: &str| -> Declare { - let agent = agent.to_owned(); - Box::new(move |b: &Job| { - templates::rebuild_nodes( - b, - &agent, - templates::RebuildOpts { - relock: true, - graceful: true, - }, - None, - ); - }) - }; - // Must append BEFORE completing the emitter (the documented contract). - q.append_subgraph(id, subgraph("a"), emitter.node_id); - q.append_subgraph(id, subgraph("b"), emitter.node_id); - q.complete_node(emitter.node_id, Ok(())); + // Both subgraphs go into the emitter's own builder, exactly as + // `run_meta_lock`'s sweep arm does. Insert-before-complete is no longer the + // caller's job to remember: it is one call, and the ordering is inside it. + let grown = q.new_job(); + for agent in ["a", "b"] { + templates::rebuild_nodes( + &grown, + agent, + templates::RebuildOpts { + relock: true, + graceful: true, + }, + None, + ); + } + q.complete_node_growing(emitter.node_id, Ok(()), grown); // 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 // take turns on the cap-1 global meta window, so drain those first — what @@ -967,7 +968,7 @@ fn rebuild_chain_nodes_suppress_crash_watch() { #[test] fn meta_update_grows_cascade_in_dag() { // The meta-update `MetaLock` grows one rebuild subgraph per affected - // agent into its OWN DAG (via append_subgraph), not child DAGs. + // agent into its OWN DAG (via the builder it is handed), not child DAGs. let spec = templates::meta_update( vec!["nixpkgs".to_owned()], Source::Manual, @@ -975,26 +976,26 @@ fn meta_update_grows_cascade_in_dag() { None, ); let q = JobQueue::new(4); - let id = submit(&q, spec); + submit(&q, spec); let meta_lock = claim_one(&q); assert_eq!(meta_lock.kind.as_str(), "meta_lock"); // 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). Both + // agents go into the one builder the node was handed, which is what + // `run_meta_lock`'s fanout arm does. + let grown = q.new_job(); for agent in ["alice", "bob"] { - let declare: Declare = Box::new(move |b: &Job| { - templates::rebuild_nodes( - b, - agent, - templates::RebuildOpts { - relock: false, - graceful: false, - }, - None, - ); - }); - q.append_subgraph(id, declare, meta_lock.node_id); + templates::rebuild_nodes( + &grown, + agent, + templates::RebuildOpts { + relock: false, + graceful: false, + }, + None, + ); } - q.complete_node(meta_lock.node_id, Ok(())); + q.complete_node_growing(meta_lock.node_id, Ok(()), grown); // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root // on the MetaLock, each on its own agent lease. The per-agent `MetaSync` // heads serialize on the global meta window (they commit to the meta repo); @@ -1233,8 +1234,8 @@ fn cancelled_power_op_runs_no_compensating_node() { for graceful in [false, true] { for running in [false, true] { let targets = vec![("agent-a".to_owned(), running)]; - // Erased to `DagSpec`: three different recipe types have to - // sit in one array. + // Erased to one boxed recipe type: three different recipe types + // have to sit in one array. let cases = [ ( "restart", @@ -1420,16 +1421,14 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { let apply = claim_one(&q); assert!(matches!(apply.kind, NodeKind::DeployApply { .. })); - // Mirrors the scheduler: the executor's `NodeOutput` subgraphs are grafted - // BEFORE the emitting node is completed. Completing first would settle the - // apply node `Done` with nothing under it, opening the tail's `AfterAny` - // gate immediately and letting the deploy "finish" before it had built. - q.append_subgraph( - id, - templates::deploy_rebuild_nodes("agent-a", 11), - apply.node_id, - ); - q.complete_node(apply.node_id, Ok(())); + // Mirrors the scheduler. The graft lands BEFORE the emitting node settles, + // and that ordering is now structural rather than a rule this call site has + // to follow: completing first would settle the apply node `Done` with + // nothing under it, opening the tail's `AfterAny` gate immediately and + // letting the deploy "finish" before it had built. + let grown = q.new_job(); + templates::deploy_rebuild_nodes(&grown, "agent-a", 11); + q.complete_node_growing(apply.node_id, Ok(()), grown); // The grafted chain runs in rebuild order. `claim_one` asserts exactly one // claimable node at each step, which also proves the `AfterAny` tail stays @@ -1481,12 +1480,9 @@ fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() { let verify = claim_one(&q); q.complete_node(verify.node_id, Ok(())); let apply = claim_one(&q); - q.append_subgraph( - id, - templates::deploy_rebuild_nodes("agent-a", 13), - apply.node_id, - ); - q.complete_node(apply.node_id, Ok(())); + let grown = q.new_job(); + templates::deploy_rebuild_nodes(&grown, "agent-a", 13); + q.complete_node_growing(apply.node_id, Ok(()), grown); for expected in ["meta_sync", "prebuild", "stop_for_update"] { let c = claim_one(&q); diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 572b455b..1e0a38e6 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -391,8 +391,7 @@ fn submit_boot_tree( n_skipped, ); - let declare: crate::job_queue::Declare = - Box::new(move |b| boot_nodes(b, any_stale, fanout, drifted)); + let declare = move |b: &crate::job_queue::Job| boot_nodes(b, any_stale, fanout, drifted); let spec = DagSpec { // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 337c6b2d..ccdd017e 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -251,6 +251,72 @@ impl Scheduler { /// propagates up the parent chain. Call [`Scheduler::settle`] again afterwards /// to start newly-unblocked work. pub fn complete(&mut self, id: NodeId, outcome: Outcome) { + self.finish(id, outcome); + } + + /// A fresh builder for a **running** node to declare more work into. + /// + /// The node runs outside this scheduler's lock — often for minutes — so it + /// cannot hold a graph reference while it works. It doesn't need one: a + /// builder is pure local state (locally-minted guids, resolved to + /// [`NodeId`]s only at insert), so it can be filled in freely and handed + /// back to [`Scheduler::complete_growing`], which inserts it under the lock. + /// + /// This is the only way to get one — [`JobBuilder::new`] is `pub(crate)` and + /// there is no `Default` impl — so a caller can declare work but never + /// insert it itself. + #[must_use] + pub fn new_job(&self) -> JobBuilder { + JobBuilder::new() + } + + /// [`Scheduler::complete`], plus whatever the node declared into the builder + /// it was handed while running. + /// + /// `grown`'s nodes are inserted **under `id`** and *before* the completion, + /// so the node cannot roll terminal with its own appended work still + /// pending — the same ordering the caller previously had to arrange by + /// hand. A job that declares nothing costs nothing: the insert is skipped + /// outright, which is the overwhelmingly common case (most nodes grow no + /// work at all). + /// + /// # Errors + /// [`BuildError`] if `grown` is malformed — **and the node is still + /// completed**. Its own work already happened; refusing to complete it + /// would misreport that, and leaving it `Running` forever would wedge the + /// DAG. So the error is returned for the caller to log, not used to abort + /// the completion. This crate has no logger of its own; the caller does. + pub fn complete_growing( + &mut self, + id: NodeId, + outcome: Outcome, + grown: JobBuilder, + ) -> Result<(), BuildError> { + // A node that is no longer in the graph grows nothing. The DAG it + // belonged to can be cancelled or evicted while it runs, and the insert + // below is *unchecked* — rooting on a departed parent would plant a + // dangling `parent` edge rather than being rejected. The host used to + // carry this guard itself, as a lookup before a separate append call; + // it belongs here, where the graph is and where it cannot be skipped. + let grew = if grown.is_empty() || self.graph.node(id).is_none() { + Ok(()) + } else { + let graph = &mut self.graph; + grown + .insert_with(Some(id), &[], |payload, deps, parent| { + graph.insert_unchecked(payload, deps, parent) + }) + .map(|_ids| ()) + }; + self.finish(id, outcome); + grew + } + + /// The completion half, shared by [`Scheduler::complete`] and + /// [`Scheduler::complete_growing`] so neither is a redirect through the + /// other: the growing form must insert *before* this runs, and the plain + /// form must not pay for an empty job. + fn finish(&mut self, id: NodeId, outcome: Outcome) { match outcome { Outcome::Failed(error) => { // Record the reason before the terminal transition so it's set From 77cc7bea6b608cc3365abb1d46149acf1bb30b6e Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 18:26:32 +0200 Subject: [PATCH 03/27] refactor(#2949): the build-log row carries its node id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `QueueInner` was `{ sched, node_rt }`, where `node_rt` held exactly one datum per node: the `build_logs` row id. It existed because a `hive_jobq` node payload is immutable after insert while the log row is created when the build starts — so the link could not ride the node. Invert it: the log row names its node (`build_logs.node_id`, one migration in the existing `schema_versions` framework). Same single-home property, in the direction the type system allows. `QueueInner` is now just the scheduler. That is the point: the queue holds no per-node side map, so nothing has to be locked alongside the graph. Deleted as a consequence, each surfaced by dead-code analysis after the edit above rather than predicted: - `NodeRuntime`, `node_rt`, `set_build_log_id`, and `build_log_id_of` (which linear-scanned the map to match a wire `u64` against opaque `NodeId`s). The lookup is an indexed query now. - `struct Ctx`, entirely. It carried `coord` + `dag_id` + `node_id` into the executors so the build-log callback could reach the queue; without the callback, `coord`/`dag_id` were never read and `node_id` was already on the `Claim` both executors receive. - `QueueInner::node_running`, which existed only for `set_build_log_id`'s "only while running" guard. - The `Fn(i64)` callbacks on `prebuild_toplevel` / `swap_update` / `priv_run_inner`, replaced by a `node_id: Option` passed down. The id travels one way now instead of being registered back. `meta.rs`'s `nix_logged` passes `None` deliberately: its callers reach it from outside the queue as well as inside, and nothing reads the link for them yet. `id_for_node` takes `MAX(id)` rather than assuming uniqueness — a retried node opens a second row and the panel wants the current attempt. The test moved to where the behaviour lives and covers that, plus survival across completion and non-collision with node-less rows. --- hive-c0re/src/dashboard/build_logs.rs | 4 +- hive-c0re/src/job_queue/exec.rs | 41 ++-------- hive-c0re/src/job_queue/mod.rs | 77 +++++------------- hive-c0re/src/job_queue/tests.rs | 27 ++----- hive-c0re/src/lifecycle/mod.rs | 49 ++++-------- hive-c0re/src/meta.rs | 7 +- hive-c0re/src/stores/build_logs.rs | 107 +++++++++++++++++++++++--- 7 files changed, 150 insertions(+), 162 deletions(-) diff --git a/hive-c0re/src/dashboard/build_logs.rs b/hive-c0re/src/dashboard/build_logs.rs index f4835878..763ee9fb 100644 --- a/hive-c0re/src/dashboard/build_logs.rs +++ b/hive-c0re/src/dashboard/build_logs.rs @@ -156,7 +156,7 @@ pub(super) async fn get_build_log_for_node( State(state): State, AxumPath(node_id): AxumPath, ) -> Response { - match state.coord.job_queue.build_log_id_of(node_id) { + match state.coord.build_logs.id_for_node(node_id) { Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await, None => ( StatusCode::NOT_FOUND, @@ -183,7 +183,7 @@ pub(super) async fn get_build_log_raw_for_node( State(state): State, AxumPath(node_id): AxumPath, ) -> Response { - match state.coord.job_queue.build_log_id_of(node_id) { + match state.coord.build_logs.id_for_node(node_id) { Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await, None => ( StatusCode::NOT_FOUND, diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 2d69a6b1..de9e1f7d 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -26,25 +26,6 @@ use crate::power::{ReconcileAction, reconcile_action}; /// N × this timeout. pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3); -/// Build-log sink for one claimed node. -struct Ctx<'a> { - coord: &'a Arc, - dag_id: u64, - node_id: super::NodeId, -} - -impl Ctx<'_> { - fn build_log(&self, log_id: i64) { - if self - .coord - .job_queue - .set_build_log_id(self.dag_id, self.node_id, log_id) - { - self.coord.emit_rebuild_queue_snapshot(); - } - } -} - /// Run one claimed node to completion. Called from a task the /// scheduler spawns per claim; the `Result` (stringified) becomes the /// node's terminal state. @@ -67,19 +48,14 @@ pub(super) async fn run_node( job: super::Job, claim: &Claim, ) -> (super::Job, Result<()>) { - let ctx = Ctx { - coord, - dag_id: claim.dag_id, - node_id: claim.node_id, - }; // 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, &ctx).await, - NodeKind::Swap { .. } => run_swap(coord, claim, &ctx).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, @@ -241,7 +217,7 @@ async fn run_meta_sync(coord: &Arc, 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, ctx: &Ctx<'_>) -> Result<()> { +async fn run_prebuild(claim: &Claim) -> Result<()> { let name = &claim.agent; // Warm the toplevel build only when the container is up — the whole // point of prebuild is to shrink the swap's downtime window. A @@ -249,8 +225,7 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { // 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, &|log_id| ctx.build_log(log_id)) - .await?; + crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(claim.node_id.get())).await?; } Ok(()) } @@ -260,17 +235,15 @@ async fn run_prebuild(claim: &Claim, ctx: &Ctx<'_>) -> 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, claim: &Claim, ctx: &Ctx<'_>) -> Result<()> { +async fn run_swap(coord: &Arc, claim: &Claim) -> Result<()> { let name = &claim.agent; // 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, &|log_id| { - ctx.build_log(log_id); - }) - .await; + let result = + crate::lifecycle::swap_update(name, &hive, &paths, Some(claim.node_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 diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 3e351673..dfe916ec 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -36,7 +36,6 @@ pub mod templates; #[cfg(test)] mod tests; -use std::collections::HashMap; use std::sync::Mutex; use chrono::{DateTime, Utc}; @@ -101,15 +100,6 @@ pub struct Claim { pub agent: String, } -/// Per-node runtime metadata the crate graph doesn't carry. Lifecycle -/// (`started_at` / `finished_at` / `error`) lives on the `hive_jobq::Node` -/// itself now, so only the build-log row link remains host-side (the -/// client fetches the log by node id). -#[derive(Debug, Default, Clone)] -struct NodeRuntime { - build_log_id: Option, -} - /// 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. @@ -119,18 +109,21 @@ struct DagMeta { created_at: DateTime, } -/// The mutable queue state behind the mutex: the crate scheduler plus the -/// per-node runtime metadata the graph can't carry. A **DAG is a single -/// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the -/// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state -/// is the DAG state, and there are no grouping side-tables: membership + meta -/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] + -/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG. +/// The mutable queue state behind the mutex: **just the crate scheduler**. +/// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) +/// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id, +/// its rolled-up state is the DAG state, and there are no grouping side-tables: +/// membership + meta are graph queries ([`QueueInner::container`] / +/// [`QueueInner::dag_meta`] + the `hive_jobq::Graph` accessors). One shared +/// crate [`Graph`] holds every DAG. +/// +/// There is deliberately **no per-node side map** any more. The last one held +/// the `build_logs` row id; that link now lives on the log row itself +/// (`build_logs.node_id`), so it survives a restart and needs no lock held +/// alongside the scheduler's — which is what lets the scheduler's own lock be +/// the only one the run loop takes. struct QueueInner { sched: Scheduler, - /// Per-node runtime metadata (the build-log id) — mutable after - /// insert, so it can't ride the immutable node payload. - node_rt: HashMap, } /// The queue. Lives on `Coordinator` (one per hive-c0re process); a single @@ -208,7 +201,6 @@ impl JobQueue { Self { inner: Mutex::new(QueueInner { sched: Scheduler::new(Graph::new(), table), - node_rt: HashMap::new(), }), notify: Notify::new(), } @@ -244,7 +236,6 @@ impl JobQueue { None, ) .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; - inner.node_rt.insert(container, NodeRuntime::default()); insert_group(&mut inner, spec.declare, Some(container))?; // Settle the container's own (no-op) logic immediately so it parks in // `Finishing` and its children become runnable — it never needs claiming @@ -374,33 +365,6 @@ impl JobQueue { true } - /// Link a `build_logs` row to a specific `Running` node. - pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool { - let mut inner = self.lock(); - if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id) - || !inner.node_running(node_id) - { - return false; - } - inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id); - true - } - - /// The `build_logs` row id linked to the wire node id `node_id`, if any — - /// the lookup behind the `GET /api/build-log/` query endpoint (the - /// client fetches a node's captured build output on demand rather than - /// receiving it inline). Takes the raw wire `u64` (the endpoint's path - /// param); `node_rt` is keyed by the opaque `NodeId`, so this scans for the - /// matching id — the map is small (live + recently-terminal nodes). - #[must_use] - pub fn build_log_id_of(&self, node_id: u64) -> Option { - self.lock() - .node_rt - .iter() - .find(|(nid, _)| nid.get() == node_id) - .and_then(|(_, rt)| rt.build_log_id) - } - /// The first failed node's error in `dag_id`, if any has failed yet. /// /// Unlike the roll-up summary this is readable *mid-flight*, which is the @@ -498,14 +462,6 @@ impl JobQueue { } impl QueueInner { - /// Whether `id` is a `Running` node. - fn node_running(&self, id: NodeId) -> bool { - self.sched - .graph() - .node(id) - .is_some_and(|n| n.state == State::Running) - } - /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals /// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. fn container(&self, dag_id: u64) -> Option { @@ -593,7 +549,12 @@ impl QueueInner { NodeKind::MetaLock { inputs, .. } => inputs.clone(), _ => Vec::new(), }; - let build_log_id = self.node_rt.get(&id).and_then(|r| r.build_log_id); + // Looked up from the log row itself (`build_logs.node_id`), not a + // host-side map. One indexed query per node in the snapshot; the + // node set is bounded by `MAX_HISTORY_DAGS` and the store is a + // local sqlite file, so this is cheaper than the lock contention + // a second shared map would reintroduce. + let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())); // `node.parent` is the structural jobq parent. Top-level nodes // have `parent == Some(container)` (direct children of the Dag // container); those become `parent: None` on the wire since the diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index f460092a..82065941 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1544,27 +1544,12 @@ fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() { assert_eq!(state_of(&q, id), State::Failed); } -// ---- build logs, history ---- - -#[test] -fn set_build_log_id_links_running_node() { - let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); - let c = claim_one(&q); - assert!(q.set_build_log_id(id, c.node_id, 42)); - q.complete_node(c.node_id, Ok(())); - assert!( - !q.set_build_log_id(id, c.node_id, 99), - "node no longer running → refused" - ); - // The log id is fetched by node id (the `GET /api/build-log/` lookup), - // not carried on the wire — it survives completion in the node runtime. - assert_eq!( - q.build_log_id_of(c.node_id.get()), - Some(42), - "log id survives completion" - ); -} +// ---- history ---- +// +// The node → build-log link is no longer queue state: the log row carries +// `node_id` and the lookup lives in `stores::build_logs` (see +// `node_link_survives_completion_and_newest_wins` there). Nothing in the queue +// needs testing for it any more, which is the point of that move. /// History retention is a **flat** newest-first cap over all terminal DAGs /// (`MAX_HISTORY_DAGS`), not a per-template bucket behind a grace window. diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index 6f816f02..a69b71a9 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -274,10 +274,10 @@ pub async fn swap_update( name: &str, hive: &HiveEnv, paths: &AgentPaths, - on_build_log_id: &(dyn Fn(i64) + Send + Sync), + node_id: Option, ) -> Result<()> { write_dropins(name, hive, paths).await?; - priv_run_inner("update", name, Some(on_build_log_id)).await + priv_run_inner("update", name, node_id).await } /// Build the `AgentSpec` list for the meta flake from `nixos-container @@ -582,14 +582,10 @@ pub async fn destroy(name: &str) -> Result<()> { /// the prebuild happens before stop, and `docs/coordinator.md::Prebuild /// attr path` for why the explicit nixosConfigurations attr is required. /// -/// `on_build_log_id` fires with the `build_logs` row id as soon as the -/// row opens, so queue-side callers can link their node to the live -/// stream. Pass `&|_| ()` when not needed. -pub async fn prebuild_toplevel( - name: &str, - flake_ref: &str, - on_build_log_id: &(dyn Fn(i64) + Send + Sync), -) -> Result<()> { +/// `node_id` is the queue node this build belongs to, when there is one — +/// it is stored on the `build_logs` row so the dashboard can find the log +/// from the node. Pass `None` for builds that run outside the queue. +pub async fn prebuild_toplevel(name: &str, flake_ref: &str, node_id: Option) -> Result<()> { use tokio::io::{AsyncBufReadExt, BufReader}; // Split `#` so we can re-emit with the explicit // `nixosConfigurations.` segment. The flake_ref shape is @@ -624,15 +620,12 @@ pub async fn prebuild_toplevel( // into the row; `finish` lands the terminal status before we bail. let logs = crate::build_logs::global(); let log_id = logs.as_ref().and_then(|h| { - h.start(name, "prebuild", &cmdline) + h.start(name, "prebuild", &cmdline, node_id) .map_err(|e| { tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)"); }) .ok() }); - if let Some(id) = log_id { - on_build_log_id(id); - } let mut child = Command::new("nix") .args(&args) @@ -784,37 +777,25 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> { priv_run_inner(kind, name, None).await } -/// Like `priv_run` but calls `on_log_id(log_id)` immediately after the -/// build-log row is opened — before the actual container op starts. -/// This lets callers surface the row id for live streaming (e.g. the -/// rebuild-queue worker sets `build_log_id` on the queue entry so the -/// dashboard can link to `/api/build-logs/id/{id}/stream`). +/// Like `priv_run` but stamps `node_id` onto the build-log row it opens, so +/// the dashboard can find the log from the queue node (and link to +/// `/api/build-logs/id/{id}/stream`). /// -/// The callback fires only when a build-log row is successfully opened -/// (i.e. the global `BuildLogs` handle is installed AND `h.start()` -/// succeeds). No-op when `on_log_id` is `None` — that's the path for -/// all callers that don't need the id. -async fn priv_run_inner( - kind: &str, - name: &str, - on_log_id: Option<&(dyn Fn(i64) + Send + Sync)>, -) -> Result<()> { +/// This used to be a `Fn(i64)` callback that handed the row id *back* to the +/// queue, which then held it in a side map. The row carries the link itself +/// now, so the id only ever travels one way. +async fn priv_run_inner(kind: &str, name: &str, node_id: Option) -> Result<()> { let container = container_name(name); let cmdline = format!("nixos-container {kind} {container}"); let logs = crate::build_logs::global(); let log_id = logs.as_ref().and_then(|h| { - h.start(name, kind, &cmdline) + h.start(name, kind, &cmdline, node_id) .map_err(|e| { tracing::warn!(error = ?e, "build_logs: start failed (priv_run log dropped)"); }) .ok() }); - // Notify the caller as soon as the log row exists so it can surface - // the id for live streaming before the container op even starts. - if let (Some(id), Some(cb)) = (log_id, on_log_id) { - cb(id); - } // For long-running ops use the streaming protocol so build_logs // receives lines in real time rather than as a batch at completion. diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 8c4100ba..6da608f3 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -1590,7 +1590,12 @@ async fn nix_logged(dir: &Path, args: &[&str], agent: &str, kind: &str) -> Resul let cmdline = format!("nix {}", nix_argv(args).join(" ")); let logs = crate::build_logs::global(); let log_id = logs.as_ref().and_then(|h| { - h.start(agent, kind, &cmdline) + // No node id: `nix_logged`'s two callers are meta-flake operations + // reached from outside the queue as well as from inside it, and the + // agent+kind+time listing is how they're surfaced today. Linking them + // to a node would mean threading the id through `meta`'s public API + // for no current reader — worth doing when something wants it. + h.start(agent, kind, &cmdline, None) .map_err(|e| { tracing::warn!(error = ?e, %kind, "build_logs: start failed (meta log dropped)"); }) diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index 24b302bd..67b28c3e 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -13,6 +13,8 @@ use serde::Serialize; use tokio::sync::broadcast; use utoipa::ToSchema; +use crate::db::Migration; + /// Process-singleton handle, set once at coordinator startup. Lets /// the `lifecycle` module's `run` / `prebuild_toplevel` access the /// writer without threading an `Arc` through every @@ -65,6 +67,27 @@ CREATE INDEX IF NOT EXISTS idx_build_logs_status_finished WHERE finished_at IS NOT NULL; "; +/// Ordered schema migrations tracked in `schema_versions` (key `"build_logs"`). +/// +/// v1 makes the log row carry its node, replacing the host-side +/// `NodeId -> build_log_id` map the job queue used to hold. The link has a +/// single home again, and the direction is the one the type system allows: +/// a `hive_jobq` node payload is immutable after insert, but the log row is +/// written when the build starts and can name the node it belongs to. +/// +/// Legacy rows keep `node_id IS NULL` — they predate the column and no node +/// still exists to link them to, so the dashboard's by-node lookup simply +/// misses them (the by-agent listing, which is how they're reached, is +/// unaffected). +const MIGRATIONS: &[Migration] = &[Migration { + sql: "BEGIN; + ALTER TABLE build_logs ADD COLUMN node_id INTEGER; + CREATE INDEX IF NOT EXISTS idx_build_logs_node + ON build_logs (node_id) WHERE node_id IS NOT NULL; + COMMIT;", + adds_column: Some(("build_logs", "node_id")), +}]; + /// Status of a finished build attempt. Stored as the literal string in /// the `status` column; `NULL` while the attempt is still in progress. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] @@ -152,6 +175,7 @@ impl BuildLogs { let conn = crate::db::open(&path, "build_logs")?; conn.execute_batch(SCHEMA) .context("apply build_logs schema")?; + crate::db::apply_versioned_migrations(&conn, "build_logs", MIGRATIONS)?; let (notify_tx, _) = broadcast::channel(NOTIFY_CAP); Ok(Self { conn: Mutex::new(conn), @@ -169,17 +193,49 @@ impl BuildLogs { /// Open a row for a new build attempt. Returns the assigned id /// — the caller threads it through `append_stdout` / `append_stderr` /// while the child runs and into `finish` once it exits. - pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result { + /// + /// `node_id` is the queue node this build belongs to, when there is one. + /// It is `None` for builds that run outside the job queue; those are + /// reachable by agent + time, just not by node. + pub fn start( + &self, + agent: &str, + kind: &str, + cmdline: &str, + node_id: Option, + ) -> Result { let now = Utc::now().timestamp(); + let node_id = node_id.and_then(|n| i64::try_from(n).ok()); let conn = self.conn.lock().unwrap(); conn.execute( - "INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)", - params![agent, kind, cmdline, now], + "INSERT INTO build_logs (agent, kind, cmdline, started_at, node_id) \ + VALUES (?1, ?2, ?3, ?4, ?5)", + params![agent, kind, cmdline, now, node_id], ) .context("insert build_logs row")?; Ok(conn.last_insert_rowid()) } + /// The most recent build-log row for `node_id`, if any. Replaces the job + /// queue's in-memory `NodeId -> build_log_id` side map: the link lives in + /// the row itself now, so it survives a restart and needs no lock held + /// alongside the scheduler's. + /// + /// `MAX(id)` rather than a uniqueness assumption — a node that is retried + /// opens a second row, and the newest is the one the panel should show. + #[must_use] + pub fn id_for_node(&self, node_id: u64) -> Option { + let node_id = i64::try_from(node_id).ok()?; + let conn = self.conn.lock().unwrap(); + conn.query_row( + "SELECT MAX(id) FROM build_logs WHERE node_id = ?1", + params![node_id], + |row| row.get::<_, Option>(0), + ) + .ok() + .flatten() + } + /// Append a single stdout line. Best-effort: errors are logged /// but never returned to the caller, so a transient sqlite blip /// never tears down a rebuild's stdout pump. @@ -453,7 +509,7 @@ mod tests { fn start_appends_finish_flow() { let (_d, db) = tmpdb(); let id = db - .start("alice", "prebuild", "nix build foo") + .start("alice", "prebuild", "nix build foo", None) .expect("start"); db.append_stdout(id, "building '/nix/store/abc.drv'"); db.append_stderr(id, "error: line 12"); @@ -482,9 +538,9 @@ mod tests { // assert id-ordering (autoincrement) is the tiebreaker — list // sorts by started_at DESC but the ORDER BY still produces the // last-inserted row first when timestamps match. - let id_a1 = db.start("alice", "run", "cmd one").expect("start"); - let _id_b = db.start("bob", "run", "cmd two").expect("start"); - let id_a2 = db.start("alice", "run", "cmd three").expect("start"); + let id_a1 = db.start("alice", "run", "cmd one", None).expect("start"); + let _id_b = db.start("bob", "run", "cmd two", None).expect("start"); + let id_a2 = db.start("alice", "run", "cmd three", None).expect("start"); db.finish(id_a1, BuildStatus::Ok); let alice_rows = db.list_recent_for_agent("alice", 10).expect("list"); @@ -515,10 +571,12 @@ mod tests { #[test] fn vacuum_drops_old_finished_only_per_status() { let (_d, db) = tmpdb(); - let id_fresh_fail = db.start("alice", "run", "fresh fail").expect("start"); - let id_old_fail = db.start("alice", "run", "old fail").expect("start"); - let id_old_ok = db.start("alice", "run", "old ok").expect("start"); - let id_running = db.start("alice", "run", "still running").expect("start"); + let id_fresh_fail = db.start("alice", "run", "fresh fail", None).expect("start"); + let id_old_fail = db.start("alice", "run", "old fail", None).expect("start"); + let id_old_ok = db.start("alice", "run", "old ok", None).expect("start"); + let id_running = db + .start("alice", "run", "still running", None) + .expect("start"); db.finish(id_fresh_fail, BuildStatus::Fail); db.finish(id_old_fail, BuildStatus::Fail); db.finish(id_old_ok, BuildStatus::Ok); @@ -550,6 +608,31 @@ mod tests { assert!(db.get_full(id_running).unwrap().is_some()); } + #[test] + fn node_link_survives_completion_and_newest_wins() { + // The queue used to hold this link in an in-memory side map, which + // meant it died with the process and needed the queue lock to read. + // On the row it outlives both the node's completion and a restart. + let (_d, db) = tmpdb(); + let first = db.start("alice", "swap", "cmd", Some(7)).expect("start"); + db.finish(first, BuildStatus::Fail); + assert_eq!( + db.id_for_node(7), + Some(first), + "link survives the build finishing" + ); + + // A retried node opens a second row; the panel wants the current + // attempt, not the first one. + let retry = db.start("alice", "swap", "cmd", Some(7)).expect("start"); + assert_eq!(db.id_for_node(7), Some(retry), "newest attempt wins"); + + // Builds that run outside the queue carry no node and are found by + // agent + time instead — they must not collide with node lookups. + db.start("alice", "run", "no node", None).expect("start"); + assert_eq!(db.id_for_node(999), None, "unknown node → no row"); + } + #[test] fn append_after_finish_still_appends() { // Defensive: if a child's stdout pump fires one last line @@ -557,7 +640,7 @@ mod tests { // append should land on the row (status already set, but the // log stays consistent with what happened). let (_d, db) = tmpdb(); - let id = db.start("alice", "run", "cmd").expect("start"); + let id = db.start("alice", "run", "cmd", None).expect("start"); db.finish(id, BuildStatus::Ok); db.append_stdout(id, "post-finish trailing line"); let full = db.get_full(id).expect("get").expect("Some"); From 9e91bf7813560f95092125bc2bb11b416f6bd2f5 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 18:30:32 +0200 Subject: [PATCH 04/27] refactor(#2949): claim_one is the primitive, settle is it in a loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One-at-a-time claiming is what lets a caller choose between claiming again immediately and backing off — a batch return cannot express that choice, and the choice is the point: the run loop wants to know there was work before it decides whether to wait. `settle()` keeps its exact meaning as `while let Some(id) = claim_one()`. A node started by an earlier iteration is `Running`, not terminal, so it cannot satisfy another node's dependency in the same sweep; it only consumes resources. The crate's ~45 existing `settle()` assertions — which cover resource borrowing, cap-1 serialisation, roll-up and cancellation — are what verify that equivalence, so it is checked rather than argued. `None` means "nothing runnable right now", which is deliberately a different statement from "nothing pending": a node can be pending and unrunnable because its resources are held elsewhere. Cost stated rather than left to be found: each `claim_one` rescans the pending set, so `settle` is O(n^2) in nodes claimed where the single-pass version was O(n). The graph is bounded by history retention. --- hive-jobq/src/scheduler.rs | 37 +++++++++++++++++++++++++++---------- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index ccdd017e..3fd46c3d 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -138,25 +138,42 @@ impl Scheduler { }) } - /// Claim every currently-runnable pending node and start it: node-deps + /// Claim **one** currently-runnable pending node and start it: node-deps /// satisfied and all resource-deps acquired atomically (all-or-nothing). - /// Each claimed node is marked `Running`, its acquired units recorded, and - /// its id returned for the runner to execute. A single pass suffices — a - /// node started here is `Running`, not terminal, so it cannot satisfy another - /// node's dependency in the same pass; it only consumes resources. + /// The node is marked `Running`, its acquired units recorded, and its id + /// returned for the caller to execute. `None` means nothing is runnable + /// right now — which is a different statement from "nothing is pending". + /// + /// One-at-a-time is the primitive on purpose: it lets the caller decide + /// between claiming again immediately and backing off, a choice a batch + /// return can't express. [`Self::settle`] is this in a loop. #[must_use] - pub fn settle(&mut self) -> Vec { + pub fn claim_one(&mut self) -> Option { let pending: Vec = self .graph .nodes() .filter(|n| n.state == State::Pending) .map(|n| n.id) .collect(); + pending + .into_iter() + .find(|&id| self.node_deps_satisfied(id) && self.try_start(id)) + } + + /// Claim every currently-runnable pending node. Equivalent to calling + /// [`Self::claim_one`] until it yields `None`: a node started by an earlier + /// iteration is `Running`, not terminal, so it cannot satisfy another + /// node's dependency here — it only consumes resources. + /// + /// ⚠️ Each iteration rescans the pending set, so this is O(n²) in the + /// number of nodes claimed where the old single-pass version was O(n). The + /// graph is bounded by history retention, so that is affordable; it is + /// stated rather than left to be discovered. + #[must_use] + pub fn settle(&mut self) -> Vec { let mut started = Vec::new(); - for id in pending { - if self.node_deps_satisfied(id) && self.try_start(id) { - started.push(id); - } + while let Some(id) = self.claim_one() { + started.push(id); } started } From d1f1a361f0a896178bb19c731e70e55f6cbfa0f0 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 18:33:24 +0200 Subject: [PATCH 05/27] =?UTF-8?q?feat(#2949):=20claim=5Fnext=20=E2=80=94?= =?UTF-8?q?=20the=20seam=20that=20cannot=20be=20half-used?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The caller supplies how to run a node and spawns what it gets back; it never touches claiming or completion. The returned future runs the node *and completes it*, so "forgot to finish the node" stops being something a caller can do — completion is inside the thing they spawn. The `Option` is answered synchronously, before anything is awaited, so the run loop learns whether there was work without waiting on the node it just started. That is what lets it choose between claiming again immediately and backing off; an id alone cannot express that choice. Locking: taken twice, briefly, and never held across the await — once to claim, once inside the future to complete. A guard alive across an await point would make the future non-`Send` and unspawnable, which is also why the node itself runs unlocked for however long it takes. `Arc` + `std::sync::Mutex` keep this runtime-agnostic: no tokio in this crate. `run` receives an owned payload rather than a borrow for the same reason a `&Job` could not be threaded through the executors: a reference parameter is live for the whole future, borrowing the graph across the await and poisoning `Send`. The output carries the insert result instead of swallowing it. This crate has no logger by design, so a malformed grown job is reported to the caller, who can log it. The node completes either way — its own work already happened. --- hive-jobq/src/scheduler.rs | 56 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 3fd46c3d..8b1e0413 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -29,7 +29,9 @@ //! is a deferred optimization — unsafe under dynamically-appended subnodes.) use std::collections::HashMap; +use std::future::Future; use std::hash::Hash; +use std::sync::{Arc, Mutex}; use crate::builder::{BuildError, JobBuilder, NodeGuid}; use crate::resources::ResourceTable; @@ -160,6 +162,60 @@ impl Scheduler { .find(|&id| self.node_deps_satisfied(id) && self.try_start(id)) } + /// Claim one runnable node and return **the work that runs it**, or `None` + /// when nothing is runnable right now. + /// + /// This is the seam: the caller supplies how to execute a node and spawns + /// the returned future, but never touches claiming or completion. The + /// future runs the node **and completes it**, so "forgot to finish the + /// node" is not expressible — completion is inside the thing you spawn. + /// + /// The `Option` is answered *synchronously*, before anything is awaited, so + /// the caller can decide "claim again immediately" vs "back off" without + /// waiting on the node it just started. + /// + /// ## Locking + /// The lock is taken twice, briefly, and **never held across the await**: + /// once here to claim, once inside the future to complete. That is what + /// keeps the returned future `Send` — a guard alive across an await point + /// would poison it — and it is why the node itself runs unlocked, for + /// however many minutes it needs. + /// + /// ## Why the payload is cloned + /// `run` gets an owned `N` rather than a borrow: a `&N` parameter is live + /// for the whole future, which both borrows the graph across the await and + /// makes the future non-`Send`. + /// + /// The output carries the insert result rather than swallowing it — this + /// crate has no logger, so a malformed grown job is reported to the caller, + /// who is the one that can log it. The node is completed either way: its + /// own work already happened. + pub fn claim_next( + sched: &Arc>, + run: F, + ) -> Option)> + use> + where + N: Clone, + F: FnOnce(NodeId, N, JobBuilder) -> Fut, + Fut: Future, Outcome)>, + { + let (id, payload) = { + let mut guard = sched.lock().expect("jobq scheduler mutex poisoned"); + let id = guard.claim_one()?; + let payload = guard.graph.node(id)?.payload.clone(); + (id, payload) + }; + let sched = Arc::clone(sched); + Some(async move { + let (grown, outcome) = run(id, payload, JobBuilder::new()).await; + let grew = sched + .lock() + .expect("jobq scheduler mutex poisoned") + .complete_growing(id, outcome, grown); + (id, grew) + }) + } + /// Claim every currently-runnable pending node. Equivalent to calling /// [`Self::claim_one`] until it yields `None`: a node started by an earlier /// iteration is `Running`, not terminal, so it cannot satisfy another From be1060e52f136a868a9eea835aca5f143ac34b0c Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 18:44:12 +0200 Subject: [PATCH 06/27] refactor(#2949): the mutex holds the scheduler, not a wrapper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `QueueInner` existed to hold the scheduler *and* a per-node side map. The map is gone, so it was a struct around one field — and worse, a struct of a type `hive_jobq` cannot drive: the crate's run-loop seam takes `&Arc>>` specifically. So `JobQueue` now holds `Arc>` directly, where `Sched` is just `Scheduler`. Its six methods become free functions over `&Sched`; all six are `DagView` projections, i.e. the code the endpoint rework is going to delete anyway, so this does not entrench them. This is the precondition for c0re calling `claim_next`, not that switch itself — `run_worker` still claims through `claim_ready`. Landing it separately keeps the type change reviewable on its own. --- hive-c0re/src/job_queue/mod.rs | 419 +++++++++++++++---------------- hive-c0re/src/job_queue/tests.rs | 1 - 2 files changed, 203 insertions(+), 217 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index dfe916ec..2780ee6a 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -36,7 +36,7 @@ pub mod templates; #[cfg(test)] mod tests; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use chrono::{DateTime, Utc}; use hive_host_sock::jobs::NodeView; @@ -109,27 +109,29 @@ struct DagMeta { created_at: DateTime, } -/// The mutable queue state behind the mutex: **just the crate scheduler**. +/// The crate scheduler, specialised to this host's node + resource types. +/// /// A **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) /// whose subtree is the DAG's work — so the container's `NodeId` is the DAG id, /// its rolled-up state is the DAG state, and there are no grouping side-tables: -/// membership + meta are graph queries ([`QueueInner::container`] / -/// [`QueueInner::dag_meta`] + the `hive_jobq::Graph` accessors). One shared -/// crate [`Graph`] holds every DAG. +/// membership + meta are graph queries ([`container`] / [`dag_meta`] + the +/// `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG. /// -/// There is deliberately **no per-node side map** any more. The last one held -/// the `build_logs` row id; that link now lives on the log row itself -/// (`build_logs.node_id`), so it survives a restart and needs no lock held -/// alongside the scheduler's — which is what lets the scheduler's own lock be -/// the only one the run loop takes. -struct QueueInner { - sched: Scheduler, -} +/// There is deliberately **no wrapper struct and no per-node side map**. The +/// last map held the `build_logs` row id; that link now lives on the log row +/// itself (`build_logs.node_id`). With nothing else to guard, the mutex holds +/// the scheduler *directly* — which is what lets `hive_jobq` drive the run loop +/// (it takes `&Arc>>`, a type a host-side wrapper could not +/// satisfy). +type Sched = Scheduler; /// The queue. Lives on `Coordinator` (one per hive-c0re process); a single /// scheduler task ([`scheduler::run_worker`]) drives it. pub struct JobQueue { - inner: Mutex, + /// The scheduler, held directly rather than behind a host-side wrapper — + /// `hive_jobq`'s run-loop seam takes `&Arc>>`, so this + /// *is* the type the crate drives. + sched: Arc>, /// Wakes the scheduler when something new arrives or state changed. pub(crate) notify: Notify, } @@ -173,12 +175,11 @@ fn outcome_of(result: Result<(), String>) -> Outcome { /// # Errors /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). fn insert_group( - inner: &mut QueueInner, + inner: &mut Sched, declare: impl FnOnce(&Job), group_parent: Option, ) -> anyhow::Result<()> { inner - .sched .insert_job(group_parent, |b| { declare(b); // c0re names no handles: a DAG is addressed by its container node, @@ -199,15 +200,13 @@ impl JobQueue { u32::try_from(build_slots.max(1)).unwrap_or(u32::MAX), ); Self { - inner: Mutex::new(QueueInner { - sched: Scheduler::new(Graph::new(), table), - }), + sched: Arc::new(Mutex::new(Scheduler::new(Graph::new(), table))), notify: Notify::new(), } } - fn lock(&self) -> std::sync::MutexGuard<'_, QueueInner> { - self.inner.lock().expect("job_queue mutex poisoned") + fn lock(&self) -> std::sync::MutexGuard<'_, Sched> { + self.sched.lock().expect("job_queue mutex poisoned") } /// Submit a DAG: insert a [`NodeKind::Dag`] **container node** carrying the @@ -225,7 +224,6 @@ impl JobQueue { pub fn submit(&self, spec: DagSpec) -> anyhow::Result { let mut inner = self.lock(); let container = inner - .sched .append( NodeKind::Dag { source: spec.source, @@ -241,7 +239,7 @@ impl JobQueue { // `Finishing` and its children become runnable — it never needs claiming // or executing, and stays out of `claim_ready`. It rolls up terminal when // its whole subtree settles (that's the DAG-done signal). - inner.sched.complete(container, Outcome::Done); + inner.complete(container, Outcome::Done); drop(inner); self.notify.notify_one(); Ok(container.get()) @@ -255,15 +253,15 @@ impl JobQueue { pub fn claim_ready(&self) -> Vec { let mut inner = self.lock(); let inner = &mut *inner; - let started = inner.sched.settle(); + let started = inner.settle(); let mut claims = Vec::with_capacity(started.len()); for id in started { - let Some(node) = inner.sched.graph().node(id) else { + 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.sched.graph().root_of(id) else { + let Some(container) = inner.graph().root_of(id) else { continue; }; claims.push(Claim { @@ -291,7 +289,7 @@ impl JobQueue { // complete) to express "grew nothing". The shared part is the outcome // mapping, and that's a free fn. let mut inner = self.lock(); - inner.sched.complete(node_id, outcome_of(result)); + inner.complete(node_id, outcome_of(result)); drop(inner); self.notify.notify_one(); } @@ -304,7 +302,7 @@ impl JobQueue { /// `Job::default()`. #[must_use] pub fn new_job(&self) -> Job { - self.lock().sched.new_job() + self.lock().new_job() } /// [`JobQueue::complete_node`] plus the work the node declared while it ran. @@ -318,10 +316,7 @@ impl JobQueue { // 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 - .sched - .complete_growing(node_id, outcome_of(result), grown) - { + if let Err(e) = inner.complete_growing(node_id, outcome_of(result), grown) { tracing::error!( node = node_id.get(), error = %e, @@ -354,10 +349,10 @@ impl JobQueue { /// just that branch. Nothing here knows about DAGs. pub fn cancel(&self, id: u64) -> bool { let mut inner = self.lock(); - let Some(node) = inner.sched.graph().resolve_id(id) else { + let Some(node) = inner.graph().resolve_id(id) else { return false; }; - if !inner.sched.cancel_node(node) { + if !inner.cancel_node(node) { return false; } drop(inner); @@ -377,12 +372,8 @@ impl JobQueue { #[must_use] pub fn first_error(&self, dag_id: u64) -> Option { let inner = self.lock(); - let container = inner.container(dag_id)?; - inner - .sched - .graph() - .first_error(container) - .map(ToOwned::to_owned) + let container = container(&inner, dag_id)?; + inner.graph().first_error(container).map(ToOwned::to_owned) } /// `(agent, label, takes_container_down)` for the live transient-pill set, @@ -414,7 +405,6 @@ impl JobQueue { pub fn running_transients(&self) -> Vec { let inner = self.lock(); inner - .sched .graph() .nodes() .filter(|n| matches!(n.state, State::Running)) @@ -443,9 +433,11 @@ impl JobQueue { #[must_use] pub fn snapshot(&self) -> Vec { let inner = self.lock(); - let mut ids = inner.visible_dags(); + let mut ids = visible_dags(&inner); ids.sort_unstable_by_key(|c| c.get()); - ids.into_iter().filter_map(|c| inner.dag_view(c)).collect() + ids.into_iter() + .filter_map(|c| dag_view(&inner, c)) + .collect() } /// Number of live (non-terminal) DAGs — tests + diagnostics. @@ -453,191 +445,186 @@ impl JobQueue { #[must_use] pub fn live_count(&self) -> usize { let inner = self.lock(); - inner - .containers() + containers(&inner) .into_iter() - .filter(|&c| inner.sched.graph().is_settled(c) == Some(false)) + .filter(|&c| inner.graph().is_settled(c) == Some(false)) .count() } } -impl QueueInner { - /// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals - /// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. - fn container(&self, dag_id: u64) -> Option { - self.sched.graph().nodes().find_map(|n| { - (n.parent.is_none() - && n.id.get() == dag_id - && matches!(n.payload, NodeKind::Dag { .. })) +/// The container node of `dag_id` — the `NodeKind::Dag` root whose id equals +/// `dag_id`. `NodeId` is un-fabricable from a raw `u64`, so this is a search. +fn container(sched: &Sched, dag_id: u64) -> Option { + sched.graph().nodes().find_map(|n| { + (n.parent.is_none() && n.id.get() == dag_id && matches!(n.payload, NodeKind::Dag { .. })) .then_some(n.id) - }) - } + }) +} - /// The container's carried domain metadata as an owned read-view. The data - /// lives solely in the [`NodeKind::Dag`] payload — this is a derived read, - /// not a stored side-table. - fn dag_meta(&self, container: NodeId) -> Option { - let NodeKind::Dag { - source, - reason, - created_at, - } = &self.sched.graph().node(container)?.payload - else { - return None; +/// The container's carried domain metadata as an owned read-view. The data +/// lives solely in the [`NodeKind::Dag`] payload — this is a derived read, +/// not a stored side-table. +fn dag_meta(sched: &Sched, container: NodeId) -> Option { + let NodeKind::Dag { + source, + reason, + created_at, + } = &sched.graph().node(container)?.payload + else { + return None; + }; + Some(DagMeta { + source: *source, + reason: reason.clone(), + created_at: *created_at, + }) +} + +/// Project a DAG into its wire [`DagView`]: a near-raw view of the +/// container's work nodes, with `Done` nodes excluded. Lifecycle +/// (`state` / `started_at` / `finished_at` / `error`) is read straight +/// off each `hive_jobq::Node`; the client derives the DAG label, roll-up +/// state, and DAG timestamps from the node set. Non-derivable per-node +/// payload (`approval_id`, meta `inputs`) rides the owning node. Returns +/// `None` when every work node is `Done` or `Skipped` — a fully-settled +/// DAG drops out of the snapshot entirely (a `Failed` one lingers until +/// aged out). +fn dag_view(sched: &Sched, container: NodeId) -> Option { + let meta = dag_meta(sched, container)?; + let mut nodes = Vec::new(); + // Whether anything in this DAG still has an outcome worth showing. + // Kept separate from `nodes` being non-empty: skipped nodes ride the + // wire so the dashboard can mark the branches that weren't taken, but + // they must not by themselves hold a finished DAG in the snapshot. + let mut any_unsettled = false; + // DAG-level timestamps are taken over *all* subtree nodes (including the + // `Done` ones excluded from the wire) — the client can't derive them + // from a `Done`-filtered node set, so the host computes them here. + let mut started: Vec> = Vec::new(); + let mut finished: Vec> = Vec::new(); + for node in sched.graph().descendants(container) { + let id = node.id; + if let Some(s) = node.started_at { + started.push(s); + } + if let Some(f) = node.finished_at { + finished.push(f); + } + // `Done` nodes drop off the wire — a finished step isn't + // interesting. `Skipped` ones stay: which branch a run *didn't* + // take is the readable half of an outcome-branched DAG. + if matches!(node.state, State::Done) { + continue; + } + any_unsettled |= !matches!(node.state, State::Skipped); + let deps: Vec = node + .deps + .iter() + .filter_map(|d| match d { + Dep::Node { id, .. } => Some(id.get()), + Dep::Resource { .. } => None, + }) + .collect(); + // Non-derivable per-node payload rides the node that owns it. Every + // deploy phase carries the approval id, but only the subtree root + // projects it onto the wire — hanging the approval link off all of + // them would render the same card once per phase. + let approval_id = match &node.payload { + NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id), + _ => None, }; - Some(DagMeta { - source: *source, - reason: reason.clone(), - created_at: *created_at, - }) + let inputs = match &node.payload { + NodeKind::MetaLock { inputs, .. } => inputs.clone(), + _ => Vec::new(), + }; + // Looked up from the log row itself (`build_logs.node_id`), not a + // host-side map. One indexed query per node in the snapshot; the + // node set is bounded by `MAX_HISTORY_DAGS` and the store is a + // local sqlite file, so this is cheaper than the lock contention + // a second shared map would reintroduce. + let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())); + // `node.parent` is the structural jobq parent. Top-level nodes + // have `parent == Some(container)` (direct children of the Dag + // container); those become `parent: None` on the wire since the + // container itself is not part of the work-node payload. Sub-nodes + // carry the id of their containing parent work-node. + let parent = node + .parent + .filter(|&p| p != container) + .map(hive_jobq::NodeId::get); + nodes.push(NodeView { + id: id.get(), + agent: node.payload.agent().to_owned(), + kind: node.payload.as_str().to_owned(), + deps, + state: node.state, + started_at: node.started_at, + finished_at: node.finished_at, + error: node.error.clone(), + approval_id, + inputs, + build_log_id, + parent, + }); } + if !any_unsettled { + return None; + } + let is_terminal = sched.graph().is_settled(container) == Some(true); + Some(DagView { + id: container.get(), + source: meta.source, + reason: meta.reason.clone(), + created_at: meta.created_at, + started_at: started.into_iter().min(), + finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), + nodes, + }) +} - /// Project a DAG into its wire [`DagView`]: a near-raw view of the - /// container's work nodes, with `Done` nodes excluded. Lifecycle - /// (`state` / `started_at` / `finished_at` / `error`) is read straight - /// off each `hive_jobq::Node`; the client derives the DAG label, roll-up - /// state, and DAG timestamps from the node set. Non-derivable per-node - /// payload (`approval_id`, meta `inputs`) rides the owning node. Returns - /// `None` when every work node is `Done` or `Skipped` — a fully-settled - /// DAG drops out of the snapshot entirely (a `Failed` one lingers until - /// aged out). - fn dag_view(&self, container: NodeId) -> Option { - let meta = self.dag_meta(container)?; - let mut nodes = Vec::new(); - // Whether anything in this DAG still has an outcome worth showing. - // Kept separate from `nodes` being non-empty: skipped nodes ride the - // wire so the dashboard can mark the branches that weren't taken, but - // they must not by themselves hold a finished DAG in the snapshot. - let mut any_unsettled = false; - // DAG-level timestamps are taken over *all* subtree nodes (including the - // `Done` ones excluded from the wire) — the client can't derive them - // from a `Done`-filtered node set, so the host computes them here. - let mut started: Vec> = Vec::new(); - let mut finished: Vec> = Vec::new(); - for node in self.sched.graph().descendants(container) { - let id = node.id; - if let Some(s) = node.started_at { - started.push(s); - } - if let Some(f) = node.finished_at { - finished.push(f); - } - // `Done` nodes drop off the wire — a finished step isn't - // interesting. `Skipped` ones stay: which branch a run *didn't* - // take is the readable half of an outcome-branched DAG. - if matches!(node.state, State::Done) { - continue; - } - any_unsettled |= !matches!(node.state, State::Skipped); - let deps: Vec = node - .deps - .iter() - .filter_map(|d| match d { - Dep::Node { id, .. } => Some(id.get()), - Dep::Resource { .. } => None, - }) - .collect(); - // Non-derivable per-node payload rides the node that owns it. Every - // deploy phase carries the approval id, but only the subtree root - // projects it onto the wire — hanging the approval link off all of - // them would render the same card once per phase. - let approval_id = match &node.payload { - NodeKind::DeployWindow { approval_id, .. } => Some(*approval_id), - _ => None, - }; - let inputs = match &node.payload { - NodeKind::MetaLock { inputs, .. } => inputs.clone(), - _ => Vec::new(), - }; - // Looked up from the log row itself (`build_logs.node_id`), not a - // host-side map. One indexed query per node in the snapshot; the - // node set is bounded by `MAX_HISTORY_DAGS` and the store is a - // local sqlite file, so this is cheaper than the lock contention - // a second shared map would reintroduce. - let build_log_id = crate::build_logs::global().and_then(|h| h.id_for_node(id.get())); - // `node.parent` is the structural jobq parent. Top-level nodes - // have `parent == Some(container)` (direct children of the Dag - // container); those become `parent: None` on the wire since the - // container itself is not part of the work-node payload. Sub-nodes - // carry the id of their containing parent work-node. - let parent = node - .parent - .filter(|&p| p != container) - .map(hive_jobq::NodeId::get); - nodes.push(NodeView { - id: id.get(), - agent: node.payload.agent().to_owned(), - kind: node.payload.as_str().to_owned(), - deps, - state: node.state, - started_at: node.started_at, - finished_at: node.finished_at, - error: node.error.clone(), - approval_id, - inputs, - build_log_id, - parent, - }); +/// When a DAG's work node finishes on `finished_at` — the max over its +/// subtree (read off the graph `Node`, as unix seconds), for the history +/// cap ordering. +fn dag_finished_at(sched: &Sched, container: NodeId) -> i64 { + sched + .graph() + .descendants(container) + .filter_map(|n| n.finished_at) + .map(|t| t.timestamp()) + .max() + .unwrap_or(0) +} + +/// Every DAG container node id in the graph. +fn containers(sched: &Sched) -> Vec { + sched + .graph() + .nodes() + .filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. })) + .map(|n| n.id) + .collect() +} + +/// The **visible** DAG set for the snapshot: every live (non-terminal) DAG, +/// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for +/// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up); +/// this filter is what bounds what the dashboard sees. +fn visible_dags(sched: &Sched) -> Vec { + let mut live: Vec = Vec::new(); + let mut terminal: Vec<(NodeId, i64)> = Vec::new(); + for c in containers(sched) { + if sched.graph().is_settled(c) == Some(true) { + terminal.push((c, dag_finished_at(sched, c))); + } else { + live.push(c); } - if !any_unsettled { - return None; - } - let is_terminal = self.sched.graph().is_settled(container) == Some(true); - Some(DagView { - id: container.get(), - source: meta.source, - reason: meta.reason.clone(), - created_at: meta.created_at, - started_at: started.into_iter().min(), - finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), - nodes, - }) - } - - /// When a DAG's work node finishes on `finished_at` — the max over its - /// subtree (read off the graph `Node`, as unix seconds), for the history - /// cap ordering. - fn dag_finished_at(&self, container: NodeId) -> i64 { - self.sched - .graph() - .descendants(container) - .filter_map(|n| n.finished_at) - .map(|t| t.timestamp()) - .max() - .unwrap_or(0) - } - - /// Every DAG container node id in the graph. - fn containers(&self) -> Vec { - self.sched - .graph() - .nodes() - .filter(|n| n.parent.is_none() && matches!(n.payload, NodeKind::Dag { .. })) - .map(|n| n.id) - .collect() - } - - /// The **visible** DAG set for the snapshot: every live (non-terminal) DAG, - /// plus the newest [`MAX_HISTORY_DAGS`] terminal ones. Crate nodes for - /// evicted DAGs linger in the graph (bounded-prune is a Stage-C follow-up); - /// this filter is what bounds what the dashboard sees. - fn visible_dags(&self) -> Vec { - let mut live: Vec = Vec::new(); - let mut terminal: Vec<(NodeId, i64)> = Vec::new(); - for c in self.containers() { - if self.sched.graph().is_settled(c) == Some(true) { - terminal.push((c, self.dag_finished_at(c))); - } else { - live.push(c); - } - } - // Newest first, so truncating to the cap keeps the most recent. - terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.0.get().cmp(&a.0.get()))); - terminal.truncate(MAX_HISTORY_DAGS); - let mut kept = live; - kept.extend(terminal.into_iter().map(|(c, _)| c)); - kept } + // Newest first, so truncating to the cap keeps the most recent. + terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.0.get().cmp(&a.0.get()))); + terminal.truncate(MAX_HISTORY_DAGS); + let mut kept = live; + kept.extend(terminal.into_iter().map(|(c, _)| c)); + kept } /// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`. diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 82065941..802372a2 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -117,7 +117,6 @@ fn settle_rebuild_tail(q: &JobQueue, agent: &str, expect_ok: bool) { fn declared_resources(q: &JobQueue, node_id: hive_jobq::NodeId) -> Vec { let inner = q.lock(); inner - .sched .graph() .node(node_id) .expect("node exists") From ab53f6710d9c0a9c942192915b3e8fb554e32ef8 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 19:01:53 +0200 Subject: [PATCH 07/27] job_queue: drop Claim, claim_ready and the completion wrappers 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. --- hive-c0re/src/job_queue/exec.rs | 164 ++++++++++++++------------- hive-c0re/src/job_queue/mod.rs | 107 ++++------------- hive-c0re/src/job_queue/scheduler.rs | 157 +++++++++++-------------- hive-c0re/src/job_queue/tests.rs | 90 ++++++++++++++- 4 files changed, 263 insertions(+), 255 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index de9e1f7d..fa303189 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -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, 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, - claim: &Claim, + dag_id: Option, 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, claim: &Claim, ok: bool) { +fn run_emit_rebuilt(coord: &Arc, agent: &str, dag_id: Option, 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, 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, claim: &Claim, up: bool) -> Result<()> { +fn run_set_wanted(coord: &Arc, agent: &str, up: bool) -> Result<()> { let wanted = if up { crate::power::Wanted::Up } else { @@ -175,8 +186,8 @@ fn run_set_wanted(coord: &Arc, 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, 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, claim: &Claim, relock: bool) -> Result<()> { - let name = &claim.agent; +async fn run_meta_sync(coord: &Arc, 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, 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, claim: &Claim) -> Result<()> { - let name = &claim.agent; +async fn run_swap(coord: &Arc, 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, 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, claim: &Claim) -> Result<()> { - let name = &claim.agent; +async fn run_post_swap(coord: &Arc, 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, 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, claim: &Claim) -> Result<()> { - let name = &claim.agent; +async fn run_provision(coord: &Arc, 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, 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, claim: &Claim) -> Result> { - 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, name: &str) -> Result> { 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, claim: &Claim) -> Result