nautilus_network/
backoff.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Exponential backoff with optional jitter for socket reconnection delays.
17//!
18//! Successive delays grow by a configurable factor up to a maximum. Random jitter reduces
19//! synchronized reconnect storms. Immediate-first mode allows the first reconnect attempt to run
20//! without delay. A rolling-window throttle enforces a minimum attempt spacing once reconnects flap,
21//! bounding the attempt rate when the stability reset would otherwise restore an immediate reconnect.
22
23use std::{collections::VecDeque, pin::pin, sync::atomic::AtomicU8, time::Duration};
24
25use nautilus_core::correctness::{check_in_range_inclusive_f64, check_predicate_true};
26use rand::RngExt;
27
28use crate::{dst, mode::ConnectionMode};
29
30// Keep public reconnect_max_attempts docs synchronized with this value
31pub(crate) const RECONNECT_STABILITY_THRESHOLD: Duration = Duration::from_secs(10);
32
33/// The minimum spacing between reconnect attempts once reconnects flap within a short window.
34///
35/// One second keeps a single client under the strictest new-connection rate among supported
36/// venues (Binance permits 300 connections per 5 minutes per IP; OKX permits 3 per second).
37pub(crate) const RECONNECT_MIN_DELAY: Duration = Duration::from_secs(1);
38
39/// The number of reconnect attempts inside [`RECONNECT_MIN_DELAY_WINDOW`] that trips the minimum
40/// delay.
41pub(crate) const RECONNECT_MIN_DELAY_ATTEMPTS: usize = 3;
42
43/// The rolling window over recent reconnect attempts.
44///
45/// Long enough to catch a venue cycling connections just past [`RECONNECT_STABILITY_THRESHOLD`],
46/// where the stability reset would otherwise restore an immediate reconnect on every cycle.
47pub(crate) const RECONNECT_MIN_DELAY_WINDOW: Duration = Duration::from_mins(2);
48
49/// Tracks recent reconnect attempt times to enforce [`RECONNECT_MIN_DELAY`] while reconnects flap.
50///
51/// The stability reset restores immediate-first reconnection after
52/// [`RECONNECT_STABILITY_THRESHOLD`], so the backoff delay alone cannot bound the attempt rate
53/// against a venue that keeps replacement connections alive just past that threshold. Once
54/// [`RECONNECT_MIN_DELAY_ATTEMPTS`] attempts occur inside [`RECONNECT_MIN_DELAY_WINDOW`], every
55/// further attempt waits at least [`RECONNECT_MIN_DELAY`] until fewer than three remain.
56#[derive(Debug, Default)]
57pub(crate) struct ReconnectThrottle {
58    recent_attempts: VecDeque<dst::time::Instant>,
59}
60
61impl ReconnectThrottle {
62    /// Returns the delay to wait before the next reconnect attempt: `backoff_delay` raised to
63    /// [`RECONNECT_MIN_DELAY`] while the rolling window holds at least
64    /// [`RECONNECT_MIN_DELAY_ATTEMPTS`] attempts.
65    pub(crate) fn gated_delay(&mut self, backoff_delay: Duration) -> Duration {
66        self.prune_expired();
67
68        if self.recent_attempts.len() >= RECONNECT_MIN_DELAY_ATTEMPTS {
69            backoff_delay.max(RECONNECT_MIN_DELAY)
70        } else {
71            backoff_delay
72        }
73    }
74
75    /// Records a reconnect attempt at the current time.
76    pub(crate) fn record_attempt(&mut self) {
77        self.prune_expired();
78        self.recent_attempts.push_back(dst::time::Instant::now());
79    }
80
81    fn prune_expired(&mut self) {
82        while self
83            .recent_attempts
84            .front()
85            .is_some_and(|oldest| oldest.elapsed() > RECONNECT_MIN_DELAY_WINDOW)
86        {
87            self.recent_attempts.pop_front();
88        }
89    }
90}
91
92#[derive(Clone, Debug)]
93pub struct ExponentialBackoff {
94    delay_initial: Duration,
95    delay_max: Duration,
96    delay_current: Duration,
97    factor: f64,
98    jitter_ms: u64,
99    immediate_reconnect: bool,
100    immediate_reconnect_original: bool,
101}
102
103/// An exponential backoff mechanism with optional jitter and immediate-first behavior.
104///
105/// The backoff starts at an initial delay, multiplies that delay by a factor after each call, and
106/// caps it at the configured maximum. Each result includes bounded random jitter. When
107/// `immediate_first` is `true`, the first call to [`Self::next_duration`] returns zero. Calling
108/// [`Self::reset`] restores both the initial delay and the original immediate-first setting.
109impl ExponentialBackoff {
110    /// Creates a new [`ExponentialBackoff]` instance.
111    ///
112    /// # Errors
113    ///
114    /// Returns an error if:
115    /// - `delay_initial` is zero.
116    /// - `delay_max` is less than `delay_initial`.
117    /// - `delay_max` exceeds `Duration::from_nanos(u64::MAX)` (≈584 years).
118    /// - `factor` is not in the range [1.0, 100.0] (to prevent reconnect spam).
119    pub fn new(
120        delay_initial: Duration,
121        delay_max: Duration,
122        factor: f64,
123        jitter_ms: u64,
124        immediate_first: bool,
125    ) -> anyhow::Result<Self> {
126        check_predicate_true(!delay_initial.is_zero(), "delay_initial must be non-zero")?;
127        check_predicate_true(
128            delay_max >= delay_initial,
129            "delay_max must be >= delay_initial",
130        )?;
131        check_predicate_true(
132            delay_max.as_nanos() <= u128::from(u64::MAX),
133            "delay_max exceeds maximum representable duration (≈584 years)",
134        )?;
135        check_in_range_inclusive_f64(factor, 1.0, 100.0, "factor")?;
136
137        Ok(Self {
138            delay_initial,
139            delay_max,
140            delay_current: delay_initial,
141            factor,
142            jitter_ms,
143            immediate_reconnect: immediate_first,
144            immediate_reconnect_original: immediate_first,
145        })
146    }
147
148    /// Returns the next backoff delay with jitter and updates the internal state.
149    ///
150    /// If the `immediate_first` flag is set and this is the first call (i.e. the current
151    /// delay equals the initial delay), it returns `Duration::ZERO` to trigger an immediate
152    /// reconnect and disables the immediate behavior for subsequent calls.
153    ///
154    /// Near the cap the jittered base is lowered to `delay_max - jitter` so
155    /// the spread survives saturation; the result is clamped into
156    /// `[min(delay_initial, delay_max), delay_max]`.
157    pub fn next_duration(&mut self) -> Duration {
158        if self.immediate_reconnect && self.delay_current == self.delay_initial {
159            self.immediate_reconnect = false;
160            return Duration::ZERO;
161        }
162
163        // Generate random jitter
164        let jitter = rand::rng().random_range(0..=self.jitter_ms); // dst-ok: transport-layer reconnect jitter, out of DST scope
165
166        // Cap the jittered base below delay_max so the spread survives saturation at the cap
167        let base = std::cmp::min(
168            self.delay_current,
169            self.delay_max
170                .saturating_sub(Duration::from_millis(self.jitter_ms)),
171        );
172        let delay_with_jitter = base + Duration::from_millis(jitter);
173
174        // The floor keeps a jitter range wider than delay_max from producing a zero delay
175        let floor = std::cmp::min(self.delay_initial, self.delay_max);
176        let clamped_delay = delay_with_jitter.clamp(floor, self.delay_max);
177
178        // The constructor guarantees both values fit in u64 nanoseconds. Float-to-integer casts
179        // saturate, so the final min preserves the configured cap even if multiplication overflows.
180        let current_nanos = self.delay_current.as_nanos() as u64;
181        let max_nanos = self.delay_max.as_nanos() as u64;
182        let next_nanos = (current_nanos as f64 * self.factor) as u64;
183        self.delay_current = Duration::from_nanos(next_nanos.min(max_nanos));
184
185        clamped_delay
186    }
187
188    /// Resets the backoff to its initial state.
189    pub const fn reset(&mut self) {
190        self.delay_current = self.delay_initial;
191        self.immediate_reconnect = self.immediate_reconnect_original;
192    }
193
194    /// Returns the current base delay without jitter.
195    /// This represents the delay that would be used as the base for the next call to `next()`,
196    /// before any jitter is applied.
197    #[must_use]
198    pub const fn current_delay(&self) -> Duration {
199        self.delay_current
200    }
201}
202
203pub(crate) async fn wait_reconnect_delay(
204    duration: Duration,
205    connection_mode: &AtomicU8,
206    state_notify: &tokio::sync::Notify,
207) -> bool {
208    if duration.is_zero() {
209        return true;
210    }
211
212    tokio::select! {
213        biased;
214        () = dst::time::sleep(duration) => true,
215        () = async {
216            loop {
217                let mut notified = pin!(state_notify.notified());
218                notified.as_mut().enable();
219
220                if !ConnectionMode::from_atomic(connection_mode).is_reconnect() {
221                    break;
222                }
223                notified.await;
224            }
225        } => false,
226    }
227}
228
229#[cfg(test)]
230mod tests {
231    use std::time::Duration;
232
233    use rstest::rstest;
234
235    use super::*;
236
237    #[rstest]
238    fn test_no_jitter_exponential_growth() {
239        let initial = Duration::from_millis(100);
240        let max = Duration::from_millis(1600);
241        let factor = 2.0;
242        let jitter = 0;
243        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
244
245        // 1st call returns the initial delay
246        let d1 = backoff.next_duration();
247        assert_eq!(d1, Duration::from_millis(100));
248
249        // 2nd call: current becomes 200ms
250        let d2 = backoff.next_duration();
251        assert_eq!(d2, Duration::from_millis(200));
252
253        // 3rd call: current becomes 400ms
254        let d3 = backoff.next_duration();
255        assert_eq!(d3, Duration::from_millis(400));
256
257        // 4th call: current becomes 800ms
258        let d4 = backoff.next_duration();
259        assert_eq!(d4, Duration::from_millis(800));
260
261        // 5th call: current would be 1600ms (800 * 2) which is within the cap
262        let d5 = backoff.next_duration();
263        assert_eq!(d5, Duration::from_millis(1600));
264
265        // 6th call: should still be capped at 1600ms
266        let d6 = backoff.next_duration();
267        assert_eq!(d6, Duration::from_millis(1600));
268    }
269
270    #[rstest]
271    fn test_reset() {
272        let initial = Duration::from_millis(100);
273        let max = Duration::from_millis(1600);
274        let factor = 2.0;
275        let jitter = 0;
276        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
277
278        // Call next() once so that the internal state updates
279        let _ = backoff.next_duration(); // current_delay becomes 200ms
280        backoff.reset();
281        let d = backoff.next_duration();
282        // After reset, the next delay should be the initial delay (100ms)
283        assert_eq!(d, Duration::from_millis(100));
284    }
285
286    #[rstest]
287    fn test_jitter_within_bounds() {
288        let initial = Duration::from_millis(100);
289        let max = Duration::from_secs(1);
290        let factor = 2.0;
291        let jitter = 50;
292        // Run several iterations to ensure that jitter stays within bounds
293        for _ in 0..10 {
294            let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
295            // Capture the expected base delay before jitter is applied
296            let base = backoff.delay_current;
297            let delay = backoff.next_duration();
298            // The returned delay must be at least the base delay and at most base + jitter
299            let min_expected = base;
300            let max_expected = base + Duration::from_millis(jitter);
301            assert!(
302                delay >= min_expected,
303                "Delay {delay:?} is less than expected minimum {min_expected:?}"
304            );
305            assert!(
306                delay <= max_expected,
307                "Delay {delay:?} exceeds expected maximum {max_expected:?}"
308            );
309        }
310    }
311
312    #[rstest]
313    fn test_factor_less_than_two() {
314        let initial = Duration::from_millis(100);
315        let max = Duration::from_millis(200);
316        let factor = 1.5;
317        let jitter = 0;
318        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
319
320        // First call returns 100ms
321        let d1 = backoff.next_duration();
322        assert_eq!(d1, Duration::from_millis(100));
323
324        // Second call: current_delay becomes 100 * 1.5 = 150ms
325        let d2 = backoff.next_duration();
326        assert_eq!(d2, Duration::from_millis(150));
327
328        // Third call: current_delay becomes 150 * 1.5 = 225ms, but capped to 200ms
329        let d3 = backoff.next_duration();
330        assert_eq!(d3, Duration::from_millis(200));
331
332        // Fourth call: remains at the max of 200ms
333        let d4 = backoff.next_duration();
334        assert_eq!(d4, Duration::from_millis(200));
335    }
336
337    #[rstest]
338    fn test_max_delay_is_respected() {
339        let initial = Duration::from_millis(500);
340        let max = Duration::from_secs(1);
341        let factor = 3.0;
342        let jitter = 0;
343        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
344
345        // 1st call returns 500ms
346        let d1 = backoff.next_duration();
347        assert_eq!(d1, Duration::from_millis(500));
348
349        // 2nd call: would be 500 * 3 = 1500ms but is capped to 1000ms
350        let d2 = backoff.next_duration();
351        assert_eq!(d2, Duration::from_secs(1));
352
353        // Subsequent calls should continue to return the max delay
354        let d3 = backoff.next_duration();
355        assert_eq!(d3, Duration::from_secs(1));
356    }
357
358    #[rstest]
359    fn test_current_delay_getter() {
360        let initial = Duration::from_millis(100);
361        let max = Duration::from_millis(1600);
362        let factor = 2.0;
363        let jitter = 0;
364        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
365
366        assert_eq!(backoff.current_delay(), initial);
367
368        let _ = backoff.next_duration();
369        assert_eq!(backoff.current_delay(), Duration::from_millis(200));
370
371        let _ = backoff.next_duration();
372        assert_eq!(backoff.current_delay(), Duration::from_millis(400));
373
374        backoff.reset();
375        assert_eq!(backoff.current_delay(), initial);
376    }
377
378    #[rstest]
379    fn test_validation_zero_initial_delay() {
380        let result = ExponentialBackoff::new(Duration::ZERO, Duration::from_secs(1), 2.0, 0, false);
381        assert!(result.is_err());
382        assert!(
383            result
384                .unwrap_err()
385                .to_string()
386                .contains("delay_initial must be non-zero")
387        );
388    }
389
390    #[rstest]
391    fn test_validation_max_less_than_initial() {
392        let result = ExponentialBackoff::new(
393            Duration::from_secs(1),
394            Duration::from_millis(500),
395            2.0,
396            0,
397            false,
398        );
399        assert!(result.is_err());
400        assert!(
401            result
402                .unwrap_err()
403                .to_string()
404                .contains("delay_max must be >= delay_initial")
405        );
406    }
407
408    #[rstest]
409    fn test_validation_factor_too_small() {
410        let result = ExponentialBackoff::new(
411            Duration::from_millis(100),
412            Duration::from_secs(1),
413            0.5,
414            0,
415            false,
416        );
417        assert!(result.is_err());
418        assert!(result.unwrap_err().to_string().contains("factor"));
419    }
420
421    #[rstest]
422    fn test_validation_factor_too_large() {
423        let result = ExponentialBackoff::new(
424            Duration::from_millis(100),
425            Duration::from_secs(1),
426            150.0,
427            0,
428            false,
429        );
430        assert!(result.is_err());
431        assert!(result.unwrap_err().to_string().contains("factor"));
432    }
433
434    #[rstest]
435    fn test_validation_delay_max_exceeds_u64_max_nanos() {
436        // Duration::from_nanos(u64::MAX) is approximately 584 years
437        // Try to create a backoff with delay_max exceeding this
438        let max_valid = Duration::from_nanos(u64::MAX);
439        let too_large = max_valid + Duration::from_nanos(1);
440
441        let result = ExponentialBackoff::new(Duration::from_millis(100), too_large, 2.0, 0, false);
442        assert!(result.is_err());
443        assert!(
444            result
445                .unwrap_err()
446                .to_string()
447                .contains("delay_max exceeds maximum representable duration")
448        );
449    }
450
451    #[rstest]
452    fn test_immediate_first() {
453        let initial = Duration::from_millis(100);
454        let max = Duration::from_millis(1600);
455        let factor = 2.0;
456        let jitter = 0;
457        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, true).unwrap();
458
459        // The first call should yield an immediate (zero) delay
460        let d1 = backoff.next_duration();
461        assert_eq!(
462            d1,
463            Duration::ZERO,
464            "Expected immediate reconnect (zero delay) on first call"
465        );
466
467        // The next call should return the current delay (i.e. the base initial delay)
468        let d2 = backoff.next_duration();
469        assert_eq!(
470            d2, initial,
471            "Expected the delay to be the initial delay after immediate reconnect"
472        );
473
474        // Subsequent calls should continue with the exponential growth
475        let d3 = backoff.next_duration();
476        let expected = initial * 2; // 100ms * 2 = 200ms
477        assert_eq!(
478            d3, expected,
479            "Expected exponential growth from the initial delay"
480        );
481    }
482
483    #[rstest]
484    fn test_reset_restores_immediate_first() {
485        let initial = Duration::from_millis(100);
486        let max = Duration::from_millis(1600);
487        let factor = 2.0;
488        let jitter = 0;
489        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, true).unwrap();
490
491        // Use immediate first
492        let d1 = backoff.next_duration();
493        assert_eq!(d1, Duration::ZERO);
494
495        // Now immediate_first should be disabled
496        let d2 = backoff.next_duration();
497        assert_eq!(d2, initial);
498
499        // Reset should restore immediate_first
500        backoff.reset();
501        let d3 = backoff.next_duration();
502        assert_eq!(
503            d3,
504            Duration::ZERO,
505            "Reset should restore immediate_first behavior"
506        );
507    }
508
509    #[rstest]
510    fn test_jitter_never_exceeds_max_delay() {
511        let initial = Duration::from_millis(100);
512        let max = Duration::from_secs(1);
513        let factor = 2.0;
514        let jitter = 500;
515
516        let mut backoff = ExponentialBackoff::new(initial, max, factor, jitter, false).unwrap();
517
518        // Run backoff until it reaches the cap
519        while backoff.current_delay() < max {
520            backoff.next_duration();
521        }
522
523        // Now that we're at the cap, verify jitter doesn't push us over delay_max
524        for _ in 0..100 {
525            let delay = backoff.next_duration();
526            assert!(
527                delay <= max,
528                "Delay with jitter {delay:?} exceeded max {max:?}"
529            );
530        }
531    }
532
533    #[rstest]
534    fn test_jitter_spreads_delays_at_cap() {
535        // Regression: clamping after adding jitter collapsed the spread to a
536        // single value once the backoff saturated, re-synchronizing clients
537        // exactly during extended outages
538        let initial = Duration::from_millis(100);
539        let max = Duration::from_secs(1);
540        let mut backoff = ExponentialBackoff::new(initial, max, 2.0, 500, false).unwrap();
541
542        while backoff.current_delay() < max {
543            backoff.next_duration();
544        }
545
546        let mut distinct = std::collections::HashSet::new();
547        for _ in 0..100 {
548            distinct.insert(backoff.next_duration());
549        }
550
551        assert!(
552            distinct.len() >= 2,
553            "Jitter must keep spreading delays once the backoff saturates at the cap"
554        );
555    }
556
557    #[rstest]
558    fn test_jitter_wider_than_max_never_returns_zero_delay() {
559        // A jitter range wider than delay_max collapses the base to zero; the
560        // floor keeps non-immediate delays positive.
561        let max = Duration::from_millis(50);
562        let mut backoff =
563            ExponentialBackoff::new(Duration::from_millis(10), max, 2.0, 100, false).unwrap();
564
565        for _ in 0..200 {
566            let delay = backoff.next_duration();
567            assert!(
568                !delay.is_zero(),
569                "Non-immediate backoff delay must be positive"
570            );
571            assert!(delay <= max, "Delay {delay:?} exceeded max {max:?}");
572        }
573    }
574
575    // Time-dependent throttle tests need an exact paused clock; under the sim build
576    // `dst::time` resolves to madsim, whose scheduler epsilon can move an attempt across
577    // the window boundary. DST-level coverage comes from the turmoil storm tests.
578    #[cfg(not(all(feature = "simulation", madsim)))]
579    #[tokio::test(flavor = "current_thread", start_paused = true)]
580    async fn test_throttle_passes_delay_through_below_attempt_threshold() {
581        let mut throttle = ReconnectThrottle::default();
582
583        for _ in 0..RECONNECT_MIN_DELAY_ATTEMPTS {
584            assert_eq!(
585                throttle.gated_delay(Duration::ZERO),
586                Duration::ZERO,
587                "Cold window must not floor an immediate reconnect"
588            );
589            throttle.record_attempt();
590        }
591    }
592
593    #[cfg(not(all(feature = "simulation", madsim)))]
594    #[tokio::test(flavor = "current_thread", start_paused = true)]
595    async fn test_throttle_floors_delay_once_threshold_trips() {
596        let mut throttle = ReconnectThrottle::default();
597
598        for _ in 0..RECONNECT_MIN_DELAY_ATTEMPTS {
599            throttle.record_attempt();
600        }
601
602        assert_eq!(
603            throttle.gated_delay(Duration::ZERO),
604            RECONNECT_MIN_DELAY,
605            "Hot window must floor an immediate reconnect"
606        );
607        assert_eq!(
608            throttle.gated_delay(Duration::from_millis(25)),
609            RECONNECT_MIN_DELAY,
610            "Hot window must raise a sub-floor backoff delay"
611        );
612        assert_eq!(
613            throttle.gated_delay(Duration::from_secs(5)),
614            Duration::from_secs(5),
615            "Hot window must not lower a backoff delay above the floor"
616        );
617    }
618
619    #[cfg(not(all(feature = "simulation", madsim)))]
620    #[tokio::test(flavor = "current_thread", start_paused = true)]
621    async fn test_throttle_lifts_floor_after_window_expires() {
622        let mut throttle = ReconnectThrottle::default();
623
624        for _ in 0..RECONNECT_MIN_DELAY_ATTEMPTS {
625            throttle.record_attempt();
626        }
627
628        assert_eq!(throttle.gated_delay(Duration::ZERO), RECONNECT_MIN_DELAY);
629
630        dst::time::sleep(RECONNECT_MIN_DELAY_WINDOW + Duration::from_secs(1)).await;
631
632        assert_eq!(
633            throttle.gated_delay(Duration::ZERO),
634            Duration::ZERO,
635            "Floor must lift once the rolling window drains"
636        );
637    }
638
639    #[cfg(not(all(feature = "simulation", madsim)))]
640    #[tokio::test(flavor = "current_thread", start_paused = true)]
641    async fn test_throttle_window_is_purely_time_based() {
642        let mut throttle = ReconnectThrottle::default();
643
644        for _ in 0..RECONNECT_MIN_DELAY_ATTEMPTS {
645            throttle.record_attempt();
646        }
647
648        // A stable connection can live between flapping attempts; the window must not
649        // treat that uptime as recovery. Only time draining the window lifts the floor.
650        dst::time::sleep(Duration::from_mins(1)).await;
651        assert_eq!(
652            throttle.gated_delay(Duration::ZERO),
653            RECONNECT_MIN_DELAY,
654            "Stable uptime inside the window must not lift the floor"
655        );
656
657        dst::time::sleep(RECONNECT_MIN_DELAY_WINDOW).await;
658        assert_eq!(
659            throttle.gated_delay(Duration::ZERO),
660            Duration::ZERO,
661            "Floor must lift once attempts drain from the window"
662        );
663    }
664
665    #[cfg(not(all(feature = "simulation", madsim)))]
666    mod throttle_proptests {
667        use proptest::prelude::*;
668        use rstest::rstest;
669
670        use super::*;
671
672        fn build_paused_runtime() -> tokio::runtime::Runtime {
673            tokio::runtime::Builder::new_current_thread()
674                .enable_time()
675                .start_paused(true)
676                .build()
677                .unwrap()
678        }
679
680        #[derive(Clone, Debug)]
681        enum ThrottleOp {
682            Attempt,
683            AdvanceMs(u64),
684            Gate(u64),
685        }
686
687        fn throttle_op_strategy() -> impl Strategy<Value = ThrottleOp> {
688            prop_oneof![
689                3 => Just(ThrottleOp::Attempt),
690                2 => (0u64..=180_000).prop_map(ThrottleOp::AdvanceMs),
691                3 => (0u64..=10_000).prop_map(ThrottleOp::Gate),
692            ]
693        }
694
695        proptest! {
696            // Pin regression files to the crate directory
697            #![proptest_config(ProptestConfig {
698                failure_persistence: Some(Box::new(
699                    proptest::test_runner::FileFailurePersistence::Direct(
700                        concat!(env!("CARGO_MANIFEST_DIR"), "/proptest-regressions/backoff.txt")
701                    )
702                )),
703                ..ProptestConfig::default()
704            })]
705
706            #[rstest]
707            fn test_throttle_threshold_boundary(
708                attempt_count in 0usize..=8,
709                input_ms in 0u64..=5_000,
710            ) {
711                let runtime = build_paused_runtime();
712                runtime.block_on(async move {
713                    let mut throttle = ReconnectThrottle::default();
714
715                    for _ in 0..attempt_count {
716                        throttle.record_attempt();
717                    }
718
719                    let input = Duration::from_millis(input_ms);
720                    let expected = if attempt_count >= RECONNECT_MIN_DELAY_ATTEMPTS {
721                        input.max(RECONNECT_MIN_DELAY)
722                    } else {
723                        input
724                    };
725                    assert_eq!(
726                        throttle.gated_delay(input),
727                        expected,
728                        "Threshold mismatch at {attempt_count} attempts in window"
729                    );
730                });
731            }
732
733            #[rstest]
734            fn test_throttle_matches_rolling_window_model(
735                ops in prop::collection::vec(throttle_op_strategy(), 1..=500),
736            ) {
737                let runtime = build_paused_runtime();
738                runtime.block_on(async move {
739                    let mut throttle = ReconnectThrottle::default();
740                    let mut attempt_times_ms: Vec<u64> = Vec::new();
741                    let mut now_ms = 0u64;
742                    let window_ms = RECONNECT_MIN_DELAY_WINDOW.as_millis() as u64;
743
744                    for op in ops {
745                        match op {
746                            ThrottleOp::Attempt => {
747                                throttle.record_attempt();
748                                attempt_times_ms.push(now_ms);
749                            }
750                            ThrottleOp::AdvanceMs(ms) => {
751                                dst::time::sleep(Duration::from_millis(ms)).await;
752                                now_ms += ms;
753                            }
754                            ThrottleOp::Gate(input_ms) => {
755                                let input = Duration::from_millis(input_ms);
756                                let kept = attempt_times_ms
757                                    .iter()
758                                    .filter(|t| now_ms - *t <= window_ms)
759                                    .count();
760                                let expected = if kept >= RECONNECT_MIN_DELAY_ATTEMPTS {
761                                    input.max(RECONNECT_MIN_DELAY)
762                                } else {
763                                    input
764                                };
765                                let gated = throttle.gated_delay(input);
766                                assert_eq!(
767                                    gated, expected,
768                                    "Model mismatch at {now_ms}ms with {kept} attempts in window"
769                                );
770                                assert!(
771                                    gated >= input,
772                                    "Floor must never lower the backoff delay"
773                                );
774                            }
775                        }
776                    }
777                });
778            }
779        }
780    }
781}
nautilus_network/backoff.rs

nautilus_network/
backoff.rs
