From b87eac0a61fe3932cf85cfdd99fa693688f96969 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 15 Jul 2026 20:21:39 +0200 Subject: [PATCH] feat(#2484): unify in-DAG growth on append_subgraph (drop append_node) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit append_subgraph is the multi-node/multi-agent generalisation of the single-node append_node, so the two in-DAG-growth channels collapse to one: the Reconcile planner now emits its mechanical Start/Stop as a single-node append_subgraph rooted on the reconcile node (stamping claim.agent on the NodeSpec, which append_node inherited implicitly). Removes NodeOutput.append_nodes + its scheduler drain loop and JobQueue::append_node. No behaviour change — a channel unification. --- hive-c0re/src/job_queue/exec.rs | 61 ++++++++++++--------------- hive-c0re/src/job_queue/mod.rs | 63 ++++++---------------------- hive-c0re/src/job_queue/scheduler.rs | 28 +++++-------- 3 files changed, 48 insertions(+), 104 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 7c3eb6bd..99b7bfc2 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -27,24 +27,19 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from /// success. #[derive(Debug, Default)] pub struct NodeOutput { - /// Mechanical sub-step nodes to append into *this same* DAG at - /// runtime, each depending `AfterOk` on the emitting node — e.g. a - /// `Reconcile` planner emitting a `Start` / `Stop`. Keeps the sub-step a - /// first-class node in the same DAG so the lease-window transient is held - /// across it. The scheduler applies these *before* the emitting node's - /// completion so the DAG never rolls terminal with the appended work - /// still pending. - pub append_nodes: Vec, /// Whole per-agent *subgraphs* to append into *this same* DAG at - /// runtime — the multi-node generalisation of `append_nodes`. Each - /// inner `Vec` is one independent subgraph whose `deps` are - /// local (0-based within that subgraph); the scheduler appends each via + /// runtime — the single in-DAG-growth channel. Each inner + /// `Vec` is one independent subgraph whose `deps` are local + /// (0-based within that subgraph); the scheduler appends each via /// [`JobQueue::append_subgraph`], which rebases the deps onto the DAG's - /// node-id space and roots the subgraph on the emitting node. Both - /// `MetaLock` flavours use this to grow one rebuild subgraph per agent - /// into their own DAG (the startup sweep's stale agents; the meta-update - /// cascade's affected agents) instead of fanning out child DAGs. Same - /// before-completion ordering as `append_nodes`. + /// node-id space and roots the subgraph 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>, } @@ -293,10 +288,7 @@ async fn run_meta_lock( .iter() .map(|agent| super::templates::rebuild_nodes(agent, true, 0)) .collect(); - return Ok(NodeOutput { - append_subgraph, - ..Default::default() - }); + return Ok(NodeOutput { append_subgraph }); } let _progress = coord.meta_update_guard(); ctx.step("nix flake update"); @@ -320,35 +312,34 @@ async fn run_meta_lock( .iter() .map(|agent| super::templates::rebuild_nodes(agent, false, 0)) .collect(); - Ok(NodeOutput { - append_subgraph, - ..Default::default() - }) + Ok(NodeOutput { append_subgraph }) } /// 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 (`NodeOutput::append_nodes`). 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. +/// *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 { let name = &claim.agent; let running = crate::lifecycle::is_running(name).await; let wanted = coord.power.get_or_seed(name, running)?; - let append_nodes = match reconcile_action(wanted, running) { - ReconcileAction::Start => vec![NodeKind::Start], - ReconcileAction::Stop => vec![NodeKind::Stop], + // One node targeting this agent, rooted on this reconcile node. The old + // `append_node` inherited the emitter's agent implicitly; `append_subgraph` + // carries it on the `NodeSpec`, so stamp `claim.agent` explicitly (same + // effect, one in-DAG-growth channel instead of two). + let sub = |kind| vec![vec![super::templates::node(name, kind, Vec::new())]]; + let append_subgraph = match reconcile_action(wanted, running) { + ReconcileAction::Start => sub(NodeKind::Start), + ReconcileAction::Stop => sub(NodeKind::Stop), ReconcileAction::Noop => { tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop"); Vec::new() } }; - Ok(NodeOutput { - append_nodes, - ..Default::default() - }) + Ok(NodeOutput { append_subgraph }) } /// Mechanical container start — the sub-step a `Reconcile` planner fans diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 1d2840bd..25d6df7d 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -154,57 +154,18 @@ impl JobQueue { Ok(id) } - /// Append a node into a *live* (non-terminal) DAG at runtime, - /// depending `AfterOk` on `dep_on` (the node that emitted it). Lets a - /// planner node — e.g. [`NodeKind::Reconcile`] — fan a mechanical - /// sub-step ([`NodeKind::Start`] / [`NodeKind::Stop`]) out as a - /// first-class node in the *same* DAG. - /// - /// Must be called *before* the emitting node's [`Self::complete_node`] - /// so the DAG doesn't roll terminal with the new node still pending — - /// that keeps the lease-window transient held across the sub-step and - /// lets the appended node's `AfterOk` dep resolve as soon as the - /// emitter settles `Done`. No-op (returns `None`) if the DAG is gone. - pub fn append_node(&self, dag_id: u64, kind: NodeKind, dep_on: NodeId) -> Option { - let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); - let dag = inner.dags.iter_mut().find(|d| d.id == dag_id)?; - // The appended sub-step targets the same agent as the node that - // emitted it (a `Reconcile` fanning out its `Start`/`Stop` acts on - // the same container), so inherit `dep_on`'s agent. - let agent = dag.node(dep_on)?.agent.clone(); - let new_id: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX); - dag.nodes.push(Node { - id: new_id, - agent, - kind, - deps: vec![model::Dep { - on: dep_on, - when: DepWhen::AfterOk, - }], - state: State::Queued, - step: None, - build_log_id: None, - started_at: None, - finished_at: None, - error: None, - }); - drop(inner); - self.notify.notify_one(); - Some(new_id) - } - - /// Append a whole *subgraph* into a live (non-terminal) DAG at runtime - /// — the multi-node, multi-agent generalisation of [`Self::append_node`]. - /// Each [`NodeSpec`] carries its own `agent` and subgraph-relative `deps` - /// (indices into `nodes`); this rebases those onto the DAG's node-id - /// space (`id == index`, an invariant `append_node` also maintains) and - /// attaches every subgraph *root* — a node with no internal deps — to - /// `dep_on` with an `AfterOk` edge. Used by the startup sweep's - /// `MetaLock` to grow per-agent rebuild subgraphs into the same boot DAG - /// instead of fanning out child DAGs. Same call-*before*-`complete_node` - /// contract as `append_node` (so the DAG can't roll terminal with the - /// appended work still pending). Returns the new node ids; empty if the - /// DAG is gone or `nodes` is empty. + /// Append a whole *subgraph* into a live (non-terminal) DAG at runtime — + /// the single in-DAG-growth primitive. Each [`NodeSpec`] carries its own + /// `agent` and subgraph-relative `deps` (indices into `nodes`); this + /// rebases those onto the DAG's node-id space (`id == index`) and attaches + /// every subgraph *root* — a node with no internal deps — to `dep_on` with + /// an `AfterOk` edge. Used both for multi-node growth (the `MetaLock` + /// growing per-agent rebuild subgraphs into the same boot / meta-update + /// DAG instead of fanning out child DAGs) and the single-node case (a + /// `Reconcile` planner's `Start` / `Stop` as a one-node subgraph). Must be + /// called *before* the emitting node's [`Self::complete_node`] so the DAG + /// can't roll terminal with the appended work still pending. Returns the + /// new node ids; empty if the DAG is gone or `nodes` is empty. pub fn append_subgraph( &self, dag_id: u64, diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index e651a505..c0f5e59c 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -9,8 +9,8 @@ //! //! In-DAG growth (a `MetaLock` growing rebuild subgraphs after the lock //! bump, a `Reconcile` fanning its `Start`/`Stop`) flows through -//! `NodeOutput.append_subgraph` / `append_nodes`, applied before the -//! emitting node completes — see `handle_completion`. +//! `NodeOutput.append_subgraph`, applied before the emitting node +//! completes — see `handle_completion`. use std::collections::HashMap; use std::sync::Arc; @@ -103,22 +103,14 @@ async fn handle_completion( node = claim.node_id, "job_queue: node done" ); - // Append any in-DAG sub-step nodes (e.g. a `Reconcile` - // planner's `Start` / `Stop`) BEFORE completing this node, so - // completing it doesn't roll the DAG terminal while the - // appended work is still pending — that keeps the lease-window - // transient held across the sub-step. Each depends `AfterOk` - // on this node, so it becomes ready the instant this one - // settles `Done` just below. - for kind in output.append_nodes { - coord - .job_queue - .append_node(claim.dag_id, kind, claim.node_id); - } - // Same before-completion ordering as `append_nodes`, but for - // whole per-agent subgraphs (the startup sweep's rebuild - // subgraphs growing into the boot DAG) — each an independent - // subgraph rooted on this node. + // 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 — that keeps the lease-window transient + // held across it. Each subgraph is independent, rooted on this + // node (`AfterOk`), so it becomes ready the instant this one + // settles `Done` just below. Covers both 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