rebuild an agent's closure when the binary cache lost it

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
+`<url>/<cache>/<hash>.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
+which runs every 12 hours (`deploy.attic.retentionPeriod`). The controller
-that agent pushes a closure the cache lost.
+[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

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://<swarm.attic.domain>`); 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 (`<name>:<base64>`)               | `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                                                                                                                                                                                                                       |

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.
       '';
     };
   };

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<Output = Result<Vec<String>>> + 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<Output = Result<Option<String>>> + Send;
+    /// `agent`'s record; `None` when it has none.
+    fn record(&self, agent: &str) -> impl Future<Output = Result<Option<Stored>>> + Send;
+    /// Whether the cache holds the closure at `store_path`. `Err` when it can't
+    /// say.
+    fn holds(&self, store_path: &str) -> impl Future<Output = Result<bool>> + Send;
+    /// Ask the builder to build `agent` at `meta_rev` and push the closure.
+    fn rebuild(
+        &self,
+        agent: &str,
+        meta_rev: &str,
+    ) -> impl Future<Output = Result<FromMetaReply, String>> + 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<Output = Result<bool>> + 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<String, (u32, Instant)>);
+
+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<Arc<Instruments>>);
+
+struct Instruments {
+    /// Exports only while it lives.
+    _provider: SdkMeterProvider,
+    rebuilds: Counter<u64>,
+}
+
+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<Option<Arc<Instruments>>> = 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<Vec<(String, Outcome)>> {
+    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<bool> {
+    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<String>,
+        url: Option<String>,
+        cache: Option<String>,
+    ) -> Option<Self> {
+        let clean = |v: Option<String>| 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<Option<Self>> {
+        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/<hash>-<name>` 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<WantedWriter>,
+    hives: Vec<String>,
+    forge: Arc<forge::Client>,
+    http: reqwest::Client,
+    facts: Facts,
+}
+
+impl World for Live {
+    async fn placed_agents(&self) -> Result<Vec<String>> {
+        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<Option<String>> {
+        self.forge.grant_builder_read().await?;
+        self.forge.head().await
+    }
+
+    async fn record(&self, agent: &str) -> Result<Option<Stored>> {
+        let (closure, revision) = crate::schedules::read_with(&self.closures, agent, |bytes| {
+            Ok(Some(serde_json::from_slice::<AgentClosure>(bytes)?))
+        })
+        .await?;
+        Ok(closure
+            .zip(revision)
+            .map(|(closure, revision)| Stored { closure, revision }))
+    }
+
+    async fn holds(&self, store_path: &str) -> Result<bool> {
+        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<FromMetaReply, String> {
+        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<bool> {
+        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<WantedWriter>,
+    hives: Vec<String>,
+    forge: Arc<forge::Client>,
+) {
+    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<WantedWriter>,
+    hives: &[String],
+    forge: &Arc<forge::Client>,
+) -> Result<Option<Live>> {
+    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<String>,
+        records: Mutex<BTreeMap<String, Stored>>,
+        cache: Mutex<BTreeSet<String>>,
+        cache_down: bool,
+        /// What a build answers, per agent; a missing agent builds to its
+        /// record's path and pushes it.
+        answers: Mutex<BTreeMap<String, Result<FromMetaReply, String>>>,
+        /// A record write that lands between the build and the replace.
+        racing: Mutex<Option<(String, AgentClosure)>>,
+        /// Whether another write beats the replace.
+        replace_loses: std::sync::atomic::AtomicBool,
+        builds: Mutex<Vec<(String, String)>>,
+        writes: Mutex<Vec<(String, AgentClosure)>>,
+    }
+
+    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<Vec<String>> {
+            Ok(self.records.lock().unwrap().keys().cloned().collect())
+        }
+
+        async fn meta_head(&self) -> Result<Option<String>> {
+            Ok(self.head.clone())
+        }
+
+        async fn record(&self, agent: &str) -> Result<Option<Stored>> {
+            Ok(self.records.lock().unwrap().get(agent).cloned())
+        }
+
+        async fn holds(&self, store_path: &str) -> Result<bool> {
+            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<FromMetaReply, String> {
+            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<bool> {
+            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<Mutex<Vec<String>>>);
+
+    impl Events {
+        fn warnings(&self) -> Vec<String> {
+            self.0.lock().unwrap().clone()
+        }
+    }
+
+    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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);
+        }
+    }
+}

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<std::collections::BTreeSet<(String, String)>> {

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;

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| {