prns_runtime/manifold/
reconnect.rs

1use core::time::Duration;
2
3const MILLIS_PER_SECOND: u128 = 1_000;
4const NANOS_PER_MILLI: u32 = 1_000_000;
5
6#[derive(Clone, Copy, Debug, Eq, PartialEq)]
7pub struct ReconnectPolicy {
8    initial_delay: Duration,
9    maximum_delay: Duration,
10    stable_reset_after: Duration,
11}
12
13#[derive(Clone, Copy, Debug, Eq, PartialEq)]
14pub enum ReconnectPolicyError {
15    ZeroInitialDelay,
16    MaximumBelowInitial,
17    ZeroStableReset,
18    SubMillisecondDelay,
19    DelayOutOfRange,
20}
21
22impl ReconnectPolicy {
23    pub const STANDARD: Self = Self {
24        initial_delay: Duration::from_millis(250),
25        maximum_delay: Duration::from_secs(5),
26        stable_reset_after: Duration::from_secs(30),
27    };
28
29    pub fn new(
30        initial_delay: Duration,
31        maximum_delay: Duration,
32        stable_reset_after: Duration,
33    ) -> Result<Self, ReconnectPolicyError> {
34        validate_millis(initial_delay)?;
35        validate_millis(maximum_delay)?;
36        validate_millis(stable_reset_after)?;
37        if initial_delay.is_zero() {
38            return Err(ReconnectPolicyError::ZeroInitialDelay);
39        }
40        if maximum_delay < initial_delay {
41            return Err(ReconnectPolicyError::MaximumBelowInitial);
42        }
43        if maximum_delay.as_millis() > u128::from(u64::MAX) * 2 / 3 {
44            return Err(ReconnectPolicyError::DelayOutOfRange);
45        }
46        if stable_reset_after.is_zero() {
47            return Err(ReconnectPolicyError::ZeroStableReset);
48        }
49        Ok(Self {
50            initial_delay,
51            maximum_delay,
52            stable_reset_after,
53        })
54    }
55
56    #[must_use]
57    pub const fn initial_delay(self) -> Duration {
58        self.initial_delay
59    }
60
61    #[must_use]
62    pub const fn maximum_delay(self) -> Duration {
63        self.maximum_delay
64    }
65
66    #[must_use]
67    pub const fn stable_reset_after(self) -> Duration {
68        self.stable_reset_after
69    }
70
71    #[must_use]
72    pub const fn schedule(self) -> ReconnectSchedule {
73        ReconnectSchedule {
74            policy: self,
75            failed_attempts: 0,
76        }
77    }
78}
79
80#[derive(Debug, Eq, PartialEq)]
81pub struct ReconnectSchedule {
82    policy: ReconnectPolicy,
83    failed_attempts: u32,
84}
85
86impl ReconnectSchedule {
87    #[must_use]
88    pub fn next_delay(&mut self, mut fill_entropy: impl FnMut(&mut [u8])) -> Duration {
89        let nominal_millis = self.nominal_delay().as_millis() as u64;
90        let mut entropy = [0u8; 8];
91        fill_entropy(&mut entropy);
92        let draw = u64::from_le_bytes(entropy);
93        let offset = ((u128::from(draw) * (u128::from(nominal_millis) + 1)) >> 64) as u64;
94        let jittered_millis = nominal_millis / 2 + offset;
95        self.failed_attempts = self.failed_attempts.saturating_add(1);
96        Duration::from_millis(jittered_millis)
97    }
98
99    pub fn record_connection_lifetime(&mut self, lifetime: Duration) {
100        if lifetime >= self.policy.stable_reset_after {
101            self.failed_attempts = 0;
102        }
103    }
104
105    #[must_use]
106    pub fn nominal_delay(&self) -> Duration {
107        let initial_millis = self.policy.initial_delay.as_millis() as u64;
108        let maximum_millis = self.policy.maximum_delay.as_millis() as u64;
109        let multiplier = 1u64.checked_shl(self.failed_attempts).unwrap_or(u64::MAX);
110        Duration::from_millis(
111            initial_millis
112                .saturating_mul(multiplier)
113                .min(maximum_millis),
114        )
115    }
116}
117
118fn validate_millis(duration: Duration) -> Result<(), ReconnectPolicyError> {
119    if !duration.subsec_nanos().is_multiple_of(NANOS_PER_MILLI) {
120        return Err(ReconnectPolicyError::SubMillisecondDelay);
121    }
122    let millis = u128::from(duration.as_secs())
123        .checked_mul(MILLIS_PER_SECOND)
124        .and_then(|millis| millis.checked_add(u128::from(duration.subsec_millis())))
125        .ok_or(ReconnectPolicyError::DelayOutOfRange)?;
126    if millis > u128::from(u64::MAX) {
127        return Err(ReconnectPolicyError::DelayOutOfRange);
128    }
129    Ok(())
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135
136    #[test]
137    fn standard_schedule_grows_to_the_nominal_plateau() {
138        let mut schedule = ReconnectPolicy::STANDARD.schedule();
139        let nominal = (0..8)
140            .map(|_| {
141                let delay = schedule.nominal_delay();
142                let _ = schedule.next_delay(|bytes| bytes.fill(0));
143                delay
144            })
145            .collect::<std::vec::Vec<_>>();
146        assert_eq!(
147            nominal,
148            std::vec![
149                Duration::from_millis(250),
150                Duration::from_millis(500),
151                Duration::from_secs(1),
152                Duration::from_secs(2),
153                Duration::from_secs(4),
154                Duration::from_secs(5),
155                Duration::from_secs(5),
156                Duration::from_secs(5),
157            ]
158        );
159    }
160
161    #[test]
162    fn standard_plateau_jitter_spans_two_and_a_half_to_seven_and_a_half_seconds() {
163        let mut low = ReconnectPolicy::STANDARD.schedule();
164        let mut high = ReconnectPolicy::STANDARD.schedule();
165        for _ in 0..5 {
166            let _ = low.next_delay(|bytes| bytes.fill(0));
167            let _ = high.next_delay(|bytes| bytes.fill(u8::MAX));
168        }
169        assert_eq!(
170            low.next_delay(|bytes| bytes.fill(0)),
171            Duration::from_millis(2_500)
172        );
173        assert_eq!(
174            high.next_delay(|bytes| bytes.fill(u8::MAX)),
175            Duration::from_millis(7_500)
176        );
177    }
178
179    #[test]
180    fn only_a_stable_connection_resets_the_schedule() {
181        let mut schedule = ReconnectPolicy::STANDARD.schedule();
182        let _ = schedule.next_delay(|bytes| bytes.fill(0));
183        let _ = schedule.next_delay(|bytes| bytes.fill(0));
184        schedule.record_connection_lifetime(Duration::from_secs(29));
185        assert_eq!(schedule.nominal_delay(), Duration::from_secs(1));
186        schedule.record_connection_lifetime(Duration::from_secs(30));
187        assert_eq!(schedule.nominal_delay(), Duration::from_millis(250));
188    }
189
190    #[test]
191    fn invalid_policies_are_typed() {
192        assert_eq!(
193            ReconnectPolicy::new(
194                Duration::ZERO,
195                Duration::from_secs(5),
196                Duration::from_secs(30)
197            ),
198            Err(ReconnectPolicyError::ZeroInitialDelay)
199        );
200        assert_eq!(
201            ReconnectPolicy::new(
202                Duration::from_secs(2),
203                Duration::from_secs(1),
204                Duration::from_secs(30)
205            ),
206            Err(ReconnectPolicyError::MaximumBelowInitial)
207        );
208        assert_eq!(
209            ReconnectPolicy::new(
210                Duration::from_millis(1),
211                Duration::from_secs(1),
212                Duration::ZERO
213            ),
214            Err(ReconnectPolicyError::ZeroStableReset)
215        );
216        assert_eq!(
217            ReconnectPolicy::new(
218                Duration::from_nanos(1),
219                Duration::from_secs(1),
220                Duration::from_secs(30)
221            ),
222            Err(ReconnectPolicyError::SubMillisecondDelay)
223        );
224        assert_eq!(
225            ReconnectPolicy::new(
226                Duration::from_millis(1),
227                Duration::from_millis(u64::MAX),
228                Duration::from_secs(30)
229            ),
230            Err(ReconnectPolicyError::DelayOutOfRange)
231        );
232    }
233}
prns_runtime/manifold/reconnect.rs

prns_runtime/manifold/
reconnect.rs
