vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

remove the hive's local build of agent closures

Browse source 
A rebuild or create downloads the closure the swarm built and applies it; with none to deploy it fails and changes nothing. hive-priv takes a store path instead of an internal/meta commit and builds nothing.
This commit is contained in:
atlas 2026-10-29 22:17:07 +01:00
commit 4590b04cac
13 changed files with 632 additions and 1351 deletions
Download patch file Download diff file Expand all files Collapse all files

59
hive-c0re/src/attic/closure.rs
View file

@ -1,62 +1,19 @@
//! The closure the swarm built for an agent at a given `internal/meta` commit,
//! as a rebuild's `Prebuild` downloads it instead of building.
//! The closure the swarm built for an agent, as a deploy reads its record
//! before downloading it.
use anyhow::{Result, anyhow};
use swarm_wire::agent_closures::AgentClosure;
use super::probe::{self, Unpullable};
use super::probe;
/// The store path of the closure `agent`'s record names, when the record is a
/// build of `rev` that the builder pushed to the cache.
/// `agent`'s record in `agent-closures`; `None` when the builder has recorded
/// no build of it.
///
/// # Errors
/// No record, one for another commit, or one the builder did not push; or the
/// queue cannot be reached. The message is what a deploy log says.
pub async fn path_for(agent: &str, rev: &str) -> Result<String> {
/// The queue cannot be reached, or the record cannot be read.
pub async fn record(agent: &str) -> Result<Option<AgentClosure>> {
let client = crate::swarm_queue::client()
.await
.ok_or_else(|| anyhow!("this hive has no connection to the swarm queue"))?;
let record = probe::read_record(&client, agent).await?;
path_in(record.as_ref(), rev)
}
fn path_in(record: Option<&AgentClosure>, rev: &str) -> Result<String> {
probe::current_record(record, Some(rev))
.map(|record| record.store_path.clone())
.map_err(|Unpullable { reason, detail }| anyhow!("{}: {detail}", reason.label()))
}
#[cfg(test)]
mod tests {
use super::*;
const REV: &str = "a1b2c3";
const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
fn record(rev: &str, cached: bool) -> AgentClosure {
AgentClosure {
meta_rev: rev.to_owned(),
store_path: PATH.to_owned(),
system: "x86_64-linux".to_owned(),
built_unix: 1,
cached,
}
}
#[test]
fn a_pushed_build_of_the_rev_names_its_path() {
assert_eq!(path_in(Some(&record(REV, true)), REV).expect("path"), PATH);
}
#[test]
fn anything_else_names_why_not() {
for (record, why) in [
(None, "no-record"),
(Some(record("other", true)), "rev-mismatch"),
(Some(record(REV, false)), "not-cached"),
] {
let err = path_in(record.as_ref(), REV).expect_err(why);
assert!(format!("{err}").starts_with(why), "{err}");
}
}
probe::read_record(&client, agent).await
}

12
hive-c0re/src/coordinator.rs
View file

@ -29,9 +29,8 @@ const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running";
pub struct Coordinator {
pub broker: Arc<Broker>,
/// Full build-log capture. `lifecycle::run` /
/// `lifecycle::prebuild_toplevel` `start()` a row per attempt,
/// pipe every stdout/stderr line into it, and `finish()` it on
/// Full build-log capture. `lifecycle::run` `start()`s a row per attempt,
/// pipes every stdout/stderr line into it, and `finish()`es it on
/// child exit. Dashboard reads it via `list_recent_for_agent` /
/// `get_full` for the per-card chip + side-panel viewer. See
/// `build_logs.rs` for retention.
@ -228,7 +227,7 @@ pub struct ServeConfig {
#[serde(flatten)]
pub env: HiveEnv,
pub model_prices: crate::hive_stats::PriceTable,
/// Number of concurrent nix-heavy job-queue nodes (prebuild /
/// Number of concurrent closure-fetching job-queue nodes (prebuild /
/// profile-swap / create / meta lock). hive-c0re-local like
/// `model_prices` — never injected into containers. Set via
/// `services.hyperhive.deploy.hive-controller.buildSlots`.
@ -374,9 +373,8 @@ impl Coordinator {
"build_logs: closed rows left in-flight by a previous run"
);
}
// Install the process-wide handle so `lifecycle::run` /
// `lifecycle::prebuild_toplevel` can write without us having
// to thread an `Arc<BuildLogs>` through every public entry
// Install the process-wide handle so `lifecycle::run` can write
// without us having to thread an `Arc<BuildLogs>` through every public entry
// point in the lifecycle surface.
crate::build_logs::install(build_logs.clone());
let power = Arc::new(crate::power::PowerStore::open(db_path).context("open agent_power")?);

37
hive-c0re/src/forge/internal_meta.rs
View file

@ -20,20 +20,13 @@ const CONFIG_BRANCH: &str = "main";
/// [`MetaSource`] over the forge API.
pub(crate) struct ForgeSource {
api: Forgejo,
/// Whether the credential file nix's `git` reads holds this hive's token.
nix_credential: bool,
}
/// A source reading as this hive, or `None` when it holds no forge token. It
/// also refreshes the credential file nix's `git` fetches with, which the
/// build after this read needs to be current.
/// A source reading as this hive, or `None` when it holds no forge token.
pub(crate) async fn internal_meta_source() -> Option<ForgeSource> {
let token: ReaderToken = super::reader_token().await?;
match token.api() {
Ok(api) => Some(ForgeSource {
api,
nix_credential: token.publish_for_nix(),
}),
Ok(api) => Some(ForgeSource { api }),
Err(e) => {
tracing::warn!(error = ?e, "internal/meta: cannot build a forge client");
None
@ -72,10 +65,6 @@ impl ForgeSource {
}
impl MetaSource for ForgeSource {
fn nix_can_fetch(&self) -> bool {
self.nix_credential
}
async fn meta_head(&self) -> anyhow::Result<Option<String>> {
use anyhow::Context as _;
@ -113,25 +102,3 @@ impl MetaSource for ForgeSource {
self.branch_head(CONFIG_ORG, agent, CONFIG_BRANCH).await
}
}
#[cfg(test)]
mod tests {
use super::*;
use forgejo_api::Auth;
fn source(nix_credential: bool) -> ForgeSource {
let url = url::Url::parse("http://forge.test").expect("url");
ForgeSource {
api: Forgejo::new(Auth::None, url).expect("client"),
nix_credential,
}
}
/// The control (`true`) shows the answer follows the field rather than
/// being a constant.
#[test]
fn nix_fetches_only_where_its_credential_was_published() {
assert!(source(true).nix_can_fetch());
assert!(!source(false).nix_can_fetch());
}
}

244
hive-c0re/src/job_queue/exec.rs
View file

@ -9,6 +9,7 @@ use std::path::PathBuf;
use std::sync::{Arc, Mutex, OnceLock};
use anyhow::{Context as _, Result};
use swarm_wire::agent_closures::AgentClosure;
use hive_jobq::NodeId;
@ -34,7 +35,7 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
const PAUSE_ACK_TIMEOUT: std::time::Duration = GRACEFUL_STOP_TIMEOUT;
/// The container-state calls the executors below make on the host: systemd,
/// `hive-priv`, the nix build, the `internal/meta` record. Production uses
/// `hive-priv`, the swarm's closure record, the `internal/meta` record. Production uses
/// [`RealHost`], which forwards to `lifecycle` / `meta_rev` / `Coordinator`
/// unchanged; tests substitute a recording fake.
pub(super) trait Host: Sync {
@ -44,28 +45,19 @@ pub(super) trait Host: Sync {
async fn kill(&self, name: &str) -> Result<()>;
/// The `internal/meta` commit `MetaSync` recorded for the agent.
fn recorded_rev(&self, name: &str) -> Result<String>;
/// The store path of the closure the swarm built for the agent at the
/// `internal/meta` commit `rev` and pushed to its cache. The error says why
/// there is none.
async fn cached_closure(&self, name: &str, rev: &str) -> Result<String>;
/// The record the swarm's builder wrote for the agent; `None` when it wrote
/// none.
async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>>;
/// Download `store_path` from the swarm's cache into the store, never
/// building it.
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()>;
/// Warm the agent's toplevel build from the `internal/meta` commit `rev`.
/// The toplevel it printed, when it printed one.
async fn prebuild_toplevel(
&self,
name: &str,
rev: &str,
node_id: u64,
) -> Result<Option<String>>;
/// Re-apply drop-ins, then `nixos-container update` at `rev`.
/// Re-apply drop-ins, then point the container at the closure `toplevel`.
async fn swap_update(
&self,
name: &str,
hive: &HiveEnv,
paths: &AgentPaths,
rev: &str,
toplevel: &str,
node_id: u64,
) -> Result<()>;
/// Write or remove the agent's pause marker.
@ -80,7 +72,8 @@ pub(super) trait Host: Sync {
async fn refresh_agent_set(&self) -> Result<()>;
/// Pin the `internal/meta` commit the agent builds from.
async fn resolve_meta(&self, name: &str) -> Result<MetaBuild>;
/// Record `build` as what `Prebuild` and `Swap` build from.
/// Record `build` as the commit `Prebuild` and `Swap` expect the swarm's
/// closure to be built from.
fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()>;
/// The build `MetaSync` recorded, if any.
fn recorded_build(&self, name: &str) -> Option<MetaBuild>;
@ -88,8 +81,8 @@ pub(super) trait Host: Sync {
fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()>;
/// First-spawn provisioning: repos, state subvolume, agent-set refresh.
async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()>;
/// `nixos-container create` at `rev`.
async fn create_only(&self, name: &str, rev: &str) -> Result<()>;
/// `nixos-container create` from the closure `toplevel`.
async fn create_only(&self, name: &str, toplevel: &str) -> Result<()>;
/// Write the nspawn and resource-limit drop-ins, then reload systemd.
async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>;
/// Start preamble, then start with the cold-start fallback.
@ -128,33 +121,23 @@ impl Host for RealHost {
Ok(crate::meta_rev::recorded_or_err(name)?.rev)
}
async fn cached_closure(&self, name: &str, rev: &str) -> Result<String> {
crate::attic::closure::path_for(name, rev).await
async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>> {
crate::attic::closure::record(name).await
}
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
crate::priv_client::realise_closure(name, store_path).await
}
async fn prebuild_toplevel(
&self,
name: &str,
rev: &str,
node_id: u64,
) -> Result<Option<String>> {
let flake_ref = crate::meta_rev::flake_ref(name, rev);
crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(node_id)).await
}
async fn swap_update(
&self,
name: &str,
hive: &HiveEnv,
paths: &AgentPaths,
rev: &str,
toplevel: &str,
node_id: u64,
) -> Result<()> {
crate::lifecycle::swap_update(name, hive, paths, rev, Some(node_id)).await
crate::lifecycle::swap_update(name, hive, paths, toplevel, Some(node_id)).await
}
async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> {
@ -198,8 +181,8 @@ impl Host for RealHost {
crate::lifecycle::provision_container(name, paths).await
}
async fn create_only(&self, name: &str, rev: &str) -> Result<()> {
crate::lifecycle::create_only(name, rev).await
async fn create_only(&self, name: &str, toplevel: &str) -> Result<()> {
crate::lifecycle::create_only(name, toplevel).await
}
async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> {
@ -300,11 +283,11 @@ async fn dispatch(
// growth executors return what to grow rather than taking the builder.
let result = match kind {
NodeKind::MetaSync { .. } => run_meta_sync(host, agent).await,
NodeKind::Prebuild { .. } => run_prebuild(host, agent, id).await,
NodeKind::Prebuild { .. } => run_prebuild(host, agent).await,
NodeKind::Swap { .. } => run_swap(host, coord, agent, id).await,
NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(host, coord, agent).await,
NodeKind::Provision { .. } => run_provision(host, agent).await,
NodeKind::Create { .. } => run_create(host, agent, id).await,
NodeKind::Create { .. } => run_create(host, agent).await,
NodeKind::DestroyContainer { .. } => run_destroy_container(host, coord, agent).await,
NodeKind::PurgeState { .. } => {
run_purge_state(host, agent).await;
@ -533,10 +516,11 @@ fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()>
}
/// The rebuild's preamble: runtime-dir prep, the agent-set refresh, and the
/// `internal/meta` commit `Prebuild` and `Swap` build from.
/// `internal/meta` commit `Prebuild` and `Swap` expect the swarm's closure to
/// be built from.
///
/// Fails when `internal/meta` does not pin the agent's current config yet:
/// there is no other flake to build it from.
/// there is no commit to deploy.
async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
// Runs while the agent is still up — the runtime dir and MCP listener
// already exist. Use the pure path accessor; no need to re-register the
@ -550,98 +534,95 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
}
host.prepare_rebuild_dirs(name, &paths).await?;
host.refresh_agent_set().await?;
// Resolved once here so `Prebuild` and `Swap` build the same flake.
// Resolved once here so `Prebuild` and `Swap` expect the same commit.
let build = host.resolve_meta(name).await?;
host.record_meta(name, &build)?;
Ok(())
}
/// Out-of-band toplevel build while the container keeps serving: warm
/// `system.build.toplevel` so the later `Swap` hits cache and skips
/// straight to the profile-swap, from the `internal/meta` commit the upstream
/// `MetaSync` node recorded. The warm build is skipped when the
/// container is already down: its only purpose is to shrink the swap's
/// downtime window, so a stopped agent (no uptime to preserve) doesn't
/// pay the double eval — `Swap` builds inline instead.
///
/// The swarm's cache is tried first: when the swarm built this agent at the
/// recorded commit and pushed the closure, it is downloaded. The local build
/// runs either way. After a download it is an evaluation and a no-op build when
/// this hive evaluates the path the swarm built, and builds it, with the agent
/// still up, when it doesn't; `Swap` never meets a build it didn't expect.
async fn run_prebuild(host: &impl Host, name: &str, id: NodeId) -> Result<()> {
// Warm the toplevel build only when the container is up — the whole
// point of prebuild is to shrink the swap's downtime window. A
// stopped agent has no uptime to preserve, so skip the (expensive)
// eval and let the downstream `Swap` build inline.
/// Download the swarm's closure for the agent while the container keeps
/// serving, so a rebuild with no closure fails here, before `StopForUpdate`
/// takes a healthy container down. `Swap` downloads the same closure again
/// (a no-op once it is in the store) and applies it. Skipped when the container
/// is already down: there is no uptime to protect, and `Swap` fails the same way.
async fn run_prebuild(host: &impl Host, name: &str) -> Result<()> {
if host.is_running(name).await {
let rev = host.recorded_rev(name)?;
let downloaded = match substitute(host, name, &rev).await {
Ok(path) => Some(path),
Err(fallback) => {
let reason = &fallback.reason;
if fallback.signature_refused {
tracing::error!(%name, %rev, "prebuild: built locally: {reason}");
} else {
tracing::warn!(%name, %rev, "prebuild: built locally: {reason}");
}
None
}
};
let evaluated = host.prebuild_toplevel(name, &rev, id.get()).await?;
if let Some(path) = downloaded {
match drift(&path, evaluated.as_deref()) {
None => tracing::info!(%name, %rev, "prebuild: substituted {path}"),
Some(note) => tracing::warn!(%name, %rev, "prebuild: {note}"),
}
}
let path = closure_for(host, name, &rev, "prebuild").await?;
tracing::info!(%name, %rev, "prebuild: substituted {path}");
}
Ok(())
}
/// What to log when the closure downloaded for a rebuild isn't the one this
/// hive's evaluation produced, which means the builder and the hive disagree
/// about the agent's flake. `None` when they agree.
fn drift(downloaded: &str, evaluated: Option<&str>) -> Option<String> {
match evaluated {
Some(path) if path == downloaded => None,
Some(path) => Some(format!(
"the downloaded closure {downloaded} is not the one this hive evaluates, {path}; built the latter locally"
)),
None => Some(format!(
"downloaded {downloaded}, but the local build printed no path to compare it with"
)),
}
/// Why the swarm's closure for an agent cannot be deployed.
enum NoClosure {
/// The builder has recorded no build of the agent.
NoRecord,
/// A build is recorded, but its closure cannot be had: the record could not
/// be read, the builder did not push the closure, or the download failed.
Unpullable(String),
}
/// Why a rebuild or create builds instead of downloading the swarm's closure.
struct LocalBuild {
reason: String,
/// The cache served a path no trusted key signed.
signature_refused: bool,
}
/// Download the closure the swarm built for `name` at `rev`; its store path.
async fn substitute(host: &impl Host, name: &str, rev: &str) -> Result<String, LocalBuild> {
let path = host
.cached_closure(name, rev)
/// The store path of the closure the swarm built for `name`, downloaded into
/// the store. The deploy applies the swarm's closure whatever commit it was
/// built from: when that is not the `rev` this hive chose, it is logged.
///
/// # Errors
/// [`without_closure`]'s refusal, after logging it.
async fn closure_for(host: &impl Host, name: &str, rev: &str, step: &str) -> Result<String> {
pull_closure(host, name, rev, step)
.await
.map_err(|e| LocalBuild {
reason: format!("{e:#}"),
signature_refused: false,
})?;
host.realise_closure(name, &path).await.map_err(|e| {
let reason = format!("{e:#}");
LocalBuild {
signature_refused: reason.contains(hive_priv_sock::SIGNATURE_REFUSED),
reason,
}
})?;
Ok(path)
.map_err(|why| without_closure(name, rev, step, &why))
}
/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb),
/// `nixos-container update`, then the post-rebuild bookkeeping tail
async fn pull_closure(
host: &impl Host,
name: &str,
rev: &str,
step: &str,
) -> Result<String, NoClosure> {
let record = host
.closure_record(name)
.await
.map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?
.ok_or(NoClosure::NoRecord)?;
if !record.cached {
return Err(NoClosure::Unpullable(
"the builder did not push the closure".to_owned(),
));
}
host.realise_closure(name, &record.store_path)
.await
.map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?;
if record.meta_rev != rev {
tracing::warn!(
%name, %rev, built_from = %record.meta_rev,
"{step}: the swarm built {} from internal/meta {}, not the {rev} this hive chose; deploying the swarm's",
record.store_path, record.meta_rev
);
}
Ok(record.store_path)
}
/// What a rebuild or create does when there is no closure to deploy: it
/// fails, and nothing changes, so a running container keeps running as it is
/// and a create makes none. The failure is logged at error level with the
/// agent's name.
///
/// This is the one place that decides what an unpullable closure (the cache
/// unreachable, a TLS error, a failed download) costs. A rebuild never builds
/// locally on the hive.
fn without_closure(name: &str, rev: &str, step: &str, why: &NoClosure) -> anyhow::Error {
let reason = match why {
NoClosure::NoRecord => "the swarm has recorded no build of this agent".to_owned(),
NoClosure::Unpullable(e) => format!("the swarm's closure cannot be pulled: {e}"),
};
tracing::error!(%name, %rev, "{step}: {reason}; nothing was changed");
anyhow::anyhow!("{name}: {reason}")
}
/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb), point the
/// container at the swarm's closure, then the post-rebuild bookkeeping tail
/// (rev marker, `Rebuilt` event, forge sync, kick, rescan).
/// The recovery-start on failure is NOT here — the DAG's tail
/// `Reconcile` runs after this node terminal ok *or* fail.
@ -657,7 +638,11 @@ async fn run_swap(
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir)?;
let rev = host.recorded_rev(name)?;
let result = host.swap_update(name, &hive, &paths, &rev, id.get()).await;
let toplevel = closure_for(host, name, &rev, "swap").await?;
tracing::info!(%name, %rev, "swap: applying {toplevel}");
let result = host
.swap_update(name, &hive, &paths, &toplevel, id.get())
.await;
// On success the Ok-only bookkeeping tail (rev marker, forge
// sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node,
// which deps `AfterOk(Swap)`. On failure `RebuildBookkeeping` is cancel-cascaded
@ -719,30 +704,13 @@ async fn run_provision(host: &impl Host, name: &str) -> Result<()> {
/// node stays purely "create", not "create + start". Marks the agent as
/// running the pins it was created from.
///
/// The swarm's cache is tried first, as in [`run_prebuild`]: when it holds the
/// closure for the resolved commit it is downloaded, and the local evaluation
/// that follows is a no-op build on a path match and builds the evaluated path
/// on a mismatch. Without a cached closure the create builds it locally.
async fn run_create(host: &impl Host, name: &str, id: NodeId) -> Result<()> {
/// The container is created from the swarm's closure for the resolved commit;
/// with none to deploy the create fails and makes no container.
async fn run_create(host: &impl Host, name: &str) -> Result<()> {
let build = host.resolve_meta(name).await?;
match substitute(host, name, &build.rev).await {
Ok(path) => {
let evaluated = host.prebuild_toplevel(name, &build.rev, id.get()).await?;
match drift(&path, evaluated.as_deref()) {
None => tracing::info!(%name, rev = %build.rev, "create: substituted {path}"),
Some(note) => tracing::warn!(%name, rev = %build.rev, "create: {note}"),
}
}
Err(fallback) => {
let reason = &fallback.reason;
if fallback.signature_refused {
tracing::error!(%name, rev = %build.rev, "create: built locally: {reason}");
} else {
tracing::warn!(%name, rev = %build.rev, "create: built locally: {reason}");
}
}
}
host.create_only(name, &build.rev).await?;
let toplevel = closure_for(host, name, &build.rev, "create").await?;
tracing::info!(%name, rev = %build.rev, "create: substituted {toplevel}");
host.create_only(name, &toplevel).await?;
host.mark_deployed(name, &build)?;
Ok(())
}

521
hive-c0re/src/job_queue/exec/tests.rs
View file

@ -11,16 +11,17 @@ use hive_jobq::NodeId;
use hive_jobq::scheduler::Scheduler;
use super::{
Host, dispatch, drift, run_create, run_destroy_bookkeeping, run_destroy_container, run_drain,
Host, dispatch, run_create, run_destroy_bookkeeping, run_destroy_container, run_drain,
run_emit_rebuilt, run_forge_sweep, run_knowledge_pull, run_meta_sync, run_pause_drain,
run_pause_signal, run_prebuild, run_provision, run_purge_state, run_rebuild_bookkeeping,
run_reconcile, run_set_wanted, run_signal, run_start, run_stop, run_stop_for_update, run_swap,
run_wanted_pull, run_write_dropin, substitute,
run_wanted_pull, run_write_dropin,
};
use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
use crate::job_queue::model::NodeKind;
use crate::meta_rev::MetaBuild;
use crate::power::Wanted;
use swarm_wire::agent_closures::AgentClosure;
const REV: &str = "2222222222222222222222222222222222222222";
const CLOSURE: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-alice";
@ -30,18 +31,12 @@ const CLOSURE: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-
struct FakeHost {
running: bool,
rev: Option<String>,
/// The store path the swarm's record names, when it has a pushed build of
/// `REV`; `None` reads as no record.
closure: Option<String>,
/// A failing `realise` says nix refused the path for want of a signature.
refusal: bool,
/// The toplevel this hive's evaluation produces; `None` is `CLOSURE`.
evaluates: Option<String>,
/// What the swarm's builder recorded for the agent; `None` reads as no
/// record.
built: Option<AgentClosure>,
/// Store paths a download put in the store.
realised: Mutex<Vec<String>>,
/// Store paths a prebuild had to build because the store lacked them.
builds: Mutex<Vec<String>>,
/// Names of the calls that fail (`kill`, `realise`, `prebuild`, `swap_update`,
/// Names of the calls that fail (`kill`, `realise`, `closure_record`, `swap_update`,
/// `set_paused`, ...). `recorded_build` fails by finding nothing.
failing: Vec<&'static str>,
/// The sweeps and pulls never finish, so only a deadline ends them.
@ -56,6 +51,7 @@ impl FakeHost {
Self {
running: true,
rev: Some(REV.to_owned()),
built: Some(closure(REV, true)),
..Self::default()
}
}
@ -64,12 +60,19 @@ impl FakeHost {
Self {
running: false,
rev: Some(REV.to_owned()),
built: Some(closure(REV, true)),
..Self::default()
}
}
fn with_closure(mut self) -> Self {
self.closure = Some(CLOSURE.to_owned());
/// The builder recorded `meta_rev` for the agent, pushed or not.
fn built(mut self, meta_rev: &str, cached: bool) -> Self {
self.built = Some(closure(meta_rev, cached));
self
}
fn without_record(mut self) -> Self {
self.built = None;
self
}
@ -98,10 +101,6 @@ impl FakeHost {
fn calls(&self) -> Vec<String> {
self.calls.lock().unwrap().clone()
}
fn builds(&self) -> Vec<String> {
self.builds.lock().unwrap().clone()
}
}
impl Host for FakeHost {
@ -125,54 +124,29 @@ impl Host for FakeHost {
.ok_or_else(|| anyhow::anyhow!("{name}: no internal/meta commit was recorded"))
}
async fn cached_closure(&self, name: &str, rev: &str) -> anyhow::Result<String> {
self.record(format!("cached_closure {name} {rev}"));
self.closure
.clone()
.ok_or_else(|| anyhow::anyhow!("no-record: no build is recorded for the agent"))
async fn closure_record(&self, name: &str) -> anyhow::Result<Option<AgentClosure>> {
self.step("closure_record", format!("closure_record {name}"))?;
Ok(self.built.clone())
}
async fn realise_closure(&self, name: &str, store_path: &str) -> anyhow::Result<()> {
self.record(format!("realise {name} {store_path}"));
if self.fails("realise") {
if self.refusal {
anyhow::bail!(
"not available: {}: lacks a signature",
hive_priv_sock::SIGNATURE_REFUSED
);
}
anyhow::bail!("not available: Couldn't connect to server");
}
self.realised.lock().unwrap().push(store_path.to_owned());
Ok(())
}
async fn prebuild_toplevel(
&self,
name: &str,
rev: &str,
node_id: u64,
) -> anyhow::Result<Option<String>> {
self.record(format!("prebuild {name} {rev} node={node_id}"));
if self.fails("prebuild") {
anyhow::bail!("prebuild failed");
}
let path = self.evaluates.clone().unwrap_or_else(|| CLOSURE.to_owned());
if !self.realised.lock().unwrap().contains(&path) {
self.builds.lock().unwrap().push(path.clone());
}
Ok(Some(path))
}
async fn swap_update(
&self,
name: &str,
_hive: &HiveEnv,
_paths: &AgentPaths,
rev: &str,
toplevel: &str,
node_id: u64,
) -> anyhow::Result<()> {
self.record(format!("swap_update {name} {rev} node={node_id}"));
self.record(format!("swap_update {name} {toplevel} node={node_id}"));
if self.fails("swap_update") {
anyhow::bail!("nixos-container update failed");
}
@ -231,8 +205,8 @@ impl Host for FakeHost {
self.step("provision", format!("provision {name}"))
}
async fn create_only(&self, name: &str, rev: &str) -> anyhow::Result<()> {
self.step("create_only", format!("create_only {name} {rev}"))
async fn create_only(&self, name: &str, toplevel: &str) -> anyhow::Result<()> {
self.step("create_only", format!("create_only {name} {toplevel}"))
}
async fn write_dropins(
@ -287,6 +261,17 @@ impl Host for FakeHost {
}
}
/// The builder's record of a closure built from `internal/meta` at `meta_rev`.
fn closure(meta_rev: &str, cached: bool) -> AgentClosure {
AgentClosure {
meta_rev: meta_rev.to_owned(),
store_path: CLOSURE.to_owned(),
system: "x86_64-linux".to_owned(),
built_unix: 1,
cached,
}
}
fn build() -> MetaBuild {
MetaBuild {
rev: REV.to_owned(),
@ -323,188 +308,185 @@ fn node_id() -> NodeId {
// --- Prebuild ---------------------------------------------------------------
const OTHER_REV: &str = "3333333333333333333333333333333333333333";
/// What a block logged, level and text: the message and each field.
#[derive(Clone, Default)]
struct Logged(Arc<Mutex<Vec<(tracing::Level, String)>>>);
impl tracing::Subscriber for Logged {
fn enabled(&self, _: &tracing::Metadata<'_>) -> bool {
true
}
fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id {
tracing::span::Id::from_u64(1)
}
fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {}
fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {}
fn event(&self, event: &tracing::Event<'_>) {
use std::fmt::Write as _;
let mut text = String::new();
event.record(
&mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| {
let _ = write!(text, "{}={value:?} ", field.name());
},
);
self.0
.lock()
.unwrap()
.push((*event.metadata().level(), text));
}
fn enter(&self, _: &tracing::span::Id) {}
fn exit(&self, _: &tracing::span::Id) {}
}
impl Logged {
fn at(&self, level: tracing::Level) -> Vec<String> {
self.0
.lock()
.unwrap()
.iter()
.filter(|(l, _)| *l == level)
.map(|(_, text)| text.clone())
.collect()
}
}
/// `fut`'s output and what it logged. The test runtime is single-threaded, so
/// the subscriber set here sees every event the future emits.
async fn logging<T>(fut: impl std::future::Future<Output = T>) -> (T, Logged) {
let logged = Logged::default();
let _guard = tracing::subscriber::set_default(logged.clone());
let out = fut.await;
(out, logged)
}
/// Every way the swarm's closure for `alice` cannot be had, with the text the
/// error log must carry and the host calls that must not happen.
fn without_a_closure() -> [(&'static str, FakeHost, &'static str); 4] {
[
("no record", FakeHost::up().without_record(), "no build"),
(
"not pushed",
FakeHost::up().built(REV, false),
"did not push",
),
(
"download fails",
FakeHost::up().failing("realise"),
"Couldn't connect",
),
(
"record unreadable",
FakeHost::up().failing("closure_record"),
"closure_record failed",
),
]
}
#[tokio::test]
async fn prebuild_warms_the_recorded_rev_when_the_container_is_up() {
async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() {
let host = FakeHost::up();
let id = node_id();
run_prebuild(&host, "alice", id).await.expect("prebuild ok");
let (result, log) = logging(run_prebuild(&host, "alice")).await;
result.expect("prebuild ok");
assert_eq!(
host.calls(),
[
"is_running alice".to_owned(),
"recorded_rev alice".to_owned(),
format!("cached_closure alice {REV}"),
format!("prebuild alice {REV} node={}", id.get()),
]
);
}
/// The download is followed by the local build either way. When this hive
/// evaluates the path the swarm built, the build finds it in the store.
#[tokio::test]
async fn a_downloaded_closure_the_hive_also_evaluates_is_found_not_built() {
let host = FakeHost::up().with_closure();
let id = node_id();
run_prebuild(&host, "alice", id).await.expect("prebuild ok");
assert_eq!(
host.calls(),
[
"is_running alice".to_owned(),
"recorded_rev alice".to_owned(),
format!("cached_closure alice {REV}"),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("prebuild alice {REV} node={}", id.get()),
]
);
assert_eq!(host.builds(), Vec::<String>::new());
}
/// When it evaluates another path, the build happens in `Prebuild`, after the
/// check that the agent is up and before `Swap` could stop it.
#[tokio::test]
async fn a_downloaded_closure_the_hive_does_not_evaluate_is_built_before_the_swap() {
let other = "/nix/store/zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-nixos-system-alice";
let host = FakeHost {
evaluates: Some(other.to_owned()),
..FakeHost::up().with_closure()
};
let id = node_id();
run_prebuild(&host, "alice", id).await.expect("prebuild ok");
let calls = host.calls();
assert_eq!(calls[0], "is_running alice", "{calls:?}");
assert_eq!(
calls.last().map(String::as_str),
Some(format!("prebuild alice {REV} node={}", id.get()).as_str()),
"{calls:?}"
);
assert_eq!(host.builds(), [other.to_owned()]);
}
#[test]
fn only_a_different_path_is_drift() {
assert_eq!(drift(CLOSURE, Some(CLOSURE)), None);
let note = drift(CLOSURE, Some("/nix/store/other")).expect("drift");
let info = log.at(tracing::Level::INFO);
assert!(
note.contains(CLOSURE) && note.contains("/nix/store/other"),
"{note}"
info.iter()
.any(|l| l.contains("prebuild: substituted") && l.contains(CLOSURE)),
"{info:?}"
);
assert!(drift(CLOSURE, None).expect("no path").contains(CLOSURE));
assert!(log.at(tracing::Level::WARN).is_empty());
assert!(log.at(tracing::Level::ERROR).is_empty());
}
/// No closure to deploy fails the node before `StopForUpdate`, so the running
/// container is untouched, and the log names the agent and why.
#[tokio::test]
async fn a_failed_download_builds_locally() {
let host = FakeHost::up().with_closure().failing("realise");
let id = node_id();
run_prebuild(&host, "alice", id)
.await
.expect("the build covers it");
let calls = host.calls();
assert_eq!(calls[3], format!("realise alice {CLOSURE}"), "{calls:?}");
assert_eq!(
calls[4],
format!("prebuild alice {REV} node={}", id.get()),
"{calls:?}"
);
assert_eq!(calls.len(), 5, "{calls:?}");
async fn a_rebuild_without_a_closure_fails_the_prebuild_and_logs_an_error() {
for (case, host, why) in without_a_closure() {
let (result, log) = logging(run_prebuild(&host, "alice")).await;
let err = result.expect_err(case);
assert!(format!("{err:#}").starts_with("alice: "), "{case}: {err:#}");
let errors = log.at(tracing::Level::ERROR);
assert_eq!(errors.len(), 1, "{case}: {errors:?}");
assert!(
errors[0].contains("alice") && errors[0].contains(why),
"{case}: {errors:?}"
);
let calls = host.calls();
assert!(
!calls.iter().any(|c| c.starts_with("kill")),
"{case}: {calls:?}"
);
assert!(
!calls
.iter()
.any(|c| c.starts_with("swap_update") || c.starts_with("create_only")),
"{case}: {calls:?}"
);
}
}
/// The fallback is the build, so a build that fails still fails the node.
/// The builder's record is not for the commit this hive chose: its closure is
/// deployed and the difference is a warning.
#[tokio::test]
async fn a_failed_download_and_a_failed_build_fail_the_node() {
let host = FakeHost::up()
.with_closure()
.failing("realise")
.failing("prebuild");
let err = run_prebuild(&host, "alice", node_id())
.await
.expect_err("nothing produced the closure");
assert!(format!("{err:#}").contains("prebuild failed"), "{err:#}");
async fn a_closure_built_from_another_commit_is_deployed_with_a_warning() {
let host = FakeHost::up().built(OTHER_REV, true);
let (result, log) = logging(run_prebuild(&host, "alice")).await;
result.expect("drift is not a refusal");
assert!(
host.calls().contains(&format!("realise alice {CLOSURE}")),
"{:?}",
host.calls()
);
let warnings = log.at(tracing::Level::WARN);
assert_eq!(warnings.len(), 1, "{warnings:?}");
assert!(
[REV, OTHER_REV, CLOSURE]
.iter()
.all(|part| warnings[0].contains(part)),
"{warnings:?}"
);
assert!(log.at(tracing::Level::ERROR).is_empty());
}
#[tokio::test]
async fn a_stopped_container_looks_up_nothing() {
let host = FakeHost::down().with_closure();
run_prebuild(&host, "alice", node_id())
.await
.expect("skipped");
assert_eq!(host.calls(), ["is_running alice"]);
}
/// The error log is for a path the cache served unsigned, not for a cache that
/// is down or a missing record.
#[tokio::test]
async fn only_a_signature_refusal_is_flagged() {
let down = substitute(
&FakeHost::up().with_closure().failing("realise"),
"alice",
REV,
)
.await
.expect_err("down");
assert!(!down.signature_refused, "{}", down.reason);
assert!(down.reason.contains("Couldn't connect"), "{}", down.reason);
let no_record = substitute(&FakeHost::up(), "alice", REV)
.await
.expect_err("no record");
assert!(!no_record.signature_refused);
assert!(
no_record.reason.starts_with("no-record"),
"{}",
no_record.reason
);
let refused = FakeHost {
refusal: true,
..FakeHost::up().with_closure().failing("realise")
};
let refused = substitute(&refused, "alice", REV)
.await
.expect_err("refused");
assert!(refused.signature_refused, "{}", refused.reason);
let ok = substitute(&FakeHost::up().with_closure(), "alice", REV).await;
assert_eq!(ok.ok().as_deref(), Some(CLOSURE));
}
#[tokio::test]
async fn prebuild_is_skipped_when_the_container_is_down() {
let host = FakeHost::down();
run_prebuild(&host, "alice", node_id())
.await
.expect("a skipped prebuild is ok");
run_prebuild(&host, "alice").await.expect("skipped");
assert_eq!(host.calls(), ["is_running alice"]);
}
#[tokio::test]
async fn prebuild_without_a_recorded_rev_fails_before_building() {
async fn prebuild_without_a_recorded_rev_fails_before_looking_up_a_closure() {
let host = FakeHost {
rev: None,
..FakeHost::up()
};
let err = run_prebuild(&host, "alice", node_id())
let err = run_prebuild(&host, "alice")
.await
.expect_err("nothing to build from");
.expect_err("nothing to deploy");
assert!(
format!("{err:#}").contains("no internal/meta commit"),
"{err:#}"
);
assert!(!host.calls().iter().any(|c| c.starts_with("prebuild")));
}
#[tokio::test]
async fn prebuild_failure_fails_the_node() {
let host = FakeHost::up().failing("prebuild");
let err = run_prebuild(&host, "alice", node_id())
.await
.expect_err("build failure propagates");
assert!(format!("{err:#}").contains("prebuild failed"), "{err:#}");
assert!(!host.calls().iter().any(|c| c.starts_with("closure_record")));
}
// --- Swap -------------------------------------------------------------------
#[tokio::test]
async fn swap_updates_at_the_recorded_rev() {
async fn swap_applies_the_swarms_closure() {
let (_dir, coord) = coordinator();
let host = FakeHost::down();
let id = node_id();
@ -513,11 +495,50 @@ async fn swap_updates_at_the_recorded_rev() {
host.calls(),
[
"recorded_rev alice".to_owned(),
format!("swap_update alice {REV} node={}", id.get()),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("swap_update alice {CLOSURE} node={}", id.get()),
]
);
}
/// A swap with no closure to deploy never updates: the container stays as it
/// was, and the log names the agent and why.
#[tokio::test]
async fn a_swap_without_a_closure_never_updates_and_logs_an_error() {
for (case, host, why) in without_a_closure() {
let (_dir, coord) = coordinator();
let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await;
result.expect_err(case);
let errors = log.at(tracing::Level::ERROR);
assert_eq!(errors.len(), 1, "{case}: {errors:?}");
assert!(
errors[0].contains("alice") && errors[0].contains(why),
"{case}: {errors:?}"
);
assert!(
!host.calls().iter().any(|c| c.starts_with("swap_update")),
"{case}: {:?}",
host.calls()
);
}
}
#[tokio::test]
async fn swap_applies_a_closure_built_from_another_commit_with_a_warning() {
let (_dir, coord) = coordinator();
let host = FakeHost::down().built(OTHER_REV, true);
let id = node_id();
let (result, log) = logging(run_swap(&host, &coord, "alice", id)).await;
result.expect("drift is not a refusal");
assert_eq!(
host.calls().last(),
Some(&format!("swap_update alice {CLOSURE} node={}", id.get()))
);
assert_eq!(log.at(tracing::Level::WARN).len(), 1);
assert!(log.at(tracing::Level::ERROR).is_empty());
}
/// A failed update fails the node and nothing else: the swap never restarts or
/// stops the container itself — recovery is the DAG's tail `Reconcile`.
#[tokio::test]
@ -529,8 +550,8 @@ async fn a_failed_swap_reports_the_error_and_touches_no_container() {
.expect_err("update failure propagates");
assert!(format!("{err:#}").contains("update failed"), "{err:#}");
let calls = host.calls();
assert_eq!(calls.len(), 2, "{calls:?}");
assert!(calls[1].starts_with("swap_update alice"), "{calls:?}");
assert_eq!(calls.len(), 4, "{calls:?}");
assert!(calls[3].starts_with("swap_update alice"), "{calls:?}");
}
#[tokio::test]
@ -824,84 +845,73 @@ async fn provision_and_write_dropin_forward_and_propagate_failure() {
}
#[tokio::test]
async fn create_without_a_record_builds_locally_at_the_resolved_rev() {
async fn create_downloads_the_swarms_closure_and_creates_from_it() {
let host = FakeHost::up();
run_create(&host, "alice", node_id()).await.expect("ok");
let (result, log) = logging(run_create(&host, "alice")).await;
result.expect("ok");
assert_eq!(
host.calls(),
[
"resolve_meta alice".to_owned(),
format!("cached_closure alice {REV}"),
format!("create_only alice {REV}"),
format!("mark_deployed alice {REV}"),
]
);
}
#[tokio::test]
async fn create_downloads_a_cached_closure_and_builds_nothing() {
let host = FakeHost::up().with_closure();
let id = node_id();
run_create(&host, "alice", id).await.expect("ok");
assert_eq!(
host.calls(),
[
"resolve_meta alice".to_owned(),
format!("cached_closure alice {REV}"),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("prebuild alice {REV} node={}", id.get()),
format!("create_only alice {REV}"),
format!("create_only alice {CLOSURE}"),
format!("mark_deployed alice {REV}"),
]
);
assert_eq!(host.builds(), Vec::<String>::new());
}
#[tokio::test]
async fn create_builds_locally_when_the_download_fails() {
let host = FakeHost::up().with_closure().failing("realise");
run_create(&host, "alice", node_id()).await.expect("ok");
assert_eq!(
host.calls(),
[
"resolve_meta alice".to_owned(),
format!("cached_closure alice {REV}"),
format!("realise alice {CLOSURE}"),
format!("create_only alice {REV}"),
format!("mark_deployed alice {REV}"),
]
assert!(
log.at(tracing::Level::INFO)
.iter()
.any(|l| l.contains("create: substituted") && l.contains(CLOSURE))
);
assert!(log.at(tracing::Level::WARN).is_empty());
}
/// A create with no closure to deploy is refused: no container is made, none is
/// marked deployed, and the log names the agent and why.
#[tokio::test]
async fn a_create_without_a_closure_is_refused_and_logs_an_error() {
for (case, host, why) in without_a_closure() {
let (result, log) = logging(run_create(&host, "alice")).await;
let err = result.expect_err(case);
assert!(format!("{err:#}").starts_with("alice: "), "{case}: {err:#}");
let errors = log.at(tracing::Level::ERROR);
assert_eq!(errors.len(), 1, "{case}: {errors:?}");
assert!(
errors[0].contains("alice") && errors[0].contains(why),
"{case}: {errors:?}"
);
let calls = host.calls();
assert!(
!calls
.iter()
.any(|c| c.starts_with("create_only") || c.starts_with("mark_deployed")),
"{case}: {calls:?}"
);
}
}
#[tokio::test]
async fn create_builds_the_evaluated_path_when_it_differs_from_the_download() {
let other = "/nix/store/zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-nixos-system-alice";
let host = FakeHost {
evaluates: Some(other.to_owned()),
..FakeHost::up().with_closure()
};
run_create(&host, "alice", node_id()).await.expect("ok");
assert_eq!(host.builds(), [other.to_owned()]);
async fn create_uses_a_closure_built_from_another_commit_with_a_warning() {
let host = FakeHost::up().built(OTHER_REV, true);
let (result, log) = logging(run_create(&host, "alice")).await;
result.expect("drift is not a refusal");
let calls = host.calls();
assert!(
calls
.iter()
.any(|c| c == &format!("create_only alice {REV}")),
calls.contains(&format!("create_only alice {CLOSURE}")),
"{calls:?}"
);
assert_eq!(log.at(tracing::Level::WARN).len(), 1);
assert!(log.at(tracing::Level::ERROR).is_empty());
}
#[tokio::test]
async fn a_failed_create_is_never_marked_deployed() {
let host = FakeHost::up().failing("create_only");
run_create(&host, "alice", node_id())
.await
.expect_err("create fails");
assert_eq!(host.calls().len(), 3);
run_create(&host, "alice").await.expect_err("create fails");
assert_eq!(host.calls().len(), 4);
let host = FakeHost::up().failing("resolve_meta");
run_create(&host, "alice", node_id())
.await
.expect_err("resolve fails");
run_create(&host, "alice").await.expect_err("resolve fails");
assert_eq!(host.calls(), ["resolve_meta alice"]);
}
@ -1165,11 +1175,19 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
),
(
alice(|agent| NodeKind::Prebuild { agent }),
&["is_running alice", "recorded_rev alice", "prebuild alice"],
&[
"is_running alice",
"recorded_rev alice",
"closure_record alice",
],
),
(
alice(|agent| NodeKind::Swap { agent }),
&["recorded_rev alice", "swap_update alice"],
&[
"recorded_rev alice",
"closure_record alice",
"swap_update alice",
],
),
(
alice(|agent| NodeKind::RebuildBookkeeping { agent }),
@ -1183,6 +1201,7 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
alice(|agent| NodeKind::Create { agent }),
&[
"resolve_meta alice",
"closure_record alice",
"create_only alice",
"mark_deployed alice",
],

4
hive-c0re/src/job_queue/model.rs
View file

@ -32,9 +32,9 @@ pub enum NodeKind {
/// The rebuild's preamble: runtime-dir prep and the `internal/meta`
/// commit the build reads.
MetaSync { agent: String },
/// Out-of-band toplevel build while the container keeps serving.
/// Download of the swarm's closure while the container keeps serving.
Prebuild { agent: String },
/// `nixos-container update` profile-swap.
/// Profile-swap onto the swarm's closure.
Swap { agent: String },
/// The post-`Swap` bookkeeping tail as a first-class node.
RebuildBookkeeping { agent: String },

6
hive-c0re/src/job_queue/templates.rs
View file

@ -190,7 +190,7 @@ fn rebuild_subtree<'a>(
.needs(Resource::Agent(a()))
.after_ok(meta_sync);
// Siblings under the brace: the build and the quiesce chain run
// Siblings under the brace: the download and the quiesce chain run
// concurrently, borrowing the brace's grants rather than declaring their
// own. Declaring the lease on both would make them mutually exclusive —
// it is single-unit — which is exactly what this shape exists to avoid.
@ -200,8 +200,8 @@ fn rebuild_subtree<'a>(
let drain = graceful.then(|| quiesce(builder, agent, agent_window));
// The container goes down here, not earlier: `AfterOk` the build so a
// failed build never stops a healthy container, and `AfterOk` the drain so
// The container goes down here, not earlier: `AfterOk` the download so a
// missing closure never stops a healthy container, and `AfterOk` the drain so
// the agent has checkpointed.
let mut stop_for_update = builder
.node(NodeKind::StopForUpdate { agent: a() })

233
hive-c0re/src/lifecycle/mod.rs
View file

@ -234,9 +234,9 @@ pub async fn provision_container(name: &str, paths: &AgentPaths) -> Result<()> {
/// The `nixos-container create` proper — the job queue's `Create` node.
/// Assumes `provision_container` already registered the agent.
/// `meta_rev` is the `internal/meta` commit to build from (`crate::meta_rev`).
pub async fn create_only(name: &str, meta_rev: &str) -> Result<()> {
priv_run_inner("create", name, Some(meta_rev), None).await
/// `toplevel` is the store path of the system closure to create it from.
pub async fn create_only(name: &str, toplevel: &str) -> Result<()> {
priv_run_inner("create", name, Some(toplevel), None).await
}
/// Rebuild-path preamble, run by the job queue's `Prebuild` node: fail fast on
@ -285,20 +285,20 @@ pub async fn prepare_rebuild_dirs(name: &str, paths: &AgentPaths) -> Result<()>
}
/// Profile-swap for an existing, stopped container: re-apply the
/// drop-ins, then `nixos-container update`. The job queue's `Swap`
/// node. Requires the container stopped (the queue's `StopForUpdate`
/// upstream); does NOT start it — the DAG's tail `Reconcile` owns
/// bringing the agent back to its wanted power state. `meta_rev` is as for
/// drop-ins, then point the container at the closure `toplevel`. The job
/// queue's `Swap` node. Requires the container stopped (the queue's
/// `StopForUpdate` upstream); does NOT start it — the DAG's tail `Reconcile` owns
/// bringing the agent back to its wanted power state. `toplevel` is as for
/// [`create_only`].
pub async fn swap_update(
name: &str,
hive: &HiveEnv,
paths: &AgentPaths,
meta_rev: &str,
toplevel: &str,
node_id: Option<u64>,
) -> Result<()> {
write_dropins(name, hive, paths).await?;
priv_run_inner("update", name, Some(meta_rev), node_id).await
priv_run_inner("update", name, Some(toplevel), node_id).await
}
/// Build the `AgentSpec` list for [`crate::agent_set::refresh`] from
@ -720,195 +720,6 @@ fn confirm_gone_after_failed_destroy(
}
}
/// Wall-clock bound on [`prebuild_toplevel`]'s `nix build`. Four times CI's
/// observed cold-cache `nix flake check` (~15 min, `.forgejo/workflows/ci.yml`),
/// which compiles this workspace's crates, the part of an agent toplevel no
/// binary cache holds. A stalled download is bounded by nix's own
/// `stalled-download-timeout`; this covers the rest, such as a wedged
/// nix-daemon holding the rebuild's build slot forever.
const PREBUILD_TIMEOUT: std::time::Duration = std::time::Duration::from_hours(1);
/// Wait for `child` until `limit`. `None` means it was still running, and its
/// whole process group was `SIGKILL`ed and the child reaped, so a grandchild
/// holding the daemon connection dies with it. The child must have been
/// spawned with `process_group(0)`.
async fn wait_bounded(
child: &mut tokio::process::Child,
limit: std::time::Duration,
) -> std::io::Result<Option<std::process::ExitStatus>> {
// Group leader, so its pid is the pgid. Read before waiting: a reaped
// child has no id.
let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());
if let Ok(status) = tokio::time::timeout(limit, child.wait()).await {
return status.map(Some);
}
if let Some(pgid) = pgid {
// SAFETY: plain `kill(2)`; `-pgid` targets the child's own process
// group, which `process_group(0)` made it lead.
unsafe {
libc::kill(-pgid, libc::SIGKILL);
}
}
child.wait().await?;
Ok(None)
}
/// Log and record a prebuild's stdout; the last non-empty line is the
/// toplevel `nix build --print-out-paths` printed.
async fn pump_prebuild_stdout(
stdout: tokio::process::ChildStdout,
cmdline: String,
logs: Option<std::sync::Arc<crate::build_logs::BuildLogs>>,
log_id: Option<i64>,
) -> Option<String> {
use tokio::io::{AsyncBufReadExt, BufReader};
let mut lines = BufReader::new(stdout).lines();
let mut out_path = None;
while let Ok(Some(line)) = lines.next_line().await {
tracing::info!(target: "nix-prebuild", cmdline = %cmdline, "{line}");
if !line.trim().is_empty() {
out_path = Some(line.trim().to_owned());
}
if let (Some(h), Some(id)) = (&logs, log_id) {
h.append_stdout(id, &line);
}
}
out_path
}
/// Pre-build `system.build.toplevel` against `meta#<name>` so the
/// subsequent `nixos-container update` finds the result cached and
/// skips straight to the profile-swap. Store-warming only — container
/// is untouched. See `docs/scheduler/coordinator.md::Rebuild path` for why
/// the prebuild happens before stop, and `docs/scheduler/coordinator.md::Prebuild
/// attr path` for why the explicit nixosConfigurations attr is required.
///
/// `node_id` is the queue node this build belongs to, when there is one —
/// it is stored on the `build_logs` row so the dashboard can find the log
/// from the node. Pass `None` for builds that run outside the queue.
///
/// Returns the toplevel `nix build` printed, `None` if it printed nothing.
pub async fn prebuild_toplevel(
name: &str,
flake_ref: &str,
node_id: Option<u64>,
) -> Result<Option<String>> {
use tokio::io::{AsyncBufReadExt, BufReader};
// Split `<root>#<name>` so we can re-emit with the explicit
// `nixosConfigurations.<name>` segment. The flake_ref shape is
// constructed by the caller and always contains exactly one
// `#`; `split_once` returning None here would be a programmer
// error we'd want to surface loudly rather than paper over.
let (flake_root, fragment) = flake_ref
.split_once('#')
.with_context(|| format!("flake_ref {flake_ref:?} missing '#<name>' fragment"))?;
// Sanity-check the fragment matches the agent name we were
// passed — guards against future calls that pass a divergent
// pair (no current callsite does, but the pair is redundant
// and worth checking once).
if fragment != name {
anyhow::bail!("prebuild_toplevel: flake_ref fragment '{fragment}' ≠ agent name '{name}'");
}
let attr = format!("{flake_root}#nixosConfigurations.{name}.config.system.build.toplevel");
let args = vec![
"--extra-experimental-features",
"nix-command flakes",
"build",
"--no-link",
"--print-out-paths",
&attr,
];
let cmdline = format!("nix {}", args.join(" "));
tracing::info!(%name, %cmdline, "prebuild: warming system toplevel");
// Open a build_logs row for this attempt (best-effort — None when
// the global handle hasn't been installed, e.g. early startup
// or standalone tests). Lines pumped from stdout/stderr append
// into the row; `finish` lands the terminal status before we bail.
let logs = crate::build_logs::global();
let log_id = logs.as_ref().and_then(|h| {
h.start(name, "prebuild", &cmdline, node_id)
.map_err(|e| {
tracing::warn!(error = ?e, "build_logs: start failed (prebuild log dropped)");
})
.ok()
});
let mut child = Command::new("nix")
.args(&args)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.process_group(0)
.spawn()
.with_context(|| format!("spawn {cmdline}"))?;
let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr");
let pump_stdout = tokio::spawn(pump_prebuild_stdout(
stdout,
cmdline.clone(),
logs.clone(),
log_id,
));
let stderr_cmdline = cmdline.clone();
let stderr_logs = logs.clone();
let pump_stderr = tokio::spawn(async move {
let mut lines = BufReader::new(stderr).lines();
while let Ok(Some(line)) = lines.next_line().await {
if is_actionable_nix_line(&line) {
tracing::warn!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
} else {
tracing::info!(target: "nix-prebuild", cmdline = %stderr_cmdline, "{line}");
}
if let (Some(h), Some(id)) = (&stderr_logs, log_id) {
h.append_stderr(id, &line);
}
}
});
let Some(status) = wait_bounded(&mut child, PREBUILD_TIMEOUT)
.await
.with_context(|| format!("wait {cmdline}"))?
else {
// A process that left the group can still hold the pipes open.
pump_stdout.abort();
pump_stderr.abort();
if let (Some(h), Some(id)) = (&logs, log_id) {
h.finish(id, crate::build_logs::BuildStatus::Fail);
}
let secs = PREBUILD_TIMEOUT.as_secs();
match log_id {
Some(id) => {
bail!("prebuild {cmdline} timed out after {secs}s; killed it; see build log #{id}")
}
None => bail!("prebuild {cmdline} timed out after {secs}s; killed it"),
}
};
let out_path = pump_stdout.await.ok().flatten();
let _ = pump_stderr.await;
let ok = status.success();
if let (Some(h), Some(id)) = (&logs, log_id) {
h.finish(
id,
if ok {
crate::build_logs::BuildStatus::Ok
} else {
crate::build_logs::BuildStatus::Fail
},
);
}
if !ok {
match log_id {
Some(id) => bail!("prebuild {cmdline} failed ({status}); see build log #{id}"),
None => bail!("prebuild {cmdline} failed ({status})"),
}
}
Ok(out_path)
}
pub async fn list() -> Result<Vec<String>> {
let stdout = crate::priv_client::list_containers().await?;
Ok(stdout
@ -990,11 +801,11 @@ fn make_log_callback(
}
}
/// A `create` or `update` asked for without an `internal/meta` commit. The
/// public entry points take the commit, so this guards the string dispatch
/// A `create` or `update` asked for without a closure. The
/// public entry points take the closure, so this guards the string dispatch
/// below only.
fn missing_meta_rev(kind: &str, name: &str) -> anyhow::Error {
anyhow::anyhow!("{name}: {kind} needs an internal/meta commit to build from")
fn missing_toplevel(kind: &str, name: &str) -> anyhow::Error {
anyhow::anyhow!("{name}: {kind} needs a closure to apply")
}
/// Execute a container operation via hive-priv and integrate with
@ -1016,7 +827,7 @@ async fn priv_run(kind: &str, name: &str) -> Result<()> {
async fn priv_run_inner(
kind: &str,
name: &str,
meta_rev: Option<&str>,
toplevel: Option<&str>,
node_id: Option<u64>,
) -> Result<()> {
let container = container_name(name);
@ -1034,27 +845,27 @@ async fn priv_run_inner(
// For long-running ops use the streaming protocol so build_logs
// receives lines in real time rather than as a batch at completion.
let result: Result<()> = match kind {
"create" => match meta_rev {
Some(rev) => {
"create" => match toplevel {
Some(toplevel) => {
crate::priv_client::create_container_streaming(
name,
rev,
toplevel,
make_log_callback(logs.clone(), log_id, cmdline.clone()),
)
.await
}
None => Err(missing_meta_rev(kind, name)),
None => Err(missing_toplevel(kind, name)),
},
"update" => match meta_rev {
Some(rev) => {
"update" => match toplevel {
Some(toplevel) => {
crate::priv_client::update_container_streaming(
name,
rev,
toplevel,
make_log_callback(logs.clone(), log_id, cmdline.clone()),
)
.await
}
None => Err(missing_meta_rev(kind, name)),
None => Err(missing_toplevel(kind, name)),
},
"start" => crate::priv_client::start_container(name).await,
"stop" => crate::priv_client::stop_container(name).await,

62
hive-c0re/src/lifecycle/tests.rs
View file

@ -349,68 +349,6 @@ fn agent_names_propagates_an_unreadable_list() {
assert!(format!("{err:#}").contains("hive-priv"), "{err:#}");
}
fn is_dead(pid: &str) -> bool {
// A killed process that is not our child stays a zombie until its own
// parent reaps it, so a zombie counts as dead.
match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
Err(_) => true,
Ok(stat) => stat
.rsplit_once(") ")
.is_some_and(|(_, rest)| rest.starts_with('Z')),
}
}
/// A child still running at the limit comes back `None`, and the kill reaches
/// its whole process group: the backgrounded grandchild dies too, not just the
/// shell that started it.
#[tokio::test]
async fn a_prebuild_past_its_limit_is_killed_with_its_group() {
use std::time::{Duration, Instant};
let dir = tempfile::tempdir().expect("tempdir");
let pidfile = dir.path().join("grandchild");
let mut child = Command::new("sh")
.arg("-c")
.arg(format!("sleep 600 & echo $! > {}; wait", pidfile.display()))
.process_group(0)
.spawn()
.expect("spawn sh");
let started = Instant::now();
let status = wait_bounded(&mut child, Duration::from_secs(2))
.await
.expect("wait_bounded");
assert!(status.is_none(), "must time out, got {status:?}");
assert!(
started.elapsed() < Duration::from_mins(1),
"the limit did not bound the wait"
);
let pid = std::fs::read_to_string(&pidfile).expect("grandchild pid");
let pid = pid.trim();
let deadline = Instant::now() + Duration::from_secs(5);
while !is_dead(pid) && Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(is_dead(pid), "grandchild {pid} outlived the timeout");
}
/// The control: a child that exits inside its limit comes back with its
/// status.
#[tokio::test]
async fn a_prebuild_inside_its_limit_returns_its_status() {
let mut child = Command::new("sh")
.arg("-c")
.arg("exit 3")
.process_group(0)
.spawn()
.expect("spawn sh");
let status = wait_bounded(&mut child, std::time::Duration::from_mins(1))
.await
.expect("wait_bounded")
.expect("a fast child must not time out");
assert_eq!(status.code(), Some(3));
}
/// The setup path creates the agent's `work/` dir beside `state/`, so the
/// bind source exists before nspawn starts the container.
#[test]

2
hive-c0re/src/main.rs
View file

@ -115,7 +115,7 @@ enum Cmd {
/// option.
#[arg(long)]
model_prices: Option<String>,
/// Override: number of concurrent nix-heavy job-queue nodes
/// Override: number of concurrent closure-fetching job-queue nodes
/// (prebuild / profile-swap / create / meta lock). Set via the
/// `services.hyperhive.deploy.hive-controller.buildSlots` NixOS option.
#[arg(long)]

82
hive-c0re/src/meta_rev.rs
View file

@ -1,27 +1,25 @@
//! The `internal/meta` commit a hive builds an agent from.
//! The `internal/meta` commit a hive deploys an agent from.
//!
//! `internal/meta` is the swarm's meta flake, committed by swarm-controller;
//! a hive has no meta flake of its own. A rebuild's `MetaSync` node resolves
//! the head ([`resolve`]) and records it ([`record`]); `Prebuild` and `Swap`
//! read the record ([`recorded`]), so both build the same flake. A create
//! resolves and uses the commit directly. When the head does not pin the
//! agent's current config, the build fails rather than falling back: the
//! swarm re-pins before it asks a hive to deploy.
//! a hive has no meta flake of its own and builds nothing from it. A rebuild's
//! `MetaSync` node resolves the head ([`resolve`]) and records it ([`record`]);
//! `Prebuild` and `Swap` read the record ([`recorded`]) for the commit the
//! swarm's closure is expected to come from. A create resolves and uses the
//! commit directly. When the head does not pin the agent's current config, the
//! deploy fails rather than falling back: the swarm re-pins before it asks a
//! hive to deploy.
//!
//! The record also names the pins a successful build deploys. [`mark_deployed`]
//! The record also names the pins a successful deploy applies. [`mark_deployed`]
//! copies them to the agent's markers, which the boot reconcile
//! ([`crate::auto_update`]) and the dashboard's update chip read.
use anyhow::{Context, Result, bail};
use hive_priv_sock::{meta_flake_url, valid_meta_rev};
use hive_priv_sock::valid_meta_rev;
use serde::{Deserialize, Serialize};
use swarm_queue_client::agent_archetype::ArchetypeRef;
/// What the forge says about `internal/meta` and an agent's config repo.
pub(crate) trait MetaSource {
/// Whether nix's own `git` can authenticate to the forge: it reads as this
/// hive's account, from the credential file the source publishes.
fn nix_can_fetch(&self) -> bool;
/// The default branch's head commit of `internal/meta`; `None` when the
/// repo is absent or has no commit.
async fn meta_head(&self) -> Result<Option<String>>;
@ -42,15 +40,13 @@ pub(crate) struct MetaBuild {
pub config: Option<String>,
}
/// The `internal/meta` commit to build `agent` from.
/// The `internal/meta` commit to deploy `agent` from.
///
/// `internal/meta` is private, so a hive whose nix cannot authenticate
/// ([`MetaSource::nix_can_fetch`]) cannot build. The head is taken only when
/// its `flake.nix` builds the agent the way it is wanted now: from
/// `archetype` when that is set, otherwise from the config commit the forge's
/// `main` is at. A head that is behind (the swarm's update has not run, or
/// failed), lacks the agent or lists it under another archetype would build an
/// older module than the one being deployed.
/// The head is taken only when its `flake.nix` builds the agent the way it is
/// wanted now: from `archetype` when that is set, otherwise from the config
/// commit the forge's `main` is at. A head that is behind (the swarm's update
/// has not run, or failed), lacks the agent or lists it under another archetype
/// would build an older module than the one being deployed.
///
/// # Errors
/// No usable head, a head that does not pin the agent's current config, or a
@ -60,9 +56,6 @@ pub(crate) async fn choose(
agent: &str,
archetype: Option<&ArchetypeRef>,
) -> Result<MetaBuild> {
if !source.nix_can_fetch() {
bail!("{agent}: this hive has no forge credential for nix to fetch internal/meta with");
}
let Some(rev) = source.meta_head().await? else {
bail!("{agent}: the swarm has not published internal/meta yet");
};
@ -174,7 +167,7 @@ fn recorded_in(dir: &std::path::Path, agent: &str) -> Option<MetaBuild> {
}
/// The record of `agent`'s build, or the error naming why a `Prebuild` or
/// `Swap` has nothing to build from.
/// `Swap` has no commit to expect.
///
/// # Errors
/// No valid record: its `MetaSync` did not run, or ran in an older hive-c0re.
@ -219,21 +212,11 @@ fn mark_deployed_in(
/// Whether an agent's `stored` marker says it runs the hyperhive commit `pin`
/// names. With no `pin`, because the hive can't read `internal/meta`, nothing
/// is stale: a rebuild would have no flake to build from either.
/// is stale: a rebuild would have no commit to deploy either.
pub(crate) fn marker_is_fresh(pin: Option<&str>, stored: Option<&str>) -> bool {
pin.is_none_or(|pin| stored == Some(pin))
}
/// The `<flake>#<agent>` ref [`crate::lifecycle::prebuild_toplevel`] builds
/// for `rev`: the same flake hive-priv builds, from the same forge URL.
pub(crate) fn flake_ref(agent: &str, rev: &str) -> String {
flake_ref_at(crate::forge::forge_http_base(), agent, rev)
}
fn flake_ref_at(forge_base: &str, agent: &str, rev: &str) -> String {
format!("{}#{agent}", meta_flake_url(forge_base, rev))
}
#[cfg(test)]
mod tests {
use super::*;
@ -249,13 +232,9 @@ mod tests {
head: Option<&'static str>,
flake: String,
config_head: Option<&'static str>,
fetchable: bool,
}
impl MetaSource for Forge {
fn nix_can_fetch(&self) -> bool {
self.fetchable
}
async fn meta_head(&self) -> Result<Option<String>> {
Ok(self.head.map(str::to_owned))
}
@ -313,7 +292,6 @@ mod tests {
head: Some(META),
flake: flake(),
config_head,
fetchable: true,
}
}
@ -369,9 +347,6 @@ mod tests {
async fn an_archetype_agent_is_taken_when_the_head_lists_that_archetype() {
struct Archetypes(Forge);
impl MetaSource for Archetypes {
fn nix_can_fetch(&self) -> bool {
self.0.nix_can_fetch()
}
async fn meta_head(&self) -> Result<Option<String>> {
self.0.meta_head().await
}
@ -422,19 +397,6 @@ mod tests {
assert_unpinned(&choose(&source, "rev", None).await.unwrap_err(), "rev");
}
/// A hive without the credential nix's git reads cannot build from the
/// repo even though its own token reads it; the first test is the control.
#[tokio::test]
async fn a_hive_whose_nix_cannot_authenticate_cannot_build() {
let mut source = forge(Some(ARGUS_HEAD));
source.fetchable = false;
let err = choose(&source, "argus", None).await.unwrap_err();
assert!(
format!("{err:#}").contains("no forge credential"),
"{err:#}"
);
}
#[tokio::test]
async fn no_repo_or_no_commit_is_an_error_naming_the_agent() {
let mut source = forge(Some(ARGUS_HEAD));
@ -568,12 +530,4 @@ mod tests {
assert!(marker_is_fresh(None, Some(STALE)), "no pin to compare with");
assert!(marker_is_fresh(None, None));
}
#[test]
fn the_prebuild_ref_is_the_flake_hive_priv_builds() {
assert_eq!(
flake_ref_at("http://forge.t.local", "argus", META),
format!("git+http://forge.t.local/internal/meta.git?rev={META}#argus")
);
}
}

8
hive-c0re/src/priv_client.rs
View file

@ -235,14 +235,14 @@ pub async fn kill_container(name: &str) -> Result<()> {
/// appending lines to `build_logs` or otherwise capturing the output.
pub async fn update_container_streaming(
name: &str,
meta_rev: &str,
toplevel: &str,
on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
ok(call_streaming(
&PrivRequest::UpdateContainer {
name: name.to_owned(),
stream: true,
meta_rev: meta_rev.to_owned(),
toplevel: toplevel.to_owned(),
},
on_line,
)
@ -253,14 +253,14 @@ pub async fn update_container_streaming(
/// arrive. Returns `Ok(())` on success.
pub async fn create_container_streaming(
name: &str,
meta_rev: &str,
toplevel: &str,
on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
ok(call_streaming(
&PrivRequest::CreateContainer {
name: name.to_owned(),
stream: true,
meta_rev: meta_rev.to_owned(),
toplevel: toplevel.to_owned(),
},
on_line,
)

2
hive-c0re/src/stores/build_logs.rs
View file

@ -16,7 +16,7 @@ use utoipa::ToSchema;
use crate::db::Migration;
/// Process-singleton handle, set once at coordinator startup. Lets
/// the `lifecycle` module's `run` / `prebuild_toplevel` access the
/// the `lifecycle` module's `run` access the
/// writer without threading an `Arc<BuildLogs>` through every
/// `pub async fn` entry point in the lifecycle surface — there are
/// 10+ callsites and the handle is the same `Arc` everywhere