wanted pull: one deadline bounds both connect waits

docs/scheduler/coordinator.md

@@ -54,7 +54,7 @@ Cheap — no build slot:
 <!-- vale write-good.Passive = NO -->
 
 | Node                 | Behavior                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
-| -------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `MetaSync`           | the rebuild's meta preamble — rebuild-dir prep, idempotent meta `sync_agents`, optional per-agent relock. Holds the `MetaWindow` resource (below); deliberately its own node so the window never covers `Prebuild`'s multi-minute build                                                                                                                                                                                                                                                                       |
 | `Provision`          | first-spawn pre-create provisioning — proposed/applied repos, state subvolume, meta registration (`sync_agents`); runs ahead of `Create` so the `nixos-container create --flake meta#<name>` ref resolves. Store/meta-only, no container yet                                                                                                                                                                                                                                                                  |
 | `Reconcile`          | idempotent power converge: read `wanted` (below) + observed state; start if `Up` & down (cold-start fallback included), stop if `Offline` & up, else noop                                                                                                                                                                                                                                                                                                                                                     |
@@ -76,7 +76,7 @@ Cheap — no build slot:
 | `ForgeSweep`         | one-shot boot-time forge user/token sweep for every container (`forge::ensure_all`) as a first-class node, so it shows as real work on the dashboard instead of running invisibly in a bare `tokio::spawn`. Agentless                                                                                                                                                                                                                                                                                         |
 | `MatrixSweep`        | matrix user/space sweep (`matrix::ensure_all`): the boot-time instance, plus one every 30 min from a loop in `main.rs`. Holds `Resource::MatrixSweep` (capacity 1), so two passes never overlap; each tick queues its own pass, which waits for the resource if one is already live. Agentless                                                                                                                                                                                                                |
 | `KnowledgePull`      | `/knowledge` pull (`knowledge::pull`): at boot (commits that landed while `hive-c0re` was down), on the swarm knowledge-changed event, and hourly as a fallback. Holds `Resource::KnowledgeTree` (capacity 1), so two pulls never overlap on the working tree; each trigger queues its own pass, which waits for the resource if one is already live. Agentless                                                                                                                                               |
-| `WantedPull`         | pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. Runs at boot, waiting up to a minute for the queue connection, and again after every attach of the wanted-state watch (which retries with a backoff when the attach fails or the stream ends); the deploy event (`swarm_status`) and the watch are the fast path, this repairs a missed one. A first deploy already queued or running for an agent isn't queued again Agentless |
+| `WantedPull`         | pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. Runs at boot, waiting up to a minute in total for the queue connection, and again after every attach of the wanted-state watch (which retries with a backoff when the attach fails or the stream ends); the deploy event (`swarm_status`) and the watch are the fast path, this repairs a missed one. A first deploy already queued or running for an agent isn't queued again. Agentless |
 
 <!-- vale write-good.Passive = YES -->
 

hive-c0re/src/workers/wanted.rs

@@ -39,7 +39,8 @@ use swarm_wire::wanted::{AgentState, HiveWanted};
 use crate::coordinator::Coordinator;
 use crate::power::Wanted;
 
-/// Longest `pull` and each watch attach wait for the swarm queue to connect.
+/// Longest `pull`, and each watch session's attach and first pull together, wait
+/// for the swarm queue to connect.
 const CONNECT_WAIT: Duration = Duration::from_mins(1);
 
 /// How often the wait for a connected client looks at the connection state.
@@ -60,9 +61,9 @@ const MAX_RETRY_DELAY: Duration = Duration::from_mins(5);
 /// this hive cannot enumerate — is an error, so the node fails visibly on the
 /// dashboard instead of reading as "nothing was declared".
 ///
-/// Waits up to [`CONNECT_WAIT`] for the queue connection: at boot the shared
+/// Waits up to [`CONNECT_WAIT`] in total, for the client and then for its
-/// client exists before it is connected, and a pull that did not wait would
+/// connection: at boot the shared client exists before it is connected, and a
-/// fail every boot.
+/// pull that did not wait would fail every boot.
 ///
 /// # Errors
 /// Propagates a failed queue connect, KV read, decode, or agent enumeration.
@@ -71,24 +72,27 @@ pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
         tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read");
         return Ok(());
     };
+    let deadline = tokio::time::Instant::now() + CONNECT_WAIT;
     let client =
