From 31cb6bfa42027400c119b25fda7183cd94e8e1d3 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 25 Oct 2026 12:56:30 +0100 Subject: [PATCH] queue the meta update only until one run succeeds --- docs/swarm/builder.md | 5 +- swarm-controller/src/jobq.rs | 118 ++++++++++++++++++++++++++--------- 2 files changed, 91 insertions(+), 32 deletions(-) diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index a88b4942..6f0274e0 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -143,8 +143,9 @@ 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 and every ten minutes after (a tick skips -while an update is pending or running), before it deploys a merged config PR, and +repo. It re-renders when it starts, and again every ten minutes while the latest +update hasn't succeeded (a tick adds nothing while one is pending or running, +or after a success), 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/src/jobq.rs b/swarm-controller/src/jobq.rs index f147855f..f66e6402 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -1467,13 +1467,14 @@ pub(crate) fn queue_builder_forge_token_mint( .map_err(|e| anyhow::anyhow!("{e}")) } -/// How often [`spawn_meta_update_timer`] queues an `UpdateMeta`. +/// How often [`spawn_meta_update_timer`] looks at the last `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. +/// Look at the last `UpdateMeta` now and every [`META_UPDATE_INTERVAL`] after, +/// and queue one when none has run yet or the latest did not succeed, so a +/// boot run that failed or hung is retried. The first tick is the boot run. +/// Once a run has succeeded a tick adds nothing: the graph keeps every node, +/// so a clock-driven node per tick would grow it without bound. pub(crate) fn spawn_meta_update_timer( sched: Arc>>, ) { @@ -1488,21 +1489,32 @@ pub(crate) fn spawn_meta_update_timer( }); } -/// Insert one `UpdateMeta` job unless one is still in flight, which reads the -/// same heads when it runs. Returns whether a job was inserted. +/// 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. fn queue_meta_update( sched: &Mutex>, ) -> Result { let mut sched = sched .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - let in_flight = sched.graph().nodes().any(|n| { + let mut update_metas = sched + .graph() + .nodes() + .filter(|n| matches!(n.payload, SwarmNodeKind::UpdateMeta)) + .collect::>(); + if update_metas.iter().any(|n| { matches!( n.state, hive_jobq::State::Pending | hive_jobq::State::Running | hive_jobq::State::Finishing - ) && matches!(n.payload, SwarmNodeKind::UpdateMeta) - }); - if in_flight { + ) + }) { + return Ok(false); + } + update_metas.sort_by_key(|n| n.id); + if update_metas + .last() + .is_some_and(|n| n.state == hive_jobq::State::Done) + { return Ok(false); } sched @@ -3115,40 +3127,86 @@ mod tests { 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. + /// A scheduler holding one `UpdateMeta` that ran and ended `outcome`. + async fn after_one_update_meta( + outcome: hive_jobq::scheduler::Outcome, + ) -> std::sync::Arc< + std::sync::Mutex>, + > { + let sched = + std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))); + assert!(super::queue_meta_update(&sched).expect("tick")); + hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move { (b, outcome) }) + .expect("runnable") + .await + .1 + .expect("no growth"); + sched + } + + fn update_metas( + sched: &std::sync::Mutex>, + ) -> usize { + sched + .lock() + .unwrap() + .graph() + .nodes() + .filter(|n| matches!(n.payload, SwarmNodeKind::UpdateMeta)) + .count() + } + + /// With none run yet the tick queues one, and queues no second while that + /// one is pending or running. #[tokio::test] - async fn the_meta_timer_queues_unless_one_is_in_flight() { + async fn the_meta_timer_queues_first_and_skips_while_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"); + assert_eq!(update_metas(&sched), 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) + (b, hive_jobq::scheduler::Outcome::Failed("late".to_owned())) }) .expect("runnable"); assert!(!super::queue_meta_update(&sched).expect("tick")); - assert_eq!(update_metas(), 1, "a running one blocks the tick"); - + assert_eq!(update_metas(&sched), 1, "a running one blocks the tick"); running.await.1.expect("no growth"); + } + + /// Once the latest run succeeded, ticks add no node. + #[tokio::test] + async fn the_meta_timer_adds_nothing_after_a_success() { + let sched = after_one_update_meta(hive_jobq::scheduler::Outcome::Done).await; + for _ in 0..3 { + assert!(!super::queue_meta_update(&sched).expect("tick")); + } + assert_eq!(update_metas(&sched), 1); + } + + /// A failed latest run is retried; a success after it ends the retries. + #[tokio::test] + async fn the_meta_timer_retries_after_a_failure_until_a_success() { + let sched = + after_one_update_meta(hive_jobq::scheduler::Outcome::Failed("no".to_owned())).await; assert!(super::queue_meta_update(&sched).expect("tick")); - assert_eq!(update_metas(), 2, "a finished one does not"); + assert_eq!(update_metas(&sched), 2); + hive_jobq::scheduler::Scheduler::claim_next(&sched, |_, _, b| async move { + (b, hive_jobq::scheduler::Outcome::Done) + }) + .expect("runnable") + .await + .1 + .expect("no growth"); + assert!(!super::queue_meta_update(&sched).expect("tick")); + assert_eq!(update_metas(&sched), 2); } }