diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index ab7fff0b..a88b4942 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -143,7 +143,8 @@ running rev (`services.hyperhive.c0re.hyperhiveRev`, from `self.rev`): The controller commits `flake.nix` and the returned `flake.lock` to `internal/meta` in one commit, and gives each hive's forge account read on the -repo. It re-renders when it starts, before it deploys a merged config PR, and +repo. It re-renders when it starts and every ten minutes after (a tick skips +while an update is pending or running), before it deploys a merged config PR, and before the first deploy of a new agent. Those deploys run whether or not the update succeeded: a hive that builds from `internal/meta` does so only while it pins the agent's current config (see diff --git a/swarm-controller/README.md b/swarm-controller/README.md index 10a8ec55..f58c9439 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -22,7 +22,9 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. reported takes its hive and declared state in `/api/agents/status` from the one hive whose declaration places it. - **Job graph** — a `hive-jobq` scheduler, served at `GET /api/jobq/graph`. - Every provisioning step below runs as a node in it. + Every provisioning step below runs as a node in it. A node that fails logs + one `warn` line, `swarm jobq: node failed`, with its `node` id, `kind` and + `error`. - **Agent creation** — `POST /api/agents` queues the SSO identity (through `swarm-authelia-bridge`), forge user, config repo, store identity, forge token and matrix account, declares the agent `paused`, then sends its hive diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 18e50825..f147855f 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -291,11 +291,28 @@ struct MetaDeps { const NEW_AGENT_WANTED_STATE: swarm_wire::wanted::AgentState = swarm_wire::wanted::AgentState::Paused; +/// Run a claimed node and log one warn when it fails, whatever its kind. +pub(crate) async fn run_swarm_node( + id: hive_jobq::NodeId, + kind: SwarmNodeKind, + builder: hive_jobq::builder::JobBuilder, + deps: WorkerDeps, +) -> ( + hive_jobq::builder::JobBuilder, + hive_jobq::scheduler::Outcome, +) { + let node_kind = hive_jobq_wire::WireNode::label(&kind); + let (builder, outcome) = run_node_work(kind, builder, deps).await; + if let hive_jobq::scheduler::Outcome::Failed(error) = &outcome { + tracing::warn!(node = id.get(), kind = %node_kind, %error, "swarm jobq: node failed"); + } + (builder, outcome) +} + /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind` /// variant turns into a real effect. -pub(crate) async fn run_swarm_node( - _id: hive_jobq::NodeId, +async fn run_node_work( kind: SwarmNodeKind, builder: hive_jobq::builder::JobBuilder, deps: WorkerDeps, @@ -431,7 +448,7 @@ async fn forge_node( /// together. const MINT_IDENTITY_FAILED: &str = "swarm jobq: mint_agent_identity failed"; -/// The `MintAgentIdentity` arm, lifted out so `run_swarm_node` stays under +/// The `MintAgentIdentity` arm, lifted out so `run_node_work` stays under /// `clippy::too_many_lines`. pub(crate) async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; @@ -446,7 +463,7 @@ pub(crate) async fn mint_identity(agent: &str) -> hive_jobq::scheduler::Outcome } } -/// The `MintAgentForgeToken` arm, lifted out so `run_swarm_node` stays under +/// The `MintAgentForgeToken` arm, lifted out so `run_node_work` stays under /// `clippy::too_many_lines`. async fn mint_forge_token( forge: Option>, @@ -681,7 +698,7 @@ fn finish_update_meta( } } -/// The `RecordFanout` arm, lifted out so `run_swarm_node` stays under the line limit. +/// The `RecordFanout` arm, lifted out so `run_node_work` stays under the line limit. async fn record_fanout( queue: Option<&async_nats::Client>, pin: &str, @@ -1450,18 +1467,42 @@ pub(crate) fn queue_builder_forge_token_mint( .map_err(|e| anyhow::anyhow!("{e}")) } -/// Insert one `UpdateMeta` job unless one is already pending, which reads the +/// How often [`spawn_meta_update_timer`] queues an `UpdateMeta`. +const META_UPDATE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(10); + +/// Queue an `UpdateMeta` now and every [`META_UPDATE_INTERVAL`] after, so a +/// boot run that failed or hung, and drift the config-merge path did not see, +/// are picked up. A tick that finds the previous run pending, running or +/// finishing queues nothing. +pub(crate) fn spawn_meta_update_timer( + sched: Arc>>, +) { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(META_UPDATE_INTERVAL); + loop { + ticker.tick().await; + if let Err(e) = queue_meta_update(&sched) { + tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed"); + } + } + }); +} + +/// Insert one `UpdateMeta` job unless one is still in flight, which reads the /// same heads when it runs. Returns whether a job was inserted. -pub(crate) fn queue_meta_update( +fn queue_meta_update( sched: &Mutex>, ) -> Result { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - let pending = sched.graph().nodes().any(|n| { - n.state == hive_jobq::State::Pending && matches!(n.payload, SwarmNodeKind::UpdateMeta) + let in_flight = sched.graph().nodes().any(|n| { + matches!( + n.state, + hive_jobq::State::Pending | hive_jobq::State::Running | hive_jobq::State::Finishing + ) && matches!(n.payload, SwarmNodeKind::UpdateMeta) }); - if pending { + if in_flight { return Ok(false); } sched @@ -3073,6 +3114,42 @@ mod tests { labels.sort(); assert_eq!(labels, ["create_forge_user", "mint_agent_forge_token"]); } + + /// The timer's tick queues an `UpdateMeta` when none is in flight, and + /// queues nothing while one is pending or running; once it is done the + /// next tick queues again. + #[tokio::test] + async fn the_meta_timer_queues_unless_one_is_in_flight() { + let sched = + std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))); + let update_metas = || { + sched + .lock() + .unwrap() + .graph() + .nodes() + .filter(|n| matches!(n.payload, SwarmNodeKind::UpdateMeta)) + .count() + }; + assert!(super::queue_meta_update(&sched).expect("tick")); + assert_eq!(update_metas(), 1); + assert!(!super::queue_meta_update(&sched).expect("tick")); + assert_eq!(update_metas(), 1, "a pending one blocks the tick"); + + let running = hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move { + (b, hive_jobq::scheduler::Outcome::Done) + }) + .expect("runnable"); + assert!(!super::queue_meta_update(&sched).expect("tick")); + assert_eq!(update_metas(), 1, "a running one blocks the tick"); + + running.await.1.expect("no growth"); + assert!(super::queue_meta_update(&sched).expect("tick")); + assert_eq!(update_metas(), 2, "a finished one does not"); + } } #[cfg(test)] @@ -3082,7 +3159,7 @@ mod mint_identity_tests { use hive_jobq::scheduler::Outcome; use tracing_subscriber::layer::SubscriberExt as _; - use super::mint_identity; + use super::{SwarmNodeKind, WorkerDeps, mint_identity, run_swarm_node}; /// One event's level and its fields, each by name. type Event = (tracing::Level, Vec<(String, String)>); @@ -3139,4 +3216,62 @@ mod mint_identity_tests { assert_eq!(field("agent"), "a/b"); assert_eq!(field("error"), error); } + + /// A node of any kind whose work returns `Failed` logs one warn naming the + /// node's id, its kind and the error, from the one place that runs nodes. + #[tokio::test] + async fn a_failed_node_logs_its_kind_id_and_error() { + let events = Events::default(); + let _guard = + tracing::subscriber::set_default(tracing_subscriber::registry().with(events.clone())); + let sched = Arc::new(Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))); + let ids = sched + .lock() + .unwrap() + .insert_job(None, |b| { + vec![ + b.node(SwarmNodeKind::SetAgentWanted { + hive: "h".to_owned(), + agent: "a".to_owned(), + }) + .guid(), + ] + }) + .expect("inserted"); + let deps = WorkerDeps { + auth: None, + forge: super::Late::default(), + queue: None, + wanted: None, + matrix_homeserver: None, + }; + hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { + run_swarm_node(id, kind, builder, deps) + }) + .expect("runnable") + .await + .1 + .expect("no growth declared"); + + let got = events.0.lock().expect("unpoisoned").clone(); + let [(level, fields)] = got.as_slice() else { + panic!("one event, got {got:?}"); + }; + assert_eq!(*level, tracing::Level::WARN); + let field = |name: &str| { + fields + .iter() + .find(|(n, _)| n == name) + .map_or_else(|| panic!("no {name} in {fields:?}"), |(_, v)| v.clone()) + }; + assert_eq!(field("kind"), "set_agent_wanted"); + assert_eq!(field("node"), ids[0].get().to_string()); + assert!( + field("error").contains("no swarm queue is configured"), + "{fields:?}" + ); + } } diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index 3e8de038..7fe1dabb 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -288,10 +288,8 @@ fn spawn_forge_workers( ) { // `internal/meta` appears, catches up with every config head that moved // while this daemon was down, and takes this daemon's rev as the hyperhive - // pin when that is newer. - if let Err(e) = crate::jobq::queue_meta_update(jobq) { - tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed"); - } + // pin when that is newer. The timer's first tick is this boot run. + crate::jobq::spawn_meta_update_timer(Arc::clone(jobq)); forge::objects::spawn(Arc::clone(&client)); if let (Some(bridge), Some(queue)) = (auth, queue) { forge::operator_profiles::spawn(Arc::clone(&client), bridge, queue);