From 263b21021afcf7de41dad4722d0e409899a6ff18 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 7 Nov 2026 11:17:13 +0100 Subject: [PATCH] swarm-controller: ask the builder for the archetypes again soon after a failed pass --- docs/2-handbook/swarm/archetypes.md | 9 ++- swarm-controller/src/archetypes.rs | 118 ++++++++++++++++++++++++++-- 2 files changed, 116 insertions(+), 11 deletions(-) diff --git a/docs/2-handbook/swarm/archetypes.md b/docs/2-handbook/swarm/archetypes.md index 1b9da25a..c133ec1d 100644 --- a/docs/2-handbook/swarm/archetypes.md +++ b/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: diff --git a/swarm-controller/src/archetypes.rs b/swarm-controller/src/archetypes.rs index 227d6dc3..9b390dcc 100644 --- a/swarm-controller/src/archetypes.rs +++ b/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 { + 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 { + 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::(&bucket, BUILTIN_SOURCE) + .await + .expect("reads"); + assert_eq!(stored.archetypes.keys().collect::>(), ["reviewer"]); + } + #[test] fn the_view_lists_each_archetype_under_its_source() { let r = evaluated(json!({