diff --git a/docs/swarm/credentials.md b/docs/swarm/credentials.md index b65ef903..6cb22dc7 100644 --- a/docs/swarm/credentials.md +++ b/docs/swarm/credentials.md @@ -61,7 +61,7 @@ of the cell says how. | `swarm/agents//matrix/` | `swarm-controller` | the agent container itself, under the certificate its hive passed in | must be stated | must be stated | | `swarm/controller/swarm-controller/matrix/appservice-token` | `swarm-matrix-ctl`, inside the `hive-matrix` container, once | `swarm-controller`, under its own certificate | ❌ `swarm-matrix-ctl` mints it once; the container keeps its copy and republishes it when the store's differs | ✅ the controller reads it on every five-minute matrix pass | | `swarm/controller/swarm-controller/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | `swarm-controller`, under its own certificate, once at start | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ❌ read once at start; the controller holds the old value until it restarts | -| `swarm/controller/swarm-controller/forge/token` | the forge, inside the `hive-forge` container, once; `swarm-bao-forge-publisher` on the forge's host copies it in under its own certificate, with a write-only grant | `swarm-controller`, under its own certificate, at start | ❌ the forge mints it once and keeps its copy in the container; the publisher writes that copy each time the forge container starts | ❌ the controller reads it at start, so a replaced token reaches it with the next restart | +| `swarm/controller/swarm-controller/forge/token` | the forge, inside the `hive-forge` container, once; `swarm-bao-forge-publisher` on the forge's host copies it in under its own certificate, with a write-only grant | `swarm-controller`, under its own certificate, at start | ❌ the forge mints it once and keeps its copy in the container; the publisher writes that copy each time the forge container starts | ❌ the controller reads it at start, and in the background with backoff until it first finds it; a replaced token reaches it with the next restart | | `swarm/agents//bao-mtls` | the store's agent PKI mount (`deploy.bao.agentPkiMountPath`), which generates the key, at `swarm-controller`'s request at agent creation | `hive-c0re`, under the hive's own certificate, when it writes the agent's container config | ✅ `swarm-controller`'s five-minute pass re-issues a live agent's leaf once it's past half its validity (45 of 90 days, read from the certificate itself) | ❌ `hive-c0re` reads it when it writes the container config, so the agent presents a new leaf from its next start; the old leaf stays valid until it expires | | `swarm/agents//queue` | `swarm-controller`, at agent creation | `hive-agent` in the agent container, under the agent's own certificate, held in memory — the identity it presents to the swarm queue, naming that one agent rather than its hive | ✅ `swarm-controller`'s five-minute pass re-mints a live agent's secret once it's 45 days old by `minted_at` on the stored object; a secret with no `minted_at` gets one stamped, value unchanged. The pass skips agents declared `Destroyed` — declaring an agent destroyed deletes every version of the path instead, the undo of the mint rather than another one | ✅ `hive-agent` reads the path before its first connect and again on every reconnect attempt, so a reconnect after a re-mint presents the new secret. An open connection keeps the secret it connected with; after a revocation the agent keeps retrying under the queue client's backoff | | `swarm/agents//forge-token` | `swarm-controller`, at agent creation and in a pass every 5 minutes over every agent with a store identity | the agent container itself, under its own certificate, fetched to `/run/hive-agent-forge-token/token` | ✅ the controller re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the agent re-fetches on a 10-minute timer | diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 5dfcdee2..4fe6ad6a 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -191,7 +191,7 @@ let # Unconditional: every swarm has a forge, reached by name wherever it runs. # The token is not here: the daemon reads it from the swarm's secret store at - # start (`forge.rs`), and runs without forge access when it cannot. + # start (`forge.rs`), and re-reads it in the background until it is there. forgeEnv = { SWARM_CONTROLLER_FORGE_URL = "https://${forgeCfg.domain}"; # `agent-configs` org avatar for the forge-objects pass diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 50b5745f..0942d67e 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -6,13 +6,17 @@ //! //! The forge's URL comes from `swarm-controller.nix`'s `forgeEnv` //! (`SWARM_CONTROLLER_FORGE_URL`). The token comes from the swarm's secret -//! store, at `swarm_secret_client::forge::controller_token_path`, read once at -//! start under this daemon's own certificate and held only in memory. The -//! forge mints it inside its container (`hive-forge/default.nix`'s +//! store, at `swarm_secret_client::forge::controller_token_path`, read under +//! this daemon's own certificate and held only in memory. The forge mints it +//! inside its container (`hive-forge/default.nix`'s //! `forgejo-swarm-controller-account`) and `swarm-bao-forge-publisher`, on the -//! forge's host, puts it there. An unset URL means [`Client::from_env`] -//! returns `None` — every caller treats that as "no forge access here" and -//! continues, the same graceful-absence shape the queue coordinates use. +//! forge's host, puts it there. +//! +//! [`enable`] fills the daemon's one client slot: at start when the token is +//! there, later from a background re-read when it is not. Every caller reads +//! the slot and treats an empty one as "no forge access here" — the same +//! graceful-absence shape the queue coordinates use — so the daemon serves +//! from the start and gains forge access without a restart. use anyhow::{Context, Result}; use forgejo_api::structs::{ @@ -28,6 +32,7 @@ use futures_util::{StreamExt as _, TryStreamExt as _}; use reqwest::StatusCode; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::{Arc, OnceLock}; use utoipa::ToSchema; use crate::webhook::DeliveryKind; @@ -130,7 +135,7 @@ const KNOWLEDGE_REPO: &str = "knowledge"; const SWARM_CONTROLLER_FORGE_USER: &str = "swarm-controller"; /// The env var `swarm-controller.nix`'s `forgeEnv` sets. Named here once so -/// [`Client::from_env`] and the nix module can't drift silently. +/// [`url_from_env`] and the nix module can't drift silently. const URL_ENV: &str = "SWARM_CONTROLLER_FORGE_URL"; /// One login and one read. An empty value is refused like an absent one: it @@ -163,34 +168,104 @@ pub struct Client { url: url::Url, } +/// The forge's URL. `Ok(None)` (not an error) when `SWARM_CONTROLLER_FORGE_URL` +/// is unset — no forge is configured for this daemon. +/// +/// # Errors +/// A URL that does not parse. Retrying cannot fix that, so it is reported once +/// rather than waited out. +pub fn url_from_env() -> Result> { + let Some(url) = std::env::var(URL_ENV).ok() else { + return Ok(None); + }; + url::Url::parse(&url) + .with_context(|| format!("parse {URL_ENV} ({url}) as a URL")) + .map(Some) +} + +/// How long the background re-read waits before its `n`-th attempt: 30s, +/// doubling, capped at 5 minutes. +fn background_delay(n: u32) -> std::time::Duration { + const FIRST_S: u64 = 30; + const CAP_S: u64 = 300; + std::time::Duration::from_secs( + FIRST_S + .saturating_mul(1u64.checked_shl(n).unwrap_or(u64::MAX)) + .min(CAP_S), + ) +} + +/// Fill `slot` with the first value `attempt` yields, sleeping +/// [`background_delay`] before each attempt. Never gives up, and logs no +/// failure: the caller logs the first one and the recovery. +async fn fill_when_ready(slot: &OnceLock, mut attempt: F) -> &T +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + let mut n = 0; + loop { + tokio::time::sleep(background_delay(n)).await; + match attempt().await { + Ok(value) => return slot.get_or_init(|| value), + Err(e) => tracing::debug!( + error = %format!("{e:#}"), + "the controller's forge token is still unavailable" + ), + } + n = n.saturating_add(1); + } +} + +/// Put a client for `url` into `slot` and hand it to `on_ready`: at once when +/// the store has the token, otherwise from a background task that re-reads it +/// until it does. Either way the daemon starts, and every forge-backed path +/// reads `slot`, so it answers "no forge" until the token lands and works +/// from then on. +/// +/// A token replaced in the store after the slot is filled reaches this daemon +/// with its next restart. +pub async fn enable( + url: url::Url, + slot: Arc>>, + on_ready: impl FnOnce(Arc) + Send + 'static, +) { + match Client::connect(&url).await { + Ok(client) => on_ready(Arc::clone(slot.get_or_init(|| Arc::new(client)))), + Err(e) => { + tracing::warn!( + error = %format!("{e:#}"), + "no forge access yet; re-reading the forge token from the swarm secret store in \ + the background" + ); + tokio::spawn(async move { + let client = Arc::clone( + fill_when_ready(&slot, || async { + Client::connect(&url).await.map(Arc::new) + }) + .await, + ); + tracing::info!("the forge token is in the swarm secret store; forge access is on"); + on_ready(client); + }); + } + } +} + impl Client { - /// Build from env and the secret store. `Ok(None)` (not an error) when - /// `SWARM_CONTROLLER_FORGE_URL` is unset — no forge is configured for this - /// daemon (see the module doc comment). + /// One read of the token from the secret store, and a client for `url` + /// holding it. /// /// # Errors - /// An unparseable URL, or a token the store still would not give up after - /// `queue_identity::RETRY_DELAYS_S`: the store refusing the login, or - /// nothing at the path because the forge's host has not published it. - pub async fn from_env() -> Result> { - let Some(url) = std::env::var(URL_ENV).ok() else { - return Ok(None); - }; - let parsed_url = - url::Url::parse(&url).with_context(|| format!("parse {URL_ENV} ({url}) as a URL"))?; - let delays = crate::queue_identity::RETRY_DELAYS_S.map(std::time::Duration::from_secs); - let token = crate::queue_identity::fetch_with_retry( - &delays, - "the controller's forge token", - fetch_token, - ) - .await?; - let api = Forgejo::new(Auth::Token(&token), parsed_url.clone()) - .context("build forgejo client")?; - Ok(Some(Self { + /// The store refusing the login, nothing at the path because the forge's + /// host has not published it yet, or the client failing to build. + async fn connect(url: &url::Url) -> Result { + let token = fetch_token().await?; + let api = Forgejo::new(Auth::Token(&token), url.clone()).context("build forgejo client")?; + Ok(Self { api, - url: parsed_url, - })) + url: url.clone(), + }) } /// Creation options for an empty repo defaulting to `main`. @@ -1306,33 +1381,59 @@ mod tests { // the same process-global env var would race each other. A single // test's body is one thread, so sequential set/clear within it is // race-free without needing a cross-test lock. - #[tokio::test] - async fn from_env_is_none_without_a_url_and_refuses_a_bad_one_before_the_store() { + #[test] + fn url_from_env_is_none_without_a_url_and_refuses_a_bad_one() { // SAFETY: test-only mutation of the env var this module itself // owns (`URL_ENV`), scoped to this one test function so there's no // cross-test race (see comment above). unsafe { std::env::remove_var(URL_ENV); } - assert!(Client::from_env().await.unwrap().is_none()); + assert_eq!(url_from_env().unwrap(), None); unsafe { std::env::set_var(URL_ENV, "not a url"); } - // Not `.unwrap_err()`: that needs `Client: Debug` for its panic - // message, and the wrapped `forgejo_api::Forgejo` doesn't derive - // it. Returning at once, with no `BAO_*` set, is what shows the URL - // is checked before the store is asked. - match Client::from_env().await { - Err(e) => assert!(e.to_string().contains(URL_ENV), "{e:#}"), - Ok(_) => panic!("expected an error for an unparseable forge URL"), - } + let err = url_from_env().expect_err("an unparseable forge URL"); + assert!(err.to_string().contains(URL_ENV), "{err:#}"); unsafe { std::env::remove_var(URL_ENV); } } + #[test] + fn the_background_re_read_backs_off_from_30s_to_a_5_minute_cap() { + let s = |n| background_delay(n).as_secs(); + assert_eq!([s(0), s(1), s(2), s(3), s(4)], [30, 60, 120, 240, 300]); + assert_eq!(s(40), 300); + assert_eq!(s(u32::MAX), 300); + } + + // Paused time: each sleep completes at once and advances the clock, so the + // elapsed time is the backoff schedule itself. + #[tokio::test(start_paused = true)] + async fn a_token_absent_at_start_fills_the_slot_once_it_arrives() { + let slot = OnceLock::new(); + let start = tokio::time::Instant::now(); + let mut calls = 0; + let got = fill_when_ready(&slot, || { + calls += 1; + let n = calls; + async move { + if n < 3 { + anyhow::bail!("nothing at the controller's forge token path") + } + Ok("client") + } + }) + .await; + assert_eq!(*got, "client"); + assert_eq!(slot.get(), Some(&"client")); + assert_eq!(calls, 3); + assert_eq!(start.elapsed().as_secs(), 30 + 60 + 120); + } + #[tokio::test] async fn page_search_results_pages_past_a_full_first_page() { #[derive(Clone)] diff --git a/swarm-controller/src/forge_account.rs b/swarm-controller/src/forge_account.rs index 637baecf..8986cd39 100644 --- a/swarm-controller/src/forge_account.rs +++ b/swarm-controller/src/forge_account.rs @@ -157,10 +157,10 @@ mod tests { hive_jobq::resources::ResourceTable::new(), ))), webhook_secret: None, - config_prs: None, + config_prs: std::sync::Arc::default(), swarm_name: None, auth: None, - forge: None, + forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: Accounts::Fake(Arc::clone(store)), } diff --git a/swarm-controller/src/github_account.rs b/swarm-controller/src/github_account.rs index c143fcad..7b5ff1c7 100644 --- a/swarm-controller/src/github_account.rs +++ b/swarm-controller/src/github_account.rs @@ -127,10 +127,10 @@ mod tests { hive_jobq::resources::ResourceTable::new(), ))), webhook_secret: None, - config_prs: None, + config_prs: std::sync::Arc::default(), swarm_name: None, auth: None, - forge: None, + forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: Accounts::Fake(Arc::clone(store)), } diff --git a/swarm-controller/src/issue_report.rs b/swarm-controller/src/issue_report.rs index edc0a6b8..c9e67026 100644 --- a/swarm-controller/src/issue_report.rs +++ b/swarm-controller/src/issue_report.rs @@ -33,7 +33,7 @@ use crate::forge; pub async fn get_repos( State(state): State, ) -> Result>, StatusUnavailable> { - let Some(client) = state.forge.as_ref() else { + let Some(client) = state.forge.get() else { return Err(StatusUnavailable( "no forge is configured on this host".to_owned(), )); @@ -64,7 +64,7 @@ pub async fn get_repos( pub async fn get_issue_report_all( State(state): State, ) -> Result>, StatusUnavailable> { - let Some(client) = state.forge.as_ref() else { + let Some(client) = state.forge.get() else { return Err(StatusUnavailable( "no forge is configured on this host".to_owned(), )); @@ -109,7 +109,7 @@ pub async fn get_issue_report( State(state): State, Path((org, repo)): Path<(String, String)>, ) -> Result>, problem_details::ProblemDetails> { - let Some(client) = state.forge.as_ref() else { + let Some(client) = state.forge.get() else { return Err(error_problem( axum::http::StatusCode::SERVICE_UNAVAILABLE, "no forge is configured on this host", diff --git a/swarm-controller/src/linked_accounts.rs b/swarm-controller/src/linked_accounts.rs index bea302a2..4f3686e6 100644 --- a/swarm-controller/src/linked_accounts.rs +++ b/swarm-controller/src/linked_accounts.rs @@ -962,10 +962,10 @@ pub(crate) mod tests { hive_jobq::resources::ResourceTable::new(), ))), webhook_secret: None, - config_prs: None, + config_prs: std::sync::Arc::default(), swarm_name: None, auth: None, - forge: None, + forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: super::Accounts::Fake(Arc::clone(store)), } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 753dcd77..a421bd25 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -234,6 +234,10 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind { } } +/// A dependency this daemon may only gain after it starts serving: empty +/// until then, filled once. Readers take what is there at the moment they ask. +type Late = Arc>; + /// Everything a claimed node's executor arm might need to reach outside /// this process — bundled into one `Clone` struct rather than growing /// `run_swarm_node`'s parameter list per node kind (three forge-shaped @@ -245,7 +249,9 @@ impl hive_jobq_wire::WireResource for SwarmResourceKind { #[derive(Clone)] struct WorkerDeps { auth: Option>, - forge: Option>, + /// Empty until the forge token has been read (`forge::enable`), so a node + /// run before then fails as "no forge access", and one run after succeeds. + forge: Late>, /// The swarm queue connection, for nodes whose effect is a published /// event rather than an API call. A handle rather than the status /// reader it is cloned from: a node that announces a deploy has no @@ -288,6 +294,7 @@ async fn run_swarm_node( ) { use hive_jobq::scheduler::Outcome; + let forge = deps.forge.get().cloned(); let outcome = match kind { SwarmNodeKind::CreateIdentity { agent } => match deps.auth { None => { @@ -311,7 +318,7 @@ async fn run_swarm_node( Err(e) => Outcome::Failed(format!("{e:#}")), }, }, - SwarmNodeKind::CreateRepo { agent } => match deps.forge { + SwarmNodeKind::CreateRepo { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" @@ -325,7 +332,7 @@ async fn run_swarm_node( Err(e) => Outcome::Failed(format!("{e:#}")), }, }, - SwarmNodeKind::CreateForgeUser { agent } => match deps.forge { + SwarmNodeKind::CreateForgeUser { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" @@ -336,7 +343,7 @@ async fn run_swarm_node( Err(e) => Outcome::Failed(format!("{e:#}")), }, }, - SwarmNodeKind::AddRepoMember { agent } => match deps.forge { + SwarmNodeKind::AddRepoMember { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" @@ -347,8 +354,8 @@ async fn run_swarm_node( Err(e) => Outcome::Failed(format!("{e:#}")), }, }, - SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(deps.forge, &agent).await, - SwarmNodeKind::InitAgentConfigRepo { agent } => match deps.forge { + SwarmNodeKind::GrantDocsRead { agent } => grant_docs_read(forge, &agent).await, + SwarmNodeKind::InitAgentConfigRepo { agent } => match forge { None => Outcome::Failed( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store)" @@ -360,16 +367,14 @@ async fn run_swarm_node( }, }, SwarmNodeKind::MintAgentIdentity { agent } => mint_identity(&agent).await, - SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(deps.forge, &agent).await, + SwarmNodeKind::MintAgentForgeToken { agent } => mint_forge_token(forge, &agent).await, SwarmNodeKind::MintAgentMatrixAccount { agent } => { mint_matrix_account(deps.matrix_homeserver.as_deref(), &agent).await } SwarmNodeKind::MintHiveSenderToken { hive } => { mint_hive_sender(deps.matrix_homeserver.as_deref(), &hive).await } - SwarmNodeKind::MintHiveForgeToken { hive } => { - mint_hive_forge_token(deps.forge, &hive).await - } + SwarmNodeKind::MintHiveForgeToken { hive } => mint_hive_forge_token(forge, &hive).await, SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await, SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted { None => Outcome::Failed( @@ -774,10 +779,10 @@ struct AppState { /// takes without a second hop through `String`. webhook_secret: Option>, /// Last successful `agent-configs/*` scan, kept current by - /// [`config_pr::spawn`]. `None` when this host has no forge configured - /// — same "absent means don't ask" shape as `status` and `forge` above, - /// not a startup failure. - config_prs: Option>, + /// [`config_pr::spawn`]. Empty while this daemon has no forge access — + /// same "absent means don't ask" shape as `status` and `forge`, not a + /// startup failure — and filled when the forge client is. + config_prs: Late>, /// The swarm's human display name (`services.hyperhive.swarm.name`), /// loaded once at startup (`load_swarm_name`). `None` when the /// operator never set it — a swarm without a display name is a @@ -788,10 +793,9 @@ struct AppState { /// The forge client itself, for the read-only repo/issue-report /// routes (`GET /api/repos`, `GET /api/repos/{org}/{repo}/issue-report`) /// and `POST /api/forge/users/{name}/admin` — distinct from `config_prs`, which holds a *cache* built off this - /// same client rather than the client. `None` under the same - /// "no forge configured on this host" shape every other - /// forge-backed field here uses. - forge: Option>, + /// same client rather than the client. Empty while this daemon has no + /// forge access, and filled the moment `forge::enable` builds the client. + forge: Late>, /// Held by `create_agent` from reading where a name is placed until its /// graph is queued, and by `set_agent_state` from that read until its /// declaration is written, so two placements of one name cannot both @@ -1433,7 +1437,7 @@ async fn get_agents_status( })?; match reader.view(&roster, std::time::SystemTime::now()).await { Ok(mut rows) => { - if let Some(cache) = state.config_prs.as_ref() { + if let Some(cache) = state.config_prs.get() { let mut config_prs = cache.snapshot(); for row in &mut rows { row.config_pr = config_prs.remove(&row.name); @@ -2178,7 +2182,7 @@ async fn get_agent_config_pr( State(state): State, Path(name): Path, ) -> Result>, StatusUnavailable> { - let Some(cache) = state.config_prs.as_ref() else { + let Some(cache) = state.config_prs.get() else { return Err(StatusUnavailable( "no forge is configured on this host".to_owned(), )); @@ -2443,16 +2447,13 @@ fn spawn_matrix_account_backfill( /// registration token, hives' read tokens, agents' `internal/docs` grants, and /// agents' tokens (which also /// sweeps legacy tokens; see `forge::legacy_tokens`) — -/// when a forge is configured. Lifted out of `main` for +/// once there is a forge client. Lifted out of `main` for /// `clippy::too_many_lines`. fn spawn_forge_workers( jobq: &Arc>>, - forge_client: Option>, + client: Arc, hives: &[HiveEntry], ) { - let Some(client) = forge_client else { - return; - }; forge::objects::spawn(Arc::clone(&client)); forge::runner_token::spawn(Arc::clone(&client)); let sched = Arc::clone(jobq); @@ -2648,7 +2649,7 @@ async fn make_forge_admin( let name = hive_types::Ident::parse(&name) .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .into_string(); - let Some(forge) = state.forge.clone() else { + let Some(forge) = state.forge.get().cloned() else { return Err(error_problem( axum::http::StatusCode::SERVICE_UNAVAILABLE, "no forge is configured on this host", @@ -2828,7 +2829,7 @@ async fn get_agent_icon( async fn get_config_prs( State(state): State, ) -> Result>, StatusUnavailable> { - let Some(cache) = state.config_prs.as_ref() else { + let Some(cache) = state.config_prs.get() else { return Err(StatusUnavailable( "no forge is configured on this host".to_owned(), )); @@ -2862,12 +2863,11 @@ const PUBLIC_URL_ENV: &str = "SWARM_CONTROLLER_PUBLIC_URL"; /// daemon keeps running — and the failure is silent at both ends, because the /// forge has nothing to report about a call that never arrived. /// -/// Silently does nothing when any of the three preconditions is missing — -/// each is a legitimate deployment shape (no forge here, no state directory -/// to hold a secret, no public vhost), and each is already logged where it -/// is discovered. -fn register_swarm_webhooks(forge: Option>, secret: Option>) { - let Some(forge) = forge else { return }; +/// Runs once there is a forge client, and silently does nothing when either +/// other precondition is missing — each is a legitimate deployment shape (no +/// state directory to hold a secret, no public vhost), and each is already +/// logged where it is discovered. +fn register_swarm_webhooks(forge: Arc, secret: Option>) { let Some(secret) = secret else { return }; let Ok(public_url) = std::env::var(PUBLIC_URL_ENV) else { tracing::info!( @@ -2953,18 +2953,32 @@ async fn connect_status_reader() -> Result>> { } } -/// Clones `forge_client` for [`AppState::forge`] before handing the -/// original off to [`register_swarm_webhooks`], which takes it by value — -/// split out of `main` purely to keep that function under clippy's -/// line-count lint (same rationale as [`connect_status_reader`]'s own doc -/// comment), no behavior change. -fn keep_forge_for_state( - forge_client: Option>, +/// Fill the forge slots and start everything that runs off the forge client — +/// its periodic passes, the config-PR poll and the webhook registration — now +/// if the token is in the store, otherwise when `forge::enable`'s background +/// re-read finds it. Nothing to start when no forge URL is configured. Lifted +/// out of `main` for `clippy::too_many_lines`. +async fn start_forge( + forge_client: Late>, + config_prs: Late>, + jobq: Arc>>, + hives: Vec, webhook_secret: Option>, -) -> Option> { - let state_forge = forge_client.clone(); - register_swarm_webhooks(forge_client, webhook_secret); - state_forge +) { + let url = match forge::url_from_env() { + Ok(Some(url)) => url, + Ok(None) => return, + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "forge misconfigured; repo provisioning is off"); + return; + } + }; + forge::enable(url, forge_client, move |client| { + spawn_forge_workers(&jobq, Arc::clone(&client), &hives); + config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client))); + register_swarm_webhooks(client, webhook_secret); + }) + .await; } #[tokio::main] @@ -3019,18 +3033,13 @@ async fn main() -> Result<()> { }; // Same shape again: a controller with no forge access still serves // everything else, and the forge-shaped node kinds give an honest - // per-job failure rather than this fn refusing to start. A token missing - // from the store lands here too, after `from_env`'s retries. - let forge_client = match forge::Client::from_env().await { - Ok(client) => client.map(Arc::new), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "no forge access; repo provisioning is off"); - None - } - }; + // per-job failure rather than this fn refusing to start. Both slots stay + // empty until `start_forge` below fills them. + let forge_client: Late> = Arc::default(); + let config_prs: Late> = Arc::default(); let deps = WorkerDeps { auth: auth.clone(), - forge: forge_client.clone(), + forge: Arc::clone(&forge_client), queue: status.as_ref().map(|s| s.queue_client()), wanted: wanted_writer(status.as_ref()), matrix_homeserver: configured_matrix_homeserver(), @@ -3094,9 +3103,14 @@ async fn main() -> Result<()> { ); } - spawn_forge_workers(&jobq, forge_client.clone(), &hives); - let config_prs = forge_client.clone().map(config_pr::spawn); - let state_forge = keep_forge_for_state(forge_client, webhook_secret.clone()); + start_forge( + Arc::clone(&forge_client), + Arc::clone(&config_prs), + Arc::clone(&jobq), + hives.clone(), + webhook_secret.clone(), + ) + .await; spawn_agent_renewal(&jobq, wanted_writer(status.as_ref()), &hives); spawn_per_agent_upkeep(status.as_ref(), &hives); @@ -3123,7 +3137,7 @@ async fn main() -> Result<()> { config_prs, swarm_name: load_swarm_name().map(Arc::from), auth, - forge: state_forge, + forge: forge_client, create_gate: Arc::default(), accounts: linked_accounts::Accounts::Live, }; @@ -3295,13 +3309,13 @@ mod tests { agent_icons: None, jobq: std::sync::Arc::clone(&sched), webhook_secret: None, - config_prs: None, + config_prs: std::sync::Arc::default(), swarm_name: None, // No bridge: these tests drive agent *creation*, which queues a // job and consults one only for a rule-breaking name. The roster // read is the verb that needs it, and it has its own test below. auth: None, - forge: None, + forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()), }; @@ -4197,7 +4211,7 @@ mod tests { let deps = WorkerDeps { auth: None, - forge: None, + forge: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: None, @@ -4250,7 +4264,7 @@ mod tests { let deps = WorkerDeps { auth: None, - forge: None, + forge: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: None, @@ -4313,7 +4327,7 @@ mod tests { let deps = WorkerDeps { auth: None, - forge: None, + forge: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: None, @@ -4364,7 +4378,7 @@ mod tests { let deps = WorkerDeps { auth: None, - forge: None, + forge: std::sync::Arc::default(), queue: None, wanted: None, matrix_homeserver: None, diff --git a/swarm-controller/src/matrix_account.rs b/swarm-controller/src/matrix_account.rs index 1817c8b1..81c603b3 100644 --- a/swarm-controller/src/matrix_account.rs +++ b/swarm-controller/src/matrix_account.rs @@ -571,10 +571,10 @@ mod tests { hive_jobq::resources::ResourceTable::new(), ))), webhook_secret: None, - config_prs: None, + config_prs: std::sync::Arc::default(), swarm_name: None, auth: None, - forge: None, + forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: Accounts::Fake(Arc::clone(store)), } diff --git a/swarm-controller/src/queue_identity.rs b/swarm-controller/src/queue_identity.rs index 56c145ed..840c71c4 100644 --- a/swarm-controller/src/queue_identity.rs +++ b/swarm-controller/src/queue_identity.rs @@ -25,7 +25,7 @@ const PREFIX: &str = "SWARM_CONTROLLER"; /// Seconds to wait after each failed fetch before the next one. About a minute /// in all, which rides out a store restarting alongside this daemon. A longer /// outage fails the start, and the unit's `Restart=` tries again from the top. -pub(crate) const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; +const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; static IDENTITY: OnceLock> = OnceLock::new(); @@ -61,9 +61,7 @@ pub async fn init() -> Result<()> { None => None, Some(coordinates) => { let delays = RETRY_DELAYS_S.map(Duration::from_secs); - Some(coordinates.with_secret( - fetch_with_retry(&delays, "the queue client secret", fetch_secret).await?, - )) + Some(coordinates.with_secret(fetch_with_retry(&delays, fetch_secret).await?)) } }; IDENTITY @@ -120,13 +118,8 @@ async fn fetch_secret() -> Result { } /// Run `attempt`, waiting each of `delays` after a failure, then once more. -/// Answers with the first success, or the last attempt's error. `what` names the -/// secret in the log line and the error. -pub(crate) async fn fetch_with_retry( - delays: &[Duration], - what: &str, - mut attempt: F, -) -> Result +/// Answers with the first success, or the last attempt's error. +async fn fetch_with_retry(delays: &[Duration], mut attempt: F) -> Result where F: FnMut() -> Fut, Fut: Future>, @@ -138,7 +131,7 @@ where tracing::warn!( error = %format!("{e:#}"), retry_in = ?delay, - "fetching {what} from the swarm secret store failed" + "fetching the queue client secret from the swarm secret store failed" ); tokio::time::sleep(*delay).await; } @@ -146,7 +139,7 @@ where } attempt().await.with_context(|| { format!( - "fetching {what} from the swarm secret store, after {} attempts", + "fetching the queue client secret from the swarm secret store, after {} attempts", delays.len() + 1 ) }) @@ -224,7 +217,7 @@ mod tests { #[tokio::test] async fn a_store_that_comes_back_within_the_window_is_waited_for() { let mut calls = 0; - let got = fetch_with_retry(&[Duration::ZERO; 3], "a secret", || { + let got = fetch_with_retry(&[Duration::ZERO; 3], || { calls += 1; let n = calls; async move { @@ -243,7 +236,7 @@ mod tests { #[tokio::test] async fn a_store_that_stays_down_fails_after_one_attempt_per_delay_plus_one() { let mut calls = 0; - let err = fetch_with_retry(&[Duration::ZERO; 3], "a secret", || { + let err = fetch_with_retry(&[Duration::ZERO; 3], || { calls += 1; async { bail!("store sealed") } }) diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 353b9c25..95c44cd5 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -364,7 +364,7 @@ pub(super) async fn post_webhook_forge( match kind { DeliveryKind::Knowledge => announce_knowledge_change(&state).await, DeliveryKind::ConfigPr => { - if let Some(cache) = state.config_prs.as_ref() { + if let Some(cache) = state.config_prs.get() { cache.apply_webhook_delivery(&body); } if let Some(merged) = crate::config_pr::merged(&body) {