diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 705a0173..5de97e56 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -259,6 +259,17 @@ pub(super) async fn run_node( builder: super::JobBuilder, id: NodeId, kind: &NodeKind, +) -> (super::JobBuilder, Result<()>) { + dispatch(&RealHost, coord, builder, id, kind).await +} + +/// [`run_node`] over any [`Host`]. +async fn dispatch( + host: &impl Host, + coord: &Arc, + builder: super::JobBuilder, + id: NodeId, + kind: &NodeKind, ) -> (super::JobBuilder, Result<()>) { // The agent this node targets rides the payload — empty for the agentless // kinds (the sweeps), which never read it. @@ -268,45 +279,43 @@ pub(super) async fn run_node( // inside an `.await` would make this future non-`Send` (see above), so the // growth executors return what to grow rather than taking the builder. let result = match kind { - NodeKind::MetaSync { .. } => run_meta_sync(&RealHost, agent).await, - NodeKind::Prebuild { .. } => run_prebuild(&RealHost, agent, id).await, - NodeKind::Swap { .. } => run_swap(&RealHost, coord, agent, id).await, - NodeKind::RebuildBookkeeping { .. } => { - run_rebuild_bookkeeping(&RealHost, coord, agent).await - } - NodeKind::Provision { .. } => run_provision(&RealHost, agent).await, - NodeKind::Create { .. } => run_create(&RealHost, agent).await, - NodeKind::DestroyContainer { .. } => run_destroy_container(&RealHost, coord, agent).await, + NodeKind::MetaSync { .. } => run_meta_sync(host, agent).await, + NodeKind::Prebuild { .. } => run_prebuild(host, agent, id).await, + NodeKind::Swap { .. } => run_swap(host, coord, agent, id).await, + NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(host, coord, agent).await, + NodeKind::Provision { .. } => run_provision(host, agent).await, + NodeKind::Create { .. } => run_create(host, agent).await, + NodeKind::DestroyContainer { .. } => run_destroy_container(host, coord, agent).await, NodeKind::PurgeState { .. } => { - run_purge_state(&RealHost, agent).await; + run_purge_state(host, agent).await; Ok(()) } NodeKind::DestroyBookkeeping { .. } => { - run_destroy_bookkeeping(&RealHost, coord, agent).await; + run_destroy_bookkeeping(host, coord, agent).await; Ok(()) } - NodeKind::Reconcile { .. } => run_reconcile(&RealHost, coord, agent).await.map(|sub| { + NodeKind::Reconcile { .. } => run_reconcile(host, coord, agent).await.map(|sub| { if let Some(kind) = sub { super::templates::fanned_out_mechanical(&builder, kind); } }), - NodeKind::Start { .. } => run_start(&RealHost, coord, agent).await, - NodeKind::Stop { .. } => run_stop(&RealHost, coord, agent).await, - NodeKind::StopForUpdate { .. } => run_stop_for_update(&RealHost, coord, agent).await, + NodeKind::Start { .. } => run_start(host, coord, agent).await, + NodeKind::Stop { .. } => run_stop(host, coord, agent).await, + NodeKind::StopForUpdate { .. } => run_stop_for_update(host, coord, agent).await, NodeKind::Signal { .. } => { run_signal(coord, agent); Ok(()) } NodeKind::Drain { .. } => run_drain(coord, agent, GRACEFUL_STOP_TIMEOUT).await, - NodeKind::PauseSignal { .. } => run_pause_signal(&RealHost, coord, agent).await, + NodeKind::PauseSignal { .. } => run_pause_signal(host, coord, agent).await, NodeKind::PauseDrain { .. } => run_pause_drain(coord, agent, PAUSE_ACK_TIMEOUT).await, - NodeKind::WriteDropin { .. } => run_write_dropin(&RealHost, coord, agent).await, + NodeKind::WriteDropin { .. } => run_write_dropin(host, 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!` // that could never fire. NodeKind::EmitRebuilt { ok, .. } => { - run_emit_rebuilt(&RealHost, coord, agent, coord.job_queue.root_of(id), *ok).await; + run_emit_rebuilt(host, coord, agent, coord.job_queue.root_of(id), *ok).await; Ok(()) } NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up), @@ -314,12 +323,10 @@ pub(super) async fn run_node( // `Finishing` so the nodes under it start. What they declare stays held // until their whole subtree settles. NodeKind::AgentWindow { .. } => Ok(()), - NodeKind::ForgeSweep => run_forge_sweep(&RealHost).await, - NodeKind::MatrixSweep => run_matrix_sweep(&RealHost, MATRIX_SWEEP_DEADLINE).await, - NodeKind::KnowledgePull => { - run_knowledge_pull(&RealHost, coord, KNOWLEDGE_PULL_DEADLINE).await - } - NodeKind::WantedPull => run_wanted_pull(&RealHost, coord).await, + NodeKind::ForgeSweep => run_forge_sweep(host).await, + NodeKind::MatrixSweep => run_matrix_sweep(host, MATRIX_SWEEP_DEADLINE).await, + NodeKind::KnowledgePull => run_knowledge_pull(host, coord, KNOWLEDGE_PULL_DEADLINE).await, + NodeKind::WantedPull => run_wanted_pull(host, coord).await, }; (builder, result) } diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index 35ddf1f3..49ccba7f 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/hive-c0re/src/job_queue/exec/tests.rs @@ -8,13 +8,15 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use hive_jobq::NodeId; +use hive_jobq::scheduler::Scheduler; use super::{ - Host, 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, + Host, dispatch, 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, }; use crate::coordinator::{AgentPaths, Coordinator, HiveEnv}; use crate::job_queue::model::NodeKind; @@ -903,3 +905,388 @@ async fn wanted_pull_passes_the_result_through() { .await .expect_err("read failure reaches the node"); } + +// --- run_node dispatch ------------------------------------------------------- + +/// What the queue ended up holding for one node: its kind, state and error. +type Settled = (&'static str, crate::job_queue::State, Option); + +/// Insert `kind` as a one-node job and run every runnable node through +/// [`dispatch`] over `host` until none is left, the way the scheduler loop +/// does. Returns each node's kind, final state and error in id order. +async fn run_dag(host: &FakeHost, coord: &Arc, kind: NodeKind) -> Vec { + coord + .job_queue + .insert_job(|b| vec![b.node(kind).guid()]) + .expect("valid shape"); + while let Some(runner) = Scheduler::claim_next(coord.job_queue.sched(), |id, kind, builder| { + let coord = coord.clone(); + async move { + let (grown, result) = dispatch(host, &coord, builder, id, &kind).await; + ( + grown, + crate::job_queue::outcome_of(result.map_err(|e| format!("{e:#}"))), + ) + } + }) { + runner.await.1.expect("grown job is well-formed"); + } + let sched = coord.job_queue.sched().lock().expect("job_queue mutex"); + sched + .graph() + .nodes() + .map(|n| (<&'static str>::from(&n.payload), n.state, n.error.clone())) + .collect() +} + +fn alice(kind: fn(String) -> NodeKind) -> NodeKind { + kind("alice".to_owned()) +} + +/// Every kind that reaches the host does so through its own executor and +/// leaves the node `Done`. +#[tokio::test] +async fn dispatch_routes_each_host_backed_kind_to_its_executor() { + let cases: [(NodeKind, &[&str]); 12] = [ + ( + alice(|agent| NodeKind::MetaSync { agent }), + &[ + "refresh_archetype alice", + "prepare_rebuild_dirs alice", + "refresh_agent_set", + "resolve_meta alice", + "record_meta alice", + ], + ), + ( + alice(|agent| NodeKind::Prebuild { agent }), + &["is_running alice", "recorded_rev alice", "prebuild alice"], + ), + ( + alice(|agent| NodeKind::Swap { agent }), + &["recorded_rev alice", "swap_update alice"], + ), + ( + alice(|agent| NodeKind::RebuildBookkeeping { agent }), + &["recorded_build alice", "mark_deployed alice"], + ), + ( + alice(|agent| NodeKind::Provision { agent }), + &["provision alice"], + ), + ( + alice(|agent| NodeKind::Create { agent }), + &[ + "resolve_meta alice", + "create_only alice", + "mark_deployed alice", + ], + ), + ( + alice(|agent| NodeKind::DestroyContainer { agent }), + &["destroy alice"], + ), + ( + alice(|agent| NodeKind::PurgeState { agent }), + &["delete_state_subvolume alice"], + ), + ( + alice(|agent| NodeKind::DestroyBookkeeping { agent }), + &["refresh_agent_set", "notify destroyed:alice"], + ), + ( + alice(|agent| NodeKind::WriteDropin { agent }), + &["write_dropins alice"], + ), + ( + alice(|agent| NodeKind::PauseSignal { agent }), + &["set_paused alice true"], + ), + ( + alice(|agent| NodeKind::StopForUpdate { agent }), + &["is_running alice", "kill alice"], + ), + ]; + for (kind, expected) in cases { + let name = <&str>::from(&kind); + let (_dir, coord) = coordinator(); + let host = FakeHost::up(); + let settled = run_dag(&host, &coord, kind.clone()).await; + assert_eq!( + settled, + [(name, crate::job_queue::State::Done, None)], + "{name}" + ); + let calls = host.calls(); + for want in expected { + assert!( + calls.iter().any(|c| c.starts_with(want)), + "{name}: missing `{want}` in {calls:?}" + ); + } + } +} + +#[tokio::test] +async fn dispatch_routes_the_agentless_and_hostless_kinds() { + let host = FakeHost::up(); + for kind in [ + NodeKind::ForgeSweep, + NodeKind::WantedPull, + NodeKind::KnowledgePull, + NodeKind::MatrixSweep, + alice(|agent| NodeKind::AgentWindow { agent }), + alice(|agent| NodeKind::Signal { agent }), + alice(|agent| NodeKind::Drain { agent }), + alice(|agent| NodeKind::PauseDrain { agent }), + ] { + let name = <&str>::from(&kind); + // A fresh coordinator per kind: a `Signal`'s fence would make a `Drain` wait out its timeout. + let (_dir, coord) = coordinator(); + let settled = run_dag(&host, &coord, kind).await; + assert!( + settled + .iter() + .any(|s| s.0 == name && s.1 == crate::job_queue::State::Done), + "{name}: {settled:?}" + ); + } + let calls = host.calls(); + for want in [ + "forge_ensure_all", + "wanted_pull", + "knowledge_pull", + "matrix_ensure_all", + ] { + assert!(calls.iter().any(|c| c == want), "{want} in {calls:?}"); + } +} + +#[tokio::test] +async fn dispatch_set_wanted_writes_the_intent_and_emit_rebuilt_announces() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up(); + run_dag( + &host, + &coord, + NodeKind::SetWanted { + agent: "alice".to_owned(), + up: false, + }, + ) + .await; + assert_eq!( + coord.power.get("alice").expect("read"), + Some(Wanted::Offline) + ); + let settled = run_dag( + &host, + &coord, + NodeKind::EmitRebuilt { + agent: "alice".to_owned(), + ok: true, + }, + ) + .await; + assert!(settled.iter().all(|s| s.1 == crate::job_queue::State::Done)); + assert!( + host.calls() + .iter() + .any(|c| c.starts_with("notify rebuilt:alice")) + ); +} + +/// A failing executor marks its node `Failed` with the error text, and the +/// kinds that swallow host errors stay `Done`. +#[tokio::test] +async fn dispatch_reports_a_failing_executor_as_a_failed_node() { + let cases = [ + ( + alice(|agent| NodeKind::DestroyContainer { agent }), + "destroy", + ), + (alice(|agent| NodeKind::Stop { agent }), "kill"), + (alice(|agent| NodeKind::Swap { agent }), "swap_update"), + (alice(|agent| NodeKind::Create { agent }), "create_only"), + (alice(|agent| NodeKind::PauseSignal { agent }), "set_paused"), + ( + alice(|agent| NodeKind::WriteDropin { agent }), + "write_dropins", + ), + (NodeKind::WantedPull, "wanted_pull"), + ]; + for (kind, failing) in cases { + let name = <&str>::from(&kind); + let (_dir, coord) = coordinator(); + let host = FakeHost::up().failing(failing); + let settled = run_dag(&host, &coord, kind).await; + assert_eq!(settled.len(), 1, "{name}"); + assert_eq!(settled[0].1, crate::job_queue::State::Failed, "{name}"); + assert!( + settled[0].2.is_some(), + "{name}: a failed node carries its error" + ); + assert!( + host.calls().iter().any(|c| c.starts_with(failing)), + "{name}: {failing} was never reached: {:?}", + host.calls() + ); + } + let (_dir, coord) = coordinator(); + let host = FakeHost::up().failing("delete_state_subvolume"); + let settled = run_dag(&host, &coord, alice(|agent| NodeKind::PurgeState { agent })).await; + assert_eq!(settled[0].1, crate::job_queue::State::Done); +} + +/// Reconcile plans, the fanned-out sub-node does the work: a start that fails +/// fails the sub-node and the planner above it, and registers no listener. +#[tokio::test] +async fn reconcile_whose_start_fails_fails_the_dag_without_registering() { + let (_dir, coord) = coordinator(); + coord.power.set("alice", Wanted::Up).expect("set"); + let host = FakeHost::down().failing("start"); + let settled = run_dag(&host, &coord, alice(|agent| NodeKind::Reconcile { agent })).await; + let by_kind = |k: &str| settled.iter().find(|s| s.0 == k).cloned(); + let start = by_kind("start").expect("reconcile fanned out a Start"); + assert_eq!(start.1, crate::job_queue::State::Failed); + assert!(start.2.is_some_and(|e| e.contains("start failed"))); + assert_eq!( + by_kind("reconcile").expect("planner").1, + crate::job_queue::State::Failed + ); + assert_eq!( + host.calls(), + ["is_running alice", "start alice"], + "no register_agent after a failed start" + ); +} + +#[tokio::test] +async fn reconcile_fans_out_a_stop_that_kills_and_announces() { + let (_dir, coord) = coordinator(); + coord.power.set("alice", Wanted::Offline).expect("set"); + let host = FakeHost::up(); + let settled = run_dag(&host, &coord, alice(|agent| NodeKind::Reconcile { agent })).await; + assert!(settled.iter().any(|s| s.0 == "stop")); + assert!(settled.iter().all(|s| s.1 == crate::job_queue::State::Done)); + assert_eq!( + host.calls(), + [ + "is_running alice", + "kill alice", + "notify killed:alice | agent 'alice' killed" + ] + ); +} + +// --- Signal / Drain / purge failure paths ----------------------------------- + +/// A signal names no container: it fences and kicks whatever the name, and a +/// name nothing runs under still resolves the node ok. +#[tokio::test] +async fn signal_to_an_unknown_agent_still_fences_and_succeeds() { + let (_dir, coord) = coordinator(); + let host = FakeHost::down(); + let settled = run_dag( + &host, + &coord, + NodeKind::Signal { + agent: "ghost".to_owned(), + }, + ) + .await; + assert_eq!(settled[0].1, crate::job_queue::State::Done); + assert!(coord.is_graceful_stop_pending("ghost")); + assert!(!coord.is_graceful_stop_pending("alice")); + assert!(host.calls().is_empty(), "a signal never touches the host"); +} + +/// The signal-then-drain pair as the scheduler runs it: the fence the signal +/// raises is what the drain waits on, and a cleared fence ends the wait. +#[tokio::test] +async fn drain_after_signal_ends_once_the_fence_is_cleared() { + let (_dir, coord) = coordinator(); + run_signal(&coord, "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)); + assert!(!coord.is_graceful_stop_pending("alice")); +} + +/// Only the named agent's fence is waited on and cleared. +#[tokio::test] +async fn drain_leaves_another_agents_fence_alone() { + let (_dir, coord) = coordinator(); + coord.mark_graceful_stop("alice"); + coord.mark_graceful_stop("bob"); + run_drain(&coord, "alice", Duration::from_millis(30)) + .await + .expect("ok"); + assert!(!coord.is_graceful_stop_pending("alice")); + assert!(coord.is_graceful_stop_pending("bob")); +} + +#[tokio::test] +async fn pause_drain_ends_early_on_acknowledgement() { + let (_dir, coord) = coordinator(); + coord.mark_pause_pending("alice"); + let acker = { + let coord = coord.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(100)).await; + coord.clear_pause_pending("alice"); + }) + }; + let start = Instant::now(); + run_pause_drain(&coord, "alice", Duration::from_secs(30)) + .await + .expect("ok"); + acker.await.expect("acker"); + assert!(start.elapsed() < Duration::from_secs(5)); +} + +/// A purge dir that cannot be removed is logged and skipped; the dirs after it +/// are still cleared. +#[tokio::test] +async fn purge_keeps_going_past_a_dir_it_cannot_remove() { + let dir = tempfile::tempdir().expect("tempdir"); + let (stuck, applied) = (dir.path().join("stuck"), dir.path().join("applied")); + std::fs::write(&stuck, b"a file, not a directory").expect("write"); + std::fs::create_dir_all(&applied).expect("mkdir"); + let host = FakeHost { + purge: vec![stuck.clone(), applied.clone()], + ..FakeHost::up() + }; + run_purge_state(&host, "alice").await; + assert!(stuck.exists(), "remove_dir_all refuses a plain file"); + assert!(!applied.exists()); +} + +#[test] +fn matrix_sweep_warning_says_how_long_it_has_been_failing() { + use crate::stats::sweep_health::SweepFailure; + let never = matrix_sweep_warning(SweepFailure { + consecutive: 3, + since_last_ok: None, + }); + assert!( + never.contains("3 consecutive, no success this session"), + "{never}" + ); + let after_ok = matrix_sweep_warning(SweepFailure { + consecutive: 2, + since_last_ok: Some(Duration::from_secs(90)), + }); + assert!(after_ok.contains("2 consecutive, last ok "), "{after_ok}"); + assert!(!after_ok.contains("no success"), "{after_ok}"); +}