diff --git a/swarm-controller/src/agent_renewal.rs b/swarm-controller/src/agent_renewal.rs index 27286640..17ec373c 100644 --- a/swarm-controller/src/agent_renewal.rs +++ b/swarm-controller/src/agent_renewal.rs @@ -382,9 +382,12 @@ async fn observe_all( } /// Wait until `connected` holds, checking every [`CONNECT_POLL`] for at most -/// [`RECONCILE_INTERVAL`]. Returns whether it held. -async fn wait_for_queue(connected: impl Fn() -> bool) -> bool { - let deadline = tokio::time::Instant::now() + RECONCILE_INTERVAL; +/// `bound`. Returns whether it held. +pub(crate) async fn wait_for_queue( + connected: impl Fn() -> bool, + bound: std::time::Duration, +) -> bool { + let deadline = tokio::time::Instant::now() + bound; loop { if connected() { return true; @@ -402,7 +405,7 @@ where P: FnMut() -> F, F: std::future::Future>>, { - if !wait_for_queue(connected).await { + if !wait_for_queue(connected, RECONCILE_INTERVAL).await { tracing::warn!( waited_s = RECONCILE_INTERVAL.as_secs(), "agent credential renewal: queue still not connected; running the first pass anyway" diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 6ac00181..d30b281f 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -790,6 +790,7 @@ async fn update_meta( .to_owned(), ); }; + await_queue_connection(|| wanted.is_connected()).await?; let placements = placements(wanted, &meta.hives) .await .map_err(|e| format!("{e:#}"))?; @@ -817,6 +818,25 @@ async fn update_meta( }) } +/// The longest `UpdateMeta` waits for the swarm queue client to connect before +/// it reads placements. The boot run starts before the client has connected, +/// and a failed run is not declared again until the next +/// [`META_UPDATE_INTERVAL`] tick. +const QUEUE_CONNECT_WAIT: std::time::Duration = std::time::Duration::from_mins(1); + +/// Wait up to [`QUEUE_CONNECT_WAIT`] for `connected` to hold; `Err` is the +/// node's failure when it never does. +async fn await_queue_connection(connected: impl Fn() -> bool) -> Result<(), String> { + if crate::agent_renewal::wait_for_queue(connected, QUEUE_CONNECT_WAIT).await { + return Ok(()); + } + Err(format!( + "the swarm queue client was still not connected after {}s, so the placed \ + agents cannot be read", + QUEUE_CONNECT_WAIT.as_secs() + )) +} + /// The `UpdateMeta` node's outcome from what [`update_meta`] decided, with the /// deploys declared into `builder` (the scheduler inserts them under the node /// when it completes). Synchronous: the builder is not `Sync`, so it cannot be @@ -4506,6 +4526,42 @@ mod tests { assert_eq!(build_prs(&sched, "atlas"), [Some(7)]); assert_eq!(build_prs(&sched, "iris"), [None]); } + + /// The boot run: a client that is `Pending` when the node starts and + /// connects a few seconds later gives a node that proceeds. + #[tokio::test(start_paused = true)] + async fn update_meta_waits_for_a_pending_queue_client_to_connect() { + use std::sync::atomic::{AtomicBool, Ordering}; + let connected = std::sync::Arc::new(AtomicBool::new(false)); + let flag = std::sync::Arc::clone(&connected); + let waiting = tokio::spawn(async move { + super::await_queue_connection(move || flag.load(Ordering::SeqCst)).await + }); + + tokio::time::sleep(std::time::Duration::from_secs(12)).await; + assert!(!waiting.is_finished(), "returned while still Pending"); + + connected.store(true, Ordering::SeqCst); + tokio::time::sleep(std::time::Duration::from_secs(6)).await; + assert_eq!(waiting.await.expect("joined"), Ok(())); + } + + /// A client that never connects fails the node once the bound passes, + /// with a message naming the queue and the wait. + #[tokio::test(start_paused = true)] + async fn update_meta_fails_clearly_when_the_queue_never_connects() { + let waiting = tokio::spawn(async { super::await_queue_connection(|| false).await }); + + tokio::time::sleep(std::time::Duration::from_secs(58)).await; + assert!(!waiting.is_finished(), "gave up before the bound"); + + tokio::time::sleep(std::time::Duration::from_secs(8)).await; + let err = waiting.await.expect("joined").expect_err("never connected"); + assert!( + err.contains("swarm queue") && err.contains("60s"), + "unclear message: {err}" + ); + } } #[cfg(test)]