diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index ef24cb9b..d554b9a8 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, NodeSpec, State, Template}; +use super::model::{NodeKind, State, Template}; use super::{Claim, TerminalDag}; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; @@ -38,16 +38,6 @@ 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. @@ -286,17 +276,8 @@ 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 { - append_subgraph, + fanout: fanout.unwrap_or_default(), ..Default::default() }); } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 6f951341..db0c3e1e 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -34,8 +34,7 @@ use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; pub use model::{ - Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, NodeSpec, PermPayload, Source, State, - Template, + Dag, DagSpec, DagView, DepWhen, Node, NodeId, NodeKind, PermPayload, Source, State, Template, }; /// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain @@ -207,71 +206,6 @@ 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 b47df1b7..eb03e98c 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, templates}; +use super::{Claim, Source, Template, templates}; use crate::coordinator::Coordinator; struct NodeDone { @@ -116,15 +116,6 @@ 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(())); @@ -166,29 +157,26 @@ async fn process_terminals( } } -/// 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`). +/// 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. fn fanout_specs(claim: &Claim, agents: Vec) -> Vec { - let reason = if let Some(approval_id) = claim.approval_id { + 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 { 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, - ) - }) + .map(|agent| templates::rebuild(&agent, source, reason.clone(), Some(claim.dag_id), relock)) .collect() } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index c2e5809d..f9a7bebd 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -216,11 +216,49 @@ pub fn meta_update( } } -// 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. +/// 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(), + )], + } +} /// 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 f321de8b..a006f1a4 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -450,58 +450,6 @@ 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 5b8d1438..9ccc19a6 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -288,14 +288,12 @@ pub async fn run(coord: Arc) -> Result<()> { Ok(()) } -/// 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. +/// 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. fn submit_boot_tree( coord: &Arc, any_stale: bool, @@ -304,61 +302,59 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source, Template}; + // 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 + }; - // 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 MetaLock - // ⇒ no meta commit on a no-change boot. The `fanout` list rides the - // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. + // start-upgrade rebuilds build against it. No stale agents ⇒ no sweep ⇒ no + // meta commit on a no-change boot. if any_stale { - nodes.push(NodeSpec { - agent: "hyperhive".to_owned(), - kind: NodeKind::MetaLock { - sweep: true, - fanout: Some(fanout), - }, - deps: Vec::new(), - }); + 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"); + } } - // One boot Reconcile per drifted agent — independent roots. for name in drifted { - 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"); + 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"); + } } coord.emit_rebuild_queue_snapshot(); }