diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 459f8982..00d1d73e 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -297,6 +297,42 @@ Mixed versions: A record written before the field existed also reads as `false`. +### Rebuild on a cache miss + +The cache drops a closure nobody fetched for its retention period, but the +record in `agent-closures` stays. Every ten minutes `swarm-controller` checks +each agent a hive places: when the record is from `internal/meta`'s head and +says the builder pushed it, the controller fetches the narinfo from +`//.narinfo` with the pull token it reads from +`secret/swarm/services/attic/` (the same four leaves it +[copies to the hives](#what-hives-do-with-the-cache); with any missing, or a URL +that isn't `https`, the pass does nothing). A 404 is a miss, and the controller +sends the builder `build-meta` for that agent and the record's own commit. The +builder builds it and pushes as for any meta build. Any other answer or a failed +request isn't a miss, and rebuilds nothing. The next ordinary build handles a record +from an older commit, or one with `cached` false. + +A pass sends at most three builds, one after the other. After an agent's rebuild +failed, got no answer or built without pushing, the pass skips it for 30 minutes, then +for twice as long after each further failure, up to 12 hours. The wait is in memory, +so a controller restart tries every agent again. The first pass runs ten +minutes after the controller starts. + +The record keeps naming what hives deploy. A rebuild that lands on the record's +path writes nothing. A rebuild that lands on another path means the build isn't +reproducible: the controller logs a warning, and replaces the record only if it +still holds the same commit and no write came since the controller read it. The +record otherwise stays as it was. + +The controller counts each rebuild in `hyperhive.builder.cache_miss_rebuilds` with +the label `outcome`: `rebuilt`, `path_changed` (the record now names the new path), +`path_changed_kept` (the record stayed), `not_pushed`, `failed` (`nix` failed) or +`unavailable` (the builder was busy, absent or gave no answer). + +Mixed versions: a builder needs no change, since a meta build already names +its commit. An old controller does nothing about a miss, as before. A hive +sees nothing new. + ### The swarm's own cache `services.hyperhive.deploy.attic.enable` runs an attic server in a `swarm-attic` @@ -333,8 +369,8 @@ host pushes to it. Storage is local disk in the container's state (`/var/lib/swarm-attic`). A store path nobody added or fetched for 14 days goes at the next garbage collection, -which runs every 12 hours (`deploy.attic.retentionPeriod`). The next build of -that agent pushes a closure the cache lost. +which runs every 12 hours (`deploy.attic.retentionPeriod`). The controller +[rebuilds a closure the cache lost](#rebuild-on-a-cache-miss). The gateway terminates TLS with a swarm services certificate and forwards to `127.0.0.1:8910` (`deploy.attic.port`) with no login in front: attic checks its diff --git a/docs/swarm/credentials-reference.md b/docs/swarm/credentials-reference.md index 7c75ab53..ee3ecc57 100644 --- a/docs/swarm/credentials-reference.md +++ b/docs/swarm/credentials-reference.md @@ -49,7 +49,7 @@ the rest of the cell says how. | `swarm/services/review-classifier/forge/token` | `swarm-controller`, at start and every 5 minutes | `swarm-controller` | ✅ when missing or no longer matching the forge | ✅ read on every review | | `swarm/services/builder/forge/token` | `swarm-controller`, at start and every 5 minutes | `swarm-builder`, `swarm-controller` | ✅ when missing or no longer matching the forge | ✅ read on every build | | `swarm/services/attic/push-token` | `swarm-bao-attic-tokens` on the cache's host, when none is stored; with an external cache, you, in the store's UI (field `value`), or `bao kv put` with `value=-` and the token on stdin | `swarm-builder` | ❌ write a new token; a token the cache refuses fails the push, and the build stays green with `cached` false | ✅ read on every push; with both `swarm.attic` options null, nothing reads it | -| `swarm/services/attic/pull-token` | `swarm-bao-attic-tokens` on the cache's host, when none is stored; read-only, ten years; with an external cache, you, in the store's UI (field `value`), or `bao kv put` with `value=-` and the value on stdin | `swarm-controller`, which copies it to each hive | ❌ write a new token | ✅ copied to the hives every 5 minutes | +| `swarm/services/attic/pull-token` | `swarm-bao-attic-tokens` on the cache's host, when none is stored; read-only, ten years; with an external cache, you, in the store's UI (field `value`), or `bao kv put` with `value=-` and the value on stdin | `swarm-controller`, to copy it and to check narinfos | ❌ write a new token | ✅ copied to the hives every 5 minutes | | `swarm/services/attic/url` | `swarm-bao-attic-tokens` on the cache's host, when none is stored (`https://`); with an external cache, you, in the store's UI (field `value`), or `bao kv put` with `value=-` and the value on stdin (scheme included, no trailing slash) | `swarm-controller`, which copies it to each hive | ❌ delete it and restart `swarm-bao-attic-tokens`, or write the new value | ✅ copied to the hives every 5 minutes | | `swarm/services/attic/cache` | `swarm-bao-attic-tokens` on the cache's host, when none is stored (`swarm.attic.cache`); with an external cache, you, in the store's UI (field `value`), or `bao kv put` with `value=-` and the value on stdin | `swarm-controller`, which copies it to each hive | ❌ as the URL | ✅ copied to the hives every 5 minutes | | `swarm/services/attic/public-key` | `swarm-bao-attic-tokens` on the cache's host, when none is stored (the key `attic cache info` prints); with an external cache, you, in the store's UI (field `value`), or `bao kv put` with `value=-` and the value on stdin (`:`) | `swarm-controller`, which copies it to each hive | ❌ as the URL. A cache recreated with a new key leaves the stored one stale: every hive then reports `bad-signature` | ✅ copied to the hives every 5 minutes | diff --git a/nix/host-modules/swarm-attic.nix b/nix/host-modules/swarm-attic.nix index d931fb5f..de203d2e 100644 --- a/nix/host-modules/swarm-attic.nix +++ b/nix/host-modules/swarm-attic.nix @@ -166,7 +166,8 @@ in description = '' How long a store path stays in the cache after it was added or last fetched. A garbage collection every twelve hours removes what is older; - a build the cache no longer holds is rebuilt on the next miss. + swarm-controller rebuilds a closure the cache no longer holds and + the builder pushes it again. ''; }; }; diff --git a/swarm-controller/src/cache_miss.rs b/swarm-controller/src/cache_miss.rs new file mode 100644 index 00000000..d313dd63 --- /dev/null +++ b/swarm-controller/src/cache_miss.rs @@ -0,0 +1,997 @@ +//! Rebuild an agent's closure when the binary cache no longer holds it. +//! +//! The cache drops a path after its retention period, but the agent's record in +//! [`swarm_wire::agent_closures`] stays, so a hive that rebuilds an idle agent +//! finds a record that names a closure nobody can download. Every +//! [`PASS_INTERVAL`] this asks the cache for the narinfo of each placed agent's +//! record and, on a 404, sends the builder the same `internal/meta` commit +//! again. The builder builds and pushes as for any meta build. +//! +//! Only a record a hive would use is checked: the one built from `internal/meta`'s +//! head, which says the builder pushed it. Only a 404 is a miss; an unreachable +//! cache or a refused token is "cannot tell" and rebuilds nothing. +//! +//! A pass sends at most [`MAX_REBUILDS_PER_PASS`] builds, one after the other. +//! An agent whose rebuild failed, or built without pushing, is left alone for +//! [`BACKOFF_BASE`], doubling per failure up to [`BACKOFF_MAX`]. The backoff +//! lives in memory, so a restart tries every agent once more. +//! +//! The record names what hives deploy, so the pass changes it as little as it +//! can. A rebuild that lands on the record's own path writes nothing. One that +//! lands on another path (a build that is not reproducible) is logged at warn, +//! and replaces the record only if the record still holds the same commit and +//! no other write came in since it was read. + +use std::collections::BTreeMap; +use std::future::Future; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant}; + +use anyhow::{Context as _, Result}; +use async_nats::jetstream::kv::Store; +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, MeterProvider as _}; +use opentelemetry_sdk::metrics::SdkMeterProvider; +use swarm_secret_client::attic::{Leaf, service_path}; +use swarm_wire::agent_closures::AgentClosure; +use swarm_wire::builder::{BuildFromMeta, FromMetaReply}; + +use crate::agent_renewal::unix_now; +use crate::forge; +use crate::meta_build::{MetaBuilds as _, MetaRepo as _}; +use crate::wanted::WantedWriter; + +/// How long the loop waits between passes, which is also the wait before the +/// first. +const PASS_INTERVAL: Duration = Duration::from_mins(10); + +/// The most builds one pass sends, whether or not they succeed. +const MAX_REBUILDS_PER_PASS: usize = 3; + +/// How long an agent is left alone after its first failed rebuild. +const BACKOFF_BASE: Duration = Duration::from_mins(30); + +/// The longest an agent is left alone. +const BACKOFF_MAX: Duration = Duration::from_hours(12); + +/// How long one narinfo request may take. +const FETCH_TIMEOUT: Duration = Duration::from_secs(20); + +/// A record with the KV revision it was read at. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Stored { + closure: AgentClosure, + revision: u64, +} + +/// What a pass reads and does outside the process, so a test can stand in for it. +trait World { + /// The agents a hive places. + fn placed_agents(&self) -> impl Future>> + Send; + /// `internal/meta`'s head, with the builder's read on it granted; `None` + /// when the repo has no commit. + fn meta_head(&self) -> impl Future>> + Send; + /// `agent`'s record; `None` when it has none. + fn record(&self, agent: &str) -> impl Future>> + Send; + /// Whether the cache holds the closure at `store_path`. `Err` when it can't + /// say. + fn holds(&self, store_path: &str) -> impl Future> + Send; + /// Ask the builder to build `agent` at `meta_rev` and push the closure. + fn rebuild( + &self, + agent: &str, + meta_rev: &str, + ) -> impl Future> + Send; + /// Replace `agent`'s record if it is still at `revision`. `Ok(false)` when + /// another write came first. + fn replace( + &self, + agent: &str, + closure: &AgentClosure, + revision: u64, + ) -> impl Future> + Send; +} + +/// What became of one rebuild: the `outcome` label of the counter. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Outcome { + /// Built and pushed, at the path the record names. + Rebuilt, + /// Built and pushed at another path, and the record now names it. + PathChanged, + /// Built and pushed at another path, and the record was left as it was. + PathChangedKept, + /// Built, but the builder's push failed or it has no cache. + NotPushed, + /// `nix` failed. + Failed, + /// The builder gave no verdict: busy, not listening, no reply. + Unavailable, +} + +impl Outcome { + const fn label(self) -> &'static str { + match self { + Self::Rebuilt => "rebuilt", + Self::PathChanged => "path_changed", + Self::PathChangedKept => "path_changed_kept", + Self::NotPushed => "not_pushed", + Self::Failed => "failed", + Self::Unavailable => "unavailable", + } + } + + /// Whether the agent is left alone for a while. + const fn backs_off(self) -> bool { + matches!(self, Self::NotPushed | Self::Failed | Self::Unavailable) + } +} + +/// Which agents are left alone until when, and after how many failures. +#[derive(Debug, Default)] +struct Backoff(BTreeMap); + +impl Backoff { + fn waiting(&self, agent: &str, now: Instant) -> bool { + self.0.get(agent).is_some_and(|(_, until)| *until > now) + } + + fn failed(&mut self, agent: &str, now: Instant) { + let failures = self.0.get(agent).map_or(1, |(n, _)| n.saturating_add(1)); + let delay = BACKOFF_BASE + .saturating_mul(2u32.saturating_pow(failures - 1)) + .min(BACKOFF_MAX); + self.0.insert(agent.to_owned(), (failures, now + delay)); + } + + fn cleared(&mut self, agent: &str) { + self.0.remove(agent); + } +} + +/// The counter `hyperhive.builder.cache_miss_rebuilds` (`outcome`), or nothing +/// where no collector is configured. +#[derive(Clone)] +struct Metrics(Option>); + +struct Instruments { + /// Exports only while it lives. + _provider: SdkMeterProvider, + rebuilds: Counter, +} + +impl Instruments { + fn new(provider: &SdkMeterProvider) -> Self { + Self { + _provider: provider.clone(), + rebuilds: provider + .meter("hyperhive.builder") + .u64_counter("hyperhive.builder.cache_miss_rebuilds") + .with_description( + "rebuilds the controller sent because the binary cache lost a closure, by outcome", + ) + .build(), + } + } +} + +static GLOBAL: OnceLock>> = OnceLock::new(); + +impl Metrics { + /// The process's counter, building its exporter on the first call. + fn global() -> Self { + Self( + GLOBAL + .get_or_init(|| { + let endpoint = crate::vcs_metrics::endpoint()?; + match crate::vcs_metrics::build_provider(crate::vcs_metrics::EXPORT_INTERVAL) { + Ok(provider) => { + tracing::info!(%endpoint, "otel cache-miss metrics: exporter enabled"); + Some(Arc::new(Instruments::new(&provider))) + } + Err(e) => { + tracing::warn!(error = ?e, "otel cache-miss metrics: exporter init failed"); + None + } + } + }) + .clone(), + ) + } + + fn count(&self, outcome: Outcome) { + if let Some(inst) = &self.0 { + inst.rebuilds + .add(1, &[KeyValue::new("outcome", outcome.label())]); + } + } +} + +/// One pass over the placed agents. The returned outcomes are the rebuilds it +/// sent, in order. +/// +/// # Errors +/// A read that failed before any agent was looked at (the placements, the +/// head). A failure on one agent is logged and the pass goes on. +async fn pass( + world: &impl World, + backoff: &mut Backoff, + metrics: &Metrics, + now: Instant, +) -> Result> { + let Some(head) = world.meta_head().await.context("reading internal/meta")? else { + return Ok(Vec::new()); + }; + let agents = world + .placed_agents() + .await + .context("reading where the hives place agents")?; + let mut sent = Vec::new(); + for agent in agents { + if backoff.waiting(&agent, now) { + continue; + } + let stored = match world.record(&agent).await { + Ok(Some(stored)) => stored, + Ok(None) => continue, + Err(e) => { + tracing::warn!(%agent, error = %format!("{e:#}"), "cache miss: reading the record failed"); + continue; + } + }; + let record = &stored.closure; + if !record.cached || record.meta_rev != head { + continue; + } + match world.holds(&record.store_path).await { + Ok(true) => { + backoff.cleared(&agent); + continue; + } + Ok(false) => {} + Err(e) => { + tracing::warn!(%agent, error = %format!("{e:#}"), "cache miss: asking the cache failed"); + continue; + } + } + if sent.len() >= MAX_REBUILDS_PER_PASS { + tracing::info!( + limit = MAX_REBUILDS_PER_PASS, + "cache miss: pass limit reached; the rest wait for the next pass" + ); + break; + } + tracing::warn!( + %agent, meta_rev = %record.meta_rev, store_path = %record.store_path, + "cache miss: the cache lost the agent's closure; rebuilding" + ); + let outcome = rebuild(world, &agent, stored).await; + metrics.count(outcome); + if outcome.backs_off() { + backoff.failed(&agent, now); + } else { + backoff.cleared(&agent); + } + sent.push((agent, outcome)); + } + Ok(sent) +} + +/// Send the builder the record's own commit, and apply the rule for the record +/// to what it answers. +async fn rebuild(world: &impl World, agent: &str, stored: Stored) -> Outcome { + let meta_rev = &stored.closure.meta_rev; + let reply = match world.rebuild(agent, meta_rev).await { + Ok(reply) => reply, + Err(why) => { + tracing::warn!(%agent, %meta_rev, %why, "cache miss: the rebuild got no answer"); + return Outcome::Unavailable; + } + }; + let (out_path, system, cached) = match reply { + FromMetaReply::Built { + out_path, + system, + cached, + } => (out_path, system, cached), + FromMetaReply::Failed { error_tail } => { + tracing::warn!(%agent, %meta_rev, %error_tail, "cache miss: the rebuild failed"); + return Outcome::Failed; + } + FromMetaReply::Unavailable { reason } => { + tracing::warn!(%agent, %meta_rev, %reason, "cache miss: the rebuild did not run"); + return Outcome::Unavailable; + } + }; + if !cached { + tracing::warn!(%agent, %meta_rev, %out_path, "cache miss: rebuilt, but the builder did not push it"); + return Outcome::NotPushed; + } + if out_path == stored.closure.store_path { + tracing::info!(%agent, %meta_rev, %out_path, "cache miss: rebuilt and pushed"); + return Outcome::Rebuilt; + } + tracing::warn!( + %agent, %meta_rev, recorded = %stored.closure.store_path, %out_path, + "cache miss: the rebuild is at another path than the record; the build is not reproducible" + ); + let now = AgentClosure { + meta_rev: meta_rev.clone(), + store_path: out_path, + system, + built_unix: unix_now(), + cached: true, + }; + match replace_if_current(world, agent, &now).await { + Ok(true) => Outcome::PathChanged, + Ok(false) => { + tracing::warn!(%agent, %meta_rev, "cache miss: the record changed meanwhile and is left as it is"); + Outcome::PathChangedKept + } + Err(e) => { + tracing::warn!(%agent, %meta_rev, error = %format!("{e:#}"), "cache miss: the record could not be replaced"); + Outcome::PathChangedKept + } + } +} + +/// Replace `agent`'s record with `now`, read again for the write: only when it +/// still holds `now`'s commit and no write has come since. +async fn replace_if_current(world: &impl World, agent: &str, now: &AgentClosure) -> Result { + let Some(current) = world.record(agent).await? else { + return Ok(false); + }; + if current.closure.meta_rev != now.meta_rev { + return Ok(false); + } + world.replace(agent, now, current.revision).await +} + +/// The cache's address and pull token, read from the store. +struct Facts { + url: String, + cache: String, + token: String, +} + +impl Facts { + /// The facts if all three are stored, non-blank, and the URL is `https`: + /// the token goes out as a basic-auth password. + fn from_values( + token: Option, + url: Option, + cache: Option, + ) -> Option { + let clean = |v: Option| v.map(|v| v.trim().to_owned()).filter(|v| !v.is_empty()); + let url = clean(url)?.trim_end_matches('/').to_owned(); + url.starts_with("https://").then_some(())?; + Some(Self { + url, + cache: clean(cache)?, + token: clean(token)?, + }) + } + + async fn read(store: &impl crate::attic_copy::FactStore) -> Result> { + let read = async |leaf| { + let path = service_path(leaf)?; + store + .read(&path) + .await + .with_context(|| format!("reading {path}")) + }; + Ok(Self::from_values( + read(Leaf::PullToken).await?, + read(Leaf::Url).await?, + read(Leaf::Cache).await?, + )) + } +} + +/// The hash `/nix/store/-` is named by in the cache. +fn path_hash(store_path: &str) -> Result<&str> { + let base = store_path.rsplit('/').next().unwrap_or_default(); + match base.split_once('-') { + Some((hash, name)) if hash.len() == 32 && !name.is_empty() => Ok(hash), + _ => anyhow::bail!("{store_path:?} is not a store path"), + } +} + +/// The real world of one pass. +struct Live { + queue: async_nats::Client, + closures: Store, + wanted: Arc, + hives: Vec, + forge: Arc, + http: reqwest::Client, + facts: Facts, +} + +impl World for Live { + async fn placed_agents(&self) -> Result> { + let placed = crate::jobq::placements(&self.wanted, &self.hives).await?; + let agents: std::collections::BTreeSet<_> = + placed.into_iter().map(|(_, agent)| agent).collect(); + Ok(agents.into_iter().collect()) + } + + async fn meta_head(&self) -> Result> { + self.forge.grant_builder_read().await?; + self.forge.head().await + } + + async fn record(&self, agent: &str) -> Result> { + let (closure, revision) = crate::schedules::read_with(&self.closures, agent, |bytes| { + Ok(Some(serde_json::from_slice::(bytes)?)) + }) + .await?; + Ok(closure + .zip(revision) + .map(|(closure, revision)| Stored { closure, revision })) + } + + async fn holds(&self, store_path: &str) -> Result { + let url = format!( + "{}/{}/{}.narinfo", + self.facts.url, + self.facts.cache, + path_hash(store_path)? + ); + let response = self + .http + .get(&url) + .basic_auth("swarm-controller", Some(&self.facts.token)) + .send() + .await + .map_err(|e| anyhow::anyhow!("GET {url}: {}", e.without_url()))?; + match response.status() { + reqwest::StatusCode::OK => Ok(true), + reqwest::StatusCode::NOT_FOUND => Ok(false), + status => anyhow::bail!("GET {url}: {status}"), + } + } + + async fn rebuild(&self, agent: &str, meta_rev: &str) -> Result { + let request = serde_json::to_vec(&BuildFromMeta { + agent: agent.to_owned(), + meta_rev: meta_rev.to_owned(), + }) + .map_err(|e| format!("encoding the build request: {e}"))?; + self.queue.build(request).await + } + + async fn replace(&self, agent: &str, closure: &AgentClosure, revision: u64) -> Result { + crate::schedules::write(&self.closures, agent, closure, Some(revision)).await + } +} + +/// Check the placed agents every [`PASS_INTERVAL`], starting one interval from +/// now. A pass that fails is logged and the next tick tries again; it never stops +/// the daemon. +pub fn spawn( + queue: async_nats::Client, + wanted: Arc, + hives: Vec, + forge: Arc, +) { + tokio::spawn(async move { + let metrics = Metrics::global(); + let mut backoff = Backoff::default(); + let mut ticker = + tokio::time::interval_at(tokio::time::Instant::now() + PASS_INTERVAL, PASS_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + ticker.tick().await; + let world = match live(&queue, &wanted, &hives, &forge).await { + Ok(Some(world)) => world, + Ok(None) => { + tracing::debug!("cache miss: the swarm has no usable cache facts"); + continue; + } + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "cache miss: setting up the pass failed"); + continue; + } + }; + match pass(&world, &mut backoff, &metrics, Instant::now()).await { + Ok(sent) => tracing::debug!(rebuilds = sent.len(), "cache miss: pass done"), + Err(e) => tracing::warn!(error = %format!("{e:#}"), "cache miss: pass failed"), + } + } + }); +} + +/// The world for one pass, or `None` when the store holds no usable cache facts. +async fn live( + queue: &async_nats::Client, + wanted: &Arc, + hives: &[String], + forge: &Arc, +) -> Result> { + let store = crate::store::connect() + .await + .context("logging in to the swarm secret store")?; + let Some(facts) = Facts::read(&store).await? else { + return Ok(None); + }; + Ok(Some(Live { + queue: queue.clone(), + closures: swarm_queue_client::agent_closures::open_or_create(queue).await?, + wanted: Arc::clone(wanted), + hives: hives.to_vec(), + forge: Arc::clone(forge), + http: reqwest::Client::builder() + .timeout(FETCH_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .build()?, + facts, + })) +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeSet; + use std::sync::Mutex; + + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader}; + use tracing_subscriber::layer::SubscriberExt as _; + + use super::*; + + const HEAD: &str = "0123456789abcdef0123456789abcdef01234567"; + const OLD: &str = "fedcba9876543210fedcba9876543210fedcba98"; + + fn path(name: &str) -> String { + format!("/nix/store/{:a<32}-{name}", "") + } + + fn record(rev: &str, name: &str, cached: bool) -> AgentClosure { + AgentClosure { + meta_rev: rev.to_owned(), + store_path: path(name), + system: "x86_64-linux".to_owned(), + built_unix: 7, + cached, + } + } + + /// The swarm as a pass sees it: records, what the cache holds, and what the + /// builder answers. Every build sent and every record write is kept. + struct Fake { + head: Option, + records: Mutex>, + cache: Mutex>, + cache_down: bool, + /// What a build answers, per agent; a missing agent builds to its + /// record's path and pushes it. + answers: Mutex>>, + /// A record write that lands between the build and the replace. + racing: Mutex>, + /// Whether another write beats the replace. + replace_loses: std::sync::atomic::AtomicBool, + builds: Mutex>, + writes: Mutex>, + } + + impl Fake { + /// Every agent has a cached record at `HEAD`, and the cache holds none + /// of the closures. + fn missing(agents: &[&str]) -> Self { + let records = agents + .iter() + .map(|a| { + let stored = Stored { + closure: record(HEAD, a, true), + revision: 4, + }; + ((*a).to_owned(), stored) + }) + .collect(); + Self { + head: Some(HEAD.to_owned()), + records: Mutex::new(records), + cache: Mutex::default(), + cache_down: false, + answers: Mutex::default(), + racing: Mutex::default(), + replace_loses: std::sync::atomic::AtomicBool::default(), + builds: Mutex::default(), + writes: Mutex::default(), + } + } + + fn holding(self, agent: &str) -> Self { + self.cache.lock().unwrap().insert(path(agent)); + self + } + + fn answering(self, agent: &str, answer: FromMetaReply) -> Self { + self.answers + .lock() + .unwrap() + .insert(agent.to_owned(), Ok(answer)); + self + } + + fn with_record(self, agent: &str, closure: AgentClosure) -> Self { + self.records.lock().unwrap().insert( + agent.to_owned(), + Stored { + closure, + revision: 4, + }, + ); + self + } + + fn builds(&self) -> Vec<(String, String)> { + self.builds.lock().unwrap().clone() + } + + fn writes(&self) -> Vec<(String, AgentClosure)> { + self.writes.lock().unwrap().clone() + } + } + + fn built(out_path: &str, cached: bool) -> FromMetaReply { + FromMetaReply::Built { + out_path: out_path.to_owned(), + system: "x86_64-linux".to_owned(), + cached, + } + } + + impl World for Fake { + async fn placed_agents(&self) -> Result> { + Ok(self.records.lock().unwrap().keys().cloned().collect()) + } + + async fn meta_head(&self) -> Result> { + Ok(self.head.clone()) + } + + async fn record(&self, agent: &str) -> Result> { + Ok(self.records.lock().unwrap().get(agent).cloned()) + } + + async fn holds(&self, store_path: &str) -> Result { + anyhow::ensure!(!self.cache_down, "the cache answered 503"); + Ok(self.cache.lock().unwrap().contains(store_path)) + } + + async fn rebuild(&self, agent: &str, meta_rev: &str) -> Result { + self.builds + .lock() + .unwrap() + .push((agent.to_owned(), meta_rev.to_owned())); + let answer = self.answers.lock().unwrap().get(agent).cloned(); + let answer = answer.unwrap_or_else(|| Ok(built(&path(agent), true))); + if let Ok(FromMetaReply::Built { + out_path, + cached: true, + .. + }) = &answer + { + self.cache.lock().unwrap().insert(out_path.clone()); + } + if let Some((racer, closure)) = self.racing.lock().unwrap().take() { + let mut records = self.records.lock().unwrap(); + records.insert( + racer, + Stored { + closure, + revision: 5, + }, + ); + } + answer + } + + async fn replace( + &self, + agent: &str, + closure: &AgentClosure, + revision: u64, + ) -> Result { + let mut records = self.records.lock().unwrap(); + if self.replace_loses.load(std::sync::atomic::Ordering::SeqCst) + || records[agent].revision != revision + { + return Ok(false); + } + records.insert( + agent.to_owned(), + Stored { + closure: closure.clone(), + revision: revision + 1, + }, + ); + self.writes + .lock() + .unwrap() + .push((agent.to_owned(), closure.clone())); + Ok(true) + } + } + + async fn pass_at(world: &Fake, backoff: &mut Backoff, now: Instant) -> Vec<(String, Outcome)> { + pass(world, backoff, &Metrics(None), now) + .await + .expect("the pass reads") + } + + fn rebuilt(agent: &str) -> (String, Outcome) { + (agent.to_owned(), Outcome::Rebuilt) + } + + #[tokio::test] + async fn a_missing_closure_is_rebuilt_once_at_the_records_commit() { + let world = Fake::missing(&["iris"]); + let mut backoff = Backoff::default(); + let now = Instant::now(); + assert_eq!(pass_at(&world, &mut backoff, now).await, [rebuilt("iris")]); + assert_eq!(world.builds(), [("iris".to_owned(), HEAD.to_owned())]); + // The build pushed it, so the next pass finds it. + assert_eq!(pass_at(&world, &mut backoff, now).await, []); + assert_eq!(world.builds().len(), 1); + assert!(world.writes().is_empty(), "the record already names it"); + } + + #[tokio::test] + async fn a_closure_the_cache_holds_is_left_alone() { + let world = Fake::missing(&["iris", "argus"]).holding("iris"); + let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await; + // Control: the other agent's miss is rebuilt in the same pass. + assert_eq!(sent, [rebuilt("argus")]); + assert_eq!(world.builds(), [("argus".to_owned(), HEAD.to_owned())]); + } + + #[tokio::test] + async fn only_the_record_a_hive_would_use_is_checked() { + let world = Fake::missing(&["iris", "argus", "atlas", "muede"]) + .with_record("iris", record(OLD, "iris", true)) + .with_record("argus", record(HEAD, "argus", false)); + let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await; + // `iris` is from another commit, `argus` was never pushed. + assert_eq!(sent, [rebuilt("atlas"), rebuilt("muede")]); + } + + #[tokio::test] + async fn a_cache_that_cannot_answer_is_not_a_miss() { + let mut world = Fake::missing(&["iris"]); + world.cache_down = true; + let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await; + assert_eq!(sent, []); + assert!(world.builds().is_empty()); + } + + #[tokio::test] + async fn no_head_means_no_pass() { + let mut world = Fake::missing(&["iris"]); + world.head = None; + assert_eq!( + pass_at(&world, &mut Backoff::default(), Instant::now()).await, + [] + ); + } + + #[tokio::test] + async fn a_pass_sends_at_most_its_limit_and_the_next_takes_the_rest() { + let agents = ["a", "b", "c", "d", "e"]; + let world = Fake::missing(&agents); + let mut backoff = Backoff::default(); + let now = Instant::now(); + assert_eq!(MAX_REBUILDS_PER_PASS, 3); + let first = pass_at(&world, &mut backoff, now).await; + assert_eq!(first, [rebuilt("a"), rebuilt("b"), rebuilt("c")]); + let second = pass_at(&world, &mut backoff, now).await; + assert_eq!(second, [rebuilt("d"), rebuilt("e")]); + assert_eq!(world.builds().len(), 5); + } + + #[tokio::test] + async fn a_failed_rebuild_waits_and_the_wait_doubles() { + let world = Fake::missing(&["iris"]); + world + .answers + .lock() + .unwrap() + .insert("iris".to_owned(), Err("no builder".to_owned())); + let mut backoff = Backoff::default(); + let t0 = Instant::now(); + let unavailable = [("iris".to_owned(), Outcome::Unavailable)]; + assert_eq!(pass_at(&world, &mut backoff, t0).await, unavailable); + // No tight loop: every pass inside the wait sends nothing. + for minutes in [1, 10, 29] { + let at = t0 + Duration::from_mins(minutes); + assert_eq!(pass_at(&world, &mut backoff, at).await, []); + } + assert_eq!(world.builds().len(), 1); + // After the wait it tries again, and the next wait is twice as long. + let t1 = t0 + BACKOFF_BASE; + assert_eq!(pass_at(&world, &mut backoff, t1).await, unavailable); + assert_eq!(pass_at(&world, &mut backoff, t1 + BACKOFF_BASE).await, []); + assert_eq!(world.builds().len(), 2); + assert_eq!( + pass_at(&world, &mut backoff, t1 + BACKOFF_BASE * 2).await, + unavailable + ); + assert_eq!(world.builds().len(), 3); + } + + #[tokio::test] + async fn the_wait_stops_growing_at_its_cap() { + let mut backoff = Backoff::default(); + let t0 = Instant::now(); + for _ in 0..40 { + backoff.failed("iris", t0); + } + assert!(backoff.waiting( + "iris", + t0 + BACKOFF_MAX.saturating_sub(Duration::from_secs(1)) + )); + assert!(!backoff.waiting("iris", t0 + BACKOFF_MAX)); + } + + #[tokio::test] + async fn a_build_that_failed_or_did_not_push_backs_off_and_writes_nothing() { + let nix_failed = FromMetaReply::Failed { + error_tail: "error".to_owned(), + }; + let world = Fake::missing(&["iris", "argus"]) + .answering("iris", nix_failed) + .answering("argus", built(&path("argus"), false)); + let mut backoff = Backoff::default(); + let now = Instant::now(); + let sent = pass_at(&world, &mut backoff, now).await; + assert_eq!( + sent, + [ + ("argus".to_owned(), Outcome::NotPushed), + ("iris".to_owned(), Outcome::Failed) + ] + ); + assert!(backoff.waiting("iris", now) && backoff.waiting("argus", now)); + assert!(world.writes().is_empty()); + } + + #[tokio::test] + async fn another_path_is_logged_and_replaces_the_record_it_matches() { + let other = path("iris-other"); + let world = Fake::missing(&["iris"]).answering("iris", built(&other, true)); + let events = Events::default(); + let sent = { + let _guard = tracing::subscriber::set_default( + tracing_subscriber::registry().with(events.clone()), + ); + pass_at(&world, &mut Backoff::default(), Instant::now()).await + }; + assert_eq!(sent, [("iris".to_owned(), Outcome::PathChanged)]); + let writes = world.writes(); + assert_eq!(writes.len(), 1); + assert_eq!(writes[0].1.store_path, other); + assert_eq!(writes[0].1.meta_rev, HEAD); + assert!(writes[0].1.cached); + assert!( + events + .warnings() + .iter() + .any(|m| m.contains("not reproducible")), + "{:?}", + events.warnings() + ); + } + + #[tokio::test] + async fn another_path_leaves_a_record_that_moved_on() { + let world = Fake::missing(&["iris"]).answering("iris", built(&path("iris-other"), true)); + // A deploy's build records a newer commit while this one builds. + *world.racing.lock().unwrap() = Some(("iris".to_owned(), record(OLD, "iris-newer", true))); + let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await; + assert_eq!(sent, [("iris".to_owned(), Outcome::PathChangedKept)]); + assert!(world.writes().is_empty()); + let kept = world.records.lock().unwrap()["iris"].clone(); + assert_eq!(kept.closure, record(OLD, "iris-newer", true)); + } + + #[tokio::test] + async fn another_path_leaves_a_record_that_a_write_beat_the_replace_to() { + let world = Fake::missing(&["iris"]).answering("iris", built(&path("iris-other"), true)); + world + .replace_loses + .store(true, std::sync::atomic::Ordering::SeqCst); + let sent = pass_at(&world, &mut Backoff::default(), Instant::now()).await; + assert_eq!(sent, [("iris".to_owned(), Outcome::PathChangedKept)]); + assert!(world.writes().is_empty()); + } + + #[test] + fn the_cache_facts_need_all_three_values_and_an_https_url() { + let some = |v: &str| Some(v.to_owned()); + let facts = Facts::from_values(some("tok"), some("https://cache.example/"), some("swarm")) + .expect("complete"); + assert_eq!(facts.url, "https://cache.example"); + assert!(Facts::from_values(None, some("https://c"), some("swarm")).is_none()); + assert!(Facts::from_values(some("tok"), some(" \n"), some("swarm")).is_none()); + assert!(Facts::from_values(some("tok"), some("https://c"), None).is_none()); + assert!(Facts::from_values(some("tok"), some("http://c"), some("swarm")).is_none()); + } + + #[test] + fn the_narinfo_is_named_by_the_stores_hash() { + let p = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system"; + assert_eq!(path_hash(p).unwrap(), "0123456789abcdfghijklmnpqrsvwxyz"); + assert!(path_hash("/nix/store/short-x").is_err()); + } + + #[test] + fn the_counter_counts_by_outcome() { + let exporter = InMemoryMetricExporter::default(); + let provider = SdkMeterProvider::builder() + .with_reader(PeriodicReader::builder(exporter.clone()).build()) + .build(); + let metrics = Metrics(Some(Arc::new(Instruments::new(&provider)))); + metrics.count(Outcome::Rebuilt); + metrics.count(Outcome::Rebuilt); + metrics.count(Outcome::Failed); + provider.force_flush().expect("flush"); + let all = exporter.get_finished_metrics().expect("metrics"); + let mut seen = BTreeMap::new(); + for metric in all + .last() + .into_iter() + .flat_map(opentelemetry_sdk::metrics::data::ResourceMetrics::scope_metrics) + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + { + assert_eq!(metric.name(), "hyperhive.builder.cache_miss_rebuilds"); + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("not a u64 sum"); + }; + for point in sum.data_points() { + let outcome = point + .attributes() + .find(|kv| kv.key.as_str() == "outcome") + .map(|kv| kv.value.to_string()) + .expect("labelled"); + seen.insert(outcome, point.value()); + } + } + assert_eq!( + seen, + BTreeMap::from([("failed".to_owned(), 1), ("rebuilt".to_owned(), 2)]) + ); + } + + /// The messages of the warn events a test emitted. + #[derive(Clone, Default)] + struct Events(Arc>>); + + impl Events { + fn warnings(&self) -> Vec { + self.0.lock().unwrap().clone() + } + } + + impl tracing_subscriber::Layer for Events { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + if *event.metadata().level() != tracing::Level::WARN { + return; + } + let mut message = String::new(); + event.record( + &mut |field: &tracing::field::Field, value: &dyn std::fmt::Debug| { + if field.name() == "message" { + message = format!("{value:?}"); + } + }, + ); + self.0.lock().unwrap().push(message); + } + } +} diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 47b103ea..ee9d2317 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -810,7 +810,7 @@ fn declare_fan_out( } /// Every `(hive, agent)` a hive in `hives` places. -async fn placements( +pub(crate) async fn placements( wanted: &wanted::WantedWriter, hives: &[String], ) -> Result> { diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index f90980dd..fc020d39 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -41,6 +41,7 @@ mod attic_copy; mod auth; mod backend_env; mod builder; +mod cache_miss; mod claude_login; mod config_pr; mod forge; diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index 81531948..a24dab0b 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -291,10 +291,18 @@ fn spawn_forge_workers( // pin when that is newer. The timer's first tick is this boot run. crate::jobq::spawn_meta_update_timer(Arc::clone(jobq)); forge::objects::spawn(Arc::clone(&client)); - if let (Some(bridge), Some(queue)) = (auth, queue) { + if let (Some(bridge), Some(queue)) = (auth, queue.clone()) { forge::operator_profiles::spawn(Arc::clone(&client), bridge, queue); } forge::runner_token::spawn(Arc::clone(&client)); + if let Some(queue) = queue { + crate::cache_miss::spawn( + queue.clone(), + Arc::new(wanted::WantedWriter::new(queue)), + hives.iter().map(|h| h.name.clone()).collect(), + Arc::clone(&client), + ); + } let sched = Arc::clone(jobq); let hive_names = hives.iter().map(|h| h.name.clone()).collect(); forge::hive_token::spawn(Arc::clone(&client), hive_names, move |hives| {