diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 3d2c3363..917d29ca 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -149,7 +149,9 @@ or after a success), before it deploys a merged config PR, and before the first deploy of a new agent. Those deploys run only after the update succeeded: a hive builds only from `internal/meta`, so after a failed update the scheduler skips the deploy node, and the graph shows the update as -failed and the deploy as skipped (see +failed and the deploy as skipped. Each ten-minute retry of a failed update +declares a fresh deploy behind it for every agent whose latest deploy the failed update +skipped, so the run that succeeds sends them (see [Config changes](../agent-lifecycle/config-changes.md#the-swarms-meta-flake)). A render equal to the committed `flake.nix` sends no lock. diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 1054a8ae..1ca5a21f 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -1488,8 +1488,53 @@ pub(crate) fn spawn_meta_update_timer( }); } +/// The deploys a failed `UpdateMeta` skipped, as `(hive, agent, rev)`: for each +/// agent, its latest `TriggerDeploy` when it waited on an `UpdateMeta` that +/// `Failed` (so the scheduler skipped it). A deploy skipped for another reason +/// (the pause declaration or the config seeding failed) is left alone: the +/// update that would unblock it was skipped too, and deploying past those is +/// the race those edges close. +fn skipped_deploys( + graph: &hive_jobq::Graph, +) -> Vec<(String, String, Option)> { + let mut latest = std::collections::BTreeMap::new(); + for n in graph.nodes() { + if let SwarmNodeKind::TriggerDeploy { hive, agent, .. } = &n.payload { + let key = (hive.clone(), agent.clone()); + if latest + .get(&key) + .is_none_or(|l: &&hive_jobq::Node<_, _>| l.id < n.id) + { + latest.insert(key, n); + } + } + } + latest + .into_values() + .filter(|n| { + n.deps.iter().any(|d| match d { + hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(|dep| { + matches!(dep.payload, SwarmNodeKind::UpdateMeta) + && dep.state == hive_jobq::State::Failed + }), + hive_jobq::Dep::Resource { .. } => false, + }) + }) + .filter_map(|n| match &n.payload { + SwarmNodeKind::TriggerDeploy { hive, agent, rev } => { + Some((hive.clone(), agent.clone(), rev.clone())) + } + _ => None, + }) + .collect() +} + /// Insert one `UpdateMeta` job when none is in flight and the latest one, by -/// node id, is not `Done` (or none exists). Returns whether a job was inserted. +/// node id, is not `Done` (or none exists), with a fresh `TriggerDeploy` +/// behind it (`after_ok`) for every deploy the failed run skipped +/// ([`skipped_deploys`]), so an agent created or merged while the update was +/// failing is deployed by the run that succeeds. Returns whether a job was +/// inserted. fn queue_meta_update( sched: &Mutex>, ) -> Result { @@ -1516,13 +1561,18 @@ fn queue_meta_update( { return Ok(false); } + let skipped = skipped_deploys(sched.graph()); sched .insert_job(None, |b| { - vec![ - b.node(SwarmNodeKind::UpdateMeta) - .needs(SwarmResourceKind::Meta) - .guid(), - ] + let update = b + .node(SwarmNodeKind::UpdateMeta) + .needs(SwarmResourceKind::Meta); + for (hive, agent, rev) in skipped { + let _ = b + .node(SwarmNodeKind::TriggerDeploy { hive, agent, rev }) + .after_ok(update); + } + vec![update.guid()] }) .map_err(|e| anyhow::anyhow!("{e}"))?; Ok(true) @@ -3241,6 +3291,143 @@ mod tests { assert!(!super::queue_meta_update(&sched).expect("tick")); assert_eq!(update_metas(&sched), 2); } + + type SharedSched = std::sync::Arc< + std::sync::Mutex>, + >; + + /// Declare `agent`'s creation, as the endpoint does, and run every node to + /// the end: the meta update fails when `update_meta_fails`, the pause + /// declaration when `set_wanted_fails`, everything else succeeds. + async fn create_and_run( + sched: &SharedSched, + agent: &str, + update_meta_fails: bool, + set_wanted_fails: bool, + ) { + sched + .lock() + .unwrap() + .insert_job(None, |b| super::declare_agent_job(b, agent, "pr1ma", false)) + .expect("the graph inserts"); + drain(sched, update_meta_fails, set_wanted_fails).await; + } + + /// Run every runnable node; outcomes as for [`create_and_run`]. + async fn drain(sched: &SharedSched, update_meta_fails: bool, set_wanted_fails: bool) { + use hive_jobq::scheduler::{Outcome, Scheduler}; + + while let Some(runner) = Scheduler::claim_next(sched, |_id, kind, builder| async move { + let outcome = match kind { + SwarmNodeKind::UpdateMeta if update_meta_fails => { + Outcome::Failed("no builder".to_owned()) + } + SwarmNodeKind::SetAgentWanted { .. } if set_wanted_fails => { + Outcome::Failed("no queue".to_owned()) + } + _ => Outcome::Done, + }; + (builder, outcome) + }) { + runner.await.1.expect("no growth declared"); + } + } + + /// The states of `agent`'s `TriggerDeploy` nodes, oldest first. + fn deploys_of(sched: &SharedSched, agent: &str) -> Vec { + let guard = sched.lock().unwrap(); + let mut nodes: Vec<_> = guard + .graph() + .nodes() + .filter(|n| matches!(&n.payload, SwarmNodeKind::TriggerDeploy { agent: a, .. } if a == agent)) + .collect(); + nodes.sort_by_key(|n| n.id); + nodes.into_iter().map(|n| n.state).collect() + } + + /// `bob` deployed under a working update; then `atlas` was created while + /// the update failed, so its deploy is `Skipped`. + async fn bob_deployed_and_atlas_skipped() -> SharedSched { + use hive_jobq::State; + + let sched: SharedSched = + std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))); + create_and_run(&sched, "bob", false, false).await; + create_and_run(&sched, "atlas", true, false).await; + assert_eq!(deploys_of(&sched, "bob"), [State::Done]); + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); + sched + } + + /// An agent created while the update was failing is deployed by the run + /// that succeeds, exactly once. + #[tokio::test] + async fn the_next_successful_update_deploys_the_agent_a_failed_one_skipped() { + use hive_jobq::State; + + let sched = bob_deployed_and_atlas_skipped().await; + assert!(super::queue_meta_update(&sched).expect("tick")); + // A retry whose update fails too skips the new deploy as well. + drain(&sched, true, false).await; + assert_eq!( + deploys_of(&sched, "atlas"), + [State::Skipped, State::Skipped] + ); + assert!(super::queue_meta_update(&sched).expect("tick")); + drain(&sched, false, false).await; + assert_eq!( + deploys_of(&sched, "atlas"), + [State::Skipped, State::Skipped, State::Done] + ); + // The success ends the retries: no tick adds a node, so no second deploy. + assert!(!super::queue_meta_update(&sched).expect("tick")); + assert_eq!( + deploys_of(&sched, "atlas"), + [State::Skipped, State::Skipped, State::Done] + ); + } + + /// An agent whose latest deploy ran gets no extra one from the retry, even + /// when an older deploy of its was skipped. + #[tokio::test] + async fn an_agent_that_was_deployed_gets_no_extra_deploy_from_the_retry() { + use hive_jobq::State; + + let sched = bob_deployed_and_atlas_skipped().await; + assert!(super::queue_meta_update(&sched).expect("tick")); + drain(&sched, false, false).await; + assert_eq!(deploys_of(&sched, "bob"), [State::Done]); + + // atlas now holds an old Skipped deploy and a Done one. Another + // creation fails its update; the retry deploys carol and only carol. + create_and_run(&sched, "carol", true, false).await; + assert!(super::queue_meta_update(&sched).expect("tick")); + drain(&sched, false, false).await; + assert_eq!(deploys_of(&sched, "carol"), [State::Skipped, State::Done]); + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped, State::Done]); + assert_eq!(deploys_of(&sched, "bob"), [State::Done]); + } + + /// A deploy skipped because the pause declaration failed stays skipped: the + /// retry must not deploy past the declaration it waits for. + #[tokio::test] + async fn a_deploy_skipped_for_another_reason_is_not_redeclared() { + use hive_jobq::State; + + let sched: SharedSched = + std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))); + create_and_run(&sched, "atlas", false, true).await; + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); + assert!(super::queue_meta_update(&sched).expect("tick")); + drain(&sched, false, false).await; + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); + } } #[cfg(test)]