90e8158f37
ruth is created from the swarm dashboard like any agent. An existing ruth container starts at boot through the boot reconcile, like every other agent. A definition of services.hyperhive.ruthless now fails evaluation with a removal message.
388 lines
16 KiB
Rust
388 lines
16 KiB
Rust
//! Boot reconcile: on `hive-c0re serve` boot, (a) run the config path
|
|
//! for agents whose per-agent rev marker is stale — one `Boot` DAG
|
|
//! (meta hyperhive lock bump) that grows a `Rebuild` subgraph for each
|
|
//! stale agent whose `wanted` power intent is `Up` — and (b)
|
|
//! converge every other drifted agent to its persisted `wanted` via
|
|
//! `Reconcile` DAGs. Two rules keep boot-time nix work minimal:
|
|
//!
|
|
//! 1. **Stale but wanted-offline agents** get no rebuild — their
|
|
//! rebuild happens the first time they're started (the start
|
|
//! submit path upgrades a stale start to rebuild+start). The sweep
|
|
//! parent still runs whenever *any* marker is stale so the meta
|
|
//! hyperhive lock is bumped for those later start-upgrades.
|
|
//! 2. **Agents whose rev marker matches** the current hyperhive flake
|
|
//! path are skipped — nothing changed, no nix work to do.
|
|
//!
|
|
//! Booting with no config change performs no meta commit — only
|
|
//! reconciles. See `docs/scheduler/coordinator.md::Boot reconcile`.
|
|
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::Result;
|
|
|
|
use crate::coordinator::Coordinator;
|
|
use crate::lifecycle::{self, AGENT_PREFIX};
|
|
|
|
/// Resolve the current rev of `hyperhive_flake`. For a path on disk we
|
|
/// canonicalize (following symlinks) so a /etc/hyperhive → /nix/store/...
|
|
/// update yields a different string. For anything else we return None.
|
|
#[must_use]
|
|
pub fn current_flake_rev(hyperhive_flake: &str) -> Option<String> {
|
|
let path = Path::new(hyperhive_flake);
|
|
if !path.exists() {
|
|
return None;
|
|
}
|
|
std::fs::canonicalize(path)
|
|
.ok()
|
|
.map(|p| p.display().to_string())
|
|
}
|
|
|
|
/// Returns true when the applied repo has commits that have not yet been
|
|
/// deployed (i.e. the applied HEAD differs from the sha currently locked in
|
|
/// meta's flake.lock). This is the semantic the dashboard `needs_update` chip
|
|
/// conveys: "there is a config change ready to apply via rebuild."
|
|
///
|
|
/// Async on purpose: this runs per agent inside `container_view::build_all`,
|
|
/// which fires on the ~10s dashboard sweep, every `AgentStatus` request, and
|
|
/// every `rescan_containers_and_emit` after a lifecycle step. A synchronous
|
|
/// `git` fork here blocks a tokio worker for the whole exec — under
|
|
/// nix-build disk saturation that's long enough that concurrent sweeps
|
|
/// starved the runtime and stalled the per-agent sockets.
|
|
pub async fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool {
|
|
let applied = crate::paths::applied_dir(name);
|
|
let applied_head = crate::lifecycle::git_command()
|
|
.args(["-C", &applied.to_string_lossy(), "rev-parse", "HEAD"])
|
|
.output()
|
|
.await
|
|
.ok()
|
|
.filter(|o| o.status.success())
|
|
.and_then(|o| String::from_utf8(o.stdout).ok())
|
|
.map(|s| s.trim().to_owned());
|
|
|
|
match (applied_head.as_deref(), deployed_sha) {
|
|
(Some(head), Some(sha)) => !head.starts_with(sha) && !sha.starts_with(head),
|
|
_ => false,
|
|
}
|
|
}
|
|
|
|
/// Boot reconcile (see the module doc): classify every agent by rev
|
|
/// freshness + persisted `wanted` intent, submit one `Boot` DAG
|
|
/// (hyperhive lock bump growing an in-DAG rebuild subgraph per stale
|
|
/// wanted-up agent) when anything is stale, and `Reconcile` DAGs for
|
|
/// agents whose observed power state drifted from `wanted`. Returns Ok even
|
|
/// if some submissions failed.
|
|
pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
|
|
let containers = match lifecycle::list().await {
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "boot reconcile: nixos-container list failed");
|
|
return Ok(());
|
|
}
|
|
};
|
|
|
|
let current_rev = current_flake_rev(&coord.hyperhive_flake);
|
|
|
|
// Resolve container names to logical agent names, then sort. The
|
|
// parent field this used to depth-sort by is gone; with no
|
|
// hierarchy left to respect, alphabetical is the whole order — and it
|
|
// is exactly what the depth sort already produced once every agent
|
|
// was a root.
|
|
let mut logical_names: Vec<String> = containers
|
|
.iter()
|
|
.filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
|
|
.collect();
|
|
logical_names.sort();
|
|
|
|
// Classify. `get_or_seed` doubles as the one-time migration: an
|
|
// agent without an `agent_power` row is seeded from its observed
|
|
// state (running ⇒ Up), after which the DB is authoritative.
|
|
let mut any_stale = false;
|
|
// stale ∧ wanted=Up → sweep rebuild. The bool is the agent's observed
|
|
// running state, used to order running agents first before submit.
|
|
let mut fanout: Vec<(String, bool)> = Vec::new();
|
|
let mut drifted: Vec<String> = Vec::new(); // fresh ∧ wanted≠observed → reconcile
|
|
let mut n_deferred = 0usize;
|
|
let mut n_skipped = 0usize;
|
|
for name in &logical_names {
|
|
let running = lifecycle::is_running(name).await;
|
|
let wanted = match coord.power.get_or_seed(name, running) {
|
|
Ok(w) => Some(w),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "agent_power read failed — reconciling so it surfaces");
|
|
None
|
|
}
|
|
};
|
|
let fresh = current_rev.as_ref().is_some_and(|rev| {
|
|
std::fs::read_to_string(crate::paths::applied_rev_marker(name))
|
|
.is_ok_and(|stored| stored == rev.as_str())
|
|
});
|
|
match boot_action(wanted, fresh, running) {
|
|
BootAction::Rebuild => {
|
|
any_stale = true;
|
|
fanout.push((name.clone(), running));
|
|
}
|
|
BootAction::Defer { reconcile } => {
|
|
any_stale = true;
|
|
n_deferred += 1;
|
|
tracing::debug!(%name, "boot reconcile: stale but offline — deferring rebuild to on-start");
|
|
if reconcile {
|
|
drifted.push(name.clone());
|
|
}
|
|
}
|
|
BootAction::Skip { reconcile } => {
|
|
n_skipped += 1;
|
|
if reconcile {
|
|
drifted.push(name.clone());
|
|
}
|
|
}
|
|
BootAction::Unreadable => drifted.push(name.clone()),
|
|
}
|
|
}
|
|
|
|
tracing::info!(
|
|
total = containers.len(),
|
|
rebuilds = fanout.len(),
|
|
reconciles = drifted.len(),
|
|
deferred = n_deferred,
|
|
up_to_date = n_skipped,
|
|
"boot reconcile"
|
|
);
|
|
|
|
// Rebuild running agents first. All fanout entries are wanted=Up;
|
|
// among them, warm the live/serving agents onto the fresh config before the
|
|
// stopped-but-wanted-up ones so the scarce build slots hit uptime-critical
|
|
// agents first. Stable sort keeps the alphabetical order within each
|
|
// running/stopped group. The `drifted` reconciles aren't sorted
|
|
// — they hold no build slot and run concurrently, so their order is moot.
|
|
fanout.sort_by_key(|(_, running)| !running);
|
|
let fanout: Vec<String> = fanout.into_iter().map(|(name, _)| name).collect();
|
|
|
|
submit_boot_tree(&coord, any_stale, fanout, drifted, n_deferred, n_skipped);
|
|
submit_startup_sweep_nodes(&coord);
|
|
Ok(())
|
|
}
|
|
|
|
/// What the boot sweep does with one agent.
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
enum BootAction {
|
|
/// Stale ∧ wanted up: rebuild against the post-bump lock. The DAG's
|
|
/// tail `Reconcile` brings the agent (back) up, covering both the
|
|
/// running-stale and stopped-but-wanted-up cases.
|
|
Rebuild,
|
|
/// Stale ∧ wanted offline: no boot-time nix work — the start submit
|
|
/// path upgrades a stale start to a rebuild.
|
|
Defer { reconcile: bool },
|
|
/// Already on the current rev.
|
|
Skip { reconcile: bool },
|
|
/// The intent could not be read.
|
|
Unreadable,
|
|
}
|
|
|
|
/// `wanted` is `None` when the store could not answer for this agent.
|
|
///
|
|
/// That case has no neutral stand-in, which is why it is a variant rather
|
|
/// than a fallback value: `Wanted::from_running(running)` is precisely the
|
|
/// value for which `reconcile_action` returns `Noop`, so guessing it here
|
|
/// puts the agent beyond every check below — a corrupt row then leaves one
|
|
/// `warn!` per boot and no other trace. The queued reconcile's own
|
|
/// `get_or_seed` fails as a per-agent node instead.
|
|
fn boot_action(wanted: Option<crate::power::Wanted>, fresh: bool, running: bool) -> BootAction {
|
|
let Some(wanted) = wanted else {
|
|
return BootAction::Unreadable;
|
|
};
|
|
let reconcile =
|
|
crate::power::reconcile_action(wanted, running) != crate::power::ReconcileAction::Noop;
|
|
if fresh {
|
|
BootAction::Skip { reconcile }
|
|
} else if wanted == crate::power::Wanted::Up {
|
|
BootAction::Rebuild
|
|
} else {
|
|
BootAction::Defer { reconcile }
|
|
}
|
|
}
|
|
|
|
/// Submit the boot-time forge/matrix/knowledge/wanted-state sweeps as DAG
|
|
/// nodes — `ForgeSweep`, `MatrixSweep`, `KnowledgePull`, `WantedPull`. Unlike
|
|
/// [`submit_boot_tree`] this runs on **every** boot, quiet or not: these
|
|
/// aren't config-drift work, they're startup housekeeping that always needs
|
|
/// to happen, and the point of moving them here is exactly so they show up
|
|
/// as real work on the dashboard instead of an invisible `tokio::spawn` that
|
|
/// only surfaces on failure. Four independent, build-slot- and lease-exempt
|
|
/// roots — no dependency edges between them, matching the existing
|
|
/// `Reconcile`-root pattern in [`boot_nodes`]. `MatrixSweep` and
|
|
/// `KnowledgePull` hold their sweep's own resource, so each queues behind any
|
|
/// pass of the same sweep already running.
|
|
fn submit_startup_sweep_nodes(coord: &Arc<Coordinator>) {
|
|
use crate::job_queue::NodeKind;
|
|
|
|
if let Err(e) = coord.job_queue.insert_job(|b| {
|
|
let _ = b.node(NodeKind::ForgeSweep);
|
|
let _ = crate::job_queue::templates::matrix_sweep(b);
|
|
let _ = crate::job_queue::templates::knowledge_pull(b);
|
|
let _ = b.node(NodeKind::WantedPull);
|
|
Vec::new()
|
|
}) {
|
|
tracing::warn!(error = ?e, "boot: startup sweep DAG insert failed");
|
|
}
|
|
}
|
|
|
|
/// The boot DAG's node declarations, split out of [`submit_boot_tree`] so they
|
|
/// can be exercised without a live [`Coordinator`].
|
|
///
|
|
/// That split is not cosmetic: this path constructs nodes outside
|
|
/// `job_queue/`, so it is the one place a resource declaration can be forgotten
|
|
/// without any in-module test noticing. It has happened once already — the
|
|
/// sweep `MetaLock` and the boot `Reconcile`s silently declared nothing when
|
|
/// kind-derived resources were removed, which drops the agent lease a boot
|
|
/// reconcile needs to not race another DAG's container ops.
|
|
pub(crate) fn boot_nodes(
|
|
builder: &crate::job_queue::JobBuilder,
|
|
any_stale: bool,
|
|
fanout: Vec<String>,
|
|
drifted: Vec<String>,
|
|
) {
|
|
use crate::job_queue::NodeKind;
|
|
use crate::job_queue::resource::Resource;
|
|
|
|
// 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.
|
|
if any_stale {
|
|
let _ = builder
|
|
.node(NodeKind::MetaLock {
|
|
sweep: true,
|
|
fanout: Some(fanout),
|
|
// A sweep bumps `hyperhive` alone (`lock_update_hyperhive`),
|
|
// so it names no inputs.
|
|
inputs: Vec::new(),
|
|
})
|
|
.needs(Resource::BuildSlot)
|
|
.needs(Resource::MetaWindow);
|
|
}
|
|
// One boot Reconcile per drifted agent — independent roots.
|
|
for name in drifted {
|
|
// Name the lease before the agent string moves into the kind.
|
|
let lease = Resource::Agent(name.clone());
|
|
let _ = builder
|
|
.node(NodeKind::Reconcile { agent: name })
|
|
.needs(lease);
|
|
}
|
|
}
|
|
|
|
/// Submit this boot's work as **one DAG** (no anchor node, 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<Coordinator>,
|
|
any_stale: bool,
|
|
fanout: Vec<String>,
|
|
drifted: Vec<String>,
|
|
n_deferred: usize,
|
|
n_skipped: usize,
|
|
) {
|
|
// Fully-quiet boot (nothing stale, nothing drifted) inserts nothing.
|
|
if !any_stale && drifted.is_empty() {
|
|
return;
|
|
}
|
|
// The summary the sweep used to hand the container as its `reason` is a log
|
|
// line now: it was only ever stored on a node nobody read, and the counts
|
|
// are worth having where they can actually be seen.
|
|
tracing::info!(
|
|
rebuilds = fanout.len(),
|
|
reconciles = drifted.len(),
|
|
deferred = n_deferred,
|
|
up_to_date = n_skipped,
|
|
"boot: sweep"
|
|
);
|
|
|
|
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
|
|
// land; the boot DAG as a whole has no terminal side effect, so no tail.
|
|
// The subgraphs also carry their own per-agent crash-watch suppression
|
|
// during their `Swap` (applied at claim time); a reconcile-only boot needs
|
|
// no transient.
|
|
if let Err(e) = coord.job_queue.insert_job(|b| {
|
|
boot_nodes(b, any_stale, fanout, drifted);
|
|
Vec::new()
|
|
}) {
|
|
tracing::warn!(error = ?e, "boot: sweep DAG insert failed");
|
|
}
|
|
coord.emit_rebuild_queue_snapshot();
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{BootAction, boot_action};
|
|
use crate::power::Wanted;
|
|
|
|
// The regression the `Unreadable` variant exists for: the arm used to
|
|
// substitute `from_running(running)`, which is `Noop` against BOTH
|
|
// observations — so no combination of inputs could reach a reconcile.
|
|
#[test]
|
|
fn an_unreadable_intent_is_never_settled_whatever_the_agent_is_doing() {
|
|
for fresh in [true, false] {
|
|
for running in [true, false] {
|
|
assert_eq!(
|
|
boot_action(None, fresh, running),
|
|
BootAction::Unreadable,
|
|
"fresh={fresh} running={running}"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_fresh_agent_doing_what_it_should_is_skipped_without_a_reconcile() {
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Up), true, true),
|
|
BootAction::Skip { reconcile: false }
|
|
);
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Offline), true, false),
|
|
BootAction::Skip { reconcile: false }
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_fresh_agent_that_drifted_from_its_intent_is_reconciled() {
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Up), true, false),
|
|
BootAction::Skip { reconcile: true }
|
|
);
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Offline), true, true),
|
|
BootAction::Skip { reconcile: true }
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_stale_agent_wanted_up_is_rebuilt_running_or_not() {
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Up), false, true),
|
|
BootAction::Rebuild
|
|
);
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Up), false, false),
|
|
BootAction::Rebuild
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_stale_agent_wanted_offline_defers_and_still_reports_drift() {
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Offline), false, false),
|
|
BootAction::Defer { reconcile: false }
|
|
);
|
|
assert_eq!(
|
|
boot_action(Some(Wanted::Offline), false, true),
|
|
BootAction::Defer { reconcile: true }
|
|
);
|
|
}
|
|
}
|