use super::*;
use std::fmt;
use std::ops::ControlFlow;
use std::time::Duration;
pub fn retry_sync<T, E, F>(rlog: RetryLog<'_>, policy: &RetryPolicy, op: F) -> Result<T, E>
where
E: fmt::Display,
F: FnMut(u32) -> Result<T, ControlFlow<E, E>>,
{
retry_sync_deadline(rlog, policy, None, op)
}
pub fn retry_sync_deadline<T, E, F>(
rlog: RetryLog<'_>,
policy: &RetryPolicy,
deadline: Option<std::time::Instant>,
mut op: F,
) -> Result<T, E>
where
E: fmt::Display,
F: FnMut(u32) -> Result<T, ControlFlow<E, E>>,
{
retry_steps_sync(rlog, policy.max_attempts, deadline, |attempt| {
controlflow_to_step(policy, attempt, op(attempt))
})
}
fn controlflow_to_step<T, E: fmt::Display>(
policy: &RetryPolicy,
attempt: u32,
result: Result<T, ControlFlow<E, E>>,
) -> RetryStep<T, E> {
match result {
Ok(v) => RetryStep::Done(v),
Err(ControlFlow::Break(e)) => RetryStep::Fail(e),
Err(ControlFlow::Continue(e)) => {
let cause = e.to_string();
RetryStep::Retry {
error: e,
delay: policy.delay_for(attempt + 1),
cause,
}
}
}
}
pub async fn retry_async<T, E, F, Fut>(
rlog: RetryLog<'_>,
policy: &RetryPolicy,
op: F,
) -> Result<T, E>
where
E: fmt::Display,
F: FnMut(u32) -> Fut,
Fut: std::future::Future<Output = Result<T, ControlFlow<E, E>>>,
{
retry_async_deadline(rlog, policy, None, op).await
}
pub async fn retry_async_deadline<T, E, F, Fut>(
rlog: RetryLog<'_>,
policy: &RetryPolicy,
deadline: Option<std::time::Instant>,
mut op: F,
) -> Result<T, E>
where
E: fmt::Display,
F: FnMut(u32) -> Fut,
Fut: std::future::Future<Output = Result<T, ControlFlow<E, E>>>,
{
retry_steps_async(rlog, policy.max_attempts, deadline, |attempt| {
let fut = op(attempt);
async move { controlflow_to_step(policy, attempt, fut.await) }
})
.await
}
pub enum RetryStep<T, E> {
Done(T),
DoneQuiet(T),
Fail(E),
Retry {
error: E,
delay: Duration,
cause: String,
},
}
pub fn retry_steps_sync<T, E, F>(
rlog: RetryLog<'_>,
max_attempts: u32,
deadline: Option<std::time::Instant>,
mut op: F,
) -> Result<T, E>
where
F: FnMut(u32) -> RetryStep<T, E>,
{
let max = max_attempts.max(1);
let mut attempt: u32 = 1;
loop {
match op(attempt) {
RetryStep::Done(v) => {
if attempt > 1 {
rlog.note_succeeded(attempt);
}
return Ok(v);
}
RetryStep::DoneQuiet(v) => return Ok(v),
RetryStep::Fail(e) => return Err(e),
RetryStep::Retry {
error,
delay,
cause,
} => {
if attempt >= max || deadline_exhausted(deadline, delay) {
rlog.warn_giving_up(attempt);
return Err(error);
}
rlog.warn_retry(attempt, max, &cause, delay);
sleep_backoff_blocking(delay);
}
}
attempt += 1;
}
}
pub async fn retry_steps_async<T, E, F, Fut>(
rlog: RetryLog<'_>,
max_attempts: u32,
deadline: Option<std::time::Instant>,
mut op: F,
) -> Result<T, E>
where
F: FnMut(u32) -> Fut,
Fut: std::future::Future<Output = RetryStep<T, E>>,
{
let max = max_attempts.max(1);
let mut attempt: u32 = 1;
loop {
match op(attempt).await {
RetryStep::Done(v) => {
if attempt > 1 {
rlog.note_succeeded(attempt);
}
return Ok(v);
}
RetryStep::DoneQuiet(v) => return Ok(v),
RetryStep::Fail(e) => return Err(e),
RetryStep::Retry {
error,
delay,
cause,
} => {
if attempt >= max || deadline_exhausted(deadline, delay) {
rlog.warn_giving_up(attempt);
return Err(error);
}
rlog.warn_retry(attempt, max, &cause, delay);
sleep_backoff_async(delay).await;
}
}
attempt += 1;
}
}
fn deadline_exhausted(deadline: Option<std::time::Instant>, delay: Duration) -> bool {
deadline.is_some_and(|d| {
std::time::Instant::now()
.checked_add(delay)
.is_none_or(|projected| projected > d)
})
}