vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

retry: share one backon helper for the client secret reads and convert the remaining doubling tables

Browse source
This commit is contained in:
damocles 2026-10-17 10:39:26 +02:00 • committed by mara
commit 9a3153cb08
14 changed files with 354 additions and 320 deletions
Download patch file Download diff file Expand all files Collapse all files

3
Cargo.lock generated
View file

@ -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",

143
hive-agent/src/swarm_queue.rs
View file

@ -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<async_nats::Client> {
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<T, C, CFut, S, SFut>(mut connect: C, mut sleep: S) -> anyhow::Result<T>
/// 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<T, C, CFut>(connect: C) -> anyhow::Result<T>
where
C: FnMut() -> CFut,
CFut: Future<Output = anyhow::Result<T>>,
S: FnMut(Duration) -> SFut,
SFut: Future<Output = ()>,
{
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>, usize, Vec<Duration>) {
/// 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>, 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")),

6
hive-sock-client/Cargo.toml
View file

@ -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"] }

104
hive-sock-client/src/lib.rs
View file

@ -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::<ResponseTimeout>() => 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::<ResponseTimeout>())
.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::<ResponseTimeout>() {
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<Duration> = retry.backoff().build().collect();
if waits.is_empty() {
return err;
}
let total_s = backoffs.iter().sum::<u64>() / 1_000;
let total_s = waits.iter().sum::<Duration>().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<Duration> = Retry::RideOutRestart.backoff().build().collect();
assert_eq!(waits, [2, 4, 8, 16, 30].map(Duration::from_secs));
assert_eq!(waits.iter().sum::<Duration>(), 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:#}"
);
}
}

26
swarm-builder/src/main.rs
View file

@ -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<String> {
}
}
/// 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<String> {
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<String> {
use swarm_secret_client::builder::{ClientSecret, client_secret_path};

26
swarm-claude-login/src/main.rs
View file

@ -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<login::Logins>) -> 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<String> {
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<String> {
use swarm_secret_client::builder::ClientSecret;

1
swarm-controller/Cargo.toml
View file

@ -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

113
swarm-controller/src/forge.rs
View file

@ -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<Option<url::Url>> {
.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<T, F, Fut>(slot: &OnceLock<T>, 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<T, F, Fut>(slot: &OnceLock<T>, attempt: F) -> &T
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<T>>,
{
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();

40
swarm-controller/src/main.rs
View file

@ -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<forge::Client>, secret: Option<Arc<str>>)
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"
);
}

80
swarm-controller/src/queue_identity.rs
View file

@ -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<Option<QueueConfig>> = 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<String> {
}
}
/// 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<F, Fut>(delays: &[Duration], mut attempt: F) -> Result<String>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<String>>,
{
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}");
}
}

1
swarm-queue-client/Cargo.toml
View file

@ -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

3
swarm-queue-client/src/lib.rs
View file

@ -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;

100
swarm-queue-client/src/startup_read.rs Normal file
View file

@ -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<T, E, F, Fut>(what: &str, read: F) -> Result<T, E>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, E>>,
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<Result<&'static str, String>> {
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)
);
}
}

26
swarm-review-classifier/src/main.rs
View file

@ -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<String> {
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<String> {
use swarm_secret_client::review_classifier::{ClientSecret, client_secret_path};