diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index d505275f..705a0173 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -5,6 +5,7 @@ //! the `Swap` tail); DAG-level failure handling is cancel-downstream in //! the queue. +use std::path::PathBuf; use std::sync::{Arc, Mutex, OnceLock}; use anyhow::{Context as _, Result}; @@ -13,6 +14,7 @@ use hive_jobq::NodeId; use super::model::NodeKind; use crate::coordinator::{AgentPaths, Coordinator, HiveEnv}; +use crate::meta_rev::MetaBuild; use crate::power::{ReconcileAction, reconcile_action}; use crate::stats::sweep_health::{self, SweepHealth}; @@ -55,6 +57,48 @@ pub(super) trait Host: Sync { ) -> Result<()>; /// Write or remove the agent's pause marker. async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()>; + /// Publish a lifecycle notice for the hive under the `core` subsystem. + async fn notify(&self, key: String, summary: String); + /// Re-read the agent's archetype from the swarm. + async fn refresh_archetype(&self, name: &str) -> Result<()>; + /// Rebuild preamble: port check, applied repo and state dirs. + async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()>; + /// Rewrite the host-side files that follow the agent set. + async fn refresh_agent_set(&self) -> Result<()>; + /// Pin the `internal/meta` commit the agent builds from. + async fn resolve_meta(&self, name: &str) -> Result; + /// Record `build` as what `Prebuild` and `Swap` build from. + fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()>; + /// The build `MetaSync` recorded, if any. + fn recorded_build(&self, name: &str) -> Option; + /// Mark the agent as running the pins of `build`. + fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()>; + /// First-spawn provisioning: repos, state subvolume, agent-set refresh. + async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()>; + /// `nixos-container create` at `rev`. + async fn create_only(&self, name: &str, rev: &str) -> Result<()>; + /// Write the nspawn and resource-limit drop-ins, then reload systemd. + async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>; + /// Start preamble, then start with the cold-start fallback. + async fn start(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()>; + /// Bind the agent's MCP listener. + fn register_agent(&self, coord: &Arc, name: &str) -> Result<()>; + /// `nixos-container destroy`. + async fn destroy(&self, name: &str) -> Result<()>; + /// The agent's ephemeral runtime dir. + fn runtime_dir(&self, name: &str) -> PathBuf; + /// Delete the agent's state subvolume through `hive-priv`. + async fn delete_state_subvolume(&self, name: &str) -> Result<()>; + /// The persistent directories a purge removes. + fn purge_dirs(&self, name: &str) -> Vec; + /// Publish the hive's forge credential for nix. + async fn forge_ensure_all(&self); + /// One matrix user/space pass; whether every agent synced. + async fn matrix_ensure_all(&self) -> bool; + /// One `/knowledge` pull. + async fn knowledge_pull(&self, coord: &Arc) -> Result<()>; + /// One swarm wanted-state pull. + async fn wanted_pull(&self, coord: &Arc) -> Result<()>; } /// [`Host`] over the real system. @@ -92,6 +136,100 @@ impl Host for RealHost { async fn set_paused(&self, agent: &hive_types::Ident, paused: bool) -> Result<()> { Coordinator::set_paused(agent, paused).await } + + async fn notify(&self, key: String, summary: String) { + crate::swarm_notices::notify("core", Some(key), summary, None).await; + } + + async fn refresh_archetype(&self, name: &str) -> Result<()> { + crate::archetypes::refresh(name).await + } + + async fn prepare_rebuild_dirs(&self, name: &str, paths: &AgentPaths) -> Result<()> { + crate::lifecycle::prepare_rebuild_dirs(name, paths).await + } + + async fn refresh_agent_set(&self) -> Result<()> { + let agents = crate::lifecycle::agents_for_meta_listing().await?; + crate::agent_set::refresh(&agents).await + } + + async fn resolve_meta(&self, name: &str) -> Result { + crate::meta_rev::resolve(name).await + } + + fn record_meta(&self, name: &str, build: &MetaBuild) -> Result<()> { + crate::meta_rev::record(name, build) + } + + fn recorded_build(&self, name: &str) -> Option { + crate::meta_rev::recorded(name) + } + + fn mark_deployed(&self, name: &str, build: &MetaBuild) -> Result<()> { + crate::meta_rev::mark_deployed(name, build) + } + + async fn provision(&self, name: &str, paths: &AgentPaths) -> Result<()> { + crate::lifecycle::provision_container(name, paths).await + } + + async fn create_only(&self, name: &str, rev: &str) -> Result<()> { + crate::lifecycle::create_only(name, rev).await + } + + async fn write_dropins(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> { + crate::lifecycle::write_dropins(name, hive, paths).await + } + + async fn start(&self, name: &str, hive: &HiveEnv, paths: &AgentPaths) -> Result<()> { + let token = crate::lifecycle::converge_start_preamble(name, hive, paths).await?; + crate::lifecycle::start_with_fallback(token).await + } + + fn register_agent(&self, coord: &Arc, name: &str) -> Result<()> { + coord.register_agent(name).map(drop) + } + + async fn destroy(&self, name: &str) -> Result<()> { + crate::lifecycle::destroy(name).await + } + + fn runtime_dir(&self, name: &str) -> PathBuf { + crate::paths::agent_runtime_dir(name) + } + + async fn delete_state_subvolume(&self, name: &str) -> Result<()> { + crate::priv_client::delete_agent_subvolume(name).await + } + + fn purge_dirs(&self, name: &str) -> Vec { + // A malformed name can't have a persistent state tree (the state dir is + // only ever created under a validated Ident), so only the applied dir + // is left to clear. + hive_types::Ident::parse(name) + .ok() + .map(|id| crate::paths::agent_state_dir(&id)) + .into_iter() + .chain([crate::paths::applied_dir(name)]) + .collect() + } + + async fn forge_ensure_all(&self) { + crate::forge::ensure_all().await; + } + + async fn matrix_ensure_all(&self) -> bool { + crate::matrix::ensure_all().await + } + + async fn knowledge_pull(&self, coord: &Arc) -> Result<()> { + crate::workers::knowledge::pull(coord).await + } + + async fn wanted_pull(&self, coord: &Arc) -> Result<()> { + crate::workers::wanted::pull(coord).await + } } /// Run one claimed node to completion. Called from a task the @@ -130,19 +268,21 @@ 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(agent).await, + 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(coord, agent).await, - NodeKind::Provision { .. } => run_provision(agent).await, - NodeKind::Create { .. } => run_create(agent).await, - NodeKind::DestroyContainer { .. } => run_destroy_container(coord, agent).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::PurgeState { .. } => { - run_purge_state(agent).await; + run_purge_state(&RealHost, agent).await; Ok(()) } NodeKind::DestroyBookkeeping { .. } => { - run_destroy_bookkeeping(coord, agent).await; + run_destroy_bookkeeping(&RealHost, coord, agent).await; Ok(()) } @@ -151,7 +291,7 @@ pub(super) async fn run_node( super::templates::fanned_out_mechanical(&builder, kind); } }), - NodeKind::Start { .. } => run_start(coord, agent).await, + 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::Signal { .. } => { @@ -161,12 +301,12 @@ pub(super) async fn run_node( 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, + NodeKind::WriteDropin { .. } => run_write_dropin(&RealHost, 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(coord, agent, coord.job_queue.root_of(id), *ok).await; + run_emit_rebuilt(&RealHost, coord, agent, coord.job_queue.root_of(id), *ok).await; Ok(()) } NodeKind::SetWanted { up, .. } => run_set_wanted(coord, agent, *up), @@ -174,10 +314,12 @@ 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().await, - NodeKind::MatrixSweep => run_matrix_sweep().await, - NodeKind::KnowledgePull => run_knowledge_pull(coord).await, - NodeKind::WantedPull => run_wanted_pull(coord).await, + 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, }; (builder, result) } @@ -187,19 +329,19 @@ pub(super) async fn run_node( /// is best-effort (a build re-publishes the credential before it fetches), so /// this wrapper has nothing left to report; it exists to make the step a /// visible unit of work. -async fn run_forge_sweep() -> Result<()> { - crate::forge::ensure_all().await; +async fn run_forge_sweep(host: &impl Host) -> Result<()> { + host.forge_ensure_all().await; Ok(()) } /// Matrix user/space sweep as a DAG node — see [`NodeKind::MatrixSweep`]. /// Every pass reports here, whoever submitted it: as the node's own error on /// the dashboard, and into the debounced warning. -async fn run_matrix_sweep() -> Result<()> { - let ok = tokio::time::timeout(MATRIX_SWEEP_DEADLINE, crate::matrix::ensure_all()) +async fn run_matrix_sweep(host: &impl Host, deadline: std::time::Duration) -> Result<()> { + let ok = tokio::time::timeout(deadline, host.matrix_ensure_all()) .await .unwrap_or_else(|_| { - tracing::warn!(deadline = ?MATRIX_SWEEP_DEADLINE, "matrix: sweep timed out"); + tracing::warn!(?deadline, "matrix: sweep timed out"); false }); let mut health = matrix_sweep_health() @@ -246,17 +388,18 @@ fn matrix_sweep_warning(ctx: sweep_health::SweepFailure) -> String { /// `/knowledge` pull as a DAG node — see [`NodeKind::KnowledgePull`]. Every /// pass reports here, whoever submitted it: as the node's own error on the /// dashboard, and into the debounced warning. -async fn run_knowledge_pull(coord: &Arc) -> Result<()> { - let result = tokio::time::timeout( - KNOWLEDGE_PULL_DEADLINE, - crate::workers::knowledge::pull(coord), - ) - .await - .unwrap_or_else(|_| { - Err(anyhow::anyhow!( - "knowledge pull timed out after {KNOWLEDGE_PULL_DEADLINE:?}" - )) - }); +async fn run_knowledge_pull( + host: &impl Host, + coord: &Arc, + deadline: std::time::Duration, +) -> Result<()> { + let result = tokio::time::timeout(deadline, host.knowledge_pull(coord)) + .await + .unwrap_or_else(|_| { + Err(anyhow::anyhow!( + "knowledge pull timed out after {deadline:?}" + )) + }); let mut health = knowledge_pull_health() .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); @@ -299,16 +442,8 @@ fn knowledge_pull_health() -> &'static Mutex { /// successful outcome the worker logs, so an error reaching here means the /// read itself failed and this hive is running blind to its declaration — /// which is exactly the state worth seeing on the dashboard. -async fn run_wanted_pull(coord: &Arc) -> Result<()> { - crate::workers::wanted::pull(coord).await -} - -/// Refresh the host-side agent-set files from whatever containers still exist -/// on disk. Idempotent. Lives here because the destroy tail is its only -/// caller; it moved with `destroy` when that became a DAG. -async fn refresh_agent_set_after_lifecycle() -> Result<()> { - let agents = crate::lifecycle::agents_for_meta_listing().await?; - crate::agent_set::refresh(&agents).await +async fn run_wanted_pull(host: &impl Host, coord: &Arc) -> Result<()> { + host.wanted_pull(coord).await } /// `nixos-container destroy`, then drop the agent from the roster and clear @@ -320,10 +455,14 @@ async fn refresh_agent_set_after_lifecycle() -> Result<()> { /// the roster claiming an agent that no longer exists. A destroy that leaves the /// container in place fails this node before un-registration, and its failure /// is what keeps the `after_ok` purge and bookkeeping from running. -async fn run_destroy_container(coord: &Arc, agent: &str) -> Result<()> { - crate::lifecycle::destroy(agent).await?; +async fn run_destroy_container( + host: &impl Host, + coord: &Arc, + agent: &str, +) -> Result<()> { + host.destroy(agent).await?; coord.unregister_agent(agent); - let runtime = crate::paths::agent_runtime_dir(agent); + let runtime = host.runtime_dir(agent); if runtime.exists() { let _ = std::fs::remove_dir_all(&runtime); } @@ -336,24 +475,15 @@ async fn run_destroy_container(coord: &Arc, agent: &str) -> Result< /// deliberate — the container is already destroyed by the time this runs, so /// failing the node would leave the operator with a half-purged agent and a red /// DAG, when what they can actually act on is the log line naming the path. -async fn run_purge_state(agent: &str) { +async fn run_purge_state(host: &impl Host, agent: &str) { // The state root may be a btrfs subvolume: a subvolume root can't be // removed with rmdir/`remove_dir_all`, so delete it via hive-priv (root) // first. No-op for plain-dir agents — the loop below then handles the // plain-dir state root plus the applied dir. - if let Err(e) = crate::priv_client::delete_agent_subvolume(agent).await { + if let Err(e) = host.delete_state_subvolume(agent).await { tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed"); } - // A malformed name can't have a persistent state tree (the state dir is - // only ever created under a validated Ident), so its removal is a no-op — - // skip the state-dir sweep and just clear the applied dir. - let state_dir = hive_types::Ident::parse(agent) - .ok() - .map(|id| crate::paths::agent_state_dir(&id)); - for dir in state_dir - .into_iter() - .chain([crate::paths::applied_dir(agent)]) - { + for dir in host.purge_dirs(agent) { if dir.exists() && let Err(e) = std::fs::remove_dir_all(&dir) { @@ -367,9 +497,9 @@ async fn run_purge_state(agent: &str) { /// none of this is undoable — a failed agent-set refresh or power-store write is a /// bookkeeping drift to log, not a reason to red a DAG whose container is /// already gone. -async fn run_destroy_bookkeeping(coord: &Arc, agent: &str) { +async fn run_destroy_bookkeeping(host: &impl Host, coord: &Arc, agent: &str) { // Drop the agent from the topology and the gateway files. - if let Err(e) = refresh_agent_set_after_lifecycle().await { + if let Err(e) = host.refresh_agent_set().await { tracing::warn!(error = ?e, %agent, "agent_set refresh after destroy failed"); } // Drop the durable power intent — a future agent of the same name seeds @@ -377,11 +507,9 @@ async fn run_destroy_bookkeeping(coord: &Arc, agent: &str) { if let Err(e) = coord.power.remove(agent) { tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed"); } - crate::swarm_notices::notify( - "core", - Some(format!("destroyed:{agent}")), + host.notify( + format!("destroyed:{agent}"), format!("agent '{agent}' destroyed"), - None, ) .await; // Container row disappeared — rescan so the dashboard fires @@ -394,12 +522,18 @@ async fn run_destroy_bookkeeping(coord: &Arc, agent: &str) { /// Emit this agent's rebuild-complete todo. `ok` is not computed — it is which /// of the tail pair the graph let run. The failure note comes from the DAG's /// first failing node, since the branch knows *that* it failed but not *why*. -async fn run_emit_rebuilt(coord: &Arc, agent: &str, dag_id: Option, ok: bool) { +async fn run_emit_rebuilt( + host: &impl Host, + coord: &Arc, + agent: &str, + dag_id: Option, + ok: bool, +) { let note = (!ok) .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag))) .flatten(); let summary = crate::coordinator::rebuilt_todo_summary(agent, ok, note.as_deref(), None, None); - crate::swarm_notices::notify("core", Some(format!("rebuilt:{agent}")), summary, None).await; + host.notify(format!("rebuilt:{agent}"), summary).await; } /// Write the agent's durable power intent — the DAG-node form of the old @@ -428,7 +562,7 @@ fn run_set_wanted(coord: &Arc, agent: &str, up: bool) -> Result<()> /// /// Fails when `internal/meta` does not pin the agent's current config yet: /// there is no other flake to build it from. -async fn run_meta_sync(name: &str) -> Result<()> { +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 // listener (event-driven: registered at start/create). @@ -436,15 +570,14 @@ async fn run_meta_sync(name: &str) -> Result<()> { let paths = Coordinator::agent_paths(name, agent_dir)?; // A failed read keeps the archetype last stored, so the agent is built as // it was. - if let Err(e) = crate::archetypes::refresh(name).await { + if let Err(e) = host.refresh_archetype(name).await { tracing::warn!(%name, error = ?e, "rebuild: reading the swarm's archetype failed; keeping the stored one"); } - crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?; - let agents = crate::lifecycle::agents_for_meta_listing().await?; - crate::agent_set::refresh(&agents).await?; + host.prepare_rebuild_dirs(name, &paths).await?; + host.refresh_agent_set().await?; // Resolved once here so `Prebuild` and `Swap` build the same flake. - let build = crate::meta_rev::resolve(name).await?; - crate::meta_rev::record(name, &build)?; + let build = host.resolve_meta(name).await?; + host.record_meta(name, &build)?; Ok(()) } @@ -503,9 +636,13 @@ async fn run_swap( /// means the profile swap succeeded. Store/forge work only — no nix /// build (build-slot-exempt); the agent lease taken at `Swap` is still held /// (the whole chain up to `Reconcile` is one agent's subgraph). -async fn run_rebuild_bookkeeping(coord: &Arc, name: &str) -> Result<()> { - if let Some(build) = crate::meta_rev::recorded(name) - && let Err(e) = crate::meta_rev::mark_deployed(name, &build) +async fn run_rebuild_bookkeeping( + host: &impl Host, + coord: &Arc, + name: &str, +) -> Result<()> { + if let Some(build) = host.recorded_build(name) + && let Err(e) = host.mark_deployed(name, &build) { tracing::warn!(%name, error = ?e, "write rev markers failed"); } @@ -528,10 +665,10 @@ async fn run_rebuild_bookkeeping(coord: &Arc, name: &str) -> Result /// First-spawn pre-create provisioning: proposed/applied repos, state /// subvolume, and the agent-set refresh. -async fn run_provision(name: &str) -> Result<()> { +async fn run_provision(host: &impl Host, name: &str) -> Result<()> { let agent_dir = crate::paths::agent_runtime_dir(name); let paths = Coordinator::agent_paths(name, agent_dir)?; - crate::lifecycle::provision_container(name, &paths).await?; + host.provision(name, &paths).await?; Ok(()) } @@ -541,10 +678,10 @@ async fn run_provision(name: &str) -> Result<()> { /// `Reconcile` (`converge_start_preamble` + `register_agent`) so this /// node stays purely "create", not "create + start". Marks the agent as /// running the pins it was created from. -async fn run_create(name: &str) -> Result<()> { - let build = crate::meta_rev::resolve(name).await?; - crate::lifecycle::create_only(name, &build.rev).await?; - crate::meta_rev::mark_deployed(name, &build)?; +async fn run_create(host: &impl Host, name: &str) -> Result<()> { + let build = host.resolve_meta(name).await?; + host.create_only(name, &build.rev).await?; + host.mark_deployed(name, &build)?; Ok(()) } @@ -582,7 +719,7 @@ async fn run_reconcile( /// Mechanical container start — the sub-step a `Reconcile` planner fans /// out when it observes `wanted = Up` and the container down. -async fn run_start(coord: &Arc, name: &str) -> Result<()> { +async fn run_start(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { // No node-local transient guard: the pill is derived from the running node // set, and `Start` reports `Starting` via `NodeKind::transient_kind`. This // used to take one "only when the DAG holds none", which was a second @@ -595,14 +732,13 @@ async fn run_start(coord: &Arc, name: &str) -> Result<()> { let agent_dir = crate::paths::agent_runtime_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir)?; - let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?; - crate::lifecycle::start_with_fallback(token).await?; + host.start(name, &hive, &paths).await?; // Bind the MCP listener immediately after starting the container. // The preamble created the runtime dir; the container is now coming // up and will connect to this socket on its first turn. Event-driven // (no background poll) — c0re owns the listener lifecycle, so // register here rather than waiting for a sweep. - coord.register_agent(name)?; + host.register_agent(coord, name)?; // No `kick_agent` here any more — `hive-agent` pushes its own "you // were just (re)started" todo on boot now, so this node doesn't need // to tell it. See the matching comment in `run_rebuild_bookkeeping`. @@ -617,13 +753,8 @@ async fn run_stop(host: &impl Host, coord: &Arc, name: &str) -> Res // own kind now. host.kill(name).await?; coord.unregister_agent(name); - crate::swarm_notices::notify( - "core", - Some(format!("killed:{name}")), - format!("agent '{name}' killed"), - None, - ) - .await; + host.notify(format!("killed:{name}"), format!("agent '{name}' killed")) + .await; coord.rescan_containers_and_emit().await; Ok(()) } @@ -735,14 +866,14 @@ async fn run_pause_drain( } /// `set_nspawn_flags` + `set_resource_limits` + daemon-reload. -async fn run_write_dropin(coord: &Arc, name: &str) -> Result<()> { +async fn run_write_dropin(host: &impl Host, coord: &Arc, name: &str) -> Result<()> { // write_dropins only needs the path value to build AgentPaths; the // dir doesn't need to exist at this point (created by ensure_agent_runtime_dir // on the upstream Prebuild/Start node). let agent_dir = crate::paths::agent_runtime_dir(name); let hive = coord.hive_env(); let paths = Coordinator::agent_paths(name, agent_dir)?; - crate::lifecycle::write_dropins(name, &hive, &paths).await?; + host.write_dropins(name, &hive, &paths).await?; Ok(()) } diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index ff6c303d..35ddf1f3 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/hive-c0re/src/job_queue/exec/tests.rs @@ -3,17 +3,22 @@ //! the host calls (order and arguments), the executor's outcome, and the //! coordinator state the executor is meant to leave behind. +use std::path::PathBuf; 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, + 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, }; use crate::coordinator::{AgentPaths, Coordinator, HiveEnv}; use crate::job_queue::model::NodeKind; +use crate::meta_rev::MetaBuild; use crate::power::Wanted; const REV: &str = "2222222222222222222222222222222222222222"; @@ -23,8 +28,13 @@ const REV: &str = "2222222222222222222222222222222222222222"; struct FakeHost { running: bool, rev: Option, - /// Names of the calls that fail: `kill`, `prebuild`, `swap_update`, `set_paused`. + /// Names of the calls that fail (`kill`, `swap_update`, ...). `recorded_build` fails by + /// finding nothing and `matrix_ensure_all` by reporting false. failing: Vec<&'static str>, + /// The sweeps and pulls never finish, so only a deadline ends them. + hang: bool, + runtime: PathBuf, + purge: Vec, calls: Mutex>, } @@ -54,6 +64,15 @@ impl FakeHost { self.failing.contains(&call) } + /// Record `line`, then fail when `call` is on the failing list. + fn step(&self, call: &str, line: String) -> anyhow::Result<()> { + self.record(line); + if self.fails(call) { + anyhow::bail!("{call} failed"); + } + Ok(()) + } + fn record(&self, call: String) { self.calls.lock().unwrap().push(call); } @@ -114,6 +133,121 @@ impl Host for FakeHost { } Ok(()) } + + async fn notify(&self, key: String, summary: String) { + self.record(format!("notify {key} | {summary}")); + } + + async fn refresh_archetype(&self, name: &str) -> anyhow::Result<()> { + self.step("refresh_archetype", format!("refresh_archetype {name}")) + } + + async fn prepare_rebuild_dirs(&self, name: &str, _paths: &AgentPaths) -> anyhow::Result<()> { + self.step( + "prepare_rebuild_dirs", + format!("prepare_rebuild_dirs {name}"), + ) + } + + async fn refresh_agent_set(&self) -> anyhow::Result<()> { + self.step("refresh_agent_set", "refresh_agent_set".to_owned()) + } + + async fn resolve_meta(&self, name: &str) -> anyhow::Result { + self.step("resolve_meta", format!("resolve_meta {name}"))?; + Ok(build()) + } + + fn record_meta(&self, name: &str, build: &MetaBuild) -> anyhow::Result<()> { + self.step("record_meta", format!("record_meta {name} {}", build.rev)) + } + + fn recorded_build(&self, name: &str) -> Option { + self.record(format!("recorded_build {name}")); + (!self.fails("recorded_build")).then(build) + } + + fn mark_deployed(&self, name: &str, build: &MetaBuild) -> anyhow::Result<()> { + self.step( + "mark_deployed", + format!("mark_deployed {name} {}", build.rev), + ) + } + + async fn provision(&self, name: &str, _paths: &AgentPaths) -> anyhow::Result<()> { + self.step("provision", format!("provision {name}")) + } + + async fn create_only(&self, name: &str, rev: &str) -> anyhow::Result<()> { + self.step("create_only", format!("create_only {name} {rev}")) + } + + async fn write_dropins( + &self, + name: &str, + _hive: &HiveEnv, + _paths: &AgentPaths, + ) -> anyhow::Result<()> { + self.step("write_dropins", format!("write_dropins {name}")) + } + + async fn start(&self, name: &str, _hive: &HiveEnv, _paths: &AgentPaths) -> anyhow::Result<()> { + self.step("start", format!("start {name}")) + } + + fn register_agent(&self, _coord: &Arc, name: &str) -> anyhow::Result<()> { + self.step("register_agent", format!("register_agent {name}")) + } + + async fn destroy(&self, name: &str) -> anyhow::Result<()> { + self.step("destroy", format!("destroy {name}")) + } + + fn runtime_dir(&self, _name: &str) -> PathBuf { + self.runtime.clone() + } + + async fn delete_state_subvolume(&self, name: &str) -> anyhow::Result<()> { + self.step( + "delete_state_subvolume", + format!("delete_state_subvolume {name}"), + ) + } + + fn purge_dirs(&self, _name: &str) -> Vec { + self.purge.clone() + } + + async fn forge_ensure_all(&self) { + self.record("forge_ensure_all".to_owned()); + } + + async fn matrix_ensure_all(&self) -> bool { + self.record("matrix_ensure_all".to_owned()); + if self.hang { + std::future::pending::<()>().await; + } + !self.fails("matrix_ensure_all") + } + + async fn knowledge_pull(&self, _coord: &Arc) -> anyhow::Result<()> { + if self.hang { + std::future::pending::<()>().await; + } + self.step("knowledge_pull", "knowledge_pull".to_owned()) + } + + async fn wanted_pull(&self, _coord: &Arc) -> anyhow::Result<()> { + self.step("wanted_pull", "wanted_pull".to_owned()) + } +} + +fn build() -> MetaBuild { + MetaBuild { + rev: REV.to_owned(), + hyperhive: "h".to_owned(), + config: Some("c".to_owned()), + } } fn coordinator() -> (tempfile::TempDir, Arc) { @@ -241,11 +375,14 @@ async fn swap_without_a_recorded_rev_never_updates() { // --- Stop / StopForUpdate --------------------------------------------------- #[tokio::test] -async fn stop_kills_the_container() { +async fn stop_kills_the_container_then_announces_it() { let (_dir, coord) = coordinator(); let host = FakeHost::up(); run_stop(&host, &coord, "alice").await.expect("stop ok"); - assert_eq!(host.calls(), ["kill alice"]); + assert_eq!( + host.calls(), + ["kill alice", "notify killed:alice | agent 'alice' killed"] + ); } #[tokio::test] @@ -454,3 +591,315 @@ async fn pause_drain_times_out_ok_and_clears_pending() { .expect("a timeout is not an error"); assert!(!coord.is_pause_pending("alice")); } + +// --- MetaSync --------------------------------------------------------------- + +#[tokio::test] +async fn meta_sync_prepares_then_resolves_and_records_last() { + let host = FakeHost::up(); + run_meta_sync(&host, "alice").await.expect("ok"); + assert_eq!( + host.calls(), + [ + "refresh_archetype alice".to_owned(), + "prepare_rebuild_dirs alice".to_owned(), + "refresh_agent_set".to_owned(), + "resolve_meta alice".to_owned(), + format!("record_meta alice {REV}"), + ] + ); +} + +#[tokio::test] +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(), 5); +} + +#[tokio::test] +async fn meta_sync_stops_at_the_first_fatal_step() { + for (fail, calls) in [ + ("prepare_rebuild_dirs", 2), + ("refresh_agent_set", 3), + ("resolve_meta", 4), + ("record_meta", 5), + ] { + let host = FakeHost::up().failing(fail); + run_meta_sync(&host, "alice") + .await + .expect_err("a fatal step fails the node"); + assert_eq!(host.calls().len(), calls, "{fail}"); + } +} + +// --- Provision / Create / WriteDropin ---------------------------------------- + +#[tokio::test] +async fn provision_and_write_dropin_forward_and_propagate_failure() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up(); + run_provision(&host, "alice").await.expect("ok"); + run_write_dropin(&host, &coord, "alice").await.expect("ok"); + assert_eq!(host.calls(), ["provision alice", "write_dropins alice"]); + let host = FakeHost::up().failing("provision").failing("write_dropins"); + run_provision(&host, "alice").await.expect_err("provision"); + run_write_dropin(&host, &coord, "alice") + .await + .expect_err("dropins"); +} + +#[tokio::test] +async fn create_builds_at_the_resolved_rev_then_marks_it_deployed() { + let host = FakeHost::up(); + run_create(&host, "alice").await.expect("ok"); + assert_eq!( + host.calls(), + [ + "resolve_meta alice".to_owned(), + format!("create_only alice {REV}"), + format!("mark_deployed alice {REV}"), + ] + ); +} + +#[tokio::test] +async fn a_failed_create_is_never_marked_deployed() { + let host = FakeHost::up().failing("create_only"); + run_create(&host, "alice").await.expect_err("create fails"); + assert_eq!(host.calls().len(), 2); + let host = FakeHost::up().failing("resolve_meta"); + run_create(&host, "alice").await.expect_err("resolve fails"); + assert_eq!(host.calls(), ["resolve_meta alice"]); +} + +// --- Start ------------------------------------------------------------------ + +#[tokio::test] +async fn start_starts_then_registers_the_listener() { + let (_dir, coord) = coordinator(); + let host = FakeHost::down(); + run_start(&host, &coord, "alice").await.expect("ok"); + assert_eq!(host.calls(), ["start alice", "register_agent alice"]); +} + +#[tokio::test] +async fn a_failed_start_registers_no_listener() { + let (_dir, coord) = coordinator(); + let host = FakeHost::down().failing("start"); + run_start(&host, &coord, "alice").await.expect_err("start"); + assert_eq!(host.calls(), ["start alice"]); +} + +#[tokio::test] +async fn a_failed_listener_registration_fails_the_start_node() { + let (_dir, coord) = coordinator(); + let host = FakeHost::down().failing("register_agent"); + run_start(&host, &coord, "alice") + .await + .expect_err("register"); +} + +// --- Destroy / Purge --------------------------------------------------------- + +#[tokio::test] +async fn destroy_container_removes_the_runtime_dir() { + let (dir, coord) = coordinator(); + let runtime = dir.path().join("run"); + std::fs::create_dir_all(runtime.join("sub")).expect("mkdir"); + let host = FakeHost { + runtime: runtime.clone(), + ..FakeHost::up() + }; + run_destroy_container(&host, &coord, "alice") + .await + .expect("ok"); + assert_eq!(host.calls(), ["destroy alice"]); + assert!(!runtime.exists()); + // A runtime dir that is already gone is fine. + run_destroy_container(&host, &coord, "alice") + .await + .expect("idempotent"); +} + +#[tokio::test] +async fn a_failed_destroy_leaves_the_runtime_dir_alone() { + let (dir, coord) = coordinator(); + let runtime = dir.path().join("run"); + std::fs::create_dir_all(&runtime).expect("mkdir"); + let host = FakeHost { + runtime: runtime.clone(), + ..FakeHost::up().failing("destroy") + }; + run_destroy_container(&host, &coord, "alice") + .await + .expect_err("destroy fails the node"); + assert!(runtime.exists()); +} + +#[tokio::test] +async fn purge_removes_every_state_dir_even_when_the_subvolume_delete_fails() { + let dir = tempfile::tempdir().expect("tempdir"); + let (state, applied, absent) = ( + dir.path().join("state"), + dir.path().join("applied"), + dir.path().join("absent"), + ); + for d in [&state, &applied] { + std::fs::create_dir_all(d.join("deep")).expect("mkdir"); + } + let host = FakeHost { + purge: vec![state.clone(), absent, applied.clone()], + ..FakeHost::up().failing("delete_state_subvolume") + }; + run_purge_state(&host, "alice").await; + assert_eq!(host.calls(), ["delete_state_subvolume alice"]); + assert!(!state.exists() && !applied.exists()); +} + +#[tokio::test] +async fn destroy_bookkeeping_drops_the_power_row_and_announces() { + let (_dir, coord) = coordinator(); + coord.power.set("alice", Wanted::Offline).expect("set"); + let host = FakeHost::up(); + run_destroy_bookkeeping(&host, &coord, "alice").await; + assert_eq!( + host.calls(), + [ + "refresh_agent_set", + "notify destroyed:alice | agent 'alice' destroyed" + ] + ); + // The Offline row is gone, so a fresh seed from "running" gives Up. + assert_eq!( + coord.power.get_or_seed("alice", true).expect("seed"), + Wanted::Up + ); +} + +#[tokio::test] +async fn destroy_bookkeeping_continues_past_a_failed_refresh() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up().failing("refresh_agent_set"); + run_destroy_bookkeeping(&host, &coord, "alice").await; + assert_eq!(host.calls().len(), 2); +} + +// --- RebuildBookkeeping / EmitRebuilt ---------------------------------------- + +#[tokio::test] +async fn rebuild_bookkeeping_marks_the_recorded_build_deployed() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up(); + run_rebuild_bookkeeping(&host, &coord, "alice") + .await + .expect("ok"); + assert_eq!( + host.calls(), + [ + "recorded_build alice".to_owned(), + format!("mark_deployed alice {REV}") + ] + ); +} + +#[tokio::test] +async fn rebuild_bookkeeping_without_a_recording_marks_nothing() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up().failing("recorded_build"); + run_rebuild_bookkeeping(&host, &coord, "alice") + .await + .expect("ok"); + assert_eq!(host.calls(), ["recorded_build alice"]); +} + +#[tokio::test] +async fn a_failed_marker_write_does_not_fail_the_bookkeeping_node() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up().failing("mark_deployed"); + run_rebuild_bookkeeping(&host, &coord, "alice") + .await + .expect("warn only"); +} + +#[tokio::test] +async fn emit_rebuilt_reports_success_and_failure_under_one_key() { + let (_dir, coord) = coordinator(); + let host = FakeHost::up(); + run_emit_rebuilt(&host, &coord, "alice", None, true).await; + run_emit_rebuilt(&host, &coord, "alice", None, false).await; + assert_eq!( + host.calls(), + [ + "notify rebuilt:alice | agent 'alice' rebuilt", + "notify rebuilt:alice | agent 'alice' rebuild FAILED: unknown error", + ] + ); +} + +// --- Sweeps and pulls --------------------------------------------------------- + +#[tokio::test] +async fn forge_sweep_runs_once_and_never_fails() { + let host = FakeHost::up(); + run_forge_sweep(&host).await.expect("ok"); + assert_eq!(host.calls(), ["forge_ensure_all"]); +} + +#[tokio::test] +async fn matrix_sweep_fails_the_node_when_an_agent_failed_sync() { + run_matrix_sweep(&FakeHost::up(), Duration::from_secs(5)) + .await + .expect("ok"); + let err = run_matrix_sweep( + &FakeHost::up().failing("matrix_ensure_all"), + Duration::from_secs(5), + ) + .await + .expect_err("failed sync"); + assert!(format!("{err:#}").contains("failed sync"), "{err:#}"); +} + +#[tokio::test] +async fn matrix_sweep_past_its_deadline_fails_instead_of_hanging() { + let host = FakeHost { + hang: true, + ..FakeHost::up() + }; + run_matrix_sweep(&host, Duration::from_millis(50)) + .await + .expect_err("deadline"); +} + +#[tokio::test] +async fn knowledge_pull_passes_the_result_through_and_times_out() { + let (_dir, coord) = coordinator(); + let secs = Duration::from_secs(5); + run_knowledge_pull(&FakeHost::up(), &coord, secs) + .await + .expect("ok"); + let err = run_knowledge_pull(&FakeHost::up().failing("knowledge_pull"), &coord, secs) + .await + .expect_err("pull fails"); + assert!( + format!("{err:#}").contains("knowledge_pull failed"), + "{err:#}" + ); + let hang = FakeHost { + hang: true, + ..FakeHost::up() + }; + let err = run_knowledge_pull(&hang, &coord, Duration::from_millis(50)) + .await + .expect_err("deadline"); + assert!(format!("{err:#}").contains("timed out"), "{err:#}"); +} + +#[tokio::test] +async fn wanted_pull_passes_the_result_through() { + let (_dir, coord) = coordinator(); + run_wanted_pull(&FakeHost::up(), &coord).await.expect("ok"); + run_wanted_pull(&FakeHost::up().failing("wanted_pull"), &coord) + .await + .expect_err("read failure reaches the node"); +}