dag_executor/advanced/
retry.rs

1//! Retry strategies.
2
3use rand::Rng;
4use std::time::Duration;
5
6/// How to space out retry attempts.
7#[derive(Debug, Clone, Copy)]
8pub enum Backoff {
9    /// Wait the same fixed duration between every attempt.
10    Fixed(Duration),
11    /// Linear growth: `base * attempt`.
12    Linear {
13        /// Per-attempt increment (the delay before the first retry).
14        base: Duration,
15        /// Upper bound on any single delay.
16        max: Duration,
17    },
18    /// Exponential growth: `base * factor^attempt`, capped at `max`.
19    Exponential {
20        /// Delay before the first retry.
21        base: Duration,
22        /// Growth multiplier per attempt.
23        factor: f64,
24        /// Upper bound on any single delay.
25        max: Duration,
26    },
27}
28
29impl Backoff {
30    /// Compute the base delay (before jitter) preceding retry `attempt`
31    /// (1-based: `attempt == 1` is the delay before the *first* retry).
32    pub fn delay(&self, attempt: u32) -> Duration {
33        let attempt = attempt.max(1);
34        match *self {
35            Backoff::Fixed(d) => d,
36            Backoff::Linear { base, max } => base.checked_mul(attempt).unwrap_or(max).min(max),
37            Backoff::Exponential { base, factor, max } => {
38                let mult = factor.powi((attempt - 1) as i32);
39                let secs = base.as_secs_f64() * mult;
40                let capped = secs.min(max.as_secs_f64());
41                Duration::from_secs_f64(capped.max(0.0))
42            }
43        }
44    }
45}
46
47/// A complete retry policy: how many attempts, how to back off, and whether to
48/// add jitter.
49#[derive(Debug, Clone, Copy)]
50pub struct RetryPolicy {
51    /// Maximum total attempts (1 = no retries).
52    pub max_attempts: u32,
53    /// Backoff schedule between attempts.
54    pub backoff: Backoff,
55    /// Add up to ±25% randomization to each delay to avoid thundering herds.
56    pub jitter: bool,
57}
58
59impl RetryPolicy {
60    /// A policy that never retries.
61    pub fn none() -> Self {
62        RetryPolicy {
63            max_attempts: 1,
64            backoff: Backoff::Fixed(Duration::ZERO),
65            jitter: false,
66        }
67    }
68
69    /// `max_attempts` total tries with a fixed delay between them.
70    pub fn fixed(max_attempts: u32, delay: Duration) -> Self {
71        RetryPolicy {
72            max_attempts,
73            backoff: Backoff::Fixed(delay),
74            jitter: false,
75        }
76    }
77
78    /// Exponential backoff with jitter — a sensible production default.
79    pub fn exponential(max_attempts: u32, base: Duration) -> Self {
80        RetryPolicy {
81            max_attempts,
82            backoff: Backoff::Exponential {
83                base,
84                factor: 2.0,
85                max: Duration::from_secs(60),
86            },
87            jitter: true,
88        }
89    }
90
91    /// Whether another attempt is allowed after `attempts_made` have completed.
92    pub fn should_retry(&self, attempts_made: u32) -> bool {
93        attempts_made < self.max_attempts
94    }
95
96    /// The delay to wait before the retry following `attempts_made`, applying
97    /// jitter if enabled.
98    pub fn delay_for(&self, attempts_made: u32) -> Duration {
99        let base = self.backoff.delay(attempts_made);
100        if !self.jitter || base.is_zero() {
101            return base;
102        }
103        let secs = base.as_secs_f64();
104        let factor = rand::thread_rng().gen_range(0.75..=1.25);
105        Duration::from_secs_f64(secs * factor)
106    }
107}
108
109impl Default for RetryPolicy {
110    fn default() -> Self {
111        RetryPolicy::exponential(3, Duration::from_millis(100))
112    }
113}
dag_executor/advanced/retry.rs

dag_executor/advanced/
retry.rs
