From 4356f3840ca009bdedabdca6549c092988d78ecb Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 2 Nov 2026 00:42:45 +0100 Subject: [PATCH] pin the declared closure once per rebuild and seed the marker from the running closure --- docs/agent-lifecycle/config-changes.md | 25 ++-- docs/agent-lifecycle/persistence.md | 5 +- docs/scheduler/coordinator.md | 16 ++- docs/swarm/archetypes.md | 6 +- docs/swarm/builder.md | 10 +- hive-c0re/src/attic/closure.rs | 178 ++++++++++++++++++++++-- hive-c0re/src/job_queue/exec.rs | 87 +++++++++--- hive-c0re/src/job_queue/exec/tests.rs | 176 ++++++++++++++++++----- hive-c0re/src/job_queue/tests.rs | 20 +++ hive-c0re/src/paths.rs | 15 +- hive-c0re/src/workers/auto_update.rs | 4 +- hive-meta-flake/README.md | 5 +- swarm-controller/src/agent_archetype.rs | 4 +- swarm-controller/src/meta_flake.rs | 2 +- 14 files changed, 457 insertions(+), 96 deletions(-) diff --git a/docs/agent-lifecycle/config-changes.md b/docs/agent-lifecycle/config-changes.md index d70b5ea2..3cdaa801 100644 --- a/docs/agent-lifecycle/config-changes.md +++ b/docs/agent-lifecycle/config-changes.md @@ -59,7 +59,7 @@ an agent. The swarm controller's `InitAgentConfigRepo` job seeds the agent's config repo with a default `agent.nix` template, then asks the target hive to deploy it. An agent created from an [archetype](../swarm/archetypes.md) gets no -config repo: its hive builds it from the archetype's module. +config repo: the swarm builds it from the archetype's module. Changing what the template seeded isn't an exception: like every later change, it's a PR on that config repo, made from a clone and @@ -146,8 +146,12 @@ the swarm builder built the agent from the updated repo (the - **Choosing the closure.** The hive doesn't choose. It deploys the closure the agent's record in `agent-closures` declares, whichever `internal/meta` commit the builder built it from, and reads no forge head. The rebuild's `MetaSync` - node only prepares it (the agent's archetype, its runtime directories, the - agent set). + node prepares it (the agent's archetype, its runtime directories, the agent + set), reads the record once and pins its store path. `Prebuild` and `Swap` use + the pin, so a record that changes mid-rebuild can't make `Swap` pull another + closure than `Prebuild` fetched, and the marker names the closure it applied. The + next autoupdate pass picks up the newer record. A first deploy has no + `MetaSync`; its `Create` reads the record once. - **The closure.** The hive builds nothing. `Prebuild`, `Swap` and `Create` read the agent's record in the swarm's `agent-closures` bucket (the controller writes it from the builder's reply) and have `hive-priv` download the closure @@ -353,11 +357,16 @@ through the same graceful rebuild subgraph. The comparison reads the record and the marker (`/var/lib/hyperhive/applied/..closure`, written after a successful -swap or create) and nothing on the forge. It rebuilds nothing when the hive -can't tell: no connection to the swarm queue, no record, a closure the builder -didn't push, or an agent with no marker. The last case covers every agent on -the first boot after the marker arrived, so that boot rebuilds nothing; the -agent's next deploy request from the swarm writes the marker. The periodic pass +swap or create) and nothing on the forge. Where the marker is missing, as for +every agent on the first boot after the marker arrived, the hive reads the +closure the container's profile +(`/nix/var/nix/profiles/per-container/h-/system`) points at and writes +that as the marker, so the first pass catches an agent that doesn't run its declared +closure. Those first-boot rebuilds go through the same job queue as +any other: each holds one of the `buildSlots`, so they don't swap all at once. +It rebuilds nothing when the hive can't tell: no connection to the swarm queue, +no record, a closure the builder didn't push, or a profile it can't read or that +leaves the nix store. The periodic pass attempts a declared store path once per process, and skips an agent while a rebuild of it waits or runs, so a closure that fails to deploy doesn't restart its agent every five minutes. Restarting `hive-c0re`, or a rebuild from diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index 05b33048..d3270136 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -545,7 +545,8 @@ unless it holds the permission, nothing else. Under `/var/lib/hyperhive/applied/` — the hive-c0re-only deploy markers `..closure`: the store path of the closure the agent was last swapped to -or created from. No config lives here; the +or created from, or, where the file was missing, the closure the container's +profile pointed at when the hive first read it. No config lives here; the forge's `agent-configs/` is the only copy. A hive that predates this finds per-agent `applied/` repos here. Nothing reads them any more, and you can delete them. @@ -597,7 +598,7 @@ this hive's archetype agents runs, as last read from the swarm's with an entry has no config repo and no `agent-` meta input; see [archetypes](../swarm/archetypes.md). -hive-c0re creates containers from `internal/meta` (`nixosConfigurations.`). +hive-c0re creates containers from the closure the swarm builds from `internal/meta` (`nixosConfigurations.`). ## Destroy vs purge diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 79a7126b..68f707cb 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -54,7 +54,7 @@ Cheap — no build slot: | Node | Behavior | | -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `MetaSync` | the rebuild's preamble — rebuild-dir prep, the archetype refresh and the agent-set refresh (`agent_set::refresh`); reads nothing from the forge. Declares no resources | +| `MetaSync` | the rebuild's preamble — rebuild-dir prep, the archetype refresh and the agent-set refresh (`agent_set::refresh`), then reads the agent's closure record once and pins its store path for `Prebuild` and `Swap`; fails when the swarm has no closure to deploy; reads nothing from the forge. Declares no resources | | `Provision` | first-spawn pre-create provisioning — the state subvolume, agent-set refresh (`agent_set::refresh`); runs ahead of `Create`. Store-only, no container yet | | `Reconcile` | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop | | `Start` | mechanical container start — runtime dir + drop-ins, `start_with_fallback`, MCP listener registration, the manager kick. Fanned out by a `Reconcile` that observed `wanted = Up` and the container down | @@ -298,8 +298,10 @@ store path the agent's `agent-closures` record declares) and persisted rebuild's tail `Reconcile` brings the agent (back) up, covering both the running-stale and stopped-but-wanted-up cases. An agent whose declaration the hive can't read or compare (no queue connection, no record, a closure the - builder didn't push, no marker) rebuilds nothing, and neither does any - agent on the first boot after the marker arrived. + builder didn't push, a running closure it can't read) rebuilds nothing. An + agent with no marker takes the closure its container profile points at as the + marker, so a first boot after the marker arrived rebuilds exactly the + agents that run something else than declared, each holding a build slot. 2. **Power path** — every agent whose observed state drifted from `wanted` gets a plain `Reconcile` DAG (`kind = reconcile`, source `auto_update`). @@ -318,7 +320,7 @@ classification (the one-time migration; thereafter the DB is authoritative). ## Meta flake Agents build from the swarm's `internal/meta` flake, which swarm-controller -commits and every hive reads (see +commits and the swarm builder builds from (see [Config changes](../agent-lifecycle/config-changes.md#the-swarms-meta-flake)). The hive keeps no flake of its own and deploys the closure the swarm builds from it, as the agent's `agent-closures` record declares (`attic/closure.rs`). @@ -349,8 +351,10 @@ one go, rather than the double-bounce a live `update` would trigger. Sequence for a rebuild DAG (each step is its own queue node): -1. `MetaSync` — rebuild-dir prep, the archetype refresh and the agent-set - refresh. Short, and it declares no resources. +1. `MetaSync` — rebuild-dir prep, the archetype refresh, the agent-set + refresh, and the pin of the swarm's declared closure (one read of the record; + `Prebuild` and `Swap` use the pin). Short, and it declares no + resources. 2. `Prebuild` — download the swarm's closure for the agent **before** stopping. The container keeps serving the previous generation while the download happens out-of-band. A missing or unpullable closure fails the DAG here, diff --git a/docs/swarm/archetypes.md b/docs/swarm/archetypes.md index b733f5a6..8b5018bf 100644 --- a/docs/swarm/archetypes.md +++ b/docs/swarm/archetypes.md @@ -125,7 +125,7 @@ swarm-controller renders `internal/meta`'s `flake.nix` with the place of the agent's own flake, and set `HIVE_ARCHETYPE` in the harness environment. -The hive builds from `internal/meta` alone. It reads the agent's key when it +The swarm builds from `internal/meta` alone and the hive deploys that closure. The hive reads the agent's key when it provisions the agent and on every rebuild, and keeps the last value read in `/var/lib/hyperhive/swarm-agent-archetypes.json`. When the read fails, a rebuild keeps the stored value. For an archetype agent the hive also: @@ -188,8 +188,8 @@ in a lock doesn't count. A hive that keeps a checkout of a custom agent's config would create one for the converted agent, under the container's running mount. A hive that lists `no-config-checkout` under `capabilities` in the status it publishes (the -document `GET /api/hives/status` serves as `snapshot`) keeps none: it builds a -custom agent from `internal/meta` alone, and the agent clones its own repo from +document `GET /api/hives/status` serves as `snapshot`) keeps none: it deploys a +custom agent's closure from the swarm alone, and the agent clones its own repo from the forge. The controller reads the agent's hive's latest status and refuses unless it's `fresh` and lists the capability; an older hive lists none. Deploy the release to every hive that places archetype agents before converting theirs. diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 90b321b8..7d4d5da3 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -199,7 +199,7 @@ it. The check runs after each update and once when the controller starts. ## Deploy gate Every deploy the swarm sends waits for a build of its agent from `internal/meta`, -the flake a hive builds from. Four paths send deploys, and each declares the same +the flake the swarm builds from. Four paths send deploys, and each declares the same build and deploy pair: a merged config PR, a new agent, the pin fan-out and the ten-minute retry. The `build_meta` node comes before the `trigger_deploy` node and needs the one `builder` slot, so it queues behind PR checks. The deploy @@ -503,10 +503,10 @@ same and creates the container from that path. A rebuild or create has two outcomes, by what the lookup and the download find. The hive deploys the record's closure whatever `internal/meta` commit the builder built it from, and reads no forge head to choose one: -| Found | What happens | -| ----------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| a record that the builder pushed | The hive downloads the closure and deploys it, then writes the store path to the agent's `applied/..closure` marker. The hive logs `prebuild: substituted ` (`create: substituted ` for a create) at info level. | -| no record, or one the hive can't use (not pushed, queue or cache unreachable, download or signature failed) | The rebuild or create fails with an error that names the agent and the reason, logged at error level, and nothing changes. A running container keeps running unchanged, because `Prebuild` fails before `StopForUpdate`. A create makes no container. | +| Found | What happens | +| ----------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| a record that the builder pushed | `MetaSync` reads the record once and pins its store path, so a record that changes during the rebuild doesn't change what `Swap` applies. The hive downloads the pinned closure and deploys it, then writes the store path to the agent's `applied/..closure` marker. The hive logs `prebuild: substituted ` (`create: substituted ` for a create) at info level. | +| no record, or one the hive can't use (not pushed, queue or cache unreachable, download or signature failed) | The rebuild or create fails with an error that names the agent and the reason, logged at error level, and nothing changes. A running container keeps running unchanged, because `Prebuild` fails before `StopForUpdate`. A create makes no container. | The hive never falls back to a local build. The second row has one owner in the code, `without_closure` in `hive-c0re/src/job_queue/exec.rs`, so a change to what diff --git a/hive-c0re/src/attic/closure.rs b/hive-c0re/src/attic/closure.rs index 6260ac0b..ebcc1396 100644 --- a/hive-c0re/src/attic/closure.rs +++ b/hive-c0re/src/attic/closure.rs @@ -1,8 +1,11 @@ //! The closure the swarm declares for an agent: its record in `agent-closures` -//! read before a deploy downloads it, and the marker of the store path the -//! hive last applied, which a changed declaration is compared with. +//! read before a deploy downloads it, the store path a rebuild pinned from it, +//! and the marker of the store path the hive last applied, which a changed +//! declaration is compared with. -use std::path::Path; +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; use anyhow::{Context as _, Result, anyhow}; use swarm_wire::agent_closures::AgentClosure; @@ -29,12 +32,71 @@ pub fn mark_deployed(agent: &str, store_path: &str) -> Result<()> { mark_deployed_in(&crate::paths::applied_closure_marker(agent), store_path) } -/// The store path `agent` was last swapped to; `None` when the hive has no -/// record of one: it never applied a closure, or last did before the marker -/// existed. +/// The store path `agent` runs: its marker, or on the first read after the +/// marker was introduced the closure its container profile points at, which is +/// then written as the marker. `None` when neither can be read: the hive cannot +/// tell what runs. #[must_use] pub fn deployed(agent: &str) -> Option { - deployed_in(&crate::paths::applied_closure_marker(agent)) + deployed_or_seed_in( + &crate::paths::applied_closure_marker(agent), + &crate::paths::container_profile(agent), + Path::new(STORE), + ) +} + +/// The nix store; a running closure must resolve into it. +const STORE: &str = "/nix/store"; + +fn deployed_or_seed_in(marker: &Path, profile: &Path, store: &Path) -> Option { + if let Some(deployed) = deployed_in(marker) { + return Some(deployed); + } + let running = running_in(profile, store)?; + if let Err(e) = mark_deployed_in(marker, &running) { + tracing::warn!(marker = %marker.display(), error = %format!("{e:#}"), "cannot seed the deployed-closure marker"); + } + Some(running) +} + +/// The store path `profile` (a container's `system` link) resolves to; `None` +/// when the link is absent or unreadable, or does not end in `store`. +fn running_in(profile: &Path, store: &Path) -> Option { + let resolved: PathBuf = std::fs::canonicalize(profile).ok()?; + resolved + .starts_with(store) + .then(|| resolved.to_str().map(str::to_owned)) + .flatten() +} + +/// The store path each agent's running rebuild pinned from its record. +static PINNED: Mutex> = Mutex::new(BTreeMap::new()); + +/// Pin `store_path` as the closure `agent`'s rebuild deploys, replacing any +/// earlier pin. +pub fn pin(agent: &str, store_path: &str) { + PINNED + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(agent.to_owned(), store_path.to_owned()); +} + +/// The store path pinned for `agent`'s rebuild. +#[must_use] +pub fn pinned(agent: &str) -> Option { + PINNED + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(agent) + .cloned() +} + +/// Drop `agent`'s pin, once its closure is applied. +pub fn unpin(agent: &str) { + PINNED + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(agent); } /// Written whole to a temp file beside the marker and renamed over it, so a @@ -58,10 +120,8 @@ fn deployed_in(marker: &Path) -> Option { /// Whether the swarm's declared closure is one the hive should now deploy: the /// builder pushed it and it differs from what `deployed` says runs. /// -/// An agent with no `deployed` marker is not pending, since the hive cannot -/// tell what it runs: every agent starts that way when the hive first reads -/// this marker, and deploying all of them at once is worse than waiting for the -/// swarm's next deploy request, which writes the marker. +/// An agent whose `deployed` closure is unknown is not pending: the hive cannot +/// tell what it runs, so it deploys nothing it was not told to. #[must_use] pub fn pending(record: &AgentClosure, deployed: Option<&str>) -> bool { record.cached && deployed.is_some_and(|deployed| deployed != record.store_path) @@ -156,8 +216,8 @@ mod tests { } /// A hive that ran the previous marker format has a config commit in its - /// old files and none at the new path: the agent is not pending, and the - /// old files are not read or touched. + /// old files and none at the new path: the old files are not read, and + /// neither seeding nor a deploy touches them. #[test] fn markers_in_the_old_format_are_neither_read_nor_removed() { let dir = tempfile::tempdir().unwrap(); @@ -165,8 +225,14 @@ mod tests { std::fs::write(&old, "1111111111111111111111111111111111111111").unwrap(); let marker = dir.path().join(".alice.closure"); assert_eq!(deployed_in(&marker), None); - assert!(!pending(&declared(true), deployed_in(&marker).as_deref())); - assert!(old.exists()); + let (store, profile) = profile_running(dir.path(), "run-alice"); + let seeded = deployed_or_seed_in(&marker, &profile, &store); + assert!(seeded.is_some(), "the running closure seeds the new marker"); + assert_eq!( + std::fs::read_to_string(&old).unwrap(), + "1111111111111111111111111111111111111111", + "seeding leaves the old marker as it was" + ); mark_deployed_in(&marker, DECLARED).unwrap(); assert!(old.exists(), "a successful deploy leaves old markers alone"); } @@ -178,4 +244,86 @@ mod tests { std::fs::write(&marker, "\n").unwrap(); assert_eq!(deployed_in(&marker), None); } + /// A store with one closure, and a profile whose `system` link reaches it + /// the way `nix-env --set` leaves it: through a generation link. + fn profile_running(dir: &Path, closure: &str) -> (PathBuf, PathBuf) { + let store = dir.join("store"); + std::fs::create_dir_all(store.join(closure)).unwrap(); + let profile = dir.join("profiles").join("h-alice"); + std::fs::create_dir_all(&profile).unwrap(); + std::os::unix::fs::symlink(store.join(closure), profile.join("system-3-link")).unwrap(); + std::os::unix::fs::symlink("system-3-link", profile.join("system")).unwrap(); + (store, profile.join("system")) + } + + #[test] + fn a_missing_marker_is_seeded_from_the_running_closure() { + let dir = tempfile::tempdir().unwrap(); + let (store, profile) = profile_running(dir.path(), "run-alice"); + let marker = dir.path().join("applied").join(".alice.closure"); + let seeded = deployed_or_seed_in(&marker, &profile, &store); + let running = std::fs::canonicalize(store.join("run-alice")).unwrap(); + assert_eq!(seeded.as_deref(), running.to_str()); + assert_eq!(deployed_in(&marker), seeded, "the seed is written"); + + // The marker wins from then on, whatever the profile says. + let (_, other) = profile_running(&dir.path().join("elsewhere"), "run-other"); + assert_eq!(deployed_or_seed_in(&marker, &other, &store), seeded); + } + + /// The point of seeding: an agent that runs something other than the + /// declaration is pending from the first read. + #[test] + fn a_seeded_agent_running_another_closure_than_declared_is_pending() { + let dir = tempfile::tempdir().unwrap(); + let (store, profile) = profile_running(dir.path(), "run-alice"); + let marker = dir.path().join(".alice.closure"); + let seeded = deployed_or_seed_in(&marker, &profile, &store); + assert!(pending(&declared(true), seeded.as_deref())); + let record = AgentClosure { + store_path: seeded.clone().unwrap(), + ..declared(true) + }; + assert!( + !pending(&record, seeded.as_deref()), + "control: running what is declared" + ); + } + + #[test] + fn a_running_closure_that_cannot_be_read_stays_unknown() { + let dir = tempfile::tempdir().unwrap(); + let (store, _) = profile_running(dir.path(), "run-alice"); + let marker = dir.path().join(".alice.closure"); + // No profile at all, as for a container that was never created. + let absent = dir.path().join("profiles").join("h-bob").join("system"); + assert_eq!(deployed_or_seed_in(&marker, &absent, &store), None); + // A dangling link, as after a garbage collection. + let dangling = dir.path().join("dangling"); + std::os::unix::fs::symlink(store.join("gone"), &dangling).unwrap(); + assert_eq!(deployed_or_seed_in(&marker, &dangling, &store), None); + // A link that leaves the store. + let outside = dir.path().join("outside"); + std::fs::create_dir_all(&outside).unwrap(); + let escape = dir.path().join("escape"); + std::os::unix::fs::symlink(&outside, &escape).unwrap(); + assert_eq!(deployed_or_seed_in(&marker, &escape, &store), None); + assert!(!marker.exists(), "nothing is seeded from an unknown"); + assert!(!pending(&declared(true), None)); + } + + #[test] + fn a_pin_lasts_until_it_is_dropped() { + assert_eq!(pinned("pin-test"), None); + pin("pin-test", DECLARED); + assert_eq!(pinned("pin-test").as_deref(), Some(DECLARED)); + pin("pin-test", RUNNING); + assert_eq!( + pinned("pin-test").as_deref(), + Some(RUNNING), + "a new pin replaces" + ); + unpin("pin-test"); + assert_eq!(pinned("pin-test"), None); + } } diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 7b5ced6d..962a8eb7 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -45,6 +45,12 @@ pub(super) trait Host: Sync { /// The record the swarm's builder wrote for the agent; `None` when it wrote /// none. async fn closure_record(&self, name: &str) -> Result>; + /// Pin `store_path` as the closure this rebuild of the agent deploys. + fn pin_closure(&self, name: &str, store_path: &str); + /// The store path pinned for the agent's rebuild. + fn pinned_closure(&self, name: &str) -> Option; + /// Drop the agent's pin. + fn unpin_closure(&self, name: &str); /// 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<()>; @@ -109,6 +115,18 @@ impl Host for RealHost { crate::attic::closure::record(name).await } + fn pin_closure(&self, name: &str, store_path: &str) { + crate::attic::closure::pin(name, store_path); + } + + fn pinned_closure(&self, name: &str) -> Option { + crate::attic::closure::pinned(name) + } + + fn unpin_closure(&self, name: &str) { + crate::attic::closure::unpin(name); + } + async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> { crate::priv_client::realise_closure(name, store_path).await } @@ -471,8 +489,13 @@ fn run_set_wanted(coord: &Arc, agent: &str, up: bool) -> Result<()> } /// 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`. +/// refresh, and the closure the rebuild deploys. The swarm's record is read +/// once, here, and its store path pinned: `Prebuild` and `Swap` use the pin, so +/// a record that changes meanwhile cannot make `Swap` pull another closure than +/// `Prebuild` fetched, and the marker names what was applied. The next +/// auto-update pass picks up the newer record. Reads nothing from the forge. +/// +/// Fails, before anything is stopped, when the swarm has no closure 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 @@ -486,6 +509,10 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> { } host.prepare_rebuild_dirs(name, &paths).await?; host.refresh_agent_set().await?; + let store_path = declared_closure(host, name) + .await + .map_err(|why| without_closure(name, "meta-sync", &why))?; + host.pin_closure(name, &store_path); Ok(()) } @@ -496,7 +523,7 @@ 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 closure = closure_for(host, name, "prebuild").await?; + let closure = realise_pinned(host, name, "prebuild").await?; tracing::info!(%name, "prebuild: substituted {closure}"); } Ok(()) @@ -512,18 +539,46 @@ enum NoClosure { } /// 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. +/// the store, for a deploy that has no `MetaSync` to pin it first (a create). +/// 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, step: &str) -> Result { - pull_closure(host, name) - .await - .map_err(|why| without_closure(name, step, &why)) + async { + let store_path = declared_closure(host, name).await?; + realise(host, name, &store_path).await?; + Ok(store_path) + } + .await + .map_err(|why| without_closure(name, step, &why)) } -async fn pull_closure(host: &impl Host, name: &str) -> Result { +/// Download the closure `MetaSync` pinned for the agent's rebuild. +/// +/// # Errors +/// No closure was pinned, or the download failed; the latter is logged by +/// [`without_closure`]. +async fn realise_pinned(host: &impl Host, name: &str, step: &str) -> Result { + let Some(store_path) = host.pinned_closure(name) else { + let why = NoClosure::Unpullable("no closure was pinned for this rebuild".to_owned()); + return Err(without_closure(name, step, &why)); + }; + realise(host, name, &store_path) + .await + .map_err(|why| without_closure(name, step, &why))?; + Ok(store_path) +} + +async fn realise(host: &impl Host, name: &str, store_path: &str) -> Result<(), NoClosure> { + host.realise_closure(name, store_path) + .await + .map_err(|e| NoClosure::Unpullable(format!("{e:#}"))) +} + +/// The store path of the record the swarm's builder wrote for `name`, when the +/// builder pushed its closure. +async fn declared_closure(host: &impl Host, name: &str) -> Result { let record = host .closure_record(name) .await @@ -534,9 +589,6 @@ async fn pull_closure(host: &impl Host, name: &str) -> Result "the builder did not push the closure".to_owned(), )); } - host.realise_closure(name, &record.store_path) - .await - .map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?; Ok(record.store_path) } @@ -574,7 +626,7 @@ 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 closure = closure_for(host, name, "swap").await?; + let closure = realise_pinned(host, name, "swap").await?; tracing::info!(%name, "swap: applying {closure}"); let result = host .swap_update(name, &hive, &paths, &closure, id.get()) @@ -582,10 +634,11 @@ async fn run_swap( // 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 Err(e) = host.mark_deployed(name, &closure) - { - tracing::warn!(%name, error = ?e, "write the deployed-closure marker failed"); + if result.is_ok() { + if let Err(e) = host.mark_deployed(name, &closure) { + tracing::warn!(%name, error = ?e, "write the deployed-closure marker failed"); + } + host.unpin_closure(name); } // On success the Ok-only bookkeeping tail (forge // sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node, diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index c742706f..3f654d31 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/hive-c0re/src/job_queue/exec/tests.rs @@ -31,7 +31,9 @@ struct FakeHost { running: bool, /// What the swarm's builder recorded for the agent; `None` reads as no /// record. - built: Option, + built: Mutex>, + /// The store path pinned for the agent's rebuild. + pinned: Mutex>, /// Store paths a download put in the store. realised: Mutex>, /// Names of the calls that fail (`kill`, `realise`, `closure_record`, `swap_update`, @@ -48,7 +50,8 @@ impl FakeHost { fn up() -> Self { Self { running: true, - built: Some(closure(true)), + built: Mutex::new(Some(closure(true))), + pinned: Mutex::new(Some(CLOSURE.to_owned())), ..Self::default() } } @@ -56,22 +59,39 @@ impl FakeHost { fn down() -> Self { Self { running: false, - built: Some(closure(true)), + built: Mutex::new(Some(closure(true))), + pinned: Mutex::new(Some(CLOSURE.to_owned())), ..Self::default() } } /// The builder recorded the closure for the agent, pushed or not. - fn built(mut self, cached: bool) -> Self { - self.built = Some(closure(cached)); + fn built(self, cached: bool) -> Self { + *self.built.lock().unwrap() = Some(closure(cached)); self } - fn without_record(mut self) -> Self { - self.built = None; + fn without_record(self) -> Self { + *self.built.lock().unwrap() = None; self } + /// The rebuild has no pinned closure, as when `MetaSync` has not run. + fn unpinned(self) -> Self { + *self.pinned.lock().unwrap() = None; + self + } + + /// The builder records another closure, at `store_path`. + fn builder_records(&self, store_path: &str) { + self.built + .lock() + .unwrap() + .as_mut() + .expect("a record") + .store_path = store_path.to_owned(); + } + fn failing(mut self, call: &'static str) -> Self { self.failing.push(call); self @@ -115,7 +135,21 @@ impl Host for FakeHost { async fn closure_record(&self, name: &str) -> anyhow::Result> { self.step("closure_record", format!("closure_record {name}"))?; - Ok(self.built.clone()) + Ok(self.built.lock().unwrap().clone()) + } + + fn pin_closure(&self, name: &str, store_path: &str) { + self.record(format!("pin_closure {name} {store_path}")); + *self.pinned.lock().unwrap() = Some(store_path.to_owned()); + } + + fn pinned_closure(&self, _name: &str) -> Option { + self.pinned.lock().unwrap().clone() + } + + fn unpin_closure(&self, name: &str) { + self.record(format!("unpin_closure {name}")); + *self.pinned.lock().unwrap() = None; } async fn realise_closure(&self, name: &str, store_path: &str) -> anyhow::Result<()> { @@ -326,15 +360,21 @@ async fn logging(fut: impl std::future::Future) -> (T, 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] { - [ +fn without_a_closure() -> Vec<(&'static str, FakeHost, &'static str)> { + let mut cases = without_a_record(); + cases.push(( + "download fails", + FakeHost::up().failing("realise"), + "Couldn't connect", + )); + cases +} + +/// The ways the swarm has no closure to pin for a rebuild. +fn without_a_record() -> Vec<(&'static str, FakeHost, &'static str)> { + vec![ ("no record", FakeHost::up().without_record(), "no build"), ("not pushed", FakeHost::up().built(false), "did not push"), - ( - "download fails", - FakeHost::up().failing("realise"), - "Couldn't connect", - ), ( "record unreadable", FakeHost::up().failing("closure_record"), @@ -343,6 +383,18 @@ fn without_a_closure() -> [(&'static str, FakeHost, &'static str); 4] { ] } +/// The ways a rebuild's `Prebuild` or `Swap` has no closure to download. +fn without_a_pinned_closure() -> Vec<(&'static str, FakeHost, &'static str)> { + vec![ + ("no pin", FakeHost::up().unpinned(), "no closure was pinned"), + ( + "download fails", + FakeHost::up().failing("realise"), + "Couldn't connect", + ), + ] +} + #[tokio::test] async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() { let host = FakeHost::up(); @@ -352,7 +404,6 @@ async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() { host.calls(), [ "is_running alice".to_owned(), - "closure_record alice".to_owned(), format!("realise alice {CLOSURE}"), ] ); @@ -370,7 +421,7 @@ async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() { /// container is untouched, and the log names the agent and why. #[tokio::test] async fn a_rebuild_without_a_closure_fails_the_prebuild_and_logs_an_error() { - for (case, host, why) in without_a_closure() { + for (case, host, why) in without_a_pinned_closure() { let (result, log) = logging(run_prebuild(&host, "alice")).await; let err = result.expect_err(case); assert!(format!("{err:#}").starts_with("alice: "), "{case}: {err:#}"); @@ -412,10 +463,10 @@ async fn swap_applies_the_swarms_closure() { assert_eq!( host.calls(), [ - "closure_record alice".to_owned(), format!("realise alice {CLOSURE}"), format!("swap_update alice {CLOSURE} node={}", id.get()), format!("mark_deployed alice {CLOSURE}"), + "unpin_closure alice".to_owned(), ] ); } @@ -424,7 +475,7 @@ async fn swap_applies_the_swarms_closure() { /// 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() { + for (case, host, why) in without_a_pinned_closure() { let (_dir, coord) = coordinator(); let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await; result.expect_err(case); @@ -450,8 +501,8 @@ async fn swap_marks_the_applied_closure_deployed_only_after_the_apply() { let id = node_id(); run_swap(&host, &coord, "alice", id).await.expect("swap ok"); assert_eq!( - host.calls().last().map(String::as_str), - Some(format!("mark_deployed alice {CLOSURE}").as_str()), + host.calls()[2], + format!("mark_deployed alice {CLOSURE}"), "{:?}", host.calls() ); @@ -489,8 +540,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(), 3, "{calls:?}"); - assert!(calls[2].starts_with("swap_update alice"), "{calls:?}"); + assert_eq!(calls.len(), 2, "{calls:?}"); + assert!(calls[1].starts_with("swap_update alice"), "{calls:?}"); } // --- Stop / StopForUpdate --------------------------------------------------- @@ -716,7 +767,7 @@ async fn pause_drain_times_out_ok_and_clears_pending() { // --- MetaSync --------------------------------------------------------------- #[tokio::test] -async fn meta_sync_only_prepares_the_rebuild() { +async fn meta_sync_prepares_the_rebuild_and_pins_the_declared_closure() { let host = FakeHost::up(); run_meta_sync(&host, "alice").await.expect("ok"); assert_eq!( @@ -725,6 +776,8 @@ async fn meta_sync_only_prepares_the_rebuild() { "refresh_archetype alice".to_owned(), "prepare_rebuild_dirs alice".to_owned(), "refresh_agent_set".to_owned(), + "closure_record alice".to_owned(), + format!("pin_closure alice {CLOSURE}"), ] ); } @@ -733,7 +786,7 @@ async fn meta_sync_only_prepares_the_rebuild() { 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(), 3); + assert_eq!(host.calls().len(), 5); } #[tokio::test] @@ -747,6 +800,69 @@ async fn meta_sync_stops_at_the_first_fatal_step() { } } +/// A rebuild without a closure to deploy fails in `MetaSync`, before anything is +/// stopped, and pins nothing. +#[tokio::test] +async fn meta_sync_without_a_closure_fails_and_pins_nothing() { + for (case, host, why) in without_a_record() { + let host = host.unpinned(); + let (result, log) = logging(run_meta_sync(&host, "alice")).await; + result.expect_err(case); + let errors = log.at(tracing::Level::ERROR); + assert_eq!(errors.len(), 1, "{case}: {errors:?}"); + assert!(errors[0].contains(why), "{case}: {errors:?}"); + assert!( + host.pinned_closure("alice").is_none(), + "{case}: {:?}", + host.calls() + ); + } +} + +/// The record is read once for a rebuild. A record that changes after `MetaSync` +/// (the builder finished another build) leaves `Prebuild` and `Swap` on the +/// closure that was pinned, and the marker names what was applied. +#[tokio::test] +async fn a_record_that_changes_mid_rebuild_does_not_change_what_swap_applies() { + const NEWER: &str = "/nix/store/zyxwvsrqpnmlkjihgfdcbadfhijklmn-nixos-system-alice"; + let (_dir, coord) = coordinator(); + let host = FakeHost::up().unpinned(); + run_meta_sync(&host, "alice").await.expect("pinned"); + run_prebuild(&host, "alice").await.expect("prebuilt"); + host.builder_records(NEWER); + let id = node_id(); + run_swap(&host, &coord, "alice", id).await.expect("swapped"); + let calls = host.calls(); + assert_eq!( + calls + .iter() + .filter(|c| c.starts_with("closure_record")) + .count(), + 1, + "the record is read once: {calls:?}" + ); + assert!( + calls.contains(&format!("swap_update alice {CLOSURE} node={}", id.get())), + "{calls:?}" + ); + assert!( + calls.contains(&format!("mark_deployed alice {CLOSURE}")), + "{calls:?}" + ); + assert!( + !calls.iter().any(|c| c.contains(NEWER)), + "control: the newer path is never touched: {calls:?}" + ); + + // Control: the changed record is what the next rebuild pins. + run_meta_sync(&host, "alice").await.expect("pinned again"); + assert!( + host.calls().contains(&format!("pin_closure alice {NEWER}")), + "{:?}", + host.calls() + ); +} + // --- Provision / Create / WriteDropin ---------------------------------------- #[tokio::test] @@ -1034,19 +1150,17 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() { "refresh_archetype alice", "prepare_rebuild_dirs alice", "refresh_agent_set", + "closure_record alice", + "pin_closure alice", ], ), ( alice(|agent| NodeKind::Prebuild { agent }), - &["is_running alice", "closure_record alice"], + &["is_running alice", "realise alice"], ), ( alice(|agent| NodeKind::Swap { agent }), - &[ - "closure_record alice", - "swap_update alice", - "mark_deployed alice", - ], + &["realise alice", "swap_update alice", "mark_deployed alice"], ), (alice(|agent| NodeKind::RebuildBookkeeping { agent }), &[]), ( diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 175ce7c9..f9149915 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1560,3 +1560,23 @@ fn a_rebuild_is_live_for_its_own_agent_only() { ); assert!(!q.rebuild_live("bob")); } + +/// Agents whose running closure differs from the declaration are rebuilt in one +/// sweep. Each rebuild's window holds a build slot, so `buildSlots` caps how +/// many swap at once, however many agents the sweep rebuilds. +#[test] +fn a_sweep_of_rebuilds_holds_one_build_slot_per_agent_window() { + let q = JobQueue::new(1); + let agents: Vec = ["alice", "bob", "carol"].map(str::to_owned).to_vec(); + insert(&q, |builder| { + crate::workers::auto_update::boot_nodes(builder, &agents, Vec::new()); + }); + assert_eq!( + declared_resources_of_kind(&q, "agent_window") + .iter() + .filter(|r| r.contains(&Resource::BuildSlot)) + .count(), + 3, + "every window declares the build slot" + ); +} diff --git a/hive-c0re/src/paths.rs b/hive-c0re/src/paths.rs index a5b770f5..ee9c4574 100644 --- a/hive-c0re/src/paths.rs +++ b/hive-c0re/src/paths.rs @@ -123,13 +123,24 @@ pub fn applied_root() -> PathBuf { } /// `applied/..closure` — marker holding the store path of the swarm's -/// closure an agent was last swapped to or created from, which -/// [`crate::attic::closure::pending`] compares the record's with. +/// closure an agent was last swapped to or created from (or, once, the one its +/// container profile pointed at), which [`crate::attic::closure::pending`] +/// compares the record's with. #[must_use] pub fn applied_closure_marker(name: &str) -> PathBuf { applied_root().join(format!(".{name}.closure")) } +/// `/nix/var/nix/profiles/per-container/h-/system` — the link `nix-env --set` +/// points at the closure an agent's container runs, as `hive-priv` writes it +/// on a swap and `nixos-container create` on a create. +#[must_use] +pub fn container_profile(name: &str) -> PathBuf { + Path::new("/nix/var/nix/profiles/per-container") + .join(format!("{}{name}", crate::lifecycle::AGENT_PREFIX)) + .join("system") +} + /// `meta/` — plain state beside the hive's other files (`topology.json`, /// the swarm limits, the permission seed files) and, on a hive that predates /// `internal/meta`, a leftover git repo. diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index a0baa8a6..9e72c68f 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -15,8 +15,8 @@ //! submit path upgrades a stale start to rebuild+start). //! 2. **Agents running the declared closure** are skipped. //! 3. **A declaration the hive cannot read or act on** (no connection to the -//! swarm queue, no record, the closure not in the cache, no marker to -//! compare with) rebuilds nothing. +//! swarm queue, no record, the closure not in the cache, no readable running +//! closure to compare with) rebuilds nothing. //! 4. **A declaration is attempted once per process** by the periodic pass, and //! never while a rebuild of the agent is queued or running, so a closure //! that fails to deploy does not restart its agent every interval. diff --git a/hive-meta-flake/README.md b/hive-meta-flake/README.md index f3a3cec8..24e37f47 100644 --- a/hive-meta-flake/README.md +++ b/hive-meta-flake/README.md @@ -6,8 +6,9 @@ archetype. Pure string rendering, with no IO and no environment reads. The caller passes the hyperhive pin, the agents with their pins, and the values every agent -container gets. swarm-controller calls `render_flake`; hive-c0re reads the -pins of the flake it renders (`pinned_rev`, `builds_agent_at`). +container gets. swarm-controller calls `render_flake` and reads the pins of the +flake it renders (`pinned_rev`). hive-c0re uses only the agent spec and port +helpers. `testdata/meta-flake.nix.in` is the render for one archetype agent with a placeholder for the hyperhive pin. A unit test keeps it equal to `render_flake`'s diff --git a/swarm-controller/src/agent_archetype.rs b/swarm-controller/src/agent_archetype.rs index d355634b..ac3abfff 100644 --- a/swarm-controller/src/agent_archetype.rs +++ b/swarm-controller/src/agent_archetype.rs @@ -2,8 +2,8 @@ //! `agent-archetype` (`swarm_queue_client::agent_archetype`), which this //! daemon alone writes. `POST /api/agents` checks the choice against the //! `agent-archetypes` record and stores it before it queues the creation; a -//! destroy deletes it. The hive reads it to build the agent from the -//! archetype's module, with no config repo. +//! destroy deletes it. The hive reads it to know the agent has no +//! config repo; the swarm builds the agent from the archetype's module. use problem_details::ProblemDetails; use swarm_queue_client::agent_archetype::{self, ArchetypeRef}; diff --git a/swarm-controller/src/meta_flake.rs b/swarm-controller/src/meta_flake.rs index c46090a3..4417a48e 100644 --- a/swarm-controller/src/meta_flake.rs +++ b/swarm-controller/src/meta_flake.rs @@ -850,7 +850,7 @@ mod tests { ); assert!( reads.contains(&("internal/meta".to_owned(), None)), - "the meta repo is read on its default branch, as hive-c0re reads it" + "the meta repo is read on its default branch" ); }