feat(#2450): collapse the startup sweep into one inline DAG
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).
This commit is contained in:
parent
b6defdeaaf
commit
4545dd312e
5 changed files with 138 additions and 112 deletions
|
|
@ -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<NodeSpec>, dep_on: NodeId) -> Vec<NodeId> {
|
||||
pub fn append_subgraph(
|
||||
&self,
|
||||
dag_id: u64,
|
||||
nodes: Vec<NodeSpec>,
|
||||
dep_on: NodeId,
|
||||
) -> Vec<NodeId> {
|
||||
if nodes.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>) -> Vec<super::DagSpec> {
|
||||
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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>) -> 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
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
Loading…
Reference in a new issue