pin the declared closure once per rebuild and seed the marker from the running closure

hive-c0re/src/job_queue/exec.rs

@@ -45,6 +45,12 @@ pub(super) trait Host: Sync {
     /// The record the swarm's builder wrote for the agent; `None` when it wrote
     /// none.
     async fn closure_record(&self, name: &str) -> Result<Option<AgentClosure>>;
+    /// Pin `store_path` as the closure this rebuild of the agent deploys.
+    fn pin_closure(&self, name: &str, store_path: &str);
+    /// The store path pinned for the agent's rebuild.
+    fn pinned_closure(&self, name: &str) -> Option<String>;
+    /// Drop the agent's pin.
+    fn unpin_closure(&self, name: &str);
     /// Download `store_path` from the swarm's cache into the store, never
     /// building it.
     async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()>;
@@ -109,6 +115,18 @@ impl Host for RealHost {
         crate::attic::closure::record(name).await
     }
 
+    fn pin_closure(&self, name: &str, store_path: &str) {
+        crate::attic::closure::pin(name, store_path);
+    }
+
+    fn pinned_closure(&self, name: &str) -> Option<String> {
+        crate::attic::closure::pinned(name)
+    }
+
+    fn unpin_closure(&self, name: &str) {
+        crate::attic::closure::unpin(name);
+    }
+
     async fn realise_closure(&self, name: &str, store_path: &str) -> Result<()> {
         crate::priv_client::realise_closure(name, store_path).await
     }
@@ -471,8 +489,13 @@ fn run_set_wanted(coord: &Arc<Coordinator>, agent: &str, up: bool) -> Result<()>
 }
 
 /// The rebuild's preamble: runtime-dir prep, the archetype and agent-set
-/// refresh. Reads nothing from the forge: what a rebuild deploys is the swarm's
-/// declared closure, read by `Prebuild` and `Swap`.
+/// refresh, and the closure the rebuild deploys. The swarm's record is read
+/// once, here, and its store path pinned: `Prebuild` and `Swap` use the pin, so
+/// a record that changes meanwhile cannot make `Swap` pull another closure than
+/// `Prebuild` fetched, and the marker names what was applied. The next
+/// auto-update pass picks up the newer record. Reads nothing from the forge.
+///
+/// Fails, before anything is stopped, when the swarm has no closure to deploy.
 async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
     // Runs while the agent is still up — the runtime dir and MCP listener
     // already exist. Use the pure path accessor; no need to re-register the
@@ -486,6 +509,10 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
     }
     host.prepare_rebuild_dirs(name, &paths).await?;
     host.refresh_agent_set().await?;
+    let store_path = declared_closure(host, name)
+        .await
+        .map_err(|why| without_closure(name, "meta-sync", &why))?;
+    host.pin_closure(name, &store_path);
     Ok(())
 }
 
