Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4545dd312e | ||
|
|
b6defdeaaf |
6 changed files with 229 additions and 114 deletions
|
|
@ -10,7 +10,7 @@ use std::sync::Arc;
|
||||||
|
|
||||||
use anyhow::{Context as _, Result};
|
use anyhow::{Context as _, Result};
|
||||||
|
|
||||||
use super::model::{NodeKind, State, Template};
|
use super::model::{NodeKind, NodeSpec, State, Template};
|
||||||
use super::{Claim, TerminalDag};
|
use super::{Claim, TerminalDag};
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
use crate::power::{ReconcileAction, reconcile_action};
|
use crate::power::{ReconcileAction, reconcile_action};
|
||||||
|
|
@ -38,6 +38,16 @@ pub struct NodeOutput {
|
||||||
/// these *before* the emitting node's completion so the DAG never
|
/// these *before* the emitting node's completion so the DAG never
|
||||||
/// rolls terminal with the appended work still pending.
|
/// rolls terminal with the appended work still pending.
|
||||||
pub append_nodes: Vec<NodeKind>,
|
pub append_nodes: Vec<NodeKind>,
|
||||||
|
/// Whole per-agent *subgraphs* to append into *this same* DAG at
|
||||||
|
/// runtime — the multi-node generalisation of `append_nodes`. Each
|
||||||
|
/// inner `Vec<NodeSpec>` 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<Vec<NodeSpec>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Step-label + build-log sink for one claimed node.
|
/// 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 {
|
if let Err(e) = crate::meta::lock_update_hyperhive().await {
|
||||||
tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
|
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 {
|
return Ok(NodeOutput {
|
||||||
fanout: fanout.unwrap_or_default(),
|
append_subgraph,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -34,7 +34,8 @@ use hive_sh4re::wire_time::now_unix;
|
||||||
use tokio::sync::Notify;
|
use tokio::sync::Notify;
|
||||||
|
|
||||||
pub use model::{
|
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
|
/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) to retain
|
||||||
|
|
@ -206,6 +207,71 @@ impl JobQueue {
|
||||||
Some(new_id)
|
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<NodeSpec>,
|
||||||
|
dep_on: NodeId,
|
||||||
|
) -> Vec<NodeId> {
|
||||||
|
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 {
|
fn push_dag(inner: &mut Inner, spec: DagSpec) -> u64 {
|
||||||
inner.next_id += 1;
|
inner.next_id += 1;
|
||||||
let id = inner.next_id;
|
let id = inner.next_id;
|
||||||
|
|
|
||||||
|
|
@ -17,7 +17,7 @@ use std::collections::HashMap;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use super::exec::{self, NodeOutput};
|
use super::exec::{self, NodeOutput};
|
||||||
use super::{Claim, Source, Template, templates};
|
use super::{Claim, Source, templates};
|
||||||
use crate::coordinator::Coordinator;
|
use crate::coordinator::Coordinator;
|
||||||
|
|
||||||
struct NodeDone {
|
struct NodeDone {
|
||||||
|
|
@ -116,6 +116,15 @@ async fn handle_completion(
|
||||||
.job_queue
|
.job_queue
|
||||||
.append_node(claim.dag_id, kind, claim.node_id);
|
.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
|
coord
|
||||||
.job_queue
|
.job_queue
|
||||||
.complete_node(claim.dag_id, claim.node_id, Ok(()));
|
.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
|
/// Child `Rebuild` specs for a completed meta-update `MetaLock` fan-out,
|
||||||
/// under the parent via `parent_id`. Meta-update children skip the
|
/// grouped under the parent via `parent_id`. Meta-update children skip the
|
||||||
/// per-agent relock (it would revert the bump the parent just
|
/// per-agent relock (`relock = false`) — it would revert the bump the parent
|
||||||
/// committed); sweep children relock like a manual rebuild.
|
/// 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> {
|
fn fanout_specs(claim: &Claim, agents: Vec<String>) -> Vec<super::DagSpec> {
|
||||||
let sweep = claim.template == Template::StartupSweep;
|
let reason = if let Some(approval_id) = claim.approval_id {
|
||||||
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")
|
format!("approval #{approval_id} meta input cascade")
|
||||||
} else {
|
} else {
|
||||||
"meta-update cascade".to_owned()
|
"meta-update cascade".to_owned()
|
||||||
};
|
};
|
||||||
agents
|
agents
|
||||||
.into_iter()
|
.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()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -216,49 +216,11 @@ pub fn meta_update(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Boot-time root anchor DAG: a single [`NodeKind::Noop`] node that groups
|
// The boot is now assembled inline in `workers/auto_update.rs::submit_boot_tree`
|
||||||
/// this boot's `StartupSweep` + per-agent `Reconcile` child DAGs (linked via
|
// as ONE DAG (a sweep `MetaLock` root that grows rebuild subgraphs in-DAG, plus
|
||||||
/// `parent_id`) into one tree so the dashboard renders the boot as one entry.
|
// a `Reconcile` root per drifted agent) — no `boot_root` Noop anchor, no
|
||||||
/// Holds no lease and does no work — the children it anchors still run
|
// `startup_sweep` parent template, no per-agent child DAGs. The old single-use
|
||||||
/// concurrently. `auto_update::run` submits it first (when there's any boot
|
// `boot_root` / `startup_sweep` builders were inlined there and removed.
|
||||||
/// 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(),
|
|
||||||
)],
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Validate a spec before it enters the queue: node ids are dense
|
/// Validate a spec before it enters the queue: node ids are dense
|
||||||
/// (index = id), deps reference existing nodes, and the dep graph is
|
/// (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 ----
|
// ---- failure: cancel-downstream + AfterAny ----
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|
|
||||||
|
|
@ -288,12 +288,14 @@ pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Submit this boot's DAGs under one `Boot` root: a `Noop` anchor with the
|
/// Submit this boot's work as **one DAG** (no `boot_root` Noop anchor, no
|
||||||
/// startup sweep + per-agent reconciles hung off it via `parent_id`, so the
|
/// per-agent child DAGs). Node 0 is the sweep `MetaLock` (only when
|
||||||
/// dashboard renders the boot as a single tree instead of N+1 rows. No-op when
|
/// something is stale) — its executor bumps the hyperhive lock, then grows
|
||||||
/// there's nothing to do. The `parent_id` link is a display grouping, not a
|
/// one rebuild subgraph per stale agent into *this same* DAG (rooted on the
|
||||||
/// dependency edge — the children run concurrently, so the reconciles never
|
/// `MetaLock`, so they build against the post-bump lock; see
|
||||||
/// wait behind the lock bump.
|
/// `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(
|
fn submit_boot_tree(
|
||||||
coord: &Arc<Coordinator>,
|
coord: &Arc<Coordinator>,
|
||||||
any_stale: bool,
|
any_stale: bool,
|
||||||
|
|
@ -302,59 +304,61 @@ fn submit_boot_tree(
|
||||||
n_deferred: usize,
|
n_deferred: usize,
|
||||||
n_skipped: usize,
|
n_skipped: usize,
|
||||||
) {
|
) {
|
||||||
// Only emit a boot root when there's actually boot work — a fully-quiet
|
use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source, Template};
|
||||||
// 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<NodeSpec> = Vec::new();
|
||||||
// Sweep whenever ANY marker is stale — even when every stale agent is
|
// Sweep whenever ANY marker is stale — even when every stale agent is
|
||||||
// wanted-offline: the hyperhive lock bump must land now so their later
|
// wanted-offline: the hyperhive lock bump must land now so their later
|
||||||
// start-upgrade rebuilds build against it. No stale agents ⇒ no sweep ⇒ no
|
// start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock
|
||||||
// meta commit on a no-change boot.
|
// ⇒ 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 {
|
if any_stale {
|
||||||
let reason = format!(
|
nodes.push(NodeSpec {
|
||||||
"startup sweep: {} rebuild(s), {} deferred (offline), {} up-to-date",
|
agent: "hyperhive".to_owned(),
|
||||||
fanout.len(),
|
kind: NodeKind::MetaLock {
|
||||||
n_deferred,
|
sweep: true,
|
||||||
n_skipped,
|
fanout: Some(fanout),
|
||||||
);
|
},
|
||||||
let mut spec = crate::job_queue::templates::startup_sweep(reason, fanout);
|
deps: Vec::new(),
|
||||||
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 {
|
for name in drifted {
|
||||||
let mut spec = crate::job_queue::templates::reconcile_only(
|
nodes.push(NodeSpec {
|
||||||
crate::job_queue::Template::Reconcile,
|
agent: name,
|
||||||
&name,
|
kind: NodeKind::Reconcile,
|
||||||
crate::job_queue::Source::AutoUpdate,
|
deps: Vec::new(),
|
||||||
"boot reconcile".to_owned(),
|
});
|
||||||
None,
|
}
|
||||||
);
|
|
||||||
spec.parent_id = boot_root_id;
|
let spec = DagSpec {
|
||||||
if let Err(e) = coord.job_queue.submit(spec) {
|
template: Template::Boot,
|
||||||
tracing::warn!(%name, error = ?e, "boot reconcile: submit failed");
|
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();
|
coord.emit_rebuild_queue_snapshot();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue