//! 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);
        }
    }
}