From 82ef06f445f64221a11346a7b582dbd11e01db25 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 17:20:43 +0200 Subject: [PATCH] =?UTF-8?q?refactor(#2949):=20kill=20Declare=20=E2=80=94?= =?UTF-8?q?=20a=20running=20node=20declares=20onto=20its=20own=20builder?= 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