swarm-controller: update_meta waits up to 60s for the queue client to connect

The boot run of the UpdateMeta node read hive placements before the queue client had connected, failed on the Pending client, and was not declared again until the next 10 minute timer tick. It now waits for the connection, bounded to 60s, and fails with a message naming the wait when the queue never connects. agent_renewal's wait_for_queue takes the bound as a parameter so both callers share it.

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<Output = Result<Vec<(String, AgentObserved)>>>,
 {
-    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"

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)]