diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml index b7cbe931..8b2a9f2f 100644 --- a/.forgejo/workflows/ci.yml +++ b/.forgejo/workflows/ci.yml @@ -18,7 +18,9 @@ jobs: steps: - uses: actions/checkout@v3 - name: check - run: nix flake check + # The job name stays "nix flake check": branch protection requires it. + # `--inputs-from .` resolves `nixpkgs` to the flake's locked input. + run: nix shell --inputs-from . nixpkgs#nix-eval-jobs nixpkgs#jq --command sh scripts/ci-flake-check.sh tracker-tags: name: tracker-tag lint diff --git a/Cargo.lock b/Cargo.lock index 2d9b9336..d40b8333 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2026,6 +2026,7 @@ name = "hive-sock-client" version = "0.1.0" dependencies = [ "anyhow", + "backon", "serde", "serde_json", "tokio", @@ -4854,6 +4855,7 @@ dependencies = [ "async-nats", "async-trait", "axum", + "backon", "base64", "bytes", "forgejo-api", @@ -5005,6 +5007,7 @@ version = "0.1.0" dependencies = [ "async-nats", "axum", + "backon", "data-encoding", "futures-util", "hive-sh4re", diff --git a/docs/scheduler/ci.md b/docs/scheduler/ci.md index c8692d1c..5330a97a 100644 --- a/docs/scheduler/ci.md +++ b/docs/scheduler/ci.md @@ -54,7 +54,7 @@ table can assert ahead of it happening. | Job | What it runs | Required? | | ----------------------------- | ------------------------------------------------------------------------------------------------------------------------- | ------------------------------- | -| **nix flake check** | treefmt formatting, `cargo clippy -D warnings`, `cargo test`, module evaluation | yes | +| **nix flake check** | `scripts/ci-flake-check.sh`: nix-eval-jobs, then builds each check (treefmt, clippy, tests, module eval) | yes | | **tracker-tag lint** | flags `#NNN` issue tags in source and comments (`scripts/check-issue-refs.sh`) | yes | | **comment-block lint** | flags contiguous comment blocks over 30 lines (`scripts/check-comment-blocks.sh`) | yes | | **doc-pointer lint** | flags `docs/…md` paths and relative markdown links that no longer resolve (`scripts/check-doc-refs.sh`) | yes | @@ -270,7 +270,7 @@ A malicious `default.nix` or build script in a PR can therefore: No forge admin token enters the container: swarm-controller holds it and publishes only the registration token, which only `swarm-ci-runner-register.service` loads, so a job step can't read it. A build process can still reach the forge over the network, and it runs as the runner's own user, so it can read `.runner` and act as this runner; it can't issue admin API calls. -Note: `nix flake check --no-build` (eval-only) reduces the attack surface but doesn't eliminate it — `builtins.fetchGit`, `builtins.fetchurl`, and import-from-derivation can reach the network and filesystem during evaluation. The default CI workflow runs full `nix flake check` (builds derivations), which is the higher-risk path. +Note: `nix flake check --no-build` (eval-only) reduces the attack surface but doesn't eliminate it — `builtins.fetchGit`, `builtins.fetchurl`, and import-from-derivation can reach the network and filesystem during evaluation. The CI workflow's `nix flake check` job builds every check derivation, which is the higher-risk path. ### Mitigation diff --git a/hive-agent/src/swarm_queue.rs b/hive-agent/src/swarm_queue.rs index c6ea3959..8a9da69d 100644 --- a/hive-agent/src/swarm_queue.rs +++ b/hive-agent/src/swarm_queue.rs @@ -7,13 +7,14 @@ //! //! **The connection is required.** [`init`] fails when the queue's address or //! the store's coordinates are missing, and [`client`] retries the first -//! connect on [`CONNECT_BACKOFF`] and then exits the process, so the harness +//! connect on a doubling backoff and then exits the process, so the harness //! unit fails and systemd restarts it. use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; use anyhow::{Context as _, anyhow}; +use backon::{ExponentialBuilder, Retryable as _}; use tokio::sync::OnceCell; use swarm_secret_client::{ @@ -34,16 +35,16 @@ const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME"; /// every reconnect behind it. const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10); -/// The waits between first-connect attempts: 5s doubling, about five minutes -/// in all. After the last attempt fails the process exits. -const CONNECT_BACKOFF: [Duration; 6] = [ - Duration::from_secs(5), - Duration::from_secs(10), - Duration::from_secs(20), - Duration::from_secs(40), - Duration::from_secs(80), - Duration::from_secs(160), -]; +/// The first wait between first-connect attempts, doubled per failure up to +/// [`CONNECT_MAX_WAIT`]. +const CONNECT_FIRST_WAIT: Duration = Duration::from_secs(5); + +/// The longest wait between first-connect attempts. +const CONNECT_MAX_WAIT: Duration = Duration::from_secs(160); + +/// Waits of 5s doubling to 160s, about five minutes in all. After the last +/// attempt fails the process exits. +const CONNECT_RETRIES: usize = 6; /// Logged at `error` once every first-connect attempt has failed, just before /// the process exits. The Grafana rule `hh-swarm-queue-connect-failed` @@ -294,7 +295,7 @@ pub fn init() -> anyhow::Result<()> { /// one client, so a second `connect` would be a second live connection for the /// same identity rather than a second credential. /// -/// The first call retries the connect on [`CONNECT_BACKOFF`], and every other +/// The first call retries the connect (see [`CONNECT_RETRIES`]), and every other /// caller waits on it. When no attempt succeeds the process exits with status /// 1, so this never returns without a client. pub async fn client() -> async_nats::Client { @@ -303,7 +304,7 @@ pub async fn client() -> async_nats::Client { async fn connect_or_exit() -> async_nats::Client { let connected = match AGENT.get() { - Some(agent) => retry_first_connect(|| connect_once(agent), tokio::time::sleep).await, + Some(agent) => retry_first_connect(|| connect_once(agent)).await, None => Err(anyhow!("swarm_queue::init has not run")), }; connected.unwrap_or_else(|e| { @@ -328,34 +329,38 @@ async fn connect_once(agent: &AgentPath) -> anyhow::Result { Ok(client) } -/// Run `connect` until it succeeds, sleeping each wait in [`CONNECT_BACKOFF`] -/// between attempts, and return the last error once every attempt has failed. -async fn retry_first_connect(mut connect: C, mut sleep: S) -> anyhow::Result +/// Run `connect` until it succeeds, waiting between attempts as +/// [`CONNECT_RETRIES`] says, and return the last error once every attempt has +/// failed. +async fn retry_first_connect(connect: C) -> anyhow::Result where C: FnMut() -> CFut, CFut: Future>, - S: FnMut(Duration) -> SFut, - SFut: Future, { - let mut waits = CONNECT_BACKOFF.iter(); let mut attempt = 1; - loop { - let e = match connect().await { - Ok(client) => return Ok(client), - Err(e) => e, - }; - let Some(&wait) = waits.next() else { - return Err(e.context(format!("{attempt} connection attempts failed"))); - }; - tracing::warn!( - error = format!("{e:#}"), - attempt, - retry_in = ?wait, - "connecting to the swarm queue failed" - ); - sleep(wait).await; - attempt += 1; - } + connect + .retry( + ExponentialBuilder::default() + .with_min_delay(CONNECT_FIRST_WAIT) + .with_max_delay(CONNECT_MAX_WAIT) + .with_max_times(CONNECT_RETRIES), + ) + .notify(|e, wait| { + tracing::warn!( + error = format!("{e:#}"), + attempt, + retry_in = ?wait, + "connecting to the swarm queue failed" + ); + attempt += 1; + }) + .await + .map_err(|e| { + e.context(format!( + "{} connection attempts failed", + CONNECT_RETRIES + 1 + )) + }) } /// Connect presenting this agent's own token as `name`. @@ -401,7 +406,7 @@ mod tests { use tracing_subscriber::layer::SubscriberExt as _; use super::{ - AgentPath, AgentSecret, CONNECT_BACKOFF, SecretSource, StoreSource, agent_path, + AgentPath, AgentSecret, CONNECT_RETRIES, SecretSource, StoreSource, agent_path, connect_as_agent, log_connect_exhausted, retry_first_connect, store_source, }; @@ -489,65 +494,55 @@ mod tests { } /// Runs `retry_first_connect` against a connect that fails `failures` - /// times and then succeeds, recording each wait instead of sleeping it. - async fn retry_with(failures: usize) -> (anyhow::Result, usize, Vec) { + /// times and then succeeds, on paused time so each wait costs nothing and + /// the elapsed time is the backoff itself. + async fn retry_with(failures: usize) -> (anyhow::Result, usize, Duration) { let mut calls = 0; - let mut waits = Vec::new(); - let result = retry_first_connect( - || { - calls += 1; - let n = calls; - async move { - if n > failures { - Ok(n) - } else { - Err(anyhow!("attempt {n} refused")) - } + let started = tokio::time::Instant::now(); + let result = retry_first_connect(|| { + calls += 1; + let n = calls; + async move { + if n > failures { + Ok(n) + } else { + Err(anyhow!("attempt {n} refused")) } - }, - |wait| { - waits.push(wait); - async {} - }, - ) + } + }) .await; - (result, calls, waits) + (result, calls, started.elapsed()) } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_first_connect_that_succeeds_waits_for_nothing() { - let (result, calls, waits) = retry_with(0).await; + let (result, calls, waited) = retry_with(0).await; assert_eq!(result.expect("connected"), 1); assert_eq!(calls, 1); - assert!(waits.is_empty(), "{waits:?}"); + assert_eq!(waited, Duration::ZERO); } /// 5s doubling, one wait per failed attempt, and the attempt after the /// last wait is the one that connects. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn a_failed_connect_is_retried_on_a_doubling_backoff() { - let (result, calls, waits) = retry_with(3).await; + let (result, calls, waited) = retry_with(3).await; assert_eq!(result.expect("connected on the fourth attempt"), 4); assert_eq!(calls, 4); - assert_eq!( - waits, - [5, 10, 20].map(Duration::from_secs), - "each wait doubles the one before" - ); + assert_eq!(waited, Duration::from_secs(5 + 10 + 20)); } /// About five minutes of waits, then the last attempt's error rather than /// another wait. - #[tokio::test] + #[tokio::test(start_paused = true)] async fn the_retry_gives_up_with_the_last_error_after_about_five_minutes() { - let (result, calls, waits) = retry_with(usize::MAX).await; + let (result, calls, waited) = retry_with(usize::MAX).await; let e = result.expect_err("every attempt failed"); - assert_eq!(calls, CONNECT_BACKOFF.len() + 1); - assert_eq!(waits, CONNECT_BACKOFF); - let total: Duration = waits.iter().sum(); + assert_eq!(calls, CONNECT_RETRIES + 1); + assert_eq!(waited, Duration::from_secs(5 + 10 + 20 + 40 + 80 + 160)); assert!( - (Duration::from_mins(4)..=Duration::from_mins(6)).contains(&total), - "{total:?}" + (Duration::from_mins(4)..=Duration::from_mins(6)).contains(&waited), + "{waited:?}" ); assert!( format!("{e:#}").contains(&format!("attempt {calls} refused")), diff --git a/hive-sock-client/Cargo.toml b/hive-sock-client/Cargo.toml index 6ea5e617..87944ec3 100644 --- a/hive-sock-client/Cargo.toml +++ b/hive-sock-client/Cargo.toml @@ -16,7 +16,13 @@ workspace = true [dependencies] anyhow.workspace = true +backon.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true tracing.workspace = true + +[dev-dependencies] +# `test-util` for `#[tokio::test(start_paused = true)]`: the ride-out test +# waits out the whole schedule on paused time. +tokio = { workspace = true, features = ["test-util"] } diff --git a/hive-sock-client/src/lib.rs b/hive-sock-client/src/lib.rs index 89a0b77e..8ceb4ae0 100644 --- a/hive-sock-client/src/lib.rs +++ b/hive-sock-client/src/lib.rs @@ -38,17 +38,20 @@ use std::path::Path; use std::time::Duration; use anyhow::{Context, Result, anyhow}; +use backon::{BackoffBuilder as _, ExponentialBuilder, Retryable as _}; use serde::Serialize; use serde::de::DeserializeOwned; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; -/// Backoff schedule for [`Retry::RideOutRestart`]. Five entries → up to 5 -/// retries on top of the initial attempt; total wall-clock cap = -/// 2+4+8+16+30 = 60s. Sized to ride out a service restart (systemd usually -/// has the unix socket back inside ~5s) without the caller having to -/// handle the transient itself. -const RIDE_OUT_RESTART_BACKOFFS_MS: &[u64] = &[2_000, 4_000, 8_000, 16_000, 30_000]; +/// Backoff for [`Retry::RideOutRestart`]: waits of 2, 4, 8, 16 and 30 seconds, +/// so up to 5 retries on top of the initial attempt and 60s of wall-clock in +/// all. Sized to ride out a service restart (systemd usually has the unix +/// socket back inside ~5s) without the caller having to handle the transient +/// itself. +const RIDE_OUT_RESTART_FIRST_WAIT: Duration = Duration::from_secs(2); +const RIDE_OUT_RESTART_MAX_WAIT: Duration = Duration::from_secs(30); +const RIDE_OUT_RESTART_RETRIES: usize = 5; /// Bound on connecting. The listener is on the same host, so a connect /// still pending after this long is not going to complete. @@ -73,7 +76,7 @@ pub enum Retry { /// backoff would stack sleeps on top of it and delay the rest of the /// batch. None, - /// Back off on `RIDE_OUT_RESTART_BACKOFFS_MS` (~60s total). For + /// Back off on the `RIDE_OUT_RESTART_*` schedule (~60s total). For /// callers with no natural retry of their own, where a surfaced /// transient costs more than the wait. RideOutRestart, @@ -81,10 +84,14 @@ pub enum Retry { impl Retry { /// Sleep schedule between attempts; its length is the retry budget. - fn backoffs(self) -> &'static [u64] { + fn backoff(self) -> ExponentialBuilder { + let schedule = ExponentialBuilder::default(); match self { - Self::None => &[], - Self::RideOutRestart => RIDE_OUT_RESTART_BACKOFFS_MS, + Self::None => schedule.with_max_times(0), + Self::RideOutRestart => schedule + .with_min_delay(RIDE_OUT_RESTART_FIRST_WAIT) + .with_max_delay(RIDE_OUT_RESTART_MAX_WAIT) + .with_max_times(RIDE_OUT_RESTART_RETRIES), } } } @@ -254,40 +261,42 @@ async fn with_retry( retry: Retry, response_timeout: Duration, ) -> Result<(String, u32)> { - let backoffs = retry.backoffs(); - let mut attempt: usize = 0; - loop { - match try_once(socket, payload, mode, response_timeout).await { - Ok(line) => return Ok((line, u32::try_from(attempt).unwrap_or(u32::MAX))), - Err(e) if e.is::() => return Err(e), - Err(e) => { - let Some(&sleep_ms) = backoffs.get(attempt) else { - return Err(exhausted(e, backoffs)); - }; - tracing::warn!( - attempt = attempt + 1, - sleep_ms, - socket = %socket.display(), - error = %e, - "hive socket attempt failed; retrying" - ); - tokio::time::sleep(Duration::from_millis(sleep_ms)).await; - attempt += 1; + let mut retries: u32 = 0; + let line = (|| try_once(socket, payload, mode, response_timeout)) + .retry(retry.backoff()) + .when(|e| !e.is::()) + .notify(|e, wait| { + retries += 1; + tracing::warn!( + attempt = retries, + sleep_ms = u64::try_from(wait.as_millis()).unwrap_or(u64::MAX), + socket = %socket.display(), + error = %e, + "hive socket attempt failed; retrying" + ); + }) + .await + .map_err(|e| { + if e.is::() { + e + } else { + exhausted(e, retry) } - } - } + })?; + Ok((line, retries)) } /// Annotate the final failure with how long the retry schedule tried, so a /// surfaced error says whether it was one shot or a full ride-out. -fn exhausted(err: anyhow::Error, backoffs: &[u64]) -> anyhow::Error { - if backoffs.is_empty() { +fn exhausted(err: anyhow::Error, retry: Retry) -> anyhow::Error { + let waits: Vec = retry.backoff().build().collect(); + if waits.is_empty() { return err; } - let total_s = backoffs.iter().sum::() / 1_000; + let total_s = waits.iter().sum::().as_secs(); err.context(format!( "gave up after {} retries over ~{total_s}s", - backoffs.len() + waits.len() )) } @@ -395,6 +404,8 @@ mod tests { use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixListener; + use backon::BackoffBuilder as _; + use super::{Retry, notify, request, request_within}; /// A fresh socket path under the test's temp dir, unique per test. @@ -513,8 +524,25 @@ mod tests { /// harness's tool callers depend on to not surface a restart. #[test] fn ride_out_restart_budget_is_a_minute() { - let total_ms: u64 = Retry::RideOutRestart.backoffs().iter().sum(); - assert_eq!(total_ms, 60_000); - assert!(Retry::None.backoffs().is_empty()); + let waits: Vec = Retry::RideOutRestart.backoff().build().collect(); + assert_eq!(waits, [2, 4, 8, 16, 30].map(Duration::from_secs)); + assert_eq!(waits.iter().sum::(), Duration::from_mins(1)); + assert_eq!(Retry::None.backoff().build().count(), 0); + } + + /// A socket that never comes back is waited on for the whole schedule, + /// and the surfaced error says so. + #[tokio::test(start_paused = true)] + async fn ride_out_restart_gives_up_after_the_whole_schedule() { + let bogus = std::path::Path::new("/nonexistent/hive/agent.sock"); + let started = tokio::time::Instant::now(); + let e = notify(bogus, &(), Retry::RideOutRestart) + .await + .expect_err("connect to a non-existent socket must fail"); + assert_eq!(started.elapsed(), Duration::from_mins(1)); + assert!( + format!("{e:#}").contains("gave up after 5 retries over ~60s"), + "{e:#}" + ); } } diff --git a/scripts/ci-flake-check.sh b/scripts/ci-flake-check.sh new file mode 100755 index 00000000..109696c3 --- /dev/null +++ b/scripts/ci-flake-check.sh @@ -0,0 +1,55 @@ +#!/bin/sh +# CI's "nix flake check" job: the same coverage as `nix flake check` on +# x86_64-linux, as two processes so the evaluator has exited before any build +# starts and no single evaluator holds the whole suite's heap. +# +# 1. nix-eval-jobs evaluates every check, package, dev shell, the formatter +# and each nixosConfiguration's toplevel, one worker at a time. A worker +# past --max-memory-size is replaced after its current attribute, so the +# peak is about the cap plus the heaviest single attribute. +# 2. nix build builds the checks' derivations. The other attributes are +# evaluated only, as `nix flake check` does. +# +# nix-eval-jobs reports an evaluation error as an `error` field on that +# attribute's line, not as its exit status, so this script fails on any. +# +# Needs nix-eval-jobs and jq on PATH; ci.yml takes both from the flake's +# pinned nixpkgs. +set -eu + +system=x86_64-linux +work="$(mktemp -d)" +trap 'rm -rf "$work"' EXIT + +nix-eval-jobs \ + --workers 1 \ + --max-memory-size 4096 \ + --gc-roots-dir "$work/gcroots" \ + --flake . \ + --select "flake: let o = flake.outputs; group = s: s // { recurseForDerivations = true; }; in { + checks = group o.checks.$system; + packages = group o.packages.$system; + devShells = group o.devShells.$system; + formatter = o.formatter.$system; + nixosConfigurations = group (builtins.mapAttrs (_: c: c.config.system.build.toplevel) o.nixosConfigurations); + }" >"$work/jobs.json" + +jq -r 'select(has("error")) | "\(.attr): \(.error)"' "$work/jobs.json" >"$work/errors" +if [ -s "$work/errors" ]; then + echo "evaluation failed for $(jq -s 'map(select(has("error"))) | length' "$work/jobs.json") attribute(s):" >&2 + cat "$work/errors" >&2 + exit 1 +fi + +jq -r 'select(.attrPath[0] == "checks") | .drvPath + "^*"' "$work/jobs.json" >"$work/build" +if [ ! -s "$work/build" ]; then + echo "nix-eval-jobs returned no checks; the --select expression matched nothing" >&2 + exit 1 +fi +echo "evaluated $(jq -s length "$work/jobs.json") attributes; building $(wc -l <"$work/build") checks" + +set -- +while IFS= read -r drv; do + set -- "$@" "$drv" +done <"$work/build" +nix build --no-link --keep-going "$@" diff --git a/swarm-builder/src/main.rs b/swarm-builder/src/main.rs index e4807936..7983b21b 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -12,7 +12,6 @@ use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; @@ -30,10 +29,6 @@ const CERT_ROLE: &str = "swarm-builder"; /// The environment prefix `swarm-builder.nix` sets. const PREFIX: &str = "SWARM_BUILDER"; -/// Seconds to wait after each failed read of the queue client secret. A -/// longer outage fails the start, and the unit's `Restart=` tries again. -const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; - #[tokio::main] async fn main() -> Result<()> { hive_log::init(); @@ -52,7 +47,11 @@ async fn main() -> Result<()> { config_org: var("CONFIG_ORG")?, deadline: BUILD_DEADLINE, }); - let secret = client_secret_with_retry(&client_id).await?; + let secret = + swarm_queue_client::read_with_startup_retry("reading the queue client secret", || { + client_secret(&client_id) + }) + .await?; let reread_id = client_id.clone(); let secret = StoredSecret::new(client_id.clone(), secret, move || { let client_id = reread_id.clone(); @@ -162,21 +161,6 @@ async fn forge_token() -> Result { } } -/// The queue client secret, read from the store, retried through a store that -/// is restarting alongside this service. -async fn client_secret_with_retry(client_id: &str) -> Result { - for delay in RETRY_DELAYS_S.map(Duration::from_secs) { - match client_secret(client_id).await { - Ok(secret) => return Ok(secret), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), retry_in = ?delay, "reading the queue client secret failed"); - tokio::time::sleep(delay).await; - } - } - } - client_secret(client_id).await -} - async fn client_secret(client_id: &str) -> Result { use swarm_secret_client::builder::{ClientSecret, client_secret_path}; diff --git a/swarm-claude-login/src/main.rs b/swarm-claude-login/src/main.rs index 6bb7977a..da3732f7 100644 --- a/swarm-claude-login/src/main.rs +++ b/swarm-claude-login/src/main.rs @@ -13,7 +13,6 @@ use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; @@ -31,10 +30,6 @@ const CERT_ROLE: &str = "swarm-claude-login"; /// The environment prefix `swarm-claude-login.nix` sets. const PREFIX: &str = "SWARM_CLAUDE_LOGIN"; -/// Seconds to wait after each failed read of the queue client secret. A -/// longer outage fails the start, and the unit's `Restart=` tries again. -const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; - /// The agent's store leaf, written under this service's certificate. struct AgentLeaf; @@ -67,7 +62,11 @@ async fn main() -> Result<()> { std::env::temp_dir(), STEP_DEADLINE, )); - let secret = client_secret_with_retry(&client_id).await?; + let secret = + swarm_queue_client::read_with_startup_retry("reading the queue client secret", || { + client_secret(&client_id) + }) + .await?; let reread_id = client_id.clone(); let secret = StoredSecret::new(client_id.clone(), secret, move || { let client_id = reread_id.clone(); @@ -166,21 +165,6 @@ async fn handle(payload: &[u8], logins: &Arc) -> Reply { reply } -/// The queue client secret, read from the store, retried through a store that -/// is restarting alongside this service. -async fn client_secret_with_retry(client_id: &str) -> Result { - for delay in RETRY_DELAYS_S.map(Duration::from_secs) { - match client_secret(client_id).await { - Ok(secret) => return Ok(secret), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), retry_in = ?delay, "reading the queue client secret failed"); - tokio::time::sleep(delay).await; - } - } - } - client_secret(client_id).await -} - async fn client_secret(client_id: &str) -> Result { use swarm_secret_client::builder::ClientSecret; diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index b6e04ee8..bae6e814 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -10,6 +10,7 @@ path = "src/main.rs" [dependencies] anyhow.workspace = true +backon.workspace = true # `kv` (which pulls `jetstream`) on top of the workspace's feature set: the # queue is this daemon's *store*, not just its transport - a hive's last # status snapshot is read out of a JetStream KV bucket. Declared here rather diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 44e2110f..f5524136 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -21,6 +21,7 @@ //! either. use anyhow::{Context, Result}; +use backon::{ExponentialBuilder, Retryable as _}; use forgejo_api::structs::{ AddCollaboratorOption, AddCollaboratorOptionPermission, ChangeFileOperation, ChangeFileOperationOperation, ChangeFilesOptions, CommitStatusState, @@ -272,52 +273,48 @@ pub fn url_from_env() -> Result> { .map(Some) } +/// How long the background re-read waits before its first attempt. The caller +/// has just made one, so an immediate retry would only repeat it. +const BACKGROUND_FIRST: std::time::Duration = std::time::Duration::from_secs(30); + /// The longest the background re-read waits between attempts. const BACKGROUND_CAP: std::time::Duration = std::time::Duration::from_mins(5); -/// How long the background re-read waits before its `n`-th attempt: 30s, -/// doubling, capped at [`BACKGROUND_CAP`]. -fn background_delay(n: u32) -> std::time::Duration { - const FIRST_S: u64 = 30; - std::time::Duration::from_secs( - FIRST_S - .saturating_mul(1u64.checked_shl(n).unwrap_or(u64::MAX)) - .min(BACKGROUND_CAP.as_secs()), - ) -} - -/// Whether the `n`-th failed attempt is logged at `info` rather than `debug`: -/// once the backoff is capped, so a controller that never gets its token says -/// so once per [`BACKGROUND_CAP`] without a line per early retry. -fn failure_logs_at_info(n: u32) -> bool { - background_delay(n) >= BACKGROUND_CAP -} - -/// Fill `slot` with the first value `attempt` yields, sleeping -/// [`background_delay`] before each attempt. Never gives up. A failure is logged -/// as [`failure_logs_at_info`] says; the caller logs the first one and the -/// recovery. -async fn fill_when_ready(slot: &OnceLock, mut attempt: F) -> &T +/// Fill `slot` with the first value `attempt` yields: after +/// [`BACKGROUND_FIRST`], then at 60s, doubling, capped at [`BACKGROUND_CAP`]. +/// Never gives up. A failure is logged at `info` once the next wait is the cap, +/// so a controller that never gets its token says so once per +/// [`BACKGROUND_CAP`] without a line per early retry, and at `debug` before +/// that. The caller logs the first failure and the recovery. +async fn fill_when_ready(slot: &OnceLock, 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) if failure_logs_at_info(n) => tracing::info!( - error = %format!("{e:#}"), - "still no forge access: the controller's forge token is unavailable" - ), - Err(e) => tracing::debug!( - error = %format!("{e:#}"), - "the controller's forge token is still unavailable" - ), - } - n = n.saturating_add(1); - } + tokio::time::sleep(BACKGROUND_FIRST).await; + let value = attempt + .retry( + ExponentialBuilder::default() + .with_min_delay(BACKGROUND_FIRST * 2) + .with_max_delay(BACKGROUND_CAP) + .without_max_times(), + ) + .notify(|e, wait| { + if wait >= BACKGROUND_CAP { + tracing::info!( + error = %format!("{e:#}"), + "still no forge access: the controller's forge token is unavailable" + ); + } else { + tracing::debug!( + error = %format!("{e:#}"), + "the controller's forge token is still unavailable" + ); + } + }) + .await + .expect("a retry with no cap on its tries returns only once it succeeds"); + slot.get_or_init(|| value) } /// The longest one unit of forge work, a periodic pass or a job node, may run @@ -2020,21 +2017,6 @@ mod tests { } } - #[test] - fn a_failed_re_read_logs_at_info_only_once_the_backoff_is_capped() { - assert!((0..4).all(|n| !failure_logs_at_info(n))); - assert!(failure_logs_at_info(4)); - assert!(failure_logs_at_info(40)); - } - - #[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)] @@ -2059,6 +2041,29 @@ mod tests { assert_eq!(start.elapsed().as_secs(), 30 + 60 + 120); } + #[tokio::test(start_paused = true)] + async fn a_token_that_stays_absent_is_retried_every_five_minutes() { + let slot = OnceLock::new(); + let start = tokio::time::Instant::now(); + let mut calls = 0; + fill_when_ready(&slot, || { + calls += 1; + let n = calls; + async move { + if n < 7 { + anyhow::bail!("nothing at the controller's forge token path") + } + Ok("client") + } + }) + .await; + assert_eq!(calls, 7); + assert_eq!( + start.elapsed().as_secs(), + 30 + 60 + 120 + 240 + 300 + 300 + 300 + ); + } + #[tokio::test(start_paused = true)] async fn a_pass_that_never_answers_is_abandoned_at_the_deadline() { let start = tokio::time::Instant::now(); diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 400fbc41..1a2d1a14 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -35,6 +35,7 @@ use axum::{ extract::{Path, State}, routing::get, }; +use backon::{ExponentialBuilder, Retryable as _}; use hive_jobq_wire::GraphWire as _; use serde::{Deserialize, Serialize}; use swarm_authelia_bridge_sock::BridgeResponse; @@ -3321,31 +3322,34 @@ fn register_swarm_webhooks(forge: Arc, secret: Option>) return; }; tokio::spawn(async move { - // Seconds to wait before each retry — a bounded schedule, not a poll - // loop: it exists to outlast a forge that is slow to start, not to + // A bounded schedule, not a poll loop: 15s doubling to 240s, five + // retries. It exists to outlast a forge that is slow to start, not to // re-register periodically. Registration is idempotent, so a repeat // costs one API call and changes nothing. - const RETRY_DELAYS_S: [u64; 5] = [15, 30, 60, 120, 240]; + const RETRIES: usize = 5; - for (attempt, delay) in RETRY_DELAYS_S.iter().enumerate() { - match forge.ensure_swarm_webhooks(&public_url, &secret).await { - Ok(()) => return, - Err(e) => tracing::info!( + let mut failed = 0; + let registered = (|| forge.ensure_swarm_webhooks(&public_url, &secret)) + .retry( + ExponentialBuilder::default() + .with_min_delay(std::time::Duration::from_secs(15)) + .with_max_delay(std::time::Duration::from_mins(4)) + .with_max_times(RETRIES), + ) + .notify(|e, wait| { + failed += 1; + tracing::info!( error = %format!("{e:#}"), - attempt = attempt + 1, - retry_in_s = *delay, + attempt = failed, + retry_in = ?wait, "registering swarm-wide forge webhooks failed; retrying" - ), - } - tokio::time::sleep(std::time::Duration::from_secs(*delay)).await; - } - - // Last attempt after the final wait, so the schedule above reads as - // "delay before the next try" rather than one entry meaning two things. - if let Err(e) = forge.ensure_swarm_webhooks(&public_url, &secret).await { + ); + }) + .await; + if let Err(e) = registered { tracing::warn!( error = %format!("{e:#}"), - attempts = RETRY_DELAYS_S.len() + 1, + attempts = RETRIES + 1, "registering swarm-wide forge webhooks failed; giving up until the next start" ); } diff --git a/swarm-controller/src/queue_identity.rs b/swarm-controller/src/queue_identity.rs index 752be84a..2820e71a 100644 --- a/swarm-controller/src/queue_identity.rs +++ b/swarm-controller/src/queue_identity.rs @@ -11,10 +11,8 @@ //! //! [`init`] runs once in `main`, before anything calls [`get`]. -use std::future::Future; use std::path::PathBuf; use std::sync::OnceLock; -use std::time::Duration; use anyhow::{Context, Result, bail}; use swarm_queue_client::stored_secret::StoredSecret; @@ -24,11 +22,6 @@ use swarm_secret_client::queue::{self, ControllerCredential}; /// The environment prefix the nix module's `queueEnv` sets. 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. -const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; - static IDENTITY: OnceLock> = OnceLock::new(); /// Everything in a [`QueueConfig`] except the secret. @@ -59,14 +52,19 @@ impl Coordinates { /// /// # Errors /// A half-set environment, which is a deployment bug and not an absent queue; -/// a secret the store still would not give up after [`RETRY_DELAYS_S`]; or a +/// a secret the store still would not give up after the startup retries; or a /// second call. pub async fn init() -> Result<()> { let identity = match coordinates_from(|name| std::env::var(name).ok())? { None => None, Some(coordinates) => { - let delays = RETRY_DELAYS_S.map(Duration::from_secs); - Some(coordinates.with_secret(fetch_with_retry(&delays, fetch_secret).await?)) + let secret = swarm_queue_client::read_with_startup_retry( + "fetching the queue client secret from the swarm secret store", + fetch_secret, + ) + .await + .context("fetching the queue client secret from the swarm secret store")?; + Some(coordinates.with_secret(secret)) } }; IDENTITY @@ -122,34 +120,6 @@ 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. -async fn fetch_with_retry(delays: &[Duration], mut attempt: F) -> Result -where - F: FnMut() -> Fut, - Fut: Future>, -{ - for delay in delays { - match attempt().await { - Ok(secret) => return Ok(secret), - Err(e) => { - tracing::warn!( - error = %format!("{e:#}"), - retry_in = ?delay, - "fetching the queue client secret from the swarm secret store failed" - ); - tokio::time::sleep(*delay).await; - } - } - } - attempt().await.with_context(|| { - format!( - "fetching the queue client secret from the swarm secret store, after {} attempts", - delays.len() + 1 - ) - }) -} - #[cfg(test)] mod tests { use std::collections::HashMap; @@ -218,38 +188,4 @@ mod tests { cfg.client_secret ); } - - #[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], || { - calls += 1; - let n = calls; - async move { - if n < 3 { - bail!("store sealed") - } - Ok("s3cret".to_owned()) - } - }) - .await - .expect("the third attempt succeeds"); - assert_eq!(got, "s3cret"); - assert_eq!(calls, 3); - } - - #[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], || { - calls += 1; - async { bail!("store sealed") } - }) - .await - .expect_err("every attempt fails"); - assert_eq!(calls, 4); - let msg = format!("{err:#}"); - assert!(msg.contains("after 4 attempts"), "{msg}"); - assert!(msg.contains("store sealed"), "{msg}"); - } } diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index ed2f85b8..e5dea8b8 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -60,6 +60,7 @@ agent-permissions = ["dep:hive-sh4re", "async-nats/jetstream"] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature # above - the connect itself needs none of them. async-nats.workspace = true +backon.workspace = true # `Shared`, so every caller refused while a secret re-read runs waits on that # one read (`stored_secret`). futures-util.workspace = true diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index a064af7e..00d8979c 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -465,6 +465,9 @@ pub mod review_classifier; pub mod stored_secret; +mod startup_read; +pub use startup_read::read_with_startup_retry; + /// The builder's build subject and the request and reply it carries. pub mod builder; diff --git a/swarm-queue-client/src/startup_read.rs b/swarm-queue-client/src/startup_read.rs new file mode 100644 index 00000000..73839f1b --- /dev/null +++ b/swarm-queue-client/src/startup_read.rs @@ -0,0 +1,100 @@ +//! The first read of a client secret, retried through a store that is +//! restarting alongside the service reading it. +//! +//! Every service that logs in to the queue reads its secret once at start. +//! The store may still be coming up, so a failed read waits and tries again; +//! a store down for longer than the schedule fails the start, and the unit's +//! `Restart=` tries again. + +use std::fmt::Display; +use std::future::Future; +use std::time::Duration; + +use backon::{ExponentialBuilder, Retryable as _}; + +/// The wait after the first failed read. +const FIRST_RETRY: Duration = Duration::from_secs(1); + +/// The longest wait between two reads. +const MAX_RETRY: Duration = Duration::from_secs(30); + +/// Reads after the first one: waits of 1, 2, 4, 8, 16 and 30 seconds. +const RETRIES: usize = 6; + +/// Run `read`, waiting after each failure up to six times, and answer +/// with the first success or the last attempt's error. Each failure is logged +/// as a warning naming `what`. +/// +/// # Errors +/// The error of the final attempt, once every read has failed. +pub async fn read_with_startup_retry(what: &str, read: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future>, + E: Display, +{ + read.retry( + ExponentialBuilder::default() + .with_min_delay(FIRST_RETRY) + .with_max_delay(MAX_RETRY) + .with_max_times(RETRIES), + ) + .notify(|e: &E, wait| { + tracing::warn!(error = %format!("{e:#}"), retry_in = ?wait, "{what} failed"); + }) + .await +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use tokio::time::Instant; + + use super::*; + + /// A store that fails `failures` times and then answers. + fn read_after( + failures: usize, + calls: &AtomicUsize, + ) -> std::future::Ready> { + std::future::ready(if calls.fetch_add(1, Ordering::SeqCst) < failures { + Err("store is down".to_owned()) + } else { + Ok("secret") + }) + } + + #[tokio::test(start_paused = true)] + async fn a_store_that_answers_at_once_costs_no_wait() { + let calls = AtomicUsize::new(0); + let started = Instant::now(); + let got = read_with_startup_retry("reading", || read_after(0, &calls)).await; + assert_eq!(got, Ok("secret")); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(started.elapsed(), Duration::ZERO); + } + + #[tokio::test(start_paused = true)] + async fn a_store_that_comes_up_late_is_waited_for() { + let calls = AtomicUsize::new(0); + let started = Instant::now(); + let got = read_with_startup_retry("reading", || read_after(3, &calls)).await; + assert_eq!(got, Ok("secret")); + assert_eq!(calls.load(Ordering::SeqCst), 4); + assert_eq!(started.elapsed(), Duration::from_secs(1 + 2 + 4)); + } + + #[tokio::test(start_paused = true)] + async fn a_store_that_stays_down_fails_after_the_whole_schedule() { + let calls = AtomicUsize::new(0); + let started = Instant::now(); + let got = read_with_startup_retry("reading", || read_after(usize::MAX, &calls)).await; + assert_eq!(got, Err("store is down".to_owned())); + assert_eq!(calls.load(Ordering::SeqCst), RETRIES + 1); + assert_eq!( + started.elapsed(), + Duration::from_secs(1 + 2 + 4 + 8 + 16 + 30) + ); + } +} diff --git a/swarm-review-classifier/src/main.rs b/swarm-review-classifier/src/main.rs index c6cab66e..04d6bc71 100644 --- a/swarm-review-classifier/src/main.rs +++ b/swarm-review-classifier/src/main.rs @@ -11,7 +11,6 @@ //! once per run, both from the store under that identity. use std::path::PathBuf; -use std::time::Duration; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; @@ -34,10 +33,6 @@ const CERT_ROLE: &str = "swarm-review-classifier"; /// The environment prefix `swarm-review-classifier.nix` sets. const PREFIX: &str = "SWARM_REVIEW_CLASSIFIER"; -/// Seconds to wait after each failed read of the queue client secret. A -/// longer outage fails the start, and the unit's `Restart=` tries again. -const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; - fn main() -> Result<()> { // Before anything else: as the trampoline, this process becomes the // model child. @@ -72,7 +67,11 @@ async fn serve() -> Result<()> { cwd: std::env::temp_dir().join("review-classifier-acp-cwd"), }, }; - let secret = client_secret_with_retry(&client_id).await?; + let secret = + swarm_queue_client::read_with_startup_retry("reading the queue client secret", || { + client_secret(&client_id) + }) + .await?; let reread_id = client_id.clone(); let secret = StoredSecret::new(client_id.clone(), secret, move || { let client_id = reread_id.clone(); @@ -156,21 +155,6 @@ fn log_run(run: &TryOut, reply: &Reply) { } } -/// The queue client secret, read from the store, retried through a store that -/// is restarting alongside this service. -async fn client_secret_with_retry(client_id: &str) -> Result { - for delay in RETRY_DELAYS_S.map(Duration::from_secs) { - match client_secret(client_id).await { - Ok(secret) => return Ok(secret), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), retry_in = ?delay, "reading the queue client secret failed"); - tokio::time::sleep(delay).await; - } - } - } - client_secret(client_id).await -} - async fn client_secret(client_id: &str) -> Result { use swarm_secret_client::review_classifier::{ClientSecret, client_secret_path};