wanted pull: one deadline bounds both connect waits

pull waits (bounded) for the swarm queue connection instead of failing at boot while the client is still connecting. The watch runs one pull after every attach, including re-attaches, and a failed attach or ended stream is retried with a bounded backoff instead of waiting for the next boot. A first deploy is not queued again while one for the agent is queued or running.

An agent whose first deploy was lost to a c0re restart is now deployed on the next boot, or the next watch re-attach, without anyone re-declaring it.

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`         | one-shot boot-time pull of the agent set the swarm controller declares for this hive (`wanted::pull`), converging the agents it names. No background loop behind this one — boot is the whole cadence; the deploy event (`swarm_status`) is the fast path, this repairs a missed one. 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/job_queue/mod.rs

@@ -190,6 +190,32 @@ impl JobQueue {
         Ok(named)
     }
 
+    /// Queue `agent`'s first deploy unless one is already queued or running.
+    ///
+    /// `None` means a `Provision` for the agent is still live. The check and
+    /// the insert share one lock, so two callers racing (the wanted-state pull
+    /// and the watch, or the deploy event) queue one deploy between them. A
+    /// deploy that finished or failed does not count: a lost one is retried.
+    ///
+    /// # Errors
+    /// Propagates a graph-insert error.
+    pub fn insert_first_deploy_once(&self, agent: &str) -> anyhow::Result<Option<Vec<NodeId>>> {
+        let mut inner = self.lock();
+        let live = inner.graph().nodes().any(|n| {
+            !n.state.is_terminal()
+                && matches!(&n.payload, model::NodeKind::Provision { agent: a } if a == agent)
+        });
+        if live {
+            return Ok(None);
+        }
+        let named = inner
+            .insert_job(None, |b| templates::first_deploy(b, agent))
+            .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
+        drop(inner);
+        self.notify.notify_one();
+        Ok(Some(named))
+    }
+
     /// The scheduler itself, for `hive_jobq`'s run-loop seam
     /// (`Scheduler::claim_next`), which takes exactly this type.
     ///

hive-c0re/src/job_queue/tests.rs

@@ -1537,3 +1537,29 @@ fn each_sweep_holds_its_own_resource() {
         vec![vec![Resource::KnowledgeTree]]
     );
 }
+
+/// The pull, the watch and the deploy event can all see one declaration, so a
+/// second first deploy for an agent that has one queued must not be inserted.
+#[test]
+fn a_first_deploy_is_not_queued_twice_for_one_agent() {
+    let q = JobQueue::new(1);
+    let first = q.insert_first_deploy_once("newbie").expect("insert");
+    assert!(first.is_some(), "control: the first deploy is queued");
+    let kinds = || declared_shape(&q).len();
+    let queued = kinds();
+
+    assert!(
+        q.insert_first_deploy_once("newbie")
+            .expect("insert")
+            .is_none(),
+        "a live first deploy for the same agent blocks another"
+    );
+    assert_eq!(kinds(), queued, "nothing was added");
+
+    assert!(
+        q.insert_first_deploy_once("other")
+            .expect("insert")
+            .is_some(),
+        "control: another agent's deploy is independent"
+    );
+}

hive-c0re/src/swarm_status.rs

@@ -322,14 +322,18 @@ async fn handle_deploy_request(
         coord
             .job_queue
             .insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true))
+            .map(Some)
     } else {
         queue_first_deploy(coord, &agent)
     };
     match inserted {
-        Ok(_) => {
+        Ok(Some(_)) => {
             tracing::info!(%agent, known, "swarm events: deploy requested by the swarm, queued");
             coord.emit_rebuild_queue_snapshot();
         }
+        Ok(None) => {
+            tracing::info!(%agent, "swarm events: first deploy already queued; not queueing another");
+        }
         Err(e) => tracing::warn!(%agent, error = %e, "swarm events: queueing the deploy failed"),
     }
 }
@@ -377,16 +381,17 @@ async fn apply_rev(agent: &str, rev: &str) -> bool {
 /// Setting the row up front closes that window. A second caller open-coding
 /// the insert would lose exactly that, and the agent would come up stopped for
 /// no visible reason.
+///
+/// `Ok(None)` when a first deploy for the agent is already queued or running:
+/// the pull, the watch and the deploy event can all see the same declaration.
 pub(crate) fn queue_first_deploy(
     coord: &std::sync::Arc<crate::coordinator::Coordinator>,
     agent: &str,
-) -> Result<Vec<hive_jobq::NodeId>> {
+) -> Result<Option<Vec<hive_jobq::NodeId>>> {
     if let Err(e) = coord.power.set(agent, crate::power::Wanted::Up) {
         tracing::warn!(%agent, error = ?e, "agent_power: seed on swarm first-deploy failed");
     }
-    coord
+    coord.job_queue.insert_first_deploy_once(agent)
-        .job_queue
-        .insert_job(|b| crate::job_queue::templates::first_deploy(b, agent))
 }
 
 /// Offer one snapshot: this hive's current readiness, under its own key.

hive-c0re/src/workers/wanted.rs

@@ -27,10 +27,11 @@
 //! behind that reconcile's back on every boot.
 //!
 //! One consequence, deliberate: a declaration outranks a local stop — an agent
-//! declared `Up` returns at the next boot, because the controller owns that.
+//! declared `Up` returns at the next boot or watch attach, because the controller owns that.
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::{Context, Result};
 use swarm_wire::wanted::{AgentState, HiveWanted};
@@ -38,6 +39,19 @@ use swarm_wire::wanted::{AgentState, HiveWanted};
 use crate::coordinator::Coordinator;
 use crate::power::Wanted;
 
+/// 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.
+const CONNECT_POLL: Duration = Duration::from_millis(250);
+
+/// Wait after the first failed watch session; doubled after each further one.
+const FIRST_RETRY_DELAY: Duration = Duration::from_secs(5);
+
+/// The longest wait between two watch sessions.
+const MAX_RETRY_DELAY: Duration = Duration::from_mins(5);
+
 /// Read this hive's declaration and converge the agents it names.
 ///
 /// `Ok(())` covers three quiet cases that are **not** an empty declaration: no
@@ -47,6 +61,10 @@ use crate::power::Wanted;
 /// 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`] in total, for the client and then for its