@@ -496,7 +523,7 @@ async fn run_meta_sync(host: &impl Host, name: &str) -> Result<()> {
 /// is already down: there is no uptime to protect, and `Swap` fails the same way.
 async fn run_prebuild(host: &impl Host, name: &str) -> Result<()> {
     if host.is_running(name).await {
-        let closure = closure_for(host, name, "prebuild").await?;
+        let closure = realise_pinned(host, name, "prebuild").await?;
         tracing::info!(%name, "prebuild: substituted {closure}");
     }
     Ok(())
@@ -512,18 +539,46 @@ enum NoClosure {
 }
 
 /// The store path of the closure the swarm declares for `name`, downloaded into
-/// the store. Whatever `internal/meta` commit it was built from, it is the one
-/// deployed.
+/// the store, for a deploy that has no `MetaSync` to pin it first (a create).
+/// Whatever `internal/meta` commit it was built from, it is the one deployed.
 ///
 /// # Errors
 /// [`without_closure`]'s refusal, after logging it.
 async fn closure_for(host: &impl Host, name: &str, step: &str) -> Result<String> {
-    pull_closure(host, name)
-        .await
-        .map_err(|why| without_closure(name, step, &why))
+    async {
+        let store_path = declared_closure(host, name).await?;
+        realise(host, name, &store_path).await?;
+        Ok(store_path)
+    }
+    .await
+    .map_err(|why| without_closure(name, step, &why))
 }
 
-async fn pull_closure(host: &impl Host, name: &str) -> Result<String, NoClosure> {
+/// Download the closure `MetaSync` pinned for the agent's rebuild.
+///
+/// # Errors
+/// No closure was pinned, or the download failed; the latter is logged by
+/// [`without_closure`].
+async fn realise_pinned(host: &impl Host, name: &str, step: &str) -> Result<String> {
+    let Some(store_path) = host.pinned_closure(name) else {
+        let why = NoClosure::Unpullable("no closure was pinned for this rebuild".to_owned());
+        return Err(without_closure(name, step, &why));
+    };
+    realise(host, name, &store_path)
+        .await
+        .map_err(|why| without_closure(name, step, &why))?;
+    Ok(store_path)
+}
+
+async fn realise(host: &impl Host, name: &str, store_path: &str) -> Result<(), NoClosure> {
+    host.realise_closure(name, store_path)
+        .await
+        .map_err(|e| NoClosure::Unpullable(format!("{e:#}")))
+}
+
+/// The store path of the record the swarm's builder wrote for `name`, when the
+/// builder pushed its closure.
+async fn declared_closure(host: &impl Host, name: &str) -> Result<String, NoClosure> {
     let record = host
         .closure_record(name)
         .await
@@ -534,9 +589,6 @@ async fn pull_closure(host: &impl Host, name: &str) -> Result<String, NoClosure>
             "the builder did not push the closure".to_owned(),
         ));
     }
-    host.realise_closure(name, &record.store_path)
-        .await
-        .map_err(|e| NoClosure::Unpullable(format!("{e:#}")))?;
     Ok(record.store_path)
 }
 
