100 lines
3.4 KiB
Rust
100 lines
3.4 KiB
Rust
//! 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)
|
|
);
|
|
}
|
|
}
|