diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml index 8b2a9f2f..b7cbe931 100644 --- a/.forgejo/workflows/ci.yml +++ b/.forgejo/workflows/ci.yml @@ -18,9 +18,7 @@ jobs: steps: - uses: actions/checkout@v3 - name: 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 + run: nix flake check tracker-tags: name: tracker-tag lint diff --git a/Cargo.lock b/Cargo.lock index d40b8333..2d9b9336 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2026,7 +2026,6 @@ name = "hive-sock-client" version = "0.1.0" dependencies = [ "anyhow", - "backon", "serde", "serde_json", "tokio", @@ -4855,7 +4854,6 @@ dependencies = [ "async-nats", "async-trait", "axum", - "backon", "base64", "bytes", "forgejo-api", @@ -5007,7 +5005,6 @@ 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 5330a97a..c8692d1c 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** | `scripts/ci-flake-check.sh`: nix-eval-jobs, then builds each check (treefmt, clippy, tests, module eval) | yes | +| **nix flake check** | treefmt formatting, `cargo clippy -D warnings`, `cargo test`, module evaluation | 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 CI workflow's `nix flake check` job builds every check derivation, 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 default CI workflow runs full `nix flake check` (builds derivations), which is the higher-risk path. ### Mitigation diff --git a/hive-agent/src/swarm_queue.rs b/hive-agent/src/swarm_queue.rs index 8a9da69d..c6ea3959 100644 --- a/hive-agent/src/swarm_queue.rs +++ b/hive-agent/src/swarm_queue.rs @@ -7,14 +7,13 @@ //! //! **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 a doubling backoff and then exits the process, so the harness +//! connect on [`CONNECT_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::{ @@ -35,16 +34,16 @@ const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME"; /// every reconnect behind it. const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10); -/// 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; +/// 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), +]; /// Logged at `error` once every first-connect attempt has failed, just before /// the process exits. The Grafana rule `hh-swarm-queue-connect-failed` @@ -295,7 +294,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 (see [`CONNECT_RETRIES`]), and every other +/// The first call retries the connect on [`CONNECT_BACKOFF`], 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 { @@ -304,7 +303,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)).await, + Some(agent) => retry_first_connect(|| connect_once(agent), tokio::time::sleep).await, None => Err(anyhow!("swarm_queue::init has not run")), }; connected.unwrap_or_else(|e| { @@ -329,38 +328,34 @@ async fn connect_once(agent: &AgentPath) -> anyhow::Result { Ok(client) } -/// 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 +/// 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 where C: FnMut() -> CFut, CFut: Future>, + S: FnMut(Duration) -> SFut, + SFut: Future, { + let mut waits = CONNECT_BACKOFF.iter(); let mut 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 - )) - }) + 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 presenting this agent's own token as `name`. @@ -406,7 +401,7 @@ mod tests { use tracing_subscriber::layer::SubscriberExt as _; use super::{ - AgentPath, AgentSecret, CONNECT_RETRIES, SecretSource, StoreSource, agent_path, + AgentPath, AgentSecret, CONNECT_BACKOFF, SecretSource, StoreSource, agent_path, connect_as_agent, log_connect_exhausted, retry_first_connect, store_source, }; @@ -494,55 +489,65 @@ mod tests { } /// Runs `retry_first_connect` against a connect that fails `failures` - /// 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) { + /// times and then succeeds, recording each wait instead of sleeping it. + async fn retry_with(failures: usize) -> (anyhow::Result, usize, Vec) { let mut calls = 0; - 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")) + 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")) + } } - } - }) + }, + |wait| { + waits.push(wait); + async {} + }, + ) .await; - (result, calls, started.elapsed()) + (result, calls, waits) } - #[tokio::test(start_paused = true)] + #[tokio::test] async fn a_first_connect_that_succeeds_waits_for_nothing() { - let (result, calls, waited) = retry_with(0).await; + let (result, calls, waits) = retry_with(0).await; assert_eq!(result.expect("connected"), 1); assert_eq!(calls, 1); - assert_eq!(waited, Duration::ZERO); + assert!(waits.is_empty(), "{waits:?}"); } /// 5s doubling, one wait per failed attempt, and the attempt after the /// last wait is the one that connects. - #[tokio::test(start_paused = true)] + #[tokio::test] async fn a_failed_connect_is_retried_on_a_doubling_backoff() { - let (result, calls, waited) = retry_with(3).await; + let (result, calls, waits) = retry_with(3).await; assert_eq!(result.expect("connected on the fourth attempt"), 4); assert_eq!(calls, 4); - assert_eq!(waited, Duration::from_secs(5 + 10 + 20)); + assert_eq!( + waits, + [5, 10, 20].map(Duration::from_secs), + "each wait doubles the one before" + ); } /// About five minutes of waits, then the last attempt's error rather than /// another wait. - #[tokio::test(start_paused = true)] + #[tokio::test] async fn the_retry_gives_up_with_the_last_error_after_about_five_minutes() { - let (result, calls, waited) = retry_with(usize::MAX).await; + let (result, calls, waits) = retry_with(usize::MAX).await; let e = result.expect_err("every attempt failed"); - assert_eq!(calls, CONNECT_RETRIES + 1); - assert_eq!(waited, Duration::from_secs(5 + 10 + 20 + 40 + 80 + 160)); + assert_eq!(calls, CONNECT_BACKOFF.len() + 1); + assert_eq!(waits, CONNECT_BACKOFF); + let total: Duration = waits.iter().sum(); assert!( - (Duration::from_mins(4)..=Duration::from_mins(6)).contains(&waited), - "{waited:?}" + (Duration::from_mins(4)..=Duration::from_mins(6)).contains(&total), + "{total:?}" ); assert!( format!("{e:#}").contains(&format!("attempt {calls} refused")), diff --git a/hive-sock-client/Cargo.toml b/hive-sock-client/Cargo.toml index 87944ec3..6ea5e617 100644 --- a/hive-sock-client/Cargo.toml +++ b/hive-sock-client/Cargo.toml @@ -16,13 +16,7 @@ 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 8ceb4ae0..89a0b77e 100644 --- a/hive-sock-client/src/lib.rs +++ b/hive-sock-client/src/lib.rs @@ -38,20 +38,17 @@ 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 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; +/// 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]; /// Bound on connecting. The listener is on the same host, so a connect /// still pending after this long is not going to complete. @@ -76,7 +73,7 @@ pub enum Retry { /// backoff would stack sleeps on top of it and delay the rest of the /// batch. None, - /// Back off on the `RIDE_OUT_RESTART_*` schedule (~60s total). For + /// Back off on `RIDE_OUT_RESTART_BACKOFFS_MS` (~60s total). For /// callers with no natural retry of their own, where a surfaced /// transient costs more than the wait. RideOutRestart, @@ -84,14 +81,10 @@ pub enum Retry { impl Retry { /// Sleep schedule between attempts; its length is the retry budget. - fn backoff(self) -> ExponentialBuilder { - let schedule = ExponentialBuilder::default(); + fn backoffs(self) -> &'static [u64] { match self { - 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), + Self::None => &[], + Self::RideOutRestart => RIDE_OUT_RESTART_BACKOFFS_MS, } } } @@ -261,42 +254,40 @@ async fn with_retry( retry: Retry, response_timeout: Duration, ) -> Result<(String, u32)> { - 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) + 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; } - })?; - 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, retry: Retry) -> anyhow::Error { - let waits: Vec = retry.backoff().build().collect(); - if waits.is_empty() { +fn exhausted(err: anyhow::Error, backoffs: &[u64]) -> anyhow::Error { + if backoffs.is_empty() { return err; } - let total_s = waits.iter().sum::().as_secs(); + let total_s = backoffs.iter().sum::() / 1_000; err.context(format!( "gave up after {} retries over ~{total_s}s", - waits.len() + backoffs.len() )) } @@ -404,8 +395,6 @@ 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. @@ -524,25 +513,8 @@ mod tests { /// harness's tool callers depend on to not surface a restart. #[test] fn ride_out_restart_budget_is_a_minute() { - 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:#}" - ); + let total_ms: u64 = Retry::RideOutRestart.backoffs().iter().sum(); + assert_eq!(total_ms, 60_000); + assert!(Retry::None.backoffs().is_empty()); } } diff --git a/scripts/ci-flake-check.sh b/scripts/ci-flake-check.sh deleted file mode 100755 index 109696c3..00000000 --- a/scripts/ci-flake-check.sh +++ /dev/null @@ -1,55 +0,0 @@ -#!/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 7983b21b..e4807936 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -12,6 +12,7 @@ use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; @@ -29,6 +30,10 @@ 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(); @@ -47,11 +52,7 @@ async fn main() -> Result<()> { config_org: var("CONFIG_ORG")?, deadline: BUILD_DEADLINE, }); - let secret = - swarm_queue_client::read_with_startup_retry("reading the queue client secret", || { - client_secret(&client_id) - }) - .await?; + let secret = client_secret_with_retry(&client_id).await?; let reread_id = client_id.clone(); let secret = StoredSecret::new(client_id.clone(), secret, move || { let client_id = reread_id.clone(); @@ -161,6 +162,21 @@ 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 da3732f7..6bb7977a 100644 --- a/swarm-claude-login/src/main.rs +++ b/swarm-claude-login/src/main.rs @@ -13,6 +13,7 @@ use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; @@ -30,6 +31,10 @@ 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; @@ -62,11 +67,7 @@ async fn main() -> Result<()> { std::env::temp_dir(), STEP_DEADLINE, )); - let secret = - swarm_queue_client::read_with_startup_retry("reading the queue client secret", || { - client_secret(&client_id) - }) - .await?; + let secret = client_secret_with_retry(&client_id).await?; let reread_id = client_id.clone(); let secret = StoredSecret::new(client_id.clone(), secret, move || { let client_id = reread_id.clone(); @@ -165,6 +166,21 @@ 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 bae6e814..b6e04ee8 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -10,7 +10,6 @@ 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 f5524136..44e2110f 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -21,7 +21,6 @@ //! either. use anyhow::{Context, Result}; -use backon::{ExponentialBuilder, Retryable as _}; use forgejo_api::structs::{ AddCollaboratorOption, AddCollaboratorOptionPermission, ChangeFileOperation, ChangeFileOperationOperation, ChangeFilesOptions, CommitStatusState, @@ -273,48 +272,52 @@ 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); -/// 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 +/// 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 where F: FnMut() -> Fut, Fut: std::future::Future>, { - 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) + 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); + } } /// The longest one unit of forge work, a periodic pass or a job node, may run @@ -2017,6 +2020,21 @@ 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)] @@ -2041,29 +2059,6 @@ 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 1a2d1a14..400fbc41 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -35,7 +35,6 @@ 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; @@ -3322,34 +3321,31 @@ fn register_swarm_webhooks(forge: Arc, secret: Option>) return; }; tokio::spawn(async move { - // 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 + // 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 // re-register periodically. Registration is idempotent, so a repeat // costs one API call and changes nothing. - const RETRIES: usize = 5; + const RETRY_DELAYS_S: [u64; 5] = [15, 30, 60, 120, 240]; - 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!( + for (attempt, delay) in RETRY_DELAYS_S.iter().enumerate() { + match forge.ensure_swarm_webhooks(&public_url, &secret).await { + Ok(()) => return, + Err(e) => tracing::info!( error = %format!("{e:#}"), - attempt = failed, - retry_in = ?wait, + attempt = attempt + 1, + retry_in_s = *delay, "registering swarm-wide forge webhooks failed; retrying" - ); - }) - .await; - if let Err(e) = registered { + ), + } + 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 { tracing::warn!( error = %format!("{e:#}"), - attempts = RETRIES + 1, + attempts = RETRY_DELAYS_S.len() + 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 2820e71a..752be84a 100644 --- a/swarm-controller/src/queue_identity.rs +++ b/swarm-controller/src/queue_identity.rs @@ -11,8 +11,10 @@ //! //! [`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; @@ -22,6 +24,11 @@ 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. @@ -52,19 +59,14 @@ 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 the startup retries; or a +/// a secret the store still would not give up after [`RETRY_DELAYS_S`]; 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 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)) + let delays = RETRY_DELAYS_S.map(Duration::from_secs); + Some(coordinates.with_secret(fetch_with_retry(&delays, fetch_secret).await?)) } }; IDENTITY @@ -120,6 +122,34 @@ 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; @@ -188,4 +218,38 @@ 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 e5dea8b8..ed2f85b8 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -60,7 +60,6 @@ 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 00d8979c..a064af7e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -465,9 +465,6 @@ 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 deleted file mode 100644 index 73839f1b..00000000 --- a/swarm-queue-client/src/startup_read.rs +++ /dev/null @@ -1,100 +0,0 @@ -//! 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 04d6bc71..c6cab66e 100644 --- a/swarm-review-classifier/src/main.rs +++ b/swarm-review-classifier/src/main.rs @@ -11,6 +11,7 @@ //! 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 _; @@ -33,6 +34,10 @@ 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. @@ -67,11 +72,7 @@ async fn serve() -> Result<()> { cwd: std::env::temp_dir().join("review-classifier-acp-cwd"), }, }; - let secret = - swarm_queue_client::read_with_startup_retry("reading the queue client secret", || { - client_secret(&client_id) - }) - .await?; + let secret = client_secret_with_retry(&client_id).await?; let reread_id = client_id.clone(); let secret = StoredSecret::new(client_id.clone(), secret, move || { let client_id = reread_id.clone(); @@ -155,6 +156,21 @@ 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};