@@ -574,7 +626,7 @@ async fn run_swap(
     let agent_dir = crate::paths::agent_runtime_dir(name);
     let hive = coord.hive_env();
     let paths = Coordinator::agent_paths(name, agent_dir)?;
-    let closure = closure_for(host, name, "swap").await?;
+    let closure = realise_pinned(host, name, "swap").await?;
     tracing::info!(%name, "swap: applying {closure}");
     let result = host
         .swap_update(name, &hive, &paths, &closure, id.get())
@@ -582,10 +634,11 @@ async fn run_swap(
     // The marker names what was applied; a failed write is not a failed swap
     // (the container runs the closure), it only means the next comparison
     // sees the declaration as new.
-    if result.is_ok()
-        && let Err(e) = host.mark_deployed(name, &closure)
-    {
-        tracing::warn!(%name, error = ?e, "write the deployed-closure marker failed");
+    if result.is_ok() {
+        if let Err(e) = host.mark_deployed(name, &closure) {
+            tracing::warn!(%name, error = ?e, "write the deployed-closure marker failed");
+        }
+        host.unpin_closure(name);
     }
     // On success the Ok-only bookkeeping tail (forge
     // sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node,

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

@@ -31,7 +31,9 @@ struct FakeHost {
     running: bool,
     /// What the swarm's builder recorded for the agent; `None` reads as no
     /// record.
-    built: Option<AgentClosure>,
+    built: Mutex<Option<AgentClosure>>,
+    /// The store path pinned for the agent's rebuild.
+    pinned: Mutex<Option<String>>,
     /// Store paths a download put in the store.
     realised: Mutex<Vec<String>>,
     /// Names of the calls that fail (`kill`, `realise`, `closure_record`, `swap_update`,
@@ -48,7 +50,8 @@ impl FakeHost {
     fn up() -> Self {
         Self {
             running: true,
-            built: Some(closure(true)),
+            built: Mutex::new(Some(closure(true))),
+            pinned: Mutex::new(Some(CLOSURE.to_owned())),
             ..Self::default()
         }
     }
@@ -56,22 +59,39 @@ impl FakeHost {
     fn down() -> Self {
         Self {
             running: false,
-            built: Some(closure(true)),
+            built: Mutex::new(Some(closure(true))),
+            pinned: Mutex::new(Some(CLOSURE.to_owned())),
             ..Self::default()
         }
     }
 
     /// The builder recorded the closure for the agent, pushed or not.
-    fn built(mut self, cached: bool) -> Self {
-        self.built = Some(closure(cached));
+    fn built(self, cached: bool) -> Self {
+        *self.built.lock().unwrap() = Some(closure(cached));
         self
     }
 
-    fn without_record(mut self) -> Self {
-        self.built = None;
+    fn without_record(self) -> Self {
+        *self.built.lock().unwrap() = None;
         self
     }
 
+    /// The rebuild has no pinned closure, as when `MetaSync` has not run.
+    fn unpinned(self) -> Self {
+        *self.pinned.lock().unwrap() = None;
+        self
+    }
+
+    /// The builder records another closure, at `store_path`.
+    fn builder_records(&self, store_path: &str) {
+        self.built
+            .lock()
+            .unwrap()
+            .as_mut()
+            .expect("a record")
+            .store_path = store_path.to_owned();
+    }
+
     fn failing(mut self, call: &'static str) -> Self {
         self.failing.push(call);
         self
@@ -115,7 +135,21 @@ impl Host for FakeHost {
 
     async fn closure_record(&self, name: &str) -> anyhow::Result<Option<AgentClosure>> {
         self.step("closure_record", format!("closure_record {name}"))?;
-        Ok(self.built.clone())
+        Ok(self.built.lock().unwrap().clone())
+    }
+
+    fn pin_closure(&self, name: &str, store_path: &str) {
+        self.record(format!("pin_closure {name} {store_path}"));
+        *self.pinned.lock().unwrap() = Some(store_path.to_owned());
+    }
+
+    fn pinned_closure(&self, _name: &str) -> Option<String> {
+        self.pinned.lock().unwrap().clone()
+    }
+
+    fn unpin_closure(&self, name: &str) {
+        self.record(format!("unpin_closure {name}"));
+        *self.pinned.lock().unwrap() = None;
     }
 
     async fn realise_closure(&self, name: &str, store_path: &str) -> anyhow::Result<()> {
@@ -326,15 +360,21 @@ async fn logging<T>(fut: impl std::future::Future<Output = T>) -> (T, Logged) {
 
 /// Every way the swarm's closure for `alice` cannot be had, with the text the
 /// error log must carry and the host calls that must not happen.
-fn without_a_closure() -> [(&'static str, FakeHost, &'static str); 4] {
-    [
+fn without_a_closure() -> Vec<(&'static str, FakeHost, &'static str)> {
+    let mut cases = without_a_record();
+    cases.push((
+        "download fails",
+        FakeHost::up().failing("realise"),
+        "Couldn't connect",
+    ));
+    cases
+}
+
+/// The ways the swarm has no closure to pin for a rebuild.
+fn without_a_record() -> Vec<(&'static str, FakeHost, &'static str)> {
+    vec![
         ("no record", FakeHost::up().without_record(), "no build"),
         ("not pushed", FakeHost::up().built(false), "did not push"),
-        (
-            "download fails",
-            FakeHost::up().failing("realise"),
-            "Couldn't connect",
-        ),
         (
             "record unreadable",
             FakeHost::up().failing("closure_record"),
@@ -343,6 +383,18 @@ fn without_a_closure() -> [(&'static str, FakeHost, &'static str); 4] {
     ]
 }
 
+/// The ways a rebuild's `Prebuild` or `Swap` has no closure to download.
+fn without_a_pinned_closure() -> Vec<(&'static str, FakeHost, &'static str)> {
+    vec![
+        ("no pin", FakeHost::up().unpinned(), "no closure was pinned"),
+        (
+            "download fails",
+            FakeHost::up().failing("realise"),
+            "Couldn't connect",
+        ),
+    ]
+}
+
 #[tokio::test]
 async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() {
     let host = FakeHost::up();
@@ -352,7 +404,6 @@ async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() {
         host.calls(),
         [
             "is_running alice".to_owned(),
-            "closure_record alice".to_owned(),
             format!("realise alice {CLOSURE}"),
         ]
     );
@@ -370,7 +421,7 @@ async fn prebuild_downloads_the_swarms_closure_when_the_container_is_up() {
 /// container is untouched, and the log names the agent and why.
 #[tokio::test]
 async fn a_rebuild_without_a_closure_fails_the_prebuild_and_logs_an_error() {
-    for (case, host, why) in without_a_closure() {
+    for (case, host, why) in without_a_pinned_closure() {
         let (result, log) = logging(run_prebuild(&host, "alice")).await;
         let err = result.expect_err(case);
         assert!(format!("{err:#}").starts_with("alice: "), "{case}: {err:#}");
@@ -412,10 +463,10 @@ async fn swap_applies_the_swarms_closure() {
     assert_eq!(
         host.calls(),
         [
-            "closure_record alice".to_owned(),
             format!("realise alice {CLOSURE}"),
             format!("swap_update alice {CLOSURE} node={}", id.get()),
             format!("mark_deployed alice {CLOSURE}"),
+            "unpin_closure alice".to_owned(),
         ]
     );
 }
@@ -424,7 +475,7 @@ async fn swap_applies_the_swarms_closure() {
 /// was, and the log names the agent and why.
 #[tokio::test]
 async fn a_swap_without_a_closure_never_updates_and_logs_an_error() {
-    for (case, host, why) in without_a_closure() {
+    for (case, host, why) in without_a_pinned_closure() {
         let (_dir, coord) = coordinator();
         let (result, log) = logging(run_swap(&host, &coord, "alice", node_id())).await;
         result.expect_err(case);
@@ -450,8 +501,8 @@ async fn swap_marks_the_applied_closure_deployed_only_after_the_apply() {
     let id = node_id();
     run_swap(&host, &coord, "alice", id).await.expect("swap ok");
     assert_eq!(
-        host.calls().last().map(String::as_str),
-        Some(format!("mark_deployed alice {CLOSURE}").as_str()),
+        host.calls()[2],
+        format!("mark_deployed alice {CLOSURE}"),
         "{:?}",
         host.calls()
     );
@@ -489,8 +540,8 @@ async fn a_failed_swap_reports_the_error_and_touches_no_container() {
         .expect_err("update failure propagates");
     assert!(format!("{err:#}").contains("update failed"), "{err:#}");
     let calls = host.calls();
-    assert_eq!(calls.len(), 3, "{calls:?}");
-    assert!(calls[2].starts_with("swap_update alice"), "{calls:?}");
+    assert_eq!(calls.len(), 2, "{calls:?}");
+    assert!(calls[1].starts_with("swap_update alice"), "{calls:?}");
 }
 
 // --- Stop / StopForUpdate ---------------------------------------------------
@@ -716,7 +767,7 @@ async fn pause_drain_times_out_ok_and_clears_pending() {
 // --- MetaSync ---------------------------------------------------------------
 
 #[tokio::test]
-async fn meta_sync_only_prepares_the_rebuild() {
+async fn meta_sync_prepares_the_rebuild_and_pins_the_declared_closure() {
     let host = FakeHost::up();
     run_meta_sync(&host, "alice").await.expect("ok");
     assert_eq!(
@@ -725,6 +776,8 @@ async fn meta_sync_only_prepares_the_rebuild() {
             "refresh_archetype alice".to_owned(),
             "prepare_rebuild_dirs alice".to_owned(),
             "refresh_agent_set".to_owned(),
+            "closure_record alice".to_owned(),
+            format!("pin_closure alice {CLOSURE}"),
         ]
     );
 }
@@ -733,7 +786,7 @@ async fn meta_sync_only_prepares_the_rebuild() {
 async fn meta_sync_keeps_going_when_the_archetype_read_fails() {
     let host = FakeHost::up().failing("refresh_archetype");
     run_meta_sync(&host, "alice").await.expect("non-fatal");
-    assert_eq!(host.calls().len(), 3);
+    assert_eq!(host.calls().len(), 5);
 }
 
 #[tokio::test]
@@ -747,6 +800,69 @@ async fn meta_sync_stops_at_the_first_fatal_step() {
     }
 }
 
+/// A rebuild without a closure to deploy fails in `MetaSync`, before anything is
+/// stopped, and pins nothing.
+#[tokio::test]
+async fn meta_sync_without_a_closure_fails_and_pins_nothing() {
+    for (case, host, why) in without_a_record() {
+        let host = host.unpinned();
+        let (result, log) = logging(run_meta_sync(&host, "alice")).await;
+        result.expect_err(case);
+        let errors = log.at(tracing::Level::ERROR);
+        assert_eq!(errors.len(), 1, "{case}: {errors:?}");
+        assert!(errors[0].contains(why), "{case}: {errors:?}");
+        assert!(
+            host.pinned_closure("alice").is_none(),
+            "{case}: {:?}",
+            host.calls()
+        );
+    }
+}
+
+/// The record is read once for a rebuild. A record that changes after `MetaSync`
+/// (the builder finished another build) leaves `Prebuild` and `Swap` on the
+/// closure that was pinned, and the marker names what was applied.
+#[tokio::test]
+async fn a_record_that_changes_mid_rebuild_does_not_change_what_swap_applies() {
+    const NEWER: &str = "/nix/store/zyxwvsrqpnmlkjihgfdcbadfhijklmn-nixos-system-alice";
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up().unpinned();
+    run_meta_sync(&host, "alice").await.expect("pinned");
+    run_prebuild(&host, "alice").await.expect("prebuilt");
+    host.builder_records(NEWER);
+    let id = node_id();
+    run_swap(&host, &coord, "alice", id).await.expect("swapped");
+    let calls = host.calls();
+    assert_eq!(
+        calls
+            .iter()
+            .filter(|c| c.starts_with("closure_record"))
+            .count(),
+        1,
+        "the record is read once: {calls:?}"
+    );
+    assert!(
+        calls.contains(&format!("swap_update alice {CLOSURE} node={}", id.get())),
+        "{calls:?}"
+    );
+    assert!(
+        calls.contains(&format!("mark_deployed alice {CLOSURE}")),
+        "{calls:?}"
+    );
+    assert!(
+        !calls.iter().any(|c| c.contains(NEWER)),
+        "control: the newer path is never touched: {calls:?}"
+    );
+
+    // Control: the changed record is what the next rebuild pins.
+    run_meta_sync(&host, "alice").await.expect("pinned again");
+    assert!(
+        host.calls().contains(&format!("pin_closure alice {NEWER}")),
+        "{:?}",
+        host.calls()
+    );
+}
+
 // --- Provision / Create / WriteDropin ----------------------------------------
 
 #[tokio::test]
@@ -1034,19 +1150,17 @@ async fn dispatch_routes_each_host_backed_kind_to_its_executor() {
                 "refresh_archetype alice",
                 "prepare_rebuild_dirs alice",
                 "refresh_agent_set",
+                "closure_record alice",
+                "pin_closure alice",
             ],
         ),
         (
             alice(|agent| NodeKind::Prebuild { agent }),
-            &["is_running alice", "closure_record alice"],
+            &["is_running alice", "realise alice"],
         ),
         (
             alice(|agent| NodeKind::Swap { agent }),
-            &[
-                "closure_record alice",
-                "swap_update alice",
-                "mark_deployed alice",
-            ],
+            &["realise alice", "swap_update alice", "mark_deployed alice"],
         ),
         (alice(|agent| NodeKind::RebuildBookkeeping { agent }), &[]),
         (

hive-c0re/src/job_queue/tests.rs

@@ -1560,3 +1560,23 @@ fn a_rebuild_is_live_for_its_own_agent_only() {
     );
     assert!(!q.rebuild_live("bob"));
 }
+
+/// Agents whose running closure differs from the declaration are rebuilt in one
+/// sweep. Each rebuild's window holds a build slot, so `buildSlots` caps how
+/// many swap at once, however many agents the sweep rebuilds.
+#[test]
+fn a_sweep_of_rebuilds_holds_one_build_slot_per_agent_window() {
+    let q = JobQueue::new(1);
+    let agents: Vec<String> = ["alice", "bob", "carol"].map(str::to_owned).to_vec();
+    insert(&q, |builder| {
+        crate::workers::auto_update::boot_nodes(builder, &agents, Vec::new());
+    });
+    assert_eq!(
+        declared_resources_of_kind(&q, "agent_window")
+            .iter()
+            .filter(|r| r.contains(&Resource::BuildSlot))
+            .count(),
+        3,
+        "every window declares the build slot"
+    );
+}