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

deploy the closure the swarm declares, read no forge head, drop the update badge

Browse source
This commit is contained in:
atlas 2026-11-01 23:56:40 +01:00 • committed by mara
commit 47c2acb640
6 changed files with 647 additions and 2001 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,5 +1,5 @@
//! Node executors — one async fn per [`NodeKind`], each a thin wrapper
//! over existing `lifecycle.rs` / `meta_rev.rs` / `actions.rs` code. Node
//! over existing `lifecycle.rs` / `actions.rs` code. Node
//! executors keep their own internal error handling where it exists
//! today (cold-start fallback inside `Reconcile`, warn-only forge sync in
//! the `Swap` tail); DAG-level failure handling is cancel-downstream in
@ -15,7 +15,6 @@ use hive_jobq::NodeId;
use super::model::NodeKind;
use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
use crate::meta_rev::MetaBuild;
use crate::power::{ReconcileAction, reconcile_action};
use crate::stats::sweep_health::{self, SweepHealth};
@ -35,21 +34,17 @@ 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 swarm's closure record, the `internal/meta` record. Production uses
/// [`RealHost`], which forwards to `lifecycle` / `meta_rev` / `Coordinator`
/// `hive-priv`, the swarm's closure record and the marker of the closure applied.
/// Production uses [`RealHost`], which forwards to `lifecycle` / `Coordinator`
/// unchanged; tests substitute a recording fake.
pub(super) trait Host: Sync {
/// Whether the agent's container unit is active.
async fn is_running(&self, name: &str) -> bool;
/// Stop the agent's container.
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 record the swarm's builder wrote for the agent; `None` when it wrote
/// none.
async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>>;
/// The pins `internal/meta` at the commit `rev` holds for the agent.
async fn pins_at(&self, name: &str, rev: &str) -> Result<MetaBuild>;
/// 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<()>;
@ -72,15 +67,8 @@ pub(super) trait Host: Sync {
async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()>;
/// Rewrite the host-side files that follow the agent set.
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 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>;
/// Mark the agent as running the pins of `build`.
fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()>;
/// Mark the agent as running the closure at `store_path`.
fn mark_deployed(&self, name: &str, store_path: &str) -> Result<()>;
/// First-spawn provisioning: repos, state subvolume, agent-set refresh.
async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()>;
/// `nixos-container create` from the closure `toplevel`.
@ -117,18 +105,10 @@ impl Host for RealHost {
crate::lifecycle::kill(name).await
}
fn recorded_rev(&self, name: &str) -> Result<String> {
Ok(crate::meta_rev::recorded_or_err(name)?.rev)
}
async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>> {
crate::attic::closure::record(name).await
}
async fn pins_at(&self, name: &str, rev: &str) -> Result<MetaBuild> {
crate::meta_rev::resolve_at(name, rev).await
}
async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
crate::priv_client::realise_closure(name, store_path).await
}
@ -165,20 +145,8 @@ impl Host for RealHost {
crate::agent_set::refresh(&agents).await
}
async fn resolve_meta(&self, name: &str) -> Result<MetaBuild> {
crate::meta_rev::resolve(name).await
}
fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()> {
crate::meta_rev::record(name, build)
}
fn recorded_build(&self, name: &str) -> Option<MetaBuild> {
crate::meta_rev::recorded(name)
}
fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()> {
crate::meta_rev::mark_deployed(name, build)
fn mark_deployed(&self, name: &str, store_path: &str) -> Result<()> {
crate::attic::closure::mark_deployed(name, store_path)
}
async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()> {
@ -283,7 +251,7 @@ async fn dispatch(
NodeKind::MetaSync { .. } => run_meta_sync(host, agent).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::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord).await,
NodeKind::Provision { .. } => run_provision(host, agent).await,
NodeKind::Create { .. } => run_create(host, agent).await,
NodeKind::DestroyContainer { .. } => run_destroy_container(host, coord, agent).await,
@ -502,12 +470,9 @@ fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()>
Ok(())
}
/// The rebuild's preamble: runtime-dir prep, the agent-set refresh, and the
/// `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 commit to deploy.
/// The rebuild's preamble: runtime-dir prep, the archetype and agent-set
/// refresh. Reads nothing from the forge: what a rebuild deploys is the swarm's
/// declared closure, read by `Prebuild` and `Swap`.
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
@ -521,9 +486,6 @@ 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` expect the same commit.
let build = host.resolve_meta(name).await?;
host.record_meta(name, &build)?;
Ok(())
}
@ -534,9 +496,8 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
/// 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 closure = closure_for(host, name, &rev, "prebuild").await?;
tracing::info!(%name, %rev, "prebuild: substituted {}", closure.path);
let closure = closure_for(host, name, "prebuild").await?;
tracing::info!(%name, "prebuild: substituted {closure}");
}
Ok(())
}
@ -550,32 +511,19 @@ enum NoClosure {
Unpullable(String),
}
/// The closure the swarm built for an agent, in the store.
struct Closure {
path: String,
/// The `internal/meta` commit the builder built it from.
meta_rev: String,
}
/// 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, and the caller records
/// the pins of the commit actually deployed ([`deployed_pins`]).
/// The store path of the closure the swarm declares for `name`, downloaded into
/// the store. Whatever `internal/meta` commit it was built from, it is the one
/// deployed.
///
/// # Errors
/// [`without_closure`]'s refusal, after logging it.
async fn closure_for(host: &impl Host, name: &str, rev: &str, step: &str) -> Result<Closure> {
pull_closure(host, name, rev, step)
async fn closure_for(host: &impl Host, name: &str, step: &str) -> Result<String> {
pull_closure(host, name)
.await
.map_err(|why| without_closure(name, rev, step, &why))
.map_err(|why| without_closure(name, step, &why))
}
async fn pull_closure(
host: &impl Host,
name: &str,
rev: &str,
step: &str,
) -> Result<Closure, NoClosure> {
async fn pull_closure(host: &impl Host, name: &str) -> Result<String, NoClosure> {
let record = host
.closure_record(name)
.await
@ -589,52 +537,7 @@ async fn pull_closure(
host.realise_closure(name, &record.store_path)
.await
.map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?;
if record.meta_rev != rev && !same_pins(host, name, &record.meta_rev, rev).await {
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(Closure {
path: record.store_path,
meta_rev: record.meta_rev,
})
}
/// Whether the agent has the same pins at the commits `built` and `chosen`. A
/// commit whose pins cannot be read counts as different.
async fn same_pins(host: &impl Host, name: &str, built: &str, chosen: &str) -> bool {
match (
host.pins_at(name, built).await,
host.pins_at(name, chosen).await,
) {
(Ok(built), Ok(chosen)) => built.same_pins(&chosen),
_ => false,
}
}
/// The pins to record for a deploy of `closure` when this hive chose the commit
/// `rev`: those of the commit the closure was built from, so the markers say
/// what runs. `None` when they are the same commit.
///
/// # Errors
/// The pins of the closure's commit could not be read. The caller fails before
/// applying anything.
async fn deployed_pins(
host: &impl Host,
name: &str,
closure: &Closure,
rev: &str,
) -> Result<Option<MetaBuild>> {
if closure.meta_rev == rev {
return Ok(None);
}
let pins = host
.pins_at(name, &closure.meta_rev)
.await
.with_context(|| format!("{name}: reading the pins of the commit the swarm built from"))?;
Ok(Some(pins))
Ok(record.store_path)
}
/// What a rebuild or create does when there is no closure to deploy: it
@ -645,18 +548,19 @@ async fn deployed_pins(
/// 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 {
fn without_closure(name: &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");
tracing::error!(%name, "{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).
/// container at the swarm's closure and, once that holds, mark the agent as
/// running it. The post-rebuild bookkeeping tail (`Rebuilt` event, forge sync,
/// kick, rescan) is the next node.
/// The recovery-start on failure is NOT here — the DAG's tail
/// `Reconcile` runs after this node terminal ok *or* fail.
async fn run_swap(
@ -670,21 +574,20 @@ async fn run_swap(
let agent_dir = crate::paths::agent_runtime_dir(name);
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir)?;
let rev = host.recorded_rev(name)?;
let closure = closure_for(host, name, &rev, "swap").await?;
let deployed = deployed_pins(host, name, &closure, &rev).await?;
tracing::info!(%name, %rev, "swap: applying {}", closure.path);
let closure = closure_for(host, name, "swap").await?;
tracing::info!(%name, "swap: applying {closure}");
let result = host
.swap_update(name, &hive, &paths, &closure.path, id.get())
.swap_update(name, &hive, &paths, &closure, id.get())
.await;
// The bookkeeping node marks what the record holds, so it holds what was
// applied.
// The marker names what was applied; a failed write is not a failed swap
// (the container runs the closure), it only means the next comparison
// sees the declaration as new.
if result.is_ok()
&& let Some(deployed) = &deployed
&& let Err(e) = host.mark_deployed(name, &closure)
{
host.record_meta(name, deployed)?;
tracing::warn!(%name, error = ?e, "write the deployed-closure marker failed");
}
// On success the Ok-only bookkeeping tail (rev marker, forge
// On success the Ok-only bookkeeping tail (forge
// sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node,
// which deps `AfterOk(Swap)`. On failure `RebuildBookkeeping` is cancel-cascaded
// and the tail `Reconcile` (`AfterAny(RebuildBookkeeping)`) handles recovery; here
@ -702,16 +605,7 @@ async fn run_swap(
/// means the profile swap succeeded. Store/forge work only — no nix
/// build (build-slot-exempt); the agent lease taken at `Swap` is still held
/// (the whole chain up to `Reconcile` is one agent's subgraph).
async fn run_rebuild_bookkeeping(
host: &impl Host,
coord: &Arc<Coordinator>,
name: &str,
) -> Result<()> {
if let Some(build) = host.recorded_build(name)
&& let Err(e) = host.mark_deployed(name, &build)
{
tracing::warn!(%name, error = ?e, "write rev markers failed");
}
async fn run_rebuild_bookkeeping(coord: &Arc<Coordinator>) -> Result<()> {
// The `Rebuilt` manager event is emitted exactly once per agent by the DAG's
// `EmitRebuilt` tail — emitting ok here and letting a failed tail `Reconcile`
// add a contradictory !ok would double-report the same rebuild.
@ -724,7 +618,7 @@ async fn run_rebuild_bookkeeping(
// alongside `run_start`'s own kick for the same boot. `hive-agent` now
// self-announces once on its own startup instead — one message per
// actual boot, not one per DAG node that happens to run during it.
// Rescan so dashboards drop the "needs update" chip.
// Rescan so dashboards see the rebuilt container.
coord.rescan_containers_and_emit().await;
Ok(())
}
@ -743,19 +637,15 @@ async fn run_provision(host: &impl Host, name: &str) -> Result<()> {
/// dir creation and MCP listener registration are deferred to the tail
/// `Reconcile` (`converge_start_preamble` + `register_agent`) so this
/// node stays purely "create", not "create + start". Marks the agent as
/// running the pins it was created from.
/// running the closure it was created from.
///
/// The container is created from the swarm's closure for the resolved commit;
/// with none to deploy the create fails and makes no container.
/// The container is created from the swarm's declared closure; 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?;
let closure = closure_for(host, name, &build.rev, "create").await?;
let build = deployed_pins(host, name, &closure, &build.rev)
.await?
.unwrap_or(build);
tracing::info!(%name, rev = %build.rev, "create: substituted {}", closure.path);
host.create_only(name, &closure.path).await?;
host.mark_deployed(name, &build)?;
let closure = closure_for(host, name, "create").await?;
tracing::info!(%name, "create: substituted {closure}");
host.create_only(name, &closure).await?;
host.mark_deployed(name, &closure)?;
Ok(())
}

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

@ -19,7 +19,6 @@ use super::{
};
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;
@ -30,19 +29,16 @@ const CLOSURE: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-
#[derive(Default)]
struct FakeHost {
running: bool,
rev: 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>>,
/// Names of the calls that fail (`kill`, `realise`, `closure_record`, `swap_update`,
/// `set_paused`, ...). `recorded_build` fails by finding nothing.
/// `set_paused`, `mark_deployed`, ...).
failing: Vec<&'static str>,
/// The sweeps and pulls never finish, so only a deadline ends them.
hang: bool,
/// The agent has the same pins at every commit.
unmoved: bool,
runtime: PathBuf,
purge: Vec<PathBuf>,
calls: Mutex<Vec<String>>,
@ -52,8 +48,7 @@ impl FakeHost {
fn up() -> Self {
Self {
running: true,
rev: Some(REV.to_owned()),
built: Some(closure(REV, true)),
built: Some(closure(true)),
..Self::default()
}
}
@ -61,20 +56,14 @@ impl FakeHost {
fn down() -> Self {
Self {
running: false,
rev: Some(REV.to_owned()),
built: Some(closure(REV, true)),
built: Some(closure(true)),
..Self::default()
}
}
/// 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 unmoved(mut self) -> Self {
self.unmoved = true;
/// The builder recorded the closure for the agent, pushed or not.
fn built(mut self, cached: bool) -> Self {
self.built = Some(closure(cached));
self
}
@ -124,31 +113,11 @@ impl Host for FakeHost {
Ok(())
}
fn recorded_rev(&self, name: &str) -> anyhow::Result<String> {
self.record(format!("recorded_rev {name}"));
self.rev
.clone()
.ok_or_else(|| anyhow::anyhow!("{name}: no internal/meta commit was recorded"))
}
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 pins_at(&self, name: &str, rev: &str) -> anyhow::Result<MetaBuild> {
self.step("pins_at", format!("pins_at {name} {rev}"))?;
Ok(MetaBuild {
rev: rev.to_owned(),
hyperhive: "dh".to_owned(),
config: Some(if self.unmoved {
"dc".to_owned()
} else {
format!("dc-{rev}")
}),
})
}
async fn realise_closure(&self, name: &str, store_path: &str) -> anyhow::Result<()> {
self.record(format!("realise {name} {store_path}"));
if self.fails("realise") {
@ -200,24 +169,10 @@ impl Host for FakeHost {
self.step("refresh_agent_set", "refresh_agent_set".to_owned())
}
async fn resolve_meta(&self, name: &str) -> anyhow::Result<MetaBuild> {
self.step("resolve_meta", format!("resolve_meta {name}"))?;
Ok(build())
}
fn record_meta(&self, name: &str, build: &MetaBuild) -> anyhow::Result<()> {
self.step("record_meta", format!("record_meta {name} {}", build.rev))
}
fn recorded_build(&self, name: &str) -> Option<MetaBuild> {
self.record(format!("recorded_build {name}"));
(!self.fails("recorded_build")).then(build)
}
fn mark_deployed(&self, name: &str, build: &MetaBuild) -> anyhow::Result<()> {
fn mark_deployed(&self, name: &str, store_path: &str) -> anyhow::Result<()> {
self.step(
"mark_deployed",
format!("mark_deployed {name} {}", build.rev),
format!("mark_deployed {name} {store_path}"),
)
}
@ -277,10 +232,10 @@ 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 {
/// The builder's record of a closure.
fn closure(cached: bool) -> AgentClosure {
AgentClosure {
meta_rev: meta_rev.to_owned(),
meta_rev: REV.to_owned(),
store_path: CLOSURE.to_owned(),
system: "x86_64-linux".to_owned(),
built_unix: 1,
@ -289,14 +244,6 @@ fn closure(meta_rev: &str, cached: bool) -> AgentClosure {
}
}
fn build() -> MetaBuild {
MetaBuild {
rev: REV.to_owned(),
hyperhive: "h".to_owned(),
config: Some("c".to_owned()),
}
}
fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
let dir = tempfile::tempdir().expect("tempdir");
let coord = Coordinator::open(
@ -325,8 +272,6 @@ 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)>>>);
@ -384,11 +329,7 @@ async fn logging<T>(fut: impl std::future::Future<Output = T>) -> (T, Logged) {
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",
),
("not pushed", FakeHost::up().built(false), "did not push"),
(
"download fails",
FakeHost::up().failing("realise"),
@ -411,7 +352,6 @@ async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() {
host.calls(),
[
"is_running alice".to_owned(),
"recorded_rev alice".to_owned(),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
]
@ -454,39 +394,6 @@ async fn a_rebuild_without_a_closure_fails_the_prebuild_and_logs_an_error() {
}
}
/// Another commit that pins the agent the same is nothing to warn about.
#[tokio::test]
async fn a_closure_built_from_a_commit_with_the_same_pins_is_deployed_quietly() {
let host = FakeHost::up().built(OTHER_REV, true).unmoved();
let (result, log) = logging(run_prebuild(&host, "alice")).await;
result.expect("drift is not a refusal");
assert!(host.calls().contains(&format!("realise alice {CLOSURE}")));
assert!(log.at(tracing::Level::WARN).is_empty());
}
/// 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_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();
@ -494,22 +401,6 @@ async fn a_stopped_container_looks_up_nothing() {
assert_eq!(host.calls(), ["is_running alice"]);
}
#[tokio::test]
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")
.await
.expect_err("nothing to deploy");
assert!(
format!("{err:#}").contains("no internal/meta commit"),
"{err:#}"
);
assert!(!host.calls().iter().any(|c| c.starts_with("closure_record")));
}
// --- Swap -------------------------------------------------------------------
#[tokio::test]
@ -521,10 +412,10 @@ async fn swap_applies_the_swarms_closure() {
assert_eq!(
host.calls(),
[
"recorded_rev alice".to_owned(),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("swap_update alice {CLOSURE} node={}", id.get()),
format!("mark_deployed alice {CLOSURE}"),
]
);
}
@ -551,53 +442,40 @@ async fn a_swap_without_a_closure_never_updates_and_logs_an_error() {
}
}
/// The marker names the closure that was applied, and only an applied one.
#[tokio::test]
async fn swap_applies_a_closure_built_from_another_commit_with_a_warning() {
async fn swap_marks_the_applied_closure_deployed_only_after_the_apply() {
let (_dir, coord) = coordinator();
let host = FakeHost::down().built(OTHER_REV, true);
let host = FakeHost::down();
let id = node_id();
let (result, log) = logging(run_swap(&host, &coord, "alice", id)).await;
result.expect("drift is not a refusal");
// The pins of the commit the closure came from are read before the apply
// and recorded after it, in place of the ones this hive chose; the
// bookkeeping node then marks those.
run_swap(&host, &coord, "alice", id).await.expect("swap ok");
assert_eq!(
host.calls(),
[
"recorded_rev alice".to_owned(),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("pins_at alice {OTHER_REV}"),
format!("pins_at alice {REV}"),
format!("pins_at alice {OTHER_REV}"),
format!("swap_update alice {CLOSURE} node={}", id.get()),
format!("record_meta alice {OTHER_REV}"),
]
host.calls().last().map(String::as_str),
Some(format!("mark_deployed alice {CLOSURE}").as_str()),
"{:?}",
host.calls()
);
let host = FakeHost::down().failing("swap_update");
run_swap(&host, &coord, "alice", node_id())
.await
.expect_err("apply fails");
assert!(
!host.calls().iter().any(|c| c.starts_with("mark_deployed")),
"{:?}",
host.calls()
);
assert_eq!(log.at(tracing::Level::WARN).len(), 1);
assert!(log.at(tracing::Level::ERROR).is_empty());
}
/// Pins that can't be read, or an apply that fails, leave the record alone.
/// The container runs the closure, so a marker that cannot be written is a
/// warning and not a failed swap.
#[tokio::test]
async fn a_drifted_swap_records_pins_only_for_an_apply_that_happened() {
for failing in ["pins_at", "swap_update"] {
let (_dir, coord) = coordinator();
let host = FakeHost::down().built(OTHER_REV, true).failing(failing);
run_swap(&host, &coord, "alice", node_id())
.await
.expect_err(failing);
let calls = host.calls();
assert!(
!calls.iter().any(|c| c.starts_with("record_meta")),
"{failing}: {calls:?}"
);
assert_eq!(
calls.iter().any(|c| c.starts_with("swap_update")),
failing == "swap_update",
"{failing}: {calls:?}"
);
}
async fn a_marker_that_cannot_be_written_does_not_fail_the_swap() {
let (_dir, coord) = coordinator();
let host = FakeHost::down().failing("mark_deployed");
let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await;
result.expect("swap ok");
assert_eq!(log.at(tracing::Level::WARN).len(), 1);
}
/// A failed update fails the node and nothing else: the swap never restarts or
@ -611,21 +489,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(), 4, "{calls:?}");
assert!(calls[3].starts_with("swap_update alice"), "{calls:?}");
}
#[tokio::test]
async fn swap_without_a_recorded_rev_never_updates() {
let (_dir, coord) = coordinator();
let host = FakeHost {
rev: None,
..FakeHost::down()
};
run_swap(&host, &coord, "alice", node_id())
.await
.expect_err("nothing to swap to");
assert_eq!(host.calls(), ["recorded_rev alice"]);
assert_eq!(calls.len(), 3, "{calls:?}");
assert!(calls[2].starts_with("swap_update alice"), "{calls:?}");
}
// --- Stop / StopForUpdate ---------------------------------------------------
@ -851,7 +716,7 @@ async fn pause_drain_times_out_ok_and_clears_pending() {
// --- MetaSync ---------------------------------------------------------------
#[tokio::test]
async fn meta_sync_prepares_then_resolves_and_records_last() {
async fn meta_sync_only_prepares_the_rebuild() {
let host = FakeHost::up();
run_meta_sync(&host, "alice").await.expect("ok");
assert_eq!(
@ -860,8 +725,6 @@ async fn meta_sync_prepares_then_resolves_and_records_last() {
"refresh_archetype alice".to_owned(),
"prepare_rebuild_dirs alice".to_owned(),
"refresh_agent_set".to_owned(),
"resolve_meta alice".to_owned(),
format!("record_meta alice {REV}"),
]
);
}
@ -870,17 +733,12 @@ async fn meta_sync_prepares_then_resolves_and_records_last() {
async fn meta_sync_keeps_going_when_the_archetype_read_fails() {
let host = FakeHost::up().failing("refresh_archetype");
run_meta_sync(&host, "alice").await.expect("non-fatal");
assert_eq!(host.calls().len(), 5);
assert_eq!(host.calls().len(), 3);
}
#[tokio::test]
async fn meta_sync_stops_at_the_first_fatal_step() {
for (fail, calls) in [
("prepare_rebuild_dirs", 2),
("refresh_agent_set", 3),
("resolve_meta", 4),
("record_meta", 5),
] {
for (fail, calls) in [("prepare_rebuild_dirs", 2), ("refresh_agent_set", 3)] {
let host = FakeHost::up().failing(fail);
run_meta_sync(&host, "alice")
.await
@ -913,11 +771,10 @@ async fn create_downloads_the_swarms_closure_and_creates_from_it() {
assert_eq!(
host.calls(),
[
"resolve_meta alice".to_owned(),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("create_only alice {CLOSURE}"),
format!("mark_deployed alice {REV}"),
format!("mark_deployed alice {CLOSURE}"),
]
);
assert!(
@ -952,39 +809,11 @@ async fn a_create_without_a_closure_is_refused_and_logs_an_error() {
}
}
#[tokio::test]
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();
// Created from the swarm's closure, and marked as the commit it was built
// from, not the one this hive chose.
assert_eq!(
calls,
[
"resolve_meta alice".to_owned(),
"closure_record alice".to_owned(),
format!("realise alice {CLOSURE}"),
format!("pins_at alice {OTHER_REV}"),
format!("pins_at alice {REV}"),
format!("pins_at alice {OTHER_REV}"),
format!("create_only alice {CLOSURE}"),
format!("mark_deployed alice {OTHER_REV}"),
]
);
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").await.expect_err("create fails");
assert_eq!(host.calls().len(), 4);
let host = FakeHost::up().failing("resolve_meta");
run_create(&host, "alice").await.expect_err("resolve fails");
assert_eq!(host.calls(), ["resolve_meta alice"]);
assert_eq!(host.calls().len(), 3);
}
// --- Start ------------------------------------------------------------------
@ -1102,38 +931,9 @@ async fn destroy_bookkeeping_continues_past_a_failed_refresh() {
// --- RebuildBookkeeping / EmitRebuilt ----------------------------------------
#[tokio::test]
async fn rebuild_bookkeeping_marks_the_recorded_build_deployed() {
async fn rebuild_bookkeeping_rescans_without_touching_the_host() {
let (_dir, coord) = coordinator();
let host = FakeHost::up();
run_rebuild_bookkeeping(&host, &coord, "alice")
.await
.expect("ok");
assert_eq!(
host.calls(),
[
"recorded_build alice".to_owned(),
format!("mark_deployed alice {REV}")
]
);
}
#[tokio::test]
async fn rebuild_bookkeeping_without_a_recording_marks_nothing() {
let (_dir, coord) = coordinator();
let host = FakeHost::up().failing("recorded_build");
run_rebuild_bookkeeping(&host, &coord, "alice")
.await
.expect("ok");
assert_eq!(host.calls(), ["recorded_build alice"]);
}
#[tokio::test]
async fn a_failed_marker_write_does_not_fail_the_bookkeeping_node() {
let (_dir, coord) = coordinator();
let host = FakeHost::up().failing("mark_deployed");
run_rebuild_bookkeeping(&host, &coord, "alice")
.await
.expect("warn only");
run_rebuild_bookkeeping(&coord).await.expect("ok");
}
#[tokio::test]
@ -1234,30 +1034,21 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
"refresh_archetype alice",
"prepare_rebuild_dirs alice",
"refresh_agent_set",
"resolve_meta alice",
"record_meta alice",
],
),
(
alice(|agent| NodeKind::Prebuild { agent }),
&[
"is_running alice",
"recorded_rev alice",
"closure_record alice",
],
&["is_running alice", "closure_record alice"],
),
(
alice(|agent| NodeKind::Swap { agent }),
&[
"recorded_rev alice",
"closure_record alice",
"swap_update alice",
"mark_deployed alice",
],
),
(
alice(|agent| NodeKind::RebuildBookkeeping { agent }),
&["recorded_build alice", "mark_deployed alice"],
),
(alice(|agent| NodeKind::RebuildBookkeeping { agent }), &[]),
(
alice(|agent| NodeKind::Provision { agent }),
&["provision alice"],
@ -1265,7 +1056,6 @@ 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",

10
hive-c0re/src/job_queue/mod.rs
View file

@ -190,6 +190,16 @@ impl JobQueue {
Ok(named)
}
/// Whether a rebuild of `agent` is queued or running: its `MetaSync` is not
/// terminal yet.
#[must_use]
pub fn rebuild_live(&self, agent: &str) -> bool {
self.lock().graph().nodes().any(|n| {
!n.state.is_terminal()
&& matches!(&n.payload, model::NodeKind::MetaSync { agent: a } if a == agent)
})
}
/// Queue `agent`'s first deploy unless one is already queued or running.
///
/// `None` means a `Provision` for the agent is still live. The check and

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

@ -16,7 +16,7 @@ pub use hive_host_sock::jobs::State;
use serde::Serialize;
/// The primitive operations — each kind maps to one executor fn in
/// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta_rev.rs`
/// `exec.rs`, a thin wrapper over existing `lifecycle.rs`
/// code. Concurrency is gated by resource classes (see
/// [`Resource`](super::resource::Resource), declared per node where the node
/// is constructed rather than derived from its kind).
@ -29,8 +29,8 @@ use serde::Serialize;
#[serde(tag = "kind", rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum NodeKind {
/// The rebuild's preamble: runtime-dir prep and the `internal/meta`
/// commit the build reads.
/// The rebuild's preamble: runtime-dir prep, the archetype and agent-set
/// refresh.
MetaSync { agent: String },
/// Download of the swarm's closure while the container keeps serving.
Prebuild { agent: String },

44
hive-c0re/src/job_queue/power.rs
View file

@ -76,8 +76,8 @@ fn stop_chain(
wanted.guid()
}
/// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev
/// agent gets the rebuild subgraph (its tail `Reconcile` starts it on
/// One agent's **start** subgraph. `SetWanted(Up)` head; a down agent whose
/// declared closure is not applied gets the rebuild subgraph (its tail `Reconcile` starts it on
/// current derivations), otherwise a plain `Reconcile` (which starts a down
/// agent and noops an already-running one).
///
@ -277,8 +277,8 @@ pub async fn restart_many(
}
/// Start `agents` in a **single** DAG. A down agent gets
/// `SetWanted(Up) → Reconcile` (or, rev stale, a rebuild-then-start so it comes
/// up on current derivations); an already-running agent gets the same shape
/// `SetWanted(Up) → Reconcile` (or, when the swarm declares a closure it has not applied, a
/// rebuild-then-start so it comes up on the declared one); an already-running agent gets the same shape
/// with the reconcile noop'ing.
///
/// # Errors
@ -287,14 +287,12 @@ pub async fn start_many(
coord: &Arc<Coordinator>,
agents: &[String],
) -> anyhow::Result<Vec<NodeId>> {
let current = crate::meta_rev::current_hyperhive_pin().await;
let mut targets = Vec::with_capacity(agents.len());
for agent in agents {
let running = lifecycle::is_running(agent).await;
let stored = read_rev_marker(&crate::paths::applied_rev_marker(agent));
let stale = !crate::meta_rev::marker_is_fresh(current.as_deref(), stored.as_deref());
let stale = crate::attic::closure::pending_path(agent).await.is_some();
if !running && stale {
tracing::info!(%agent, "start: rev stale + agent down — rebuild-then-start");
tracing::info!(%agent, "start: closure declared + agent down — rebuild-then-start");
}
targets.push((agent.clone(), running, stale));
}
@ -303,20 +301,6 @@ pub async fn start_many(
Ok(ids)
}
/// The rev `marker` records, or `None` when there is none. A missing file is
/// the normal never-applied case; any other read error is logged, and the
/// caller still treats the rev as stale.
fn read_rev_marker(marker: &std::path::Path) -> Option<String> {
match std::fs::read_to_string(marker) {
Ok(rev) => Some(rev),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
Err(e) => {
tracing::warn!(path = %marker.display(), error = %e, "start: applied-rev marker unreadable; treating rev as stale");
None
}
}
}
/// Stop `agents` in a **single** DAG. A running agent gets
/// `SetWanted(Off) → [Signal → Drain →](graceful) Reconcile`; a down agent
/// skips the pointless quiesce but keeps the `Reconcile` as the race-up
@ -366,19 +350,3 @@ pub fn pause_many(coord: &Arc<Coordinator>, agents: &[String]) -> anyhow::Result
coord.emit_rebuild_queue_snapshot();
Ok(ids)
}
#[cfg(test)]
mod tests {
use super::read_rev_marker;
#[test]
fn a_rev_marker_reads_as_its_contents_and_none_when_absent_or_unreadable() {
let dir = tempfile::tempdir().unwrap();
let marker = dir.path().join("rev");
assert_eq!(read_rev_marker(&marker), None);
std::fs::write(&marker, "abc").unwrap();
assert_eq!(read_rev_marker(&marker).as_deref(), Some("abc"));
// A directory where the file belongs fails with a non-NotFound error.
assert_eq!(read_rev_marker(dir.path()), None);
}
}

15
hive-c0re/src/job_queue/tests.rs
View file

@ -1545,3 +1545,18 @@ fn a_first_deploy_is_not_queued_twice_for_one_agent() {
"control: another agent's deploy is independent"
);
}
/// The auto-update pass queues no second rebuild for an agent that has one in
/// flight; another agent's rebuild does not count.
#[test]
fn a_rebuild_is_live_for_its_own_agent_only() {
let q = JobQueue::new(1);
assert!(!q.rebuild_live("alice"), "nothing queued yet");
q.insert_job(|b| templates::rebuild(b, "alice"))
.expect("insert");
assert!(
q.rebuild_live("alice"),
"control: the queued rebuild is seen"
);
assert!(!q.rebuild_live("bob"));
}