add a host seam to the job-queue executors and test them

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

@@ -0,0 +1,456 @@
+//! Executor tests: the container-state executors run against a recording
+//! [`Host`] fake and a real `Coordinator` on a temp database. Assertions read
+//! the host calls (order and arguments), the executor's outcome, and the
+//! coordinator state the executor is meant to leave behind.
+
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use hive_jobq::NodeId;
+
+use super::{
+    Host, run_drain, run_pause_drain, run_pause_signal, run_prebuild, run_reconcile,
+    run_set_wanted, run_signal, run_stop, run_stop_for_update, run_swap,
+};
+use crate::coordinator::{AgentPaths, Coordinator, HiveEnv};
+use crate::job_queue::model::NodeKind;
+use crate::power::Wanted;
+
+const REV: &str = "2222222222222222222222222222222222222222";
+
+/// 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`, `prebuild`, `swap_update`, `set_paused`.
+    failing: Vec<&'static str>,
+    calls: Mutex<Vec<String>>,
+}
+
+impl FakeHost {
+    fn up() -> Self {
+        Self {
+            running: true,
+            rev: Some(REV.to_owned()),
+            ..Self::default()
+        }
+    }
+
+    fn down() -> Self {
+        Self {
+            running: false,
+            rev: Some(REV.to_owned()),
+            ..Self::default()
+        }
+    }
+
+    fn failing(mut self, call: &'static str) -> Self {
+        self.failing.push(call);
+        self
+    }
+
+    fn fails(&self, call: &str) -> bool {
+        self.failing.contains(&call)
+    }
+
+    fn record(&self, call: String) {
+        self.calls.lock().unwrap().push(call);
+    }
+
+    fn calls(&self) -> Vec<String> {
+        self.calls.lock().unwrap().clone()
+    }
+}
+
+impl Host for FakeHost {
+    async fn is_running(&self, name: &str) -> bool {
+        self.record(format!("is_running {name}"));
+        self.running
+    }
+
+    async fn kill(&self, name: &str) -> anyhow::Result<()> {
+        self.record(format!("kill {name}"));
+        if self.fails("kill") {
+            anyhow::bail!("stop failed");
+        }
+        Ok(())
+    }
+
+    fn recorded_rev(&self, name: &str) -> anyhow::Result<String> {
+        self.record(format!("recorded_rev {name}"));
+        self.rev
+            .clone()
+            .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<()> {
+        self.record(format!("prebuild {name} {rev} node={node_id}"));
+        if self.fails("prebuild") {
+            anyhow::bail!("prebuild failed");
+        }
+        Ok(())
+    }
+
+    async fn swap_update(
+        &self,
+        name: &str,
+        _hive: &HiveEnv,
+        _paths: &AgentPaths,
+        rev: &str,
+        node_id: u64,
+    ) -> anyhow::Result<()> {
+        self.record(format!("swap_update {name} {rev} node={node_id}"));
+        if self.fails("swap_update") {
+            anyhow::bail!("nixos-container update failed");
+        }
+        Ok(())
+    }
+
+    async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> anyhow::Result<()> {
+        self.record(format!("set_paused {} {paused}", agent.as_str()));
+        if self.fails("set_paused") {
+            anyhow::bail!("hive-priv unreachable");
+        }
+        Ok(())
+    }
+}
+
+fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let coord = Coordinator::open(
+        &dir.path().join("broker.sqlite"),
+        HiveEnv::default(),
+        crate::hive_stats::PriceTable::new(),
+        1,
+    )
+    .expect("open coordinator");
+    (dir, Arc::new(coord))
+}
+
+/// A real node id, taken from a one-node job in a throwaway queue.
+fn node_id() -> NodeId {
+    let q = crate::job_queue::JobQueue::new(1);
+    q.insert_job(|b| {
+        vec![
+            b.node(NodeKind::AgentWindow {
+                agent: "alice".to_owned(),
+            })
+            .guid(),
+        ]
+    })
+    .expect("valid shape")[0]
+}
+
+// --- Prebuild ---------------------------------------------------------------
+
+#[tokio::test]
+async fn prebuild_warms_the_recorded_rev_when_the_container_is_up() {
+    let host = FakeHost::up();
+    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!("prebuild alice {REV} node={}", id.get()),
+        ]
+    );
+}
+
+#[tokio::test]
+async fn prebuild_is_skipped_when_the_container_is_down() {
+    let host = FakeHost::down();
+    run_prebuild(&host, "alice", node_id())
+        .await
+        .expect("a skipped prebuild is ok");
+    assert_eq!(host.calls(), ["is_running alice"]);
+}
+
+#[tokio::test]
+async fn prebuild_without_a_recorded_rev_fails_before_building() {
+    let host = FakeHost {
+        rev: None,
+        ..FakeHost::up()
+    };
+    let err = run_prebuild(&host, "alice", node_id())
+        .await
+        .expect_err("nothing to build from");
+    assert!(
+        format!("{err:#}").contains("no internal/meta commit"),
+        "{err:#}"
+    );
+    assert!(!host.calls().iter().any(|c| c.starts_with("prebuild")));
+}
+
+#[tokio::test]
+async fn prebuild_failure_fails_the_node() {
+    let host = FakeHost::up().failing("prebuild");
+    let err = run_prebuild(&host, "alice", node_id())
+        .await
+        .expect_err("build failure propagates");
+    assert!(format!("{err:#}").contains("prebuild failed"), "{err:#}");
+}
+
+// --- Swap -------------------------------------------------------------------
+
+#[tokio::test]
+async fn swap_updates_at_the_recorded_rev() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::down();
+    let id = node_id();
+    run_swap(&host, &coord, "alice", id).await.expect("swap ok");
+    assert_eq!(
+        host.calls(),
+        [
+            "recorded_rev alice".to_owned(),
+            format!("swap_update alice {REV} node={}", id.get()),
+        ]
+    );
+}
+
+/// A failed update fails the node and nothing else: the swap never restarts or
+/// stops the container itself — recovery is the DAG's tail `Reconcile`.
+#[tokio::test]
+async fn a_failed_swap_reports_the_error_and_touches_no_container() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::down().failing("swap_update");
+    let err = run_swap(&host, &coord, "alice", node_id())
+        .await
+        .expect_err("update failure propagates");
+    assert!(format!("{err:#}").contains("update failed"), "{err:#}");
+    let calls = host.calls();
+    assert_eq!(calls.len(), 2, "{calls:?}");
+    assert!(calls[1].starts_with("swap_update alice"), "{calls:?}");
+}
+
+#[tokio::test]
+async fn swap_without_a_recorded_rev_never_updates() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost {
+        rev: None,
+        ..FakeHost::down()
+    };
+    run_swap(&host, &coord, "alice", node_id())
+        .await
+        .expect_err("nothing to swap to");
+    assert_eq!(host.calls(), ["recorded_rev alice"]);
+}
+
+// --- Stop / StopForUpdate ---------------------------------------------------
+
+#[tokio::test]
+async fn stop_kills_the_container() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up();
+    run_stop(&host, &coord, "alice").await.expect("stop ok");
+    assert_eq!(host.calls(), ["kill alice"]);
+}
+
+#[tokio::test]
+async fn a_failed_stop_fails_the_node() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up().failing("kill");
+    let err = run_stop(&host, &coord, "alice")
+        .await
+        .expect_err("a wedged container fails the stop");
+    assert!(format!("{err:#}").contains("stop failed"), "{err:#}");
+}
+
+#[tokio::test]
+async fn stop_for_update_kills_a_running_container_and_seeds_its_intent_up() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up();
+    run_stop_for_update(&host, &coord, "alice")
+        .await
+        .expect("stop ok");
+    assert_eq!(host.calls(), ["is_running alice", "kill alice"]);
+    // Seeded from the pre-stop observation, so the tail Reconcile restarts it.
+    assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up));
+}
+
+#[tokio::test]
+async fn stop_for_update_keeps_an_existing_offline_intent() {
+    let (_dir, coord) = coordinator();
+    coord.power.set("alice", Wanted::Offline).expect("seed");
+    let host = FakeHost::up();
+    run_stop_for_update(&host, &coord, "alice")
+        .await
+        .expect("stop ok");
+    assert_eq!(
+        coord.power.get("alice").expect("read"),
+        Some(Wanted::Offline)
+    );
+}
+
+#[tokio::test]
+async fn stop_for_update_is_a_noop_on_a_stopped_container() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::down();
+    run_stop_for_update(&host, &coord, "alice")
+        .await
+        .expect("noop ok");
+    assert_eq!(host.calls(), ["is_running alice"]);
+    assert_eq!(coord.power.get("alice").expect("read"), None);
+}
+
+#[tokio::test]
+async fn stop_for_update_failure_fails_the_node_after_seeding() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up().failing("kill");
+    run_stop_for_update(&host, &coord, "alice")
+        .await
+        .expect_err("a failed stop fails the node");
+    assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up));
+}
+
+// --- Reconcile --------------------------------------------------------------
+
+#[tokio::test]
+async fn reconcile_fans_out_start_for_wanted_up_and_down() {
+    let (_dir, coord) = coordinator();
+    coord.power.set("alice", Wanted::Up).expect("set");
+    let sub = run_reconcile(&FakeHost::down(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(matches!(sub, Some(NodeKind::Start { agent }) if agent == "alice"));
+}
+
+#[tokio::test]
+async fn reconcile_fans_out_stop_for_wanted_offline_and_up() {
+    let (_dir, coord) = coordinator();
+    coord.power.set("alice", Wanted::Offline).expect("set");
+    let sub = run_reconcile(&FakeHost::up(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(matches!(sub, Some(NodeKind::Stop { agent }) if agent == "alice"));
+}
+
+#[tokio::test]
+async fn reconcile_is_a_noop_when_state_matches_intent() {
+    let (_dir, coord) = coordinator();
+    coord.power.set("alice", Wanted::Up).expect("set");
+    let sub = run_reconcile(&FakeHost::up(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(sub.is_none());
+}
+
+#[tokio::test]
+async fn reconcile_seeds_a_missing_intent_from_the_observed_state() {
+    let (_dir, coord) = coordinator();
+    let sub = run_reconcile(&FakeHost::up(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(sub.is_none());
+    assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up));
+}
+
+// --- Set wanted -------------------------------------------------------------
+
+#[tokio::test]
+async fn set_wanted_writes_the_intent() {
+    let (_dir, coord) = coordinator();
+    run_set_wanted(&coord, "alice", false).expect("ok");
+    assert_eq!(
+        coord.power.get("alice").expect("read"),
+        Some(Wanted::Offline)
+    );
+    run_set_wanted(&coord, "alice", true).expect("ok");
+    assert_eq!(coord.power.get("alice").expect("read"), Some(Wanted::Up));
+}
+
+// --- Signal / Drain ---------------------------------------------------------
+
+#[tokio::test]
+async fn signal_sets_the_graceful_fence() {
+    let (_dir, coord) = coordinator();
+    run_signal(&coord, "alice");
+    assert!(coord.is_graceful_stop_pending("alice"));
+}
+
+#[tokio::test]
+async fn drain_returns_at_once_when_no_fence_is_set() {
+    let (_dir, coord) = coordinator();
+    let start = Instant::now();
+    run_drain(&coord, "alice", Duration::from_mins(1))
+        .await
+        .expect("ok");
+    assert!(start.elapsed() < Duration::from_secs(1));
+}
+
+/// A wedged agent costs the timeout and nothing more: the node still resolves
+/// ok and the fence is cleared so the downstream stop proceeds.
+#[tokio::test]
+async fn drain_times_out_ok_and_clears_the_fence() {
+    let (_dir, coord) = coordinator();
+    coord.mark_graceful_stop("alice");
+    let start = Instant::now();
+    run_drain(&coord, "alice", Duration::from_millis(50))
+        .await
+        .expect("a timeout is not an error");
+    assert!(start.elapsed() >= Duration::from_millis(50));
+    assert!(!coord.is_graceful_stop_pending("alice"));
+}
+
+#[tokio::test]
+async fn drain_ends_early_when_the_harness_clears_the_fence() {
+    let (_dir, coord) = coordinator();
+    coord.mark_graceful_stop("alice");
+    let clearer = {
+        let coord = coord.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(100)).await;
+            coord.clear_graceful_stop("alice");
+        })
+    };
+    let start = Instant::now();
+    run_drain(&coord, "alice", Duration::from_secs(30))
+        .await
+        .expect("ok");
+    clearer.await.expect("clearer");
+    assert!(start.elapsed() < Duration::from_secs(5));
+}
+
+// --- Pause ------------------------------------------------------------------
+
+#[tokio::test]
+async fn pause_signal_writes_the_marker_and_marks_pending() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up();
+    run_pause_signal(&host, &coord, "alice").await.expect("ok");
+    assert_eq!(host.calls(), ["set_paused alice true"]);
+    assert!(coord.is_pause_pending("alice"));
+}
+
+#[tokio::test]
+async fn pause_signal_failure_leaves_nothing_pending() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up().failing("set_paused");
+    run_pause_signal(&host, &coord, "alice")
+        .await
+        .expect_err("marker write failure fails the node");
+    assert!(!coord.is_pause_pending("alice"));
+}
+
+#[tokio::test]
+async fn pause_signal_rejects_a_malformed_name_without_calling_the_host() {
+    let (_dir, coord) = coordinator();
+    let host = FakeHost::up();
+    let err = run_pause_signal(&host, &coord, "../evil")
+        .await
+        .expect_err("bad name");
+    assert!(format!("{err:#}").contains("invalid agent name"), "{err:#}");
+    assert!(host.calls().is_empty());
+}
+
+#[tokio::test]
+async fn pause_drain_times_out_ok_and_clears_pending() {
+    let (_dir, coord) = coordinator();
+    coord.mark_pause_pending("alice");
+    run_pause_drain(&coord, "alice", Duration::from_millis(50))
+        .await
+        .expect("a timeout is not an error");
+    assert!(!coord.is_pause_pending("alice"));
+}