From 61292e865ea491a6eb7cb418e7aba000114aed4f Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 27 Oct 2026 19:03:54 +0100 Subject: [PATCH] add a host seam to the job-queue executors and test them --- hive-c0re/src/job_queue/exec.rs | 143 ++++++-- hive-c0re/src/job_queue/exec/tests.rs | 456 ++++++++++++++++++++++++++ 2 files changed, 568 insertions(+), 31 deletions(-) create mode 100644 hive-c0re/src/job_queue/exec/tests.rs diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 2ce086c9..d505275f 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -12,7 +12,7 @@ use anyhow::{Context as _, Result}; use hive_jobq::NodeId; use super::model::NodeKind; -use crate::coordinator::Coordinator; +use crate::coordinator::{AgentPaths, Coordinator, HiveEnv}; use crate::power::{ReconcileAction, reconcile_action}; use crate::stats::sweep_health::{self, SweepHealth}; @@ -31,6 +31,69 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from /// change to one doesn't silently retune the other. const PAUSE_ACK_TIMEOUT: std::time::Duration = GRACEFUL_STOP_TIMEOUT; +/// The container-state calls the executors below make on the host: systemd, +/// `hive-priv`, the nix build, the `internal/meta` record. Production uses +/// [`RealHost`], which forwards to `lifecycle` / `meta_rev` / `Coordinator` +/// unchanged; tests substitute a recording fake. +pub(super) trait Host: Sync { + /// Whether the agent's container unit is active. + async fn is_running(&self, name: &str) -> bool; + /// Stop the agent's container. + async fn kill(&self, name: &str) -> Result<()>; + /// The `internal/meta` commit `MetaSync` recorded for the agent. + fn recorded_rev(&self, name: &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<()>; + /// Re-apply drop-ins, then `nixos-container update` at `rev`. + async fn swap_update( + &self, + name: &str, + hive: &HiveEnv, + paths: &AgentPaths, + rev: &str, + node_id: u64, + ) -> Result<()>; + /// Write or remove the agent's pause marker. + async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()>; +} + +/// [`Host`] over the real system. +pub(super) struct RealHost; + +impl Host for RealHost { + async fn is_running(&self, name: &str) -> bool { + crate::lifecycle::is_running(name).await + } + + async fn kill(&self, name: &str) -> Result<()> { + crate::lifecycle::kill(name).await + } + + fn recorded_rev(&self, name: &str) -> Result { + Ok(crate::meta_rev::recorded_or_err(name)?.rev) + } + + async fn prebuild_toplevel(&self, name: &str, rev: &str, node_id: u64) -> Result<()> { + let flake_ref = crate::meta_rev::flake_ref(name, rev); + crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(node_id)).await + } + + async fn swap_update( + &self, + name: &str, + hive: &HiveEnv, + paths: &AgentPaths, + rev: &str, + node_id: u64, + ) -> Result<()> { + crate::lifecycle::swap_update(name, hive, paths, rev, Some(node_id)).await + } + + async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> { + Coordinator::set_paused(agent, paused).await + } +} + /// Run one claimed node to completion. Called from a task the /// scheduler spawns per claim; the `Result` (stringified) becomes the /// node's terminal state. @@ -68,8 +131,8 @@ pub(super) async fn run_node( // growth executors return what to grow rather than taking the builder. let result = match kind { NodeKind::MetaSync { .. } => run_meta_sync(agent).await, - NodeKind::Prebuild { .. } => run_prebuild(agent, id).await, - NodeKind::Swap { .. } => run_swap(coord, agent, id).await, + NodeKind::Prebuild { .. } => run_prebuild(&RealHost, agent, id).await, + NodeKind::Swap { .. } => run_swap(&RealHost, coord, agent, id).await, NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord, agent).await, NodeKind::Provision { .. } => run_provision(agent).await, NodeKind::Create { .. } => run_create(agent).await, @@ -83,21 +146,21 @@ pub(super) async fn run_node( Ok(()) } - NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| { + NodeKind::Reconcile { .. } => run_reconcile(&RealHost, coord, agent).await.map(|sub| { if let Some(kind) = sub { super::templates::fanned_out_mechanical(&builder, kind); } }), NodeKind::Start { .. } => run_start(coord, agent).await, - NodeKind::Stop { .. } => run_stop(coord, agent).await, - NodeKind::StopForUpdate { .. } => run_stop_for_update(coord, agent).await, + NodeKind::Stop { .. } => run_stop(&RealHost, coord, agent).await, + NodeKind::StopForUpdate { .. } => run_stop_for_update(&RealHost, coord, agent).await, NodeKind::Signal { .. } => { run_signal(coord, agent); Ok(()) } - NodeKind::Drain { .. } => run_drain(coord, agent).await, - NodeKind::PauseSignal { .. } => run_pause_signal(coord, agent).await, - NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent).await, + NodeKind::Drain { .. } => run_drain(coord, agent, GRACEFUL_STOP_TIMEOUT).await, + NodeKind::PauseSignal { .. } => run_pause_signal(&RealHost, coord, agent).await, + NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent, PAUSE_ACK_TIMEOUT).await, NodeKind::WriteDropin { .. } => run_write_dropin(coord, agent).await, // The payload rides the node and is destructured here, so the executor // takes it directly instead of re-matching the kind behind a `bail!` @@ -392,15 +455,14 @@ async fn run_meta_sync(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. -async fn run_prebuild(name: &str, id: NodeId) -> Result<()> { +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 // stopped agent has no uptime to preserve, so skip the (expensive) // eval and let the downstream `Swap` build inline. - if crate::lifecycle::is_running(name).await { - let build = crate::meta_rev::recorded_or_err(name)?; - let flake_ref = crate::meta_rev::flake_ref(name, &build.rev); - crate::lifecycle::prebuild_toplevel(name, &flake_ref, Some(id.get())).await?; + if host.is_running(name).await { + let rev = host.recorded_rev(name)?; + host.prebuild_toplevel(name, &rev, id.get()).await?; } Ok(()) } @@ -410,15 +472,19 @@ async fn run_prebuild(name: &str, id: NodeId) -> Result<()> { /// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan). /// The recovery-start on failure is NOT here — the DAG's tail /// `Reconcile` runs after this node terminal ok *or* fail. -async fn run_swap(coord: &Arc, name: &str, id: NodeId) -> Result<()> { +async fn run_swap( + host: &impl Host, + coord: &Arc, + name: &str, + id: NodeId, +) -> Result<()> { // Swap runs on an already-existing (stopped) container — runtime dir // and listener were created earlier. Pure path accessor suffices. let agent_dir = crate::paths::agent_runtime_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir)?; - let build = crate::meta_rev::recorded_or_err(name)?; - let result = - crate::lifecycle::swap_update(name, &hive, &paths, &build.rev, Some(id.get())).await; + let rev = host.recorded_rev(name)?; + let result = host.swap_update(name, &hive, &paths, &rev, id.get()).await; // On success the Ok-only bookkeeping tail (rev marker, forge/matrix // sync, kick, rescan, snapshot) runs in the sibling `RebuildBookkeeping` node, // which deps `AfterOk(Swap)`. On failure `RebuildBookkeeping` is cancel-cascaded @@ -493,8 +559,12 @@ async fn run_create(name: &str) -> Result<()> { /// declaring it — the declaration has to happen outside any `.await`, see /// [`run_node`]. `NodeKind` carries the agent it targets, so this node's agent /// is stamped into the fanned-out kind here. -async fn run_reconcile(coord: &Arc, name: &str) -> Result> { - let running = crate::lifecycle::is_running(name).await; +async fn run_reconcile( + host: &impl Host, + coord: &Arc, + name: &str, +) -> Result> { + let running = host.is_running(name).await; let wanted = coord.power.get_or_seed(name, running)?; Ok(match reconcile_action(wanted, running) { ReconcileAction::Start => Some(NodeKind::Start { @@ -542,10 +612,10 @@ async fn run_start(coord: &Arc, name: &str) -> Result<()> { /// Mechanical container stop — the sub-step a `Reconcile` planner fans /// out when it observes `wanted = Offline` and the container up. -async fn run_stop(coord: &Arc, name: &str) -> Result<()> { +async fn run_stop(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { // See `run_start`: no node-local guard — `Stop` reports `Stopping` from its // own kind now. - crate::lifecycle::kill(name).await?; + host.kill(name).await?; coord.unregister_agent(name); crate::swarm_notices::notify( "core", @@ -560,8 +630,8 @@ async fn run_stop(coord: &Arc, name: &str) -> Result<()> { /// Mechanical stop for the profile swap. Never *changes* `wanted`; /// noop when already stopped. -async fn run_stop_for_update(coord: &Arc, name: &str) -> Result<()> { - if crate::lifecycle::is_running(name).await { +async fn run_stop_for_update(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { + if host.is_running(name).await { // Seed a missing agent_power row from the PRE-stop observation // — the DAG's tail `Reconcile` observes only the mechanically // stopped state and would otherwise seed a running-but-unknown @@ -569,7 +639,7 @@ async fn run_stop_for_update(coord: &Arc, name: &str) -> Result<()> if let Err(e) = coord.power.get_or_seed(name, true) { tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed"); } - crate::lifecycle::kill(name).await?; + host.kill(name).await?; coord.rescan_containers_and_emit().await; } Ok(()) @@ -595,8 +665,12 @@ fn run_signal(coord: &Arc, name: &str) { /// Await the harness clearing the fence (`GracefulStopComplete`) or /// the timeout — either way the downstream `Reconcile` proceeds with /// the actual stop. -async fn run_drain(coord: &Arc, name: &str) -> Result<()> { - let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT; +async fn run_drain( + coord: &Arc, + name: &str, + timeout: std::time::Duration, +) -> Result<()> { + let deadline = std::time::Instant::now() + timeout; while coord.is_graceful_stop_pending(name) { if std::time::Instant::now() >= deadline { tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping"); @@ -621,11 +695,11 @@ async fn run_drain(coord: &Arc, name: &str) -> Result<()> { /// no edge for `run_pause_drain` to observe — marking pending here /// would just burn its timeout every time an operator re-confirms a /// pause that already took. -async fn run_pause_signal(coord: &Arc, name: &str) -> Result<()> { +async fn run_pause_signal(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { let agent = hive_types::Ident::parse(name) .map_err(|e| anyhow::anyhow!("invalid agent name for pause {name:?}: {e}"))?; let already_paused = Coordinator::is_paused(&agent); - Coordinator::set_paused(&agent, true).await?; + host.set_paused(&agent, true).await?; if !already_paused { coord.mark_pause_pending(name); } @@ -643,8 +717,12 @@ async fn run_pause_signal(coord: &Arc, name: &str) -> Result<()> { /// a timed-out wait doesn't leave the agent un-paused, just leaves the /// dashboard's "pausing…" badge running a little longer than it needed /// to. -async fn run_pause_drain(coord: &Arc, name: &str) -> Result<()> { - let deadline = std::time::Instant::now() + PAUSE_ACK_TIMEOUT; +async fn run_pause_drain( + coord: &Arc, + name: &str, + timeout: std::time::Duration, +) -> Result<()> { + let deadline = std::time::Instant::now() + timeout; while coord.is_pause_pending(name) { if std::time::Instant::now() >= deadline { tracing::warn!(agent = %name, "pause: ack wait timed out — marker is set regardless"); @@ -667,3 +745,6 @@ async fn run_write_dropin(coord: &Arc, name: &str) -> Result<()> { crate::lifecycle::write_dropins(name, &hive, &paths).await?; Ok(()) } + +#[cfg(test)] +mod tests; diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs new file mode 100644 index 00000000..ff6c303d --- /dev/null +++ b/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, + /// Names of the calls that fail: `kill`, `prebuild`, `swap_update`, `set_paused`. + failing: Vec<&'static str>, + calls: Mutex>, +} + +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 { + 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 { + 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) { + 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")); +}