+/// connection: at boot the shared client exists before it is connected, and a
+/// pull that did not wait would fail every boot.
+///
 /// # Errors
 /// Propagates a failed queue connect, KV read, decode, or agent enumeration.
 pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
@@ -54,16 +72,29 @@ pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
         tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read");
         return Ok(());
     };
-    let Some(client) = crate::swarm_queue::client().await else {
+    let deadline = tokio::time::Instant::now() + CONNECT_WAIT;
-        // Absent or failed — either way already reported by the shared
+    let client =
-        // connector (an `info` log or a `swarm_queue_config` banner).
+        match tokio::time::timeout_at(deadline, crate::swarm_queue::wait_for_client()).await {
-        return Ok(());
+            // 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()),
         };
-    // An unconnected client does not fail a JetStream request, it hangs on it.
+    pull_with(coord, &client, &hive, deadline).await
-    // A hung node is worse than a failed one: the dashboard shows it running.
+}
-    swarm_queue_client::ensure_connected(&client)?;
 
-    let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive)
+/// [`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, deadline).await?;
+
+    let Some(store) = swarm_queue_client::wanted::open_read_only(client, hive)
         .await
         .context("opening this hive's wanted-state bucket")?
     else {
@@ -71,7 +102,7 @@ pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
         return Ok(());
     };
     let Some(raw) = store
-        .get(&hive)
+        .get(hive)
         .await
         .context("reading this hive's wanted state")?
     else {
@@ -84,11 +115,90 @@ pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
     converge(coord, &declared).await
 }
 
+/// 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, deadline: tokio::time::Instant) -> Result<()> {
+    while swarm_queue_client::ensure_connected(client).is_err()
+        && tokio::time::Instant::now() < deadline
+    {
+        tokio::time::sleep(CONNECT_POLL).await;
+    }
+    Ok(swarm_queue_client::ensure_connected(client)?)
+}
+
+/// How one watch session ended.
+#[derive(Debug, PartialEq, Eq)]
+enum SessionEnd {
+    /// The hive is shutting down.
+    Shutdown,
+    /// The watch could not be attached (`attached == false`) or its stream
+    /// ended.
+    Lost { attached: bool },
+}
+
+/// The wait before the next watch session after `failures` sessions in a row
+/// never attached: [`FIRST_RETRY_DELAY`] doubling up to [`MAX_RETRY_DELAY`].
+fn retry_delay(failures: u32) -> Duration {
+    FIRST_RETRY_DELAY
+        .saturating_mul(1_u32 << failures.min(16))
+        .min(MAX_RETRY_DELAY)
+}
+
+/// Attach the watch, then pull once, and hand back the watch.
+///
+/// The watch comes first: a put landing between a pull and a later attach
+/// would be seen by neither. One landing between the two is seen by both, which
+/// `converge` tolerates. `None` when the attach did not happen; a failed pull is
+/// logged and does not drop the watch, since the next attach pulls again.
+async fn attach_then_pull<U, A, P>(attach: A, pull: P) -> Option<U>
+where
+    A: Future<Output = Option<U>>,
+    P: Future<Output = Result<()>>,
+{
+    let updates = attach.await?;
+    if let Err(e) = pull.await {
+        tracing::warn!(error = ?e, "wanted state: pull after attaching the watch failed");
+    }
+    Some(updates)
+}
+
+/// Run watch sessions until shutdown, waiting [`retry_delay`] between them.
+async fn supervise<F, Fut>(mut session: F, mut shutdown: tokio::sync::watch::Receiver<bool>)
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = SessionEnd>,
+{
+    let mut failures = 0_u32;
+    loop {
+        match session().await {
+            SessionEnd::Shutdown => return,
+            SessionEnd::Lost { attached } => {
+                failures = if attached {
+                    0
+                } else {
+                    failures.saturating_add(1)
+                };
+            }
+        }
+        tokio::select! {
+            () = tokio::time::sleep(retry_delay(failures)) => {}
+            _ = shutdown.changed() => return,
+        }
+    }
+}
+
 /// Converge again each time the controller republishes this hive's declaration.
 ///
-/// The **fast** path, and it does not replace [`pull`]: this hears only what is
+/// The **fast** path. Every attach, the first included, is followed by one
-/// published while it is listening, so a hive that was down still learns the
+/// [`pull`]: the watch hears only what is published while it is attached, so
-/// current declaration from the boot read. Both, not either.
+/// what was declared before an attach (a hive that was down, a deploy lost to
+/// a restart) would otherwise wait for the next boot. A failed attach or an
+/// ended stream is retried with a bounded backoff, forever.
 ///
 /// ⚠️ A refused watch and a quiet one are told apart here, unlike the core-NATS
 /// subscriptions in [`crate::swarm_status`]: a watch is a `JetStream` consumer,
