laminar_connectors/retry/
mod.rs

1//! Retry/backoff helper shared by connectors.
2//!
3//! Centralises exponential backoff with jitter so every reconnect path
4//! behaves the same way: capped exponent (no shift overflow), capped
5//! delay, jittered to break thundering-herd on broker-wide outages.
6
7use std::time::Duration;
8
9use rand::RngExt;
10
11/// Exponential backoff schedule with jitter and a cap.
12#[derive(Debug, Clone, Copy)]
13pub struct Backoff {
14    initial: Duration,
15    max: Duration,
16    /// Multiplicative jitter range, expressed as a fraction in `[0.0, 1.0]`.
17    /// `0.25` means the actual delay is uniform on
18    /// `[delay * 0.75, delay * 1.25]`.
19    jitter: f64,
20}
21
22impl Backoff {
23    /// New backoff with the given bounds and jitter fraction.
24    #[must_use]
25    pub const fn new(initial: Duration, max: Duration, jitter: f64) -> Self {
26        Self {
27            initial,
28            max,
29            jitter,
30        }
31    }
32
33    /// Default broker-reconnect schedule: 1s → 30s, ±25 % jitter.
34    /// Jitter prevents reconnect storms during simultaneous endpoint restarts.
35    #[must_use]
36    pub const fn broker_reconnect() -> Self {
37        Self::new(Duration::from_secs(1), Duration::from_secs(30), 0.25)
38    }
39
40    /// Compute the delay for `attempt` (0-indexed). Caps the exponent at
41    /// 30 to prevent shift overflow with adversarial inputs, then caps
42    /// the resulting duration at `self.max`, then applies jitter.
43    #[must_use]
44    pub fn delay(&self, attempt: u32) -> Duration {
45        // 2^30 seconds is ~34 years; capping the exponent at 30 keeps
46        // the multiplication in u64 range for any sane `initial`.
47        let shift = attempt.min(30);
48        let factor = 1u64 << shift;
49        let raw_nanos = self
50            .initial
51            .as_nanos()
52            .saturating_mul(u128::from(factor))
53            .min(u128::from(u64::MAX));
54        #[allow(clippy::cast_possible_truncation)]
55        let raw = Duration::from_nanos(raw_nanos as u64).min(self.max);
56
57        if self.jitter <= 0.0 {
58            return raw;
59        }
60        let mut rng = rand::rng();
61        let frac: f64 = rng.random_range(-self.jitter..=self.jitter);
62        #[allow(
63            clippy::cast_precision_loss,
64            clippy::cast_sign_loss,
65            clippy::cast_possible_truncation
66        )]
67        let jittered_nanos = (raw.as_nanos() as f64 * (1.0 + frac)).max(0.0) as u64;
68        Duration::from_nanos(jittered_nanos).min(self.max)
69    }
70}
71
72#[cfg(test)]
73mod tests;
laminar_connectors/retry/mod.rs

laminar_connectors/retry/
mod.rs
