download an agent's cached closure in prebuild, building locally when that fails

hive-c0re/src/job_queue/exec.rs

@@ -44,8 +44,21 @@ pub(super) trait Host: Sync {
     async fn kill(&self, name: &str) -> Result<()>;
     /// The `internal/meta` commit `MetaSync` recorded for the agent.
     fn recorded_rev(&self, name: &str) -> Result<String>;
+    /// The store path of the closure the swarm built for the agent at the
+    /// `internal/meta` commit `rev` and pushed to its cache. The error says why
+    /// there is none.
+    async fn cached_closure(&self, name: &str, rev: &str) -> Result<String>;
+    /// Download `store_path` from the swarm's cache into the store, never
+    /// building it.
+    async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()>;
     /// Warm the agent's toplevel build from the `internal/meta` commit `rev`.
-    async fn prebuild_toplevel(&self, name: &str, rev: &str, node_id: u64) -> Result<()>;
+    /// The toplevel it printed, when it printed one.
+    async fn prebuild_toplevel(
+        &self,
+        name: &str,
+        rev: &str,
+        node_id: u64,
+    ) -> Result<Option<String>>;
     /// Re-apply drop-ins, then `nixos-container update` at `rev`.
     async fn swap_update(
         &self,
@@ -117,7 +130,20 @@ impl Host for RealHost {
         Ok(crate::meta_rev::recorded_or_err(name)?.rev)
     }
 
-    async fn prebuild_toplevel(&self, name: &str, rev: &str, node_id: u64) -> Result<()> {
+    async fn cached_closure(&self, name: &str, rev: &str) -> Result<String> {
+        crate::attic::closure::path_for(name, rev).await
+    }
+
+    async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
+        crate::priv_client::realise_closure(name, store_path).await
+    }
+
+    async fn prebuild_toplevel(
+        &self,
+        name: &str,
+        rev: &str,
+        node_id: u64,
+    ) -> Result<Option<String>> {
         let flake_ref = crate::meta_rev::flake_ref(name, rev);
         crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(node_id)).await
     }
@@ -595,6 +621,12 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
 /// container is already down: its only purpose is to shrink the swap's
 /// downtime window, so a stopped agent (no uptime to preserve) doesn't
 /// pay the double eval — `Swap` builds inline instead.
+///
+/// The swarm's cache is tried first: when the swarm built this agent at the
+/// recorded commit and pushed the closure, it is downloaded. The local build
+/// runs either way. After a download it is an evaluation and a no-op build when
+/// this hive evaluates the path the swarm built, and builds it, with the agent
+/// still up, when it doesn't; `Swap` never meets a build it didn't expect.
 async fn run_prebuild(host: &impl Host, name: &str, id: NodeId) -> Result<()> {
     // Warm the toplevel build only when the container is up — the whole
     // point of prebuild is to shrink the swap's downtime window. A
@@ -602,11 +634,70 @@ async fn run_prebuild(host: &impl Host, name: &str, id: NodeId) -> Result<()> {
     // eval and let the downstream `Swap` build inline.
     if host.is_running(name).await {
         let rev = host.recorded_rev(name)?;
-        host.prebuild_toplevel(name, &rev, id.get()).await?;
+        let downloaded = match substitute(host, name, &rev).await {
+            Ok(path) => Some(path),
+            Err(fallback) => {
+                let reason = &fallback.reason;
+                if fallback.signature_refused {
+                    tracing::error!(%name, %rev, "prebuild: built locally: {reason}");
+                } else {
+                    tracing::warn!(%name, %rev, "prebuild: built locally: {reason}");
+                }
+                None
+            }
+        };
+        let evaluated = host.prebuild_toplevel(name, &rev, id.get()).await?;
+        if let Some(path) = downloaded {
+            match drift(&path, evaluated.as_deref()) {
+                None => tracing::info!(%name, %rev, "prebuild: substituted {path}"),
+                Some(note) => tracing::warn!(%name, %rev, "prebuild: {note}"),
+            }
+        }
     }
     Ok(())
 }
 
+/// What to log when the closure downloaded for a rebuild isn't the one this
+/// hive's evaluation produced, which means the builder and the hive disagree
+/// about the agent's flake. `None` when they agree.
+fn drift(downloaded: &str, evaluated: Option<&str>) -> Option<String> {
+    match evaluated {
+        Some(path) if path == downloaded => None,
+        Some(path) => Some(format!(
+            "the downloaded closure {downloaded} is not the one this hive evaluates, {path}; built the latter locally"
+        )),
+        None => Some(format!(
+            "downloaded {downloaded}, but the local build printed no path to compare it with"
+        )),
+    }
+}
+
+/// Why a rebuild builds instead of downloading the swarm's closure.
+struct LocalBuild {
+    reason: String,
+    /// The cache served a path no trusted key signed.
+    signature_refused: bool,
+}
+
+/// Download the closure the swarm built for `name` at `rev`; its store path.
+async fn substitute(host: &impl Host, name: &str, rev: &str) -> Result<String, LocalBuild> {
+    let path = host
+        .cached_closure(name, rev)
+        .await
+        .map_err(|e| LocalBuild {
+            reason: format!("{e:#}"),
+            signature_refused: false,
+        })?;
+    host.realise_closure(name, &path).await.map_err(|e| {
+        let reason = format!("{e:#}");
+        LocalBuild {
+            signature_refused: reason.contains(hive_priv_sock::SIGNATURE_REFUSED),
+            reason,
+        }
+    })?;
+    Ok(path)
+}
+
 /// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb),
 /// `nixos-container update`, then the post-rebuild bookkeeping tail
 /// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan).

hive-c0re/src/job_queue/exec/tests.rs

@@ -11,12 +11,12 @@ use hive_jobq::NodeId;
 use hive_jobq::scheduler::Scheduler;
 
 use super::{
-    Host, dispatch, matrix_sweep_warning, run_create, run_destroy_bookkeeping,
+    Host, dispatch, drift, matrix_sweep_warning, run_create, run_destroy_bookkeeping,
     run_destroy_container, run_drain, run_emit_rebuilt, run_forge_sweep, run_knowledge_pull,
     run_matrix_sweep, run_meta_sync, run_pause_drain, run_pause_signal, run_prebuild,
     run_provision, run_purge_state, run_rebuild_bookkeeping, run_reconcile, run_set_wanted,
     run_signal, run_start, run_stop, run_stop_for_update, run_swap, run_wanted_pull,
-    run_write_dropin,
+    run_write_dropin, substitute,
 };
 use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
 use crate::job_queue::model::NodeKind;
@@ -24,14 +24,27 @@ use crate::meta_rev::MetaBuild;
 use crate::power::Wanted;
 
 const REV: &str = "2222222222222222222222222222222222222222";
+const CLOSURE: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-alice";
 
 /// A [`Host`] that records every call as one string and fails on request.
 #[derive(Default)]
 struct FakeHost {
     running: bool,
     rev: Option<String>,
-    /// Names of the calls that fail (`kill`, `swap_update`, ...). `recorded_build` fails by
-    /// finding nothing and `matrix_ensure_all` by reporting false.
+    /// The store path the swarm's record names, when it has a pushed build of
+    /// `REV`; `None` reads as no record.
+    closure: Option<String>,
+    /// A failing `realise` says nix refused the path for want of a signature.
+    refusal: bool,
+    /// The toplevel this hive's evaluation produces; `None` is `CLOSURE`.
+    evaluates: Option<String>,
+    /// Store paths a download put in the store.
+    realised: Mutex<Vec<String>>,
+    /// Store paths a prebuild had to build because the store lacked them.
+    builds: Mutex<Vec<String>>,
+    /// Names of the calls that fail (`kill`, `realise`, `prebuild`, `swap_update`,
+    /// `set_paused`, ...). `recorded_build` fails by finding nothing and
+    /// `matrix_ensure_all` by reporting false.
     failing: Vec<&'static str>,
     /// The sweeps and pulls never finish, so only a deadline ends them.
     hang: bool,
@@ -57,6 +70,11 @@ impl FakeHost {
         }
     }
 
+    fn with_closure(mut self) -> Self {
+        self.closure = Some(CLOSURE.to_owned());
+        self
+    }
+
     fn failing(mut self, call: &'static str) -> Self {
         self.failing.push(call);
         self
@@ -82,6 +100,10 @@ impl FakeHost {
     fn calls(&self) -> Vec<String> {
         self.calls.lock().unwrap().clone()
     }
+
+    fn builds(&self) -> Vec<String> {
+        self.builds.lock().unwrap().clone()
+    }
 }
 
 impl Host for FakeHost {
@@ -105,12 +127,43 @@ impl Host for FakeHost {
             .ok_or_else(|| anyhow::anyhow!("{name}: no internal/meta commit was recorded"))
     }
 
-    async fn prebuild_toplevel(&self, name: &str, rev: &str, node_id: u64) -> anyhow::Result<()> {
+    async fn cached_closure(&self, name: &str, rev: &str) -> anyhow::Result<String> {
+        self.record(format!("cached_closure {name} {rev}"));
+        self.closure
+            .clone()
+            .ok_or_else(|| anyhow::anyhow!("no-record: no build is recorded for the agent"))
+    }
+
+    async fn realise_closure(&self, name: &str, store_path: &str) -> anyhow::Result<()> {
+        self.record(format!("realise {name} {store_path}"));
+        if self.fails("realise") {
+            if self.refusal {
+                anyhow::bail!(
+                    "not available: {}: lacks a signature",
+                    hive_priv_sock::SIGNATURE_REFUSED
+                );
+            }
+            anyhow::bail!("not available: Couldn't connect to server");
+        }
+        self.realised.lock().unwrap().push(store_path.to_owned());
+        Ok(())
+    }
+
+    async fn prebuild_toplevel(
+        &self,
+        name: &str,
+        rev: &str,
+        node_id: u64,
+    ) -> anyhow::Result<Option<String>> {
         self.record(format!("prebuild {name} {rev} node={node_id}"));
         if self.fails("prebuild") {
             anyhow::bail!("prebuild failed");
         }
-        Ok(())
+        let path = self.evaluates.clone().unwrap_or_else(|| CLOSURE.to_owned());
+        if !self.realised.lock().unwrap().contains(&path) {
+            self.builds.lock().unwrap().push(path.clone());
+        }
+        Ok(Some(path))
     }
 
     async fn swap_update(
@@ -290,11 +343,140 @@ async fn prebuild_warms_the_recorded_rev_when_the_container_is_up() {
         [
             "is_running alice".to_owned(),
             "recorded_rev alice".to_owned(),
+            format!("cached_closure alice {REV}"),
             format!("prebuild alice {REV} node={}", id.get()),
         ]
     );
 }
 
+/// The download is followed by the local build either way. When this hive
+/// evaluates the path the swarm built, the build finds it in the store.
+#[tokio::test]
+async fn a_downloaded_closure_the_hive_also_evaluates_is_found_not_built() {
+    let host = FakeHost::up().with_closure();
+    let id = node_id();
+    run_prebuild(&host, "alice", id).await.expect("prebuild ok");
+    assert_eq!(
+        host.calls(),
+        [
+            "is_running alice".to_owned(),
+            "recorded_rev alice".to_owned(),
+            format!("cached_closure alice {REV}"),
+            format!("realise alice {CLOSURE}"),
+            format!("prebuild alice {REV} node={}", id.get()),
+        ]
+    );
+    assert_eq!(host.builds(), Vec::<String>::new());
+}
+
+/// When it evaluates another path, the build happens in `Prebuild`, after the
+/// check that the agent is up and before `Swap` could stop it.
+#[tokio::test]
+async fn a_downloaded_closure_the_hive_does_not_evaluate_is_built_before_the_swap() {
+    let other = "/nix/store/zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-nixos-system-alice";
+    let host = FakeHost {
+        evaluates: Some(other.to_owned()),
+        ..FakeHost::up().with_closure()
+    };
+    let id = node_id();
+    run_prebuild(&host, "alice", id).await.expect("prebuild ok");
+    let calls = host.calls();
+    assert_eq!(calls[0], "is_running alice", "{calls:?}");
+    assert_eq!(
+        calls.last().map(String::as_str),
+        Some(format!("prebuild alice {REV} node={}", id.get()).as_str()),
+        "{calls:?}"
+    );
+    assert_eq!(host.builds(), [other.to_owned()]);
+}
+
+#[test]
+fn only_a_different_path_is_drift() {
+    assert_eq!(drift(CLOSURE, Some(CLOSURE)), None);
+    let note = drift(CLOSURE, Some("/nix/store/other")).expect("drift");
+    assert!(
+        note.contains(CLOSURE) && note.contains("/nix/store/other"),
+        "{note}"
+    );
+    assert!(drift(CLOSURE, None).expect("no path").contains(CLOSURE));
+}
+
+#[tokio::test]
+async fn a_failed_download_builds_locally() {
+    let host = FakeHost::up().with_closure().failing("realise");
+    let id = node_id();
+    run_prebuild(&host, "alice", id)
+        .await
+        .expect("the build covers it");
+    let calls = host.calls();
+    assert_eq!(calls[3], format!("realise alice {CLOSURE}"), "{calls:?}");
+    assert_eq!(
+        calls[4],
+        format!("prebuild alice {REV} node={}", id.get()),
+        "{calls:?}"
+    );
+    assert_eq!(calls.len(), 5, "{calls:?}");
+}
+
+/// The fallback is the build, so a build that fails still fails the node.
+#[tokio::test]
+async fn a_failed_download_and_a_failed_build_fail_the_node() {
+    let host = FakeHost::up()
+        .with_closure()
+        .failing("realise")
+        .failing("prebuild");
+    let err = run_prebuild(&host, "alice", node_id())
+        .await
+        .expect_err("nothing produced the closure");
+    assert!(format!("{err:#}").contains("prebuild failed"), "{err:#}");
+}
+
+#[tokio::test]
+async fn a_stopped_container_looks_up_nothing() {
+    let host = FakeHost::down().with_closure();
+    run_prebuild(&host, "alice", node_id())
+        .await
+        .expect("skipped");
+    assert_eq!(host.calls(), ["is_running alice"]);
+}
+
+/// The error log is for a path the cache served unsigned, not for a cache that
+/// is down or a missing record.
+#[tokio::test]
+async fn only_a_signature_refusal_is_flagged() {
+    let down = substitute(
+        &FakeHost::up().with_closure().failing("realise"),
+        "alice",
+        REV,
+    )
+    .await
+    .expect_err("down");
+    assert!(!down.signature_refused, "{}", down.reason);
+    assert!(down.reason.contains("Couldn't connect"), "{}", down.reason);
+
+    let no_record = substitute(&FakeHost::up(), "alice", REV)
+        .await
+        .expect_err("no record");
+    assert!(!no_record.signature_refused);
+    assert!(
+        no_record.reason.starts_with("no-record"),
+        "{}",
+        no_record.reason
+    );
+
+    let refused = FakeHost {
+        refusal: true,
+        ..FakeHost::up().with_closure().failing("realise")
+    };
+    let refused = substitute(&refused, "alice", REV)
+        .await
+        .expect_err("refused");
+    assert!(refused.signature_refused, "{}", refused.reason);
+
+    let ok = substitute(&FakeHost::up().with_closure(), "alice", REV).await;
+    assert_eq!(ok.ok().as_deref(), Some(CLOSURE));
+}
+
 #[tokio::test]
 async fn prebuild_is_skipped_when_the_container_is_down() {
     let host = FakeHost::down();