From 4545dd312e90504a69c17de51eed275bc935be65 Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 15 Jul 2026 17:29:23 +0200 Subject: [PATCH] feat(#2450): collapse the startup sweep into one inline DAG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The boot is now ONE DAG, assembled inline in submit_boot_tree — no boot_root Noop anchor, no per-agent child DAGs, no display-only parent_id grouping: a sweep MetaLock root (only when something is stale) that grows one rebuild subgraph per stale agent into the same DAG (via append_subgraph, previous commit), plus one Reconcile root per drifted agent (independent — a boot reconcile needs no lock bump). - submit_boot_tree builds the DagSpec inline; removed the single-use templates::{boot_root, startup_sweep} builders (inlined per the operator's "don't force single-use shapes into templates.rs" steer). - fanout_specs simplified to the meta-update cascade path only — the startup sweep no longer fans out child DAGs, so its branch was dead. - test: append_subgraph_roots_on_emitter_and_rebases_local_deps. Vestigial after this (deliberately left as follow-ups, flagged in the PR): NodeKind::Noop is now unconstructed (contained to hive-c0re, removable); Template::StartupSweep is unconstructed but a hive-sh4re wire type (frontend-coordinated removal). --- hive-c0re/src/job_queue/mod.rs | 7 +- hive-c0re/src/job_queue/scheduler.rs | 33 ++++---- hive-c0re/src/job_queue/templates.rs | 48 ++---------- hive-c0re/src/job_queue/tests.rs | 52 +++++++++++++ hive-c0re/src/workers/auto_update.rs | 110 ++++++++++++++------------- 5 files changed, 138 insertions(+), 112 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index eddf4f81..6f951341 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -219,7 +219,12 @@ impl JobQueue { /// 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 { + pub fn append_subgraph( + &self, + dag_id: u64, + nodes: Vec, + dep_on: NodeId, + ) -> Vec { if nodes.is_empty() { return Vec::new(); } diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 90e3bda6..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 { @@ -166,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(); }