From 2b3130f63c41a38e04e23f4d4bae65a628a4713f Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 15 Jul 2026 18:56:40 +0200 Subject: [PATCH] feat(#2476): grow the meta-update cascade in-DAG instead of child DAGs MetaLock's non-sweep completion now grows one rebuild subgraph per affected agent into the same DAG (append_subgraph), replacing the fan-out-child-DAGs + cancel_children dance. Drops NodeOutput.fanout and scheduler's fanout_specs. meta_update DAG carries Rebuilding transient so each cascade agent gets crash-watch suppression at Swap (the property the old child Rebuild DAGs held via their own transient); MetaLock head needs no lease so the pseudo-agent gets no pill. append_children/parent_id and child-DAG tests are intentionally left for the #2453 capstone. --- docs/coordinator.md | 18 ++++++---- hive-c0re/src/job_queue/exec.rs | 34 +++++++++++-------- hive-c0re/src/job_queue/scheduler.rs | 50 +++++----------------------- hive-c0re/src/job_queue/templates.rs | 16 ++++++--- hive-c0re/src/job_queue/tests.rs | 50 ++++++++++++++++++++++++++++ 5 files changed, 102 insertions(+), 66 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index 81b4e440..1d87059e 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -104,9 +104,9 @@ restart(a..): online a: SetWanted(a,Up) → [Signal→Drain→ if graceful] Sto start(a..): a: SetWanted(a,Up) → Reconcile(a) (down+stale ⇒ SetWanted(a,Up) → «rebuild subgraph») spawn(a): [wanted=Up at approve] Create(a) → WriteDropin(a) → Reconcile(a) perm-change(a): WritePermFile(a) → «rebuild subgraph» -meta-update(inp): MetaLock(inp) → «fan-out rebuild(a) per affected agent» -boot: (if any rev marker stale) MetaLock(hyperhive) → «fan-out rebuild»; - plus Reconcile(a) for every drifted agent +meta-update(inp): MetaLock(inp) →(in-DAG) «rebuild subgraph» per affected agent +boot: (if any rev marker stale) MetaLock(hyperhive) →(in-DAG) «rebuild subgraph» per stale agent; + plus Reconcile(a) for every drifted agent (all ONE DAG) ``` Notable collapses: @@ -123,10 +123,14 @@ Notable collapses: - **Graceful stop needs no watcher thread**: `Signal`/`Drain` are cheap, so a whole-hive graceful stop fires every agent's signal immediately and all drains overlap; each DAG's tail `Reconcile` does the actual stop. -- **The meta-update cascade fans out on completion**: `MetaLock`'s executor - computes the affected agent set after the bump lands and appends child - `rebuild` DAGs (`parent_id` set, `relock = false` so the children don't - revert the bump). A failed bump fans out nothing — no cancel-children dance. +- **The meta-update cascade grows in the same DAG on completion**: + `MetaLock`'s executor computes the affected agent set after the bump lands + and grows one `rebuild` subgraph per agent into its *own* DAG via + `append_subgraph` (rooted on the `MetaLock`, `relock = false` so the cascade + doesn't revert the bump). Not child DAGs — one DAG, no `parent_id`. A failed + bump appends nothing (no cancel-children dance). Same shape as the startup + sweep; the meta-update DAG carries the `Rebuilding` transient so each cascade + agent keeps crash-watch suppression during its `Swap`. ### Desired-state (spec vs status) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index ef24cb9b..7c3eb6bd 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -27,26 +27,24 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from /// success. #[derive(Debug, Default)] pub struct NodeOutput { - /// Agents to fan child `Rebuild` DAGs out for (`MetaLock` only). - pub fanout: Vec, /// 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`. A separate - /// channel from `fanout` (which appends whole *child* DAGs) so the - /// sub-step stays a first-class node in the same DAG and 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. + /// `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 /// [`JobQueue::append_subgraph`], which rebases the deps onto the DAG's - /// node-id space and roots the subgraph on the emitting node. The - /// startup sweep's `MetaLock` uses this to grow one stale-agent rebuild - /// subgraph per agent into the same boot DAG instead of fanning out - /// child DAGs. Same before-completion ordering as `append_nodes`. + /// 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`. pub append_subgraph: Vec>, } @@ -312,8 +310,18 @@ async fn run_meta_lock( Some(list) => list, None => meta_update_cascade_agents(&claim.inputs).await, }; + // Grow one rebuild subgraph per affected agent into *this* meta-update + // DAG (rooted on this `MetaLock`, so they build against the post-bump + // lock), rather than fanning out child DAGs. `relock = false` — the + // 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| super::templates::rebuild_nodes(agent, false, 0)) + .collect(); Ok(NodeOutput { - fanout: cascade, + append_subgraph, ..Default::default() }) } diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index b47df1b7..e651a505 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -3,21 +3,20 @@ //! executor task per claim, and on any completion re-evaluate. //! Concurrency comes from the build-slot count, not multiple workers. //! -//! Also owns the two DAG-lifetime side channels the sync queue core -//! can't hold itself: -//! - the per-DAG transient guard (dashboard pill + crash-watch -//! suppression), created when a DAG acquires its agent lease and -//! dropped when the DAG settles terminal; -//! - the `MetaLock` fan-out: appending child `Rebuild` DAGs once the -//! lock bump lands, so children build against the post-bump lock -//! (and a failed bump fans out nothing — replacing the old -//! pre-enqueue + cancel-children dance). +//! Also owns the per-DAG transient guard (dashboard pill + crash-watch +//! suppression) that the sync queue core can't hold itself — created when a +//! DAG acquires its agent lease, dropped when the DAG settles terminal. +//! +//! 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`. use std::collections::HashMap; use std::sync::Arc; +use super::Claim; use super::exec::{self, NodeOutput}; -use super::{Claim, Source, templates}; use crate::coordinator::Coordinator; struct NodeDone { @@ -128,10 +127,6 @@ async fn handle_completion( coord .job_queue .complete_node(claim.dag_id, claim.node_id, Ok(())); - if !output.fanout.is_empty() { - let specs = fanout_specs(&claim, output.fanout); - coord.job_queue.append_children(specs); - } } Err(e) => { let msg = format!("{e:#}"); @@ -165,30 +160,3 @@ async fn process_terminals( exec::on_dag_terminal(coord, &terminal).await; } } - -/// Child `Rebuild` specs for a completed meta-update `MetaLock` fan-out, -/// grouped under the parent via `parent_id`. Meta-update children skip the -/// per-agent relock (`relock = false`) — it would revert the bump the parent -/// just committed. This is now the meta-update cascade path only: the startup -/// sweep no longer fans out child DAGs — it grows one rebuild subgraph per -/// stale agent into its own DAG via `append_subgraph` (see -/// `exec::run_meta_lock`). -fn fanout_specs(claim: &Claim, agents: Vec) -> Vec { - let reason = if let Some(approval_id) = claim.approval_id { - format!("approval #{approval_id} meta input cascade") - } else { - "meta-update cascade".to_owned() - }; - agents - .into_iter() - .map(|agent| { - templates::rebuild( - &agent, - Source::MetaUpdate, - reason.clone(), - Some(claim.dag_id), - false, - ) - }) - .collect() -} diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index c2e5809d..b7e17a7c 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -186,10 +186,16 @@ pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPay } } -/// Meta-input lock bump. Child `Rebuild` DAGs fan out on completion — -/// appended *after* the bump lands so their prebuilds run against the -/// post-bump lock (and so a failed bump simply fans out nothing, -/// replacing the old pre-enqueue + `cancel_children` dance). +/// 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 +/// run against the post-bump lock, and a failed bump appends nothing +/// (replacing the old fan-out-child-DAGs + `cancel_children` dance). +/// `transient = Rebuilding` because those appended subgraphs are rebuilds: +/// it's applied per-agent at claim time (the `MetaLock` head needs no lease, +/// so the "hyperhive" pseudo-agent gets no pill), giving each cascade agent +/// crash-watch suppression during its `Swap` — the property the old child +/// `Rebuild` DAGs carried via their own transient. pub fn meta_update( inputs: Vec, source: Source, @@ -204,7 +210,7 @@ pub fn meta_update( approval_id, inputs, perm_payload: None, - transient: None, + transient: Some(TransientKind::Rebuilding), nodes: vec![node( "hyperhive", NodeKind::MetaLock { diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index f321de8b..2258d865 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -502,6 +502,56 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { ); } +#[test] +fn meta_update_carries_rebuilding_transient_and_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. + // The DAG carries `Rebuilding` so the folded rebuilds keep crash-watch + // suppression (the property the old child Rebuild DAGs had via their own + // transient). + let spec = templates::meta_update( + vec!["nixpkgs".to_owned()], + Source::Manual, + "bump".to_owned(), + None, + ); + assert!( + matches!( + spec.transient, + Some(crate::coordinator::TransientKind::Rebuilding) + ), + "meta-update DAG must carry Rebuilding so cascade rebuilds get suppression" + ); + let q = JobQueue::new(4); + let id = 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). + for agent in ["alice", "bob"] { + q.append_subgraph( + id, + templates::rebuild_nodes(agent, false, 0), + meta_lock.node_id, + ); + } + q.complete_node(id, meta_lock.node_id, Ok(())); + // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root + // on the MetaLock, each on its own agent lease. + assert_eq!(q.snapshot().len(), 1); + let next = q.claim_ready(); + let mut kinds: Vec<(&str, &str)> = next + .iter() + .map(|c| (c.agent.as_str(), c.kind.as_str())) + .collect(); + kinds.sort_unstable(); + assert_eq!( + kinds, + vec![("alice", "prebuild"), ("bob", "prebuild")], + "cascade rebuilds grow in the meta-update DAG, concurrent per agent" + ); +} + // ---- failure: cancel-downstream + AfterAny ---- #[test]