swarm-controller: ask the builder for the archetypes again soon after a failed pass

docs/2-handbook/swarm/archetypes.md

@@ -56,7 +56,9 @@ an archetype from any other source name.
 
 1. swarm-controller asks the builder for the list on the queue subject
    `$SWARM.builder.archetypes` once the queue connects, and every 10 minutes
-   after that.
+   after that. The controller asks again after a pass that gets no answer or
+   a failed evaluation: after 10 seconds, then 20, doubling up to 10 minutes,
+   until one succeeds.
 2. The builder evaluates the `meta` of each `hyperhiveArchetypes` entry on
    its own hyperhive and nixpkgs sources, and answers with the JSON or the
    last lines `nix` printed:
@@ -76,8 +78,9 @@ an archetype from any other source name.
 
 A builder host redeploy can change the list. The next pass, within 10
 minutes, picks it up. When no builder answers, the controller keeps the
-stored list and logs a warning on each pass. A swarm with no builder enabled
-shows no archetype select.
+stored list and logs a warning on each pass. A swarm whose
+builder connects after the controller has no list until a retry succeeds. A swarm with no builder enabled shows no
+archetype select.
 
 The stored value:
 

swarm-controller/src/archetypes.rs

@@ -6,7 +6,8 @@
 //! asks the builder for the `meta` of its hyperhive source's archetypes, checks
 //! each one the way `POST /api/agents` checks its settings, and writes the
 //! result under [`BUILTIN_SOURCE`] when it differs from what is stored. A
-//! failed or unanswered evaluation keeps the stored record.
+//! failed or unanswered evaluation keeps the stored record and is asked again
+//! after [`RETRY_FIRST`], doubling up to [`REFRESH_INTERVAL`].
 
 use std::time::Duration;
 
@@ -32,6 +33,16 @@ use crate::{agent_limits, agent_permissions, inference};
 /// screen.
 const REFRESH_INTERVAL: Duration = Duration::from_mins(10);
 
+/// How soon a failed pass is repeated; each further failure doubles the wait,
+/// up to [`REFRESH_INTERVAL`]. The builder connects to the queue on its own
+/// schedule, so the first pass can find nobody listening.
+const RETRY_FIRST: Duration = Duration::from_secs(10);
+
+/// The wait after a failed pass, given the wait after the failure before it.
+fn retry_after(previous: Option<Duration>) -> Duration {
+    previous.map_or(RETRY_FIRST, |wait| (wait * 2).min(REFRESH_INTERVAL))
+}
+
 /// One archetype's `meta` as a source flake declares it. Keys this daemon
 /// does not know, `ui` among them, are ignored, so an archetype written for a
 /// newer swarm still lists.
@@ -144,27 +155,40 @@ async fn store(client: &async_nats::Client, value: &SourceRecord) -> anyhow::Res
 }
 
 /// Refresh the record once the queue is connected, then every
-/// [`REFRESH_INTERVAL`].
+/// [`REFRESH_INTERVAL`]; a failed pass is repeated sooner, see [`retry_after`].
 pub(crate) fn spawn(client: async_nats::Client) {
     tokio::spawn(async move {
         const CONNECT_POLL: Duration = Duration::from_secs(5);
+        let mut retry = None;
         loop {
             while swarm_queue_client::ensure_connected(&client).is_err() {
                 tokio::time::sleep(CONNECT_POLL).await;
             }
-            refresh(&client).await;
-            tokio::time::sleep(REFRESH_INTERVAL).await;
+            tokio::time::sleep(pass(&client, &mut retry).await).await;
         }
     });
 }
 