-        match tokio::time::timeout(CONNECT_WAIT, crate::swarm_queue::wait_for_client()).await {
+        match tokio::time::timeout_at(deadline, crate::swarm_queue::wait_for_client()).await {
             // Absent — already reported by the shared connector (an `info` log or
             // a `swarm_queue_config` banner).
             Ok(None) => return Ok(()),
             Ok(Some(client)) => client,
             Err(_) => anyhow::bail!("no swarm queue client within {}s", CONNECT_WAIT.as_secs()),
         };
-    pull_with(coord, &client, &hive).await
+    pull_with(coord, &client, &hive, deadline).await
 }
 
-/// [`pull`] over a client the caller already has.
+/// [`pull`] over a client the caller already has, waiting for its connection
+/// no later than `deadline`.
 async fn pull_with(
     coord: &Arc<Coordinator>,
     client: &async_nats::Client,
     hive: &str,
+    deadline: tokio::time::Instant,
 ) -> Result<()> {
-    wait_connected(client, CONNECT_WAIT).await?;
+    wait_connected(client, deadline).await?;
 
     let Some(store) = swarm_queue_client::wanted::open_read_only(client, hive)
         .await
@@ -111,15 +115,14 @@ async fn pull_with(
     converge(coord, &declared).await
 }
 
-/// Wait up to `limit` for `client` to be connected.
+/// Wait until `deadline` for `client` to be connected.
 ///
 /// `swarm_queue::wait_for_client` returns a client that may still be
 /// connecting. An unconnected client does not fail a `JetStream` request, it
 /// hangs on it, and a hung node shows as running on the dashboard, so every
 /// request here is preceded by this bounded wait and its
 /// `ensure_connected` verdict.
-async fn wait_connected(client: &async_nats::Client, limit: Duration) -> Result<()> {
+async fn wait_connected(client: &async_nats::Client, deadline: tokio::time::Instant) -> Result<()> {
-    let deadline = tokio::time::Instant::now() + limit;
     while swarm_queue_client::ensure_connected(client).is_err()
         && tokio::time::Instant::now() < deadline
     {
@@ -223,8 +226,9 @@ async fn watch_session(
 ) -> SessionEnd {
     use futures_util::StreamExt as _;
 
+    let deadline = tokio::time::Instant::now() + CONNECT_WAIT;
     let attach = async {
-        if let Err(e) = wait_connected(client, CONNECT_WAIT).await {
+        if let Err(e) = wait_connected(client, deadline).await {
             tracing::warn!(%hive, error = ?e, "wanted state: cannot watch this hive's bucket; retrying");
             return None;
         }
@@ -244,8 +248,11 @@ async fn watch_session(
             }
         }
     };
-    let Some(mut updates) =
+    let Some(mut updates) = Box::pin(attach_then_pull(
-        Box::pin(attach_then_pull(attach, pull_with(coord, client, hive))).await
+        attach,
+        pull_with(coord, client, hive, deadline),
+    ))
+    .await
     else {
         return SessionEnd::Lost { attached: false };
     };
@@ -940,6 +947,33 @@ mod tests {
         }
     }
 
+    /// A client that never connects: initial-connect retries run in the
+    /// background and `ensure_connected` keeps failing.
+    async fn never_connected() -> async_nats::Client {
+        async_nats::ConnectOptions::new()
+            .retry_on_initial_connect()
+            .connect("nats://127.0.0.1:1")
+            .await
+            .expect("a client that retries its first connect is returned unconnected")
+    }
+
+    /// One deadline bounds the whole connect wait: time already spent before
+    /// `wait_connected` counts against it, so `pull`'s client wait and its
+    /// connection wait together stay inside one `CONNECT_WAIT`.
+    #[tokio::test(start_paused = true)]
+    async fn the_connect_wait_is_bounded_by_one_deadline() {
+        let client = never_connected().await;
+        let start = tokio::time::Instant::now();
+        let deadline = start + super::CONNECT_WAIT;
+        tokio::time::sleep(std::time::Duration::from_secs(59)).await;
+        assert!(super::wait_connected(&client, deadline).await.is_err());
+        assert!(
+            start.elapsed() <= super::CONNECT_WAIT + super::CONNECT_POLL,
+            "waited {:?} in total",
+            start.elapsed()
+        );
+    }
+
     mod watch_supervision {
         use std::sync::{Arc, Mutex};
         use std::time::Duration;