use crate::error::Result;
use std::time::Duration;
use tokio::time::sleep;
use tracing::{debug, warn};
#[derive(Debug, Clone)]
pub struct RetryConfig {
pub max_attempts: u32,
pub initial_backoff: Duration,
pub max_backoff: Duration,
pub backoff_multiplier: f64,
pub respect_retry_after: bool,
}
impl Default for RetryConfig {
fn default() -> Self {
Self {
max_attempts: 3,
initial_backoff: Duration::from_millis(500),
max_backoff: Duration::from_secs(60),
backoff_multiplier: 2.0,
respect_retry_after: true,
}
}
}
impl RetryConfig {
pub fn new() -> Self {
Self::default()
}
pub fn with_max_attempts(mut self, attempts: u32) -> Self {
self.max_attempts = attempts;
self
}
pub fn with_initial_backoff(mut self, duration: Duration) -> Self {
self.initial_backoff = duration;
self
}
pub fn with_max_backoff(mut self, duration: Duration) -> Self {
self.max_backoff = duration;
self
}
pub fn with_backoff_multiplier(mut self, multiplier: f64) -> Self {
self.backoff_multiplier = multiplier;
self
}
fn calculate_backoff(&self, attempt: u32, retry_after: Option<u64>) -> Duration {
if self.respect_retry_after {
if let Some(seconds) = retry_after {
return Duration::from_secs(seconds).min(self.max_backoff);
}
}
let backoff_secs =
self.initial_backoff.as_secs_f64() * self.backoff_multiplier.powi(attempt as i32);
Duration::from_secs_f64(backoff_secs.min(self.max_backoff.as_secs_f64()))
}
}
pub async fn retry_with_backoff<F, Fut, T>(config: RetryConfig, mut operation: F) -> Result<T>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<T>>,
{
let mut attempt = 0;
loop {
attempt += 1;
debug!("Attempt {}/{}", attempt, config.max_attempts);
match operation().await {
Ok(result) => {
if attempt > 1 {
debug!("Request succeeded after {} attempts", attempt);
}
return Ok(result);
}
Err(error) => {
if !error.is_retryable() {
debug!("Error is not retryable: {:?}", error);
return Err(error);
}
if attempt >= config.max_attempts {
warn!(
"Max retry attempts ({}) reached, failing",
config.max_attempts
);
return Err(error);
}
let retry_after = error.retry_after();
let backoff = config.calculate_backoff(attempt - 1, retry_after);
warn!(
"Request failed (attempt {}/{}): {:?}. Retrying in {:?}",
attempt, config.max_attempts, error, backoff
);
sleep(backoff).await;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::Error;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
#[test]
fn test_default_config() {
let config = RetryConfig::default();
assert_eq!(config.max_attempts, 3);
assert_eq!(config.initial_backoff, Duration::from_millis(500));
assert_eq!(config.max_backoff, Duration::from_secs(60));
}
#[test]
fn test_config_builder() {
let config = RetryConfig::new()
.with_max_attempts(5)
.with_initial_backoff(Duration::from_secs(1))
.with_max_backoff(Duration::from_secs(30));
assert_eq!(config.max_attempts, 5);
assert_eq!(config.initial_backoff, Duration::from_secs(1));
assert_eq!(config.max_backoff, Duration::from_secs(30));
}
#[test]
fn test_calculate_backoff() {
let config = RetryConfig::new()
.with_initial_backoff(Duration::from_secs(1))
.with_backoff_multiplier(2.0);
assert_eq!(config.calculate_backoff(0, None), Duration::from_secs(1));
assert_eq!(config.calculate_backoff(1, None), Duration::from_secs(2));
assert_eq!(config.calculate_backoff(2, None), Duration::from_secs(4));
}
#[test]
fn test_respect_retry_after() {
let config = RetryConfig::new().with_initial_backoff(Duration::from_secs(1));
let backoff = config.calculate_backoff(0, Some(10));
assert_eq!(backoff, Duration::from_secs(10));
}
#[test]
fn test_max_backoff_cap() {
let config = RetryConfig::new()
.with_initial_backoff(Duration::from_secs(1))
.with_max_backoff(Duration::from_secs(5))
.with_backoff_multiplier(10.0);
let backoff = config.calculate_backoff(10, None);
assert!(backoff <= Duration::from_secs(5));
}
#[tokio::test]
async fn test_retry_succeeds_first_attempt() {
let config = RetryConfig::new();
let call_count = Arc::new(AtomicU32::new(0));
let count = call_count.clone();
let result = retry_with_backoff(config, || {
let count = count.clone();
async move {
count.fetch_add(1, Ordering::SeqCst);
Ok::<_, Error>("success")
}
})
.await;
assert!(result.is_ok());
assert_eq!(call_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn test_retry_succeeds_after_failures() {
let config = RetryConfig::new().with_initial_backoff(Duration::from_millis(10));
let call_count = Arc::new(AtomicU32::new(0));
let count = call_count.clone();
let result = retry_with_backoff(config, || {
let count = count.clone();
async move {
let current = count.fetch_add(1, Ordering::SeqCst) + 1;
if current < 3 {
Err(Error::Server {
status: 503,
message: "Service unavailable".into(),
})
} else {
Ok::<_, Error>("success")
}
}
})
.await;
assert!(result.is_ok());
assert_eq!(call_count.load(Ordering::SeqCst), 3);
}
#[tokio::test]
async fn test_retry_fails_non_retryable() {
let config = RetryConfig::new();
let call_count = Arc::new(AtomicU32::new(0));
let count = call_count.clone();
let result = retry_with_backoff(config, || {
let count = count.clone();
async move {
count.fetch_add(1, Ordering::SeqCst);
Err::<String, _>(Error::Authentication("Bad key".into()))
}
})
.await;
assert!(result.is_err());
assert_eq!(call_count.load(Ordering::SeqCst), 1); }
#[tokio::test]
async fn test_retry_exhausts_attempts() {
let config = RetryConfig::new()
.with_max_attempts(2)
.with_initial_backoff(Duration::from_millis(10));
let call_count = Arc::new(AtomicU32::new(0));
let count = call_count.clone();
let result = retry_with_backoff(config, || {
let count = count.clone();
async move {
count.fetch_add(1, Ordering::SeqCst);
Err::<String, _>(Error::Server {
status: 500,
message: "Error".into(),
})
}
})
.await;
assert!(result.is_err());
assert_eq!(call_count.load(Ordering::SeqCst), 2);
}
}