diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index d554b9a8..ef24cb9b 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::model::{NodeKind, State, Template}; +use super::model::{NodeKind, NodeSpec, State, Template}; use super::{Claim, TerminalDag}; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; @@ -38,6 +38,16 @@ pub struct NodeOutput { /// 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`. + pub append_subgraph: Vec>, } /// Step-label + build-log sink for one claimed node. @@ -276,8 +286,17 @@ async fn run_meta_lock( if let Err(e) = crate::meta::lock_update_hyperhive().await { tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed"); } + // Grow one rebuild subgraph per stale agent into *this* boot DAG + // (rooted on this `MetaLock`, so they build against the post-bump + // lock), rather than fanning out child DAGs. `relock = true` — a + // boot sweep relocks per-agent like a manual rebuild. + let append_subgraph = fanout + .unwrap_or_default() + .iter() + .map(|agent| super::templates::rebuild_nodes(agent, true, 0)) + .collect(); return Ok(NodeOutput { - fanout: fanout.unwrap_or_default(), + append_subgraph, ..Default::default() }); } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index db0c3e1e..6f951341 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -34,7 +34,8 @@ use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; pub use model::{ - Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template, + Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, NodeSpec, PermPayload, Source, State, + Template, }; /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain @@ -206,6 +207,71 @@ impl JobQueue { 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. + pub fn append_subgraph( + &self, + dag_id: u64, + nodes: Vec, + dep_on: NodeId, + ) -> Vec { + if nodes.is_empty() { + return Vec::new(); + } + let mut inner = self.inner.lock().expect("job_queue mutex poisoned"); + let Some(dag) = inner.dags.iter_mut().find(|d| d.id == dag_id) else { + return Vec::new(); + }; + let base: NodeId = u32::try_from(dag.nodes.len()).unwrap_or(u32::MAX); + let mut new_ids = Vec::with_capacity(nodes.len()); + for (i, spec) in nodes.into_iter().enumerate() { + let new_id: NodeId = base + u32::try_from(i).unwrap_or(u32::MAX); + // Subgraph roots (no internal deps) hang off the emitting node; + // internal deps rebase from subgraph-relative onto the DAG id + // space (both start at `base`). + let deps = if spec.deps.is_empty() { + vec![model::Dep { + on: dep_on, + when: DepWhen::AfterOk, + }] + } else { + spec.deps + .into_iter() + .map(|d| model::Dep { + on: base + d.on, + when: d.when, + }) + .collect() + }; + dag.nodes.push(Node { + id: new_id, + agent: spec.agent, + kind: spec.kind, + deps, + state: State::Queued, + step: None, + build_log_id: None, + started_at: None, + finished_at: None, + error: None, + }); + new_ids.push(new_id); + } + drop(inner); + self.notify.notify_one(); + new_ids + } + fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 { inner.next_id += 1; let id = inner.next_id; diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index eb03e98c..b47df1b7 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -17,7 +17,7 @@ use std::collections::HashMap; use std::sync::Arc; use super::exec::{self, NodeOutput}; -use super::{Claim, Source, Template, templates}; +use super::{Claim, Source, templates}; use crate::coordinator::Coordinator; struct NodeDone { @@ -116,6 +116,15 @@ async fn handle_completion( .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. + for subgraph in output.append_subgraph { + coord + .job_queue + .append_subgraph(claim.dag_id, subgraph, claim.node_id); + } coord .job_queue .complete_node(claim.dag_id, claim.node_id, Ok(())); @@ -157,26 +166,29 @@ async fn process_terminals( } } -/// Child `Rebuild` specs for a completed `MetaLock` fan-out, grouped -/// under the parent via `parent_id`. Meta-update children skip the -/// per-agent relock (it would revert the bump the parent just -/// committed); sweep children relock like a manual rebuild. +/// 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 sweep = claim.template == Template::StartupSweep; - let (source, relock) = if sweep { - (Source::StartupSweep, true) - } else { - (Source::MetaUpdate, false) - }; - let reason = if sweep { - "startup sweep".to_owned() - } else if let Some(approval_id) = claim.approval_id { + 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, reason.clone(), Some(claim.dag_id), relock)) + .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 f9a7bebd..c2e5809d 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -216,49 +216,11 @@ pub fn meta_update( } } -/// Boot-time root anchor DAG: a single [`NodeKind::Noop`] node that groups -/// this boot's `StartupSweep` + per-agent `Reconcile` child DAGs (linked via -/// `parent_id`) into one tree so the dashboard renders the boot as one entry. -/// Holds no lease and does no work — the children it anchors still run -/// concurrently. `auto_update::run` submits it first (when there's any boot -/// work), then parents the sweep + reconciles onto its id. -pub fn boot_root(reason: String) -> DagSpec { - DagSpec { - template: Template::Boot, - source: Source::AutoUpdate, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: None, - nodes: vec![node("hyperhive", NodeKind::Noop, Vec::new())], - } -} - -/// Boot-time sweep parent: bump meta's hyperhive input (non-fatal), -/// then fan out `Rebuild` children for the precomputed stale agent -/// list (topology-sorted by the caller). -pub fn startup_sweep(reason: String, stale_agents: Vec) -> DagSpec { - DagSpec { - template: Template::StartupSweep, - source: Source::AutoUpdate, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: None, - nodes: vec![node( - "hyperhive", - NodeKind::MetaLock { - sweep: true, - fanout: Some(stale_agents), - }, - Vec::new(), - )], - } -} +// The boot is now assembled inline in `workers/auto_update.rs::submit_boot_tree` +// as ONE DAG (a sweep `MetaLock` root that grows rebuild subgraphs in-DAG, plus +// a `Reconcile` root per drifted agent) — no `boot_root` Noop anchor, no +// `startup_sweep` parent template, no per-agent child DAGs. The old single-use +// `boot_root` / `startup_sweep` builders were inlined there and removed. /// Validate a spec before it enters the queue: node ids are dense /// (index = id), deps reference existing nodes, and the dep graph is diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index a006f1a4..f321de8b 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -450,6 +450,58 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { ); } +#[test] +fn append_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. + let q = JobQueue::new(4); + let spec = DagSpec { + template: Template::Boot, + source: Source::AutoUpdate, + reason: "sweep".to_owned(), + parent_id: None, + approval_id: None, + inputs: Vec::new(), + perm_payload: None, + transient: None, + nodes: vec![NodeSpec { + agent: "hyperhive".to_owned(), + kind: NodeKind::MetaLock { + sweep: true, + fanout: None, + }, + deps: Vec::new(), + }], + }; + let id = 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 (`rebuild_nodes(_, true, 0)`: root Prebuild → + // StopForUpdate → Swap → Reconcile, local 0-based deps), so this test + // tracks any drift in that builder's root-first (`base = 0`) shape. + let subgraph = |agent: &str| templates::rebuild_nodes(agent, true, 0); + // 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(id, emitter.node_id, Ok(())); + // Still ONE DAG; both subgraph roots become ready once the emitter is + // Done (rooted on it), 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![("a", "prebuild"), ("b", "prebuild")], + "both rebuild subgraphs root on the emitter and run concurrently in one DAG" + ); +} + // ---- failure: cancel-downstream + AfterAny ---- #[test] diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 9ccc19a6..5b8d1438 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -288,12 +288,14 @@ pub async fn run(coord: Arc) -> Result<()> { Ok(()) } -/// Submit this boot's DAGs under one `Boot` root: a `Noop` anchor with the -/// startup sweep + per-agent reconciles hung off it via `parent_id`, so the -/// dashboard renders the boot as a single tree instead of N+1 rows. No-op when -/// there's nothing to do. The `parent_id` link is a display grouping, not a -/// dependency edge — the children run concurrently, so the reconciles never -/// wait behind the lock bump. +/// Submit this boot's work as **one DAG** (no `boot_root` Noop anchor, no +/// per-agent child DAGs). Node 0 is the sweep `MetaLock` (only when +/// something is stale) — its executor bumps the hyperhive lock, then grows +/// one rebuild subgraph per stale agent into *this same* DAG (rooted on the +/// `MetaLock`, so they build against the post-bump lock; see +/// `exec::run_meta_lock`). Every drifted agent gets a boot `Reconcile` as an +/// independent root — a boot reconcile needs no lock bump, so it converges +/// concurrently with the sweep. No-op when there's nothing to do. fn submit_boot_tree( coord: &Arc, any_stale: bool, @@ -302,59 +304,61 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - // Only emit a boot root when there's actually boot work — a fully-quiet - // boot (nothing stale, nothing drifted) submits nothing, exactly as before. - let boot_root_id = if any_stale || !drifted.is_empty() { - let reason = format!( - "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date", - fanout.len(), - drifted.len(), - n_deferred, - n_skipped, - ); - match coord - .job_queue - .submit(crate::job_queue::templates::boot_root(reason)) - { - Ok(id) => Some(id), - Err(e) => { - tracing::warn!(error = ?e, "boot reconcile: boot-root submit failed"); - None - } - } - } else { - None - }; + use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source, Template}; + // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. + if !any_stale && drifted.is_empty() { + return; + } + let reason = format!( + "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date", + fanout.len(), + drifted.len(), + n_deferred, + n_skipped, + ); + + let mut nodes: Vec = Vec::new(); // Sweep whenever ANY marker is stale — even when every stale agent is // wanted-offline: the hyperhive lock bump must land now so their later - // start-upgrade rebuilds build against it. No stale agents ⇒ no sweep ⇒ no - // meta commit on a no-change boot. + // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock + // ⇒ no meta commit on a no-change boot. The `fanout` list rides the + // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. if any_stale { - let reason = format!( - "startup sweep: {} rebuild(s), {} deferred (offline), {} up-to-date", - fanout.len(), - n_deferred, - n_skipped, - ); - let mut spec = crate::job_queue::templates::startup_sweep(reason, fanout); - spec.parent_id = boot_root_id; - if let Err(e) = coord.job_queue.submit(spec) { - tracing::warn!(error = ?e, "boot reconcile: sweep submit failed"); - } + nodes.push(NodeSpec { + agent: "hyperhive".to_owned(), + kind: NodeKind::MetaLock { + sweep: true, + fanout: Some(fanout), + }, + deps: Vec::new(), + }); } + // One boot Reconcile per drifted agent — independent roots. for name in drifted { - let mut spec = crate::job_queue::templates::reconcile_only( - crate::job_queue::Template::Reconcile, - &name, - crate::job_queue::Source::AutoUpdate, - "boot reconcile".to_owned(), - None, - ); - spec.parent_id = boot_root_id; - if let Err(e) = coord.job_queue.submit(spec) { - tracing::warn!(%name, error = ?e, "boot reconcile: submit failed"); - } + nodes.push(NodeSpec { + agent: name, + kind: NodeKind::Reconcile, + deps: Vec::new(), + }); + } + + let spec = DagSpec { + template: Template::Boot, + source: Source::AutoUpdate, + reason, + parent_id: None, + approval_id: None, + inputs: Vec::new(), + perm_payload: None, + // Rebuilding when the sweep will grow rebuild subgraphs (per-agent + // crash-watch suppression during their Swap, applied at claim time); + // a reconcile-only boot needs no transient. + transient: any_stale.then_some(crate::coordinator::TransientKind::Rebuilding), + nodes, + }; + if let Err(e) = coord.job_queue.submit(spec) { + tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); } coord.emit_rebuild_queue_snapshot(); }