rget/
retry.rs

1//! Exponential backoff with jitter (PRD §14).
2
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::time::Duration;
5
6use crate::error::TransferError;
7
8#[derive(Debug, Clone, Copy)]
9pub struct RetryPolicy {
10    /// Total attempts per range, including the first one.
11    pub max_attempts: u32,
12    pub base: Duration,
13    pub max_delay: Duration,
14}
15
16impl Default for RetryPolicy {
17    fn default() -> Self {
18        Self {
19            max_attempts: 10,
20            base: Duration::from_millis(500),
21            max_delay: Duration::from_secs(30),
22        }
23    }
24}
25
26#[derive(Debug, PartialEq, Eq)]
27pub enum Decision {
28    /// Wait this long, then try again. `attempt` is the number of the attempt
29    /// about to be made (1-based).
30    Retry {
31        delay: Duration,
32        attempt: u32,
33    },
34    GiveUp,
35}
36
37impl RetryPolicy {
38    /// `attempts_made` counts attempts already completed and failed.
39    pub fn decide(&self, err: &TransferError, attempts_made: u32) -> Decision {
40        if !err.is_retryable() || attempts_made >= self.max_attempts {
41            return Decision::GiveUp;
42        }
43        let delay = err
44            .retry_after()
45            .map(|d| d.min(self.max_delay))
46            .unwrap_or_else(|| self.backoff(attempts_made));
47        Decision::Retry {
48            delay,
49            attempt: attempts_made + 1,
50        }
51    }
52
53    /// Full jitter: `rand(0, min(max, base * 2^n))`. Full jitter beats
54    /// equal jitter for de-synchronising a fleet of workers that all failed at
55    /// the same instant, which is exactly our situation when a network drops.
56    fn backoff(&self, attempts_made: u32) -> Duration {
57        let exp = attempts_made.min(20);
58        let ceiling = self
59            .base
60            .saturating_mul(1u32.checked_shl(exp).unwrap_or(u32::MAX))
61            .min(self.max_delay);
62        let ceil_ms = ceiling.as_millis() as u64;
63        if ceil_ms == 0 {
64            return Duration::ZERO;
65        }
66        // Keep at least half the ceiling so we do not hammer a struggling
67        // server with a run of near-zero delays.
68        let floor_ms = ceil_ms / 2;
69        Duration::from_millis(floor_ms + jitter(ceil_ms - floor_ms + 1))
70    }
71}
72
73/// xorshift64*, seeded once from the clock. We need spread, not
74/// unpredictability, so pulling in a CSPRNG would be overkill.
75fn jitter(modulo: u64) -> u64 {
76    static STATE: AtomicU64 = AtomicU64::new(0);
77    let mut x = STATE.load(Ordering::Relaxed);
78    if x == 0 {
79        x = std::time::SystemTime::now()
80            .duration_since(std::time::UNIX_EPOCH)
81            .map(|d| d.as_nanos() as u64)
82            .unwrap_or(0x9E3779B97F4A7C15)
83            | 1;
84    }
85    x ^= x >> 12;
86    x ^= x << 25;
87    x ^= x >> 27;
88    STATE.store(x, Ordering::Relaxed);
89    x.wrapping_mul(0x2545F4914F6CDD1D) % modulo.max(1)
90}
91
92#[cfg(test)]
93mod tests {
94    use super::*;
95
96    #[test]
97    fn gives_up_on_permanent_errors() {
98        let p = RetryPolicy::default();
99        let err = TransferError::Status {
100            status: 404,
101            retry_after: None,
102        };
103        assert_eq!(p.decide(&err, 0), Decision::GiveUp);
104    }
105
106    #[test]
107    fn gives_up_after_max_attempts() {
108        let p = RetryPolicy {
109            max_attempts: 3,
110            ..Default::default()
111        };
112        let err = TransferError::Network("reset".into());
113        assert!(matches!(p.decide(&err, 2), Decision::Retry { .. }));
114        assert_eq!(p.decide(&err, 3), Decision::GiveUp);
115    }
116
117    #[test]
118    fn delay_grows_and_is_capped() {
119        let p = RetryPolicy {
120            max_attempts: 30,
121            base: Duration::from_millis(100),
122            max_delay: Duration::from_secs(5),
123        };
124        let err = TransferError::Network("reset".into());
125        let mut prev = Duration::ZERO;
126        for n in 0..4 {
127            let Decision::Retry { delay, attempt } = p.decide(&err, n) else {
128                panic!("expected retry");
129            };
130            assert_eq!(attempt, n + 1);
131            // Floor is half the ceiling, so growth is monotonic despite jitter.
132            assert!(delay >= prev, "delay {delay:?} < prev {prev:?}");
133            prev = delay;
134        }
135        for n in 10..20 {
136            let Decision::Retry { delay, .. } = p.decide(&err, n) else {
137                panic!("expected retry");
138            };
139            assert!(delay <= p.max_delay);
140        }
141    }
142
143    #[test]
144    fn honours_retry_after() {
145        let p = RetryPolicy::default();
146        let err = TransferError::Status {
147            status: 429,
148            retry_after: Some(Duration::from_secs(7)),
149        };
150        assert_eq!(
151            p.decide(&err, 0),
152            Decision::Retry {
153                delay: Duration::from_secs(7),
154                attempt: 1
155            }
156        );
157    }
158
159    #[test]
160    fn clamps_absurd_retry_after() {
161        let p = RetryPolicy::default();
162        let err = TransferError::Status {
163            status: 503,
164            retry_after: Some(Duration::from_secs(86_400)),
165        };
166        let Decision::Retry { delay, .. } = p.decide(&err, 0) else {
167            panic!("expected retry");
168        };
169        assert_eq!(delay, p.max_delay);
170    }
171
172    #[test]
173    fn jitter_spreads() {
174        let values: Vec<u64> = (0..50).map(|_| jitter(1000)).collect();
175        let distinct: std::collections::HashSet<_> = values.iter().collect();
176        assert!(distinct.len() > 20, "jitter is not spreading: {distinct:?}");
177        assert!(values.iter().all(|v| *v < 1000));
178    }
179}
rget/retry.rs

rget/
retry.rs
