test the run_node dispatch and the reconcile, signal, drain and purge failure paths

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<Coordinator>,
+    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::MetaSync { .. } => run_meta_sync(host, agent).await,
-        NodeKind::Prebuild { .. } => run_prebuild(&RealHost, agent, id).await,
+        NodeKind::Prebuild { .. } => run_prebuild(host, agent, id).await,
-        NodeKind::Swap { .. } => run_swap(&RealHost, coord, agent, id).await,
+        NodeKind::Swap { .. } => run_swap(host, coord, agent, id).await,
-        NodeKind::RebuildBookkeeping { .. } => {
+        NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(host, coord, agent).await,
-            run_rebuild_bookkeeping(&RealHost, coord, agent).await
+        NodeKind::Provision { .. } => run_provision(host, agent).await,
-        }
+        NodeKind::Create { .. } => run_create(host, agent).await,
-        NodeKind::Provision { .. } => run_provision(&RealHost, agent).await,
+        NodeKind::DestroyContainer { .. } => run_destroy_container(host, coord, agent).await,
-        NodeKind::Create { .. } => run_create(&RealHost, agent).await,
-        NodeKind::DestroyContainer { .. } => run_destroy_container(&RealHost, 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::Start { .. } => run_start(host, coord, agent).await,
-        NodeKind::Stop { .. } => run_stop(&RealHost, coord, agent).await,
+        NodeKind::Stop { .. } => run_stop(host, coord, agent).await,
-        NodeKind::StopForUpdate { .. } => run_stop_for_update(&RealHost, 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::ForgeSweep => run_forge_sweep(host).await,
-        NodeKind::MatrixSweep => run_matrix_sweep(&RealHost, MATRIX_SWEEP_DEADLINE).await,
+        NodeKind::MatrixSweep => run_matrix_sweep(host, MATRIX_SWEEP_DEADLINE).await,
-        NodeKind::KnowledgePull => {
+        NodeKind::KnowledgePull => run_knowledge_pull(host, coord, KNOWLEDGE_PULL_DEADLINE).await,
-            run_knowledge_pull(&RealHost, coord, KNOWLEDGE_PULL_DEADLINE).await
+        NodeKind::WantedPull => run_wanted_pull(host, coord).await,
-        }
-        NodeKind::WantedPull => run_wanted_pull(&RealHost, coord).await,
     };
     (builder, result)
 }

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,
+    Host, dispatch, matrix_sweep_warning, run_create, run_destroy_bookkeeping,
-    run_forge_sweep, run_knowledge_pull, run_matrix_sweep, run_meta_sync, run_pause_drain,
+    run_destroy_container, run_drain, run_emit_rebuilt, run_forge_sweep, run_knowledge_pull,
-    run_pause_signal, run_prebuild, run_provision, run_purge_state, run_rebuild_bookkeeping,
+    run_matrix_sweep, run_meta_sync, run_pause_drain, run_pause_signal, run_prebuild,
-    run_reconcile, run_set_wanted, run_signal, run_start, run_stop, run_stop_for_update, run_swap,
+    run_provision, run_purge_state, run_rebuild_bookkeeping, run_reconcile, run_set_wanted,
-    run_wanted_pull, run_write_dropin,
+    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<String>);
+
+/// 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<Coordinator>, kind: NodeKind) -> Vec<Settled> {
+    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}");
+}