@@ -98,29 +208,53 @@ pub async fn watch_declarations(
     client: async_nats::Client,
     coord: Arc<Coordinator>,
     hive: String,
-    mut shutdown: tokio::sync::watch::Receiver<bool>,
+    shutdown: tokio::sync::watch::Receiver<bool>,
 ) {
+    Box::pin(supervise(
+        || watch_session(&client, &coord, &hive, shutdown.clone()),
+        shutdown.clone(),
+    ))
+    .await;
+}
+
+/// One attach, one pull, then the updates until the stream ends.
+async fn watch_session(
+    client: &async_nats::Client,
+    coord: &Arc<Coordinator>,
+    hive: &str,
+    mut shutdown: tokio::sync::watch::Receiver<bool>,
+) -> SessionEnd {
     use futures_util::StreamExt as _;
 
-    let mut updates = match swarm_queue_client::wanted::watch(&client, &hive).await {
+    let deadline = tokio::time::Instant::now() + CONNECT_WAIT;
-        Ok(Some(updates)) => updates,
+    let attach = async {
+        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;
+        }
+        match swarm_queue_client::wanted::watch(client, hive).await {
+            Ok(Some(updates)) => Some(updates),
             Ok(None) => {
-            tracing::warn!(
+                tracing::warn!(%hive, "wanted state: cannot watch this hive's bucket; retrying");
-                %hive,
+                None
-                "wanted state: cannot watch this hive's bucket; changes will be \
-                 picked up at the next boot instead"
-            );
-            return;
             }
             Err(e) => {
                 tracing::warn!(
                     %hive,
                     error = %swarm_queue_client::chain(&e),
-                "wanted state: cannot watch this hive's bucket; changes will be \
+                    "wanted state: cannot watch this hive's bucket; retrying"
-                 picked up at the next boot instead"
                 );
-            return;
+                None
             }
+        }
+    };
+    let Some(mut updates) = Box::pin(attach_then_pull(
+        attach,
+        pull_with(coord, client, hive, deadline),
+    ))
+    .await
+    else {
+        return SessionEnd::Lost { attached: false };
     };
     tracing::info!(%hive, "wanted state: watching for declarations");
 
@@ -128,15 +262,14 @@ pub async fn watch_declarations(
         tokio::select! {
             entry = updates.next() => {
                 let Some(entry) = entry else {
-                    // Same reading as the sibling subscriptions: `async-nats`
+                    // `async-nats` reconnects underneath a live watch, so an
-                    // reconnects underneath a live watch, so an ended stream is
+                    // ended stream is the connection going away for good
-                    // the connection going away for good rather than a blip to
+                    // rather than a blip to spin on.
-                    // spin on.
+                    tracing::warn!(%hive, "wanted state: watch closed; re-attaching");
-                    tracing::warn!(%hive, "wanted state: watch closed");
+                    return SessionEnd::Lost { attached: true };
-                    return;
                 };
                 match entry {
-                    Ok(entry) => apply_entry(&coord, &hive, &entry).await,
+                    Ok(entry) => apply_entry(coord, hive, &entry).await,
                     // The watch survives one bad entry; the stream ending is
                     // the case above.
                     Err(e) => tracing::warn!(%hive, error = %e, "wanted state: watch error"),
@@ -144,7 +277,7 @@ pub async fn watch_declarations(
             }
             _ = shutdown.changed() => {
                 tracing::info!("wanted state: shutdown signal received");
-                return;
+                return SessionEnd::Shutdown;
             }
         }
     }
@@ -255,7 +388,9 @@ async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()>
     let mut queued = false;
     for agent in &plan.deploy {
         match crate::swarm_status::queue_first_deploy(coord, agent) {
-            Ok(_) => queued = true,
+            Ok(Some(_)) => queued = true,
+            // Already queued or running: the pull and the watch overlap.
+            Ok(None) => {}
             // Per agent rather than the whole sweep: one agent that cannot be
             // queued is not a reason to leave the others undeclared.
             Err(e) => tracing::warn!(%agent, error = %e, "wanted state: queueing a deploy failed"),
@@ -811,4 +946,171 @@ 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;
+
+        use super::super::{MAX_RETRY_DELAY, SessionEnd, attach_then_pull, retry_delay, supervise};
+
+        #[tokio::test]
+        async fn the_watch_attaches_before_the_pull_runs() {
+            let log = Arc::new(Mutex::new(Vec::new()));
+            let (a, p) = (Arc::clone(&log), Arc::clone(&log));
+            let got = attach_then_pull(
+                async move {
+                    a.lock().unwrap().push("attach");
+                    Some(7)
+                },
+                async move {
+                    p.lock().unwrap().push("pull");
+                    Ok(())
+                },
+            )
+            .await;
+            assert_eq!(got, Some(7));
+            assert_eq!(*log.lock().unwrap(), ["attach", "pull"]);
+        }
+
+        #[tokio::test]
+        async fn a_failed_attach_does_not_pull() {
+            let log = Arc::new(Mutex::new(Vec::new()));
+            let p = Arc::clone(&log);
+            let got: Option<u8> = attach_then_pull(async { None }, async move {
+                p.lock().unwrap().push("pull");
+                Ok(())
+            })
+            .await;
+            assert_eq!(got, None);
+            assert!(log.lock().unwrap().is_empty(), "no watch, no pull");
+        }
+
+        #[tokio::test]
+        async fn a_failed_pull_keeps_the_watch() {
+            let got = attach_then_pull(async { Some(1) }, async { anyhow::bail!("down") }).await;
+            assert_eq!(got, Some(1));
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn every_re_attach_pulls_again() {
+            let log = Arc::new(Mutex::new(Vec::new()));
+            let (_tx, shutdown) = tokio::sync::watch::channel(false);
+            let l = Arc::clone(&log);
+            supervise(
+                move || {
+                    let l = Arc::clone(&l);
+                    async move {
+                        let (a, p) = (Arc::clone(&l), Arc::clone(&l));
+                        let _ = attach_then_pull(
+                            async move {
+                                a.lock().unwrap().push("attach");
+                                Some(())
+                            },
+                            async move {
+                                p.lock().unwrap().push("pull");
+                                Ok(())
+                            },
+                        )
+                        .await;
+                        if l.lock().unwrap().len() >= 6 {
+                            SessionEnd::Shutdown
+                        } else {
+                            SessionEnd::Lost { attached: true }
+                        }
+                    }
+                },
+                shutdown,
+            )
+            .await;
+            assert_eq!(
+                *log.lock().unwrap(),
+                ["attach", "pull", "attach", "pull", "attach", "pull"]
+            );
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn failed_attaches_are_retried_not_abandoned() {
+            let tries = Arc::new(Mutex::new(0_u32));
+            let (_tx, shutdown) = tokio::sync::watch::channel(false);
+            let t = Arc::clone(&tries);
+            supervise(
+                move || {
+                    let t = Arc::clone(&t);
+                    async move {
+                        let mut n = t.lock().unwrap();
+                        *n += 1;
+                        if *n >= 5 {
+                            SessionEnd::Shutdown
+                        } else {
+                            SessionEnd::Lost { attached: false }
+                        }
+                    }
+                },
+                shutdown,
+            )
+            .await;
+            assert_eq!(*tries.lock().unwrap(), 5);
+        }
+
+        #[tokio::test(start_paused = true)]
+        async fn an_attach_resets_the_backoff() {
+            let start = tokio::time::Instant::now();
+            let starts = Arc::new(Mutex::new(Vec::new()));
+            let (_tx, shutdown) = tokio::sync::watch::channel(false);
+            let s = Arc::clone(&starts);
+            supervise(
+                move || {
+                    let s = Arc::clone(&s);
+                    async move {
+                        let mut s = s.lock().unwrap();
+                        s.push(start.elapsed().as_secs());
+                        match s.len() {
+                            1 | 2 => SessionEnd::Lost { attached: false },
+                            3 => SessionEnd::Lost { attached: true },
+                            _ => SessionEnd::Shutdown,
+                        }
+                    }
+                },
+                shutdown,
+            )
+            .await;
+            // Waits: 10 and 20 after two misses, then 5 once an attach worked.
+            assert_eq!(*starts.lock().unwrap(), [0, 10, 30, 35]);
+        }
+
+        #[test]
+        fn the_retry_delay_doubles_and_is_capped() {
+            let secs: Vec<u64> = (0..8).map(|n| retry_delay(n).as_secs()).collect();
+            assert_eq!(secs, [5, 10, 20, 40, 80, 160, 300, 300]);
+            assert_eq!(retry_delay(u32::MAX), MAX_RETRY_DELAY);
+            assert!(retry_delay(1_000) <= Duration::from_mins(5));
+        }
+    }
 }