-/// One evaluation, stored when it changed anything.
-async fn refresh(client: &async_nats::Client) {
+/// One refresh, and how long to wait before the next. `retry` is the wait
+/// after the last failed pass, `None` after a pass that succeeded.
+async fn pass(client: &async_nats::Client, retry: &mut Option<Duration>) -> Duration {
+    if refresh(client).await {
+        *retry = None;
+        return REFRESH_INTERVAL;
+    }
+    let wait = retry_after(*retry);
+    *retry = Some(wait);
+    wait
+}
+
+/// One evaluation, stored when it changed anything. Answers whether the
+/// stored record is now what the builder evaluated.
+async fn refresh(client: &async_nats::Client) -> bool {
     let value = match evaluate(client).await {
         Ok(value) => value,
         Err(error) => {
             tracing::warn!(%error, "archetypes: kept the stored list");
-            return;
+            return false;
         }
     };
     for (name, why) in &value.refused {
@@ -177,8 +201,12 @@ async fn refresh(client: &async_nats::Client) {
             "archetypes: stored"
         ),
         Ok(false) => {}
-        Err(e) => tracing::warn!(error = %format!("{e:#}"), "archetypes: storing failed"),
+        Err(e) => {
+            tracing::warn!(error = %format!("{e:#}"), "archetypes: storing failed");
+            return false;
+        }
     }
+    true
 }
 
 /// One archetype the create screen can offer, with the settings it prefills.
@@ -419,6 +447,80 @@ mod tests {
         assert_eq!(stored, changed);
     }
 
+    #[test]
+    fn a_failed_pass_is_repeated_sooner_each_time_up_to_the_interval() {
+        let mut previous = None;
+        let mut waits = Vec::new();
+        for _ in 0..8 {
+            let wait = retry_after(previous);
+            waits.push(wait.as_secs());
+            previous = Some(wait);
+        }
+        assert_eq!(waits, [10, 20, 40, 80, 160, 320, 600, 600]);
+    }
+
+    /// Against a real server: with no builder listening a pass keeps failing
+    /// and asks again after a longer wait each time; the first answer is
+    /// stored and puts the wait back to the interval.
+    #[tokio::test]
+    async fn a_pass_asks_again_until_the_builder_answers() {
+        use futures_util::StreamExt as _;
+
+        let nats = crate::test_nats::TestNats::start();
+        let client = nats.client().await;
+        let mut retry = None;
+        assert_eq!(pass(&client, &mut retry).await, Duration::from_secs(10));
+        assert_eq!(pass(&client, &mut retry).await, Duration::from_secs(20));
+        assert_eq!(retry, Some(Duration::from_secs(20)));
+
+        let builder = nats.client().await;
+        let mut requests = builder
+            .queue_subscribe(
+                ARCHETYPES_SUBJECT,
+                swarm_wire::builder::QUEUE_GROUP.to_owned(),
+            )
+            .await
+            .expect("subscribe");
+        // A message the builder sends itself and receives back proves the
+        // server processed the subscription before it.
+        let mut probe = builder
+            .subscribe("archetypes-test.probe")
+            .await
+            .expect("subscribe");
+        builder
+            .publish("archetypes-test.probe", "".into())
+            .await
+            .expect("publish");
+        probe.next().await.expect("the probe comes back");
+        tokio::spawn(async move {
+            let msg = requests.next().await.expect("a request");
+            let serde_json::Value::Object(archetypes) = json!({
+                "reviewer": { "description": "reviews" }
+            }) else {
+                unreachable!()
+            };
+            let reply = serde_json::to_vec(&ArchetypesReply::Evaluated {
+                source: "/nix/store/h".to_owned(),
+                archetypes,
+            })
+            .expect("encodes");
+            builder
+                .publish(msg.reply.expect("a reply subject"), reply.into())
+                .await
+                .expect("publish");
+        });
+        assert_eq!(pass(&client, &mut retry).await, REFRESH_INTERVAL);
+        assert_eq!(retry, None);
+        let bucket = swarm_queue_client::archetypes::open(&client)
+            .await
+            .expect("opens")
+            .expect("exists");
+        let (stored, _, _) = read_or_default::<SourceRecord>(&bucket, BUILTIN_SOURCE)
+            .await
+            .expect("reads");
+        assert_eq!(stored.archetypes.keys().collect::<Vec<_>>(), ["reviewer"]);
+    }
+
     #[test]
     fn the_view_lists_each_archetype_under_its_source() {
         let r = evaluated(json!({