Add jitter support to retry behavior (#41)

According to our streaming spec, the retry behavior should also include
some form of jitter. This jitter will cause the delay to be AT MOST
halved (e.g. a 30 second delay, post jitter will be in the range of
[15, 30] seconds).

Author: keelerm84

eventsource-client/Cargo.toml

@@ -20,6 +20,7 @@ log = "0.4.6"
 pin-project = "1.0.10"
 tokio = { version = "1.17.0", features = ["time"] }
 hyper-timeout = "0.4.1"
+rand = "0.8.5"
 
 [dev-dependencies]
 env_logger = "0.7.1"

eventsource-client/src/client.rs

@@ -318,6 +318,7 @@ impl<C> ReconnectingRequest<C> {
                 reconnect_delay,
                 delay_max,
                 backoff_factor,
+                true,
             )),
             redirect_count: 0,
             current_url: url,

eventsource-client/src/retry.rs

@@ -1,5 +1,7 @@
 use std::time::{Duration, Instant};
 
+use rand::{thread_rng, Rng};
+
 pub(crate) trait RetryStrategy {
     /// Return the next amount of time a failed request should delay before re-attempting.
     fn next_delay(&mut self, current_time: Instant) -> Duration;
@@ -17,18 +19,25 @@ pub(crate) struct BackoffRetry {
     base_delay: Duration,
     max_delay: Duration,
     backoff_factor: u32,
+    include_jitter: bool,
 
     reset_interval: Duration,
     next_delay: Duration,
     good_since: Option<Instant>,
 }
 
 impl BackoffRetry {
-    pub fn new(base_delay: Duration, max_delay: Duration, backoff_factor: u32) -> Self {
+    pub fn new(
+        base_delay: Duration,
+        max_delay: Duration,
+        backoff_factor: u32,
+        include_jitter: bool,
+    ) -> Self {
         Self {
             base_delay,
             max_delay,
             backoff_factor,
+            include_jitter,
             reset_interval: DEFAULT_RESET_RETRY_INTERVAL,
             next_delay: base_delay,
             good_since: None,
@@ -38,18 +47,22 @@ impl BackoffRetry {
 
 impl RetryStrategy for BackoffRetry {
     fn next_delay(&mut self, current_time: Instant) -> Duration {
-        let mut next_delay = self.next_delay;
+        let mut current_delay = self.next_delay;
 
         if let Some(good_since) = self.good_since {
             if current_time - good_since >= self.reset_interval {
-                next_delay = self.base_delay;
+                current_delay = self.base_delay;
             }
         }
 
         self.good_since = None;
-        self.next_delay = std::cmp::min(self.max_delay, next_delay * self.backoff_factor);
+        self.next_delay = std::cmp::min(self.max_delay, current_delay * self.backoff_factor);
 
-        next_delay
+        if self.include_jitter {
+            thread_rng().gen_range(current_delay / 2..=current_delay)
+        } else {
+            current_delay
+        }
     }
 
     fn change_base_delay(&mut self, base_delay: Duration) {
@@ -76,7 +89,7 @@ mod tests {
     #[test]
     fn test_fixed_retry() {
         let base = Duration::from_secs(10);
-        let mut retry = BackoffRetry::new(base, Duration::from_secs(30), 1);
+        let mut retry = BackoffRetry::new(base, Duration::from_secs(30), 1, false);
         let start = Instant::now() - Duration::from_secs(60);
 
         assert_eq!(retry.next_delay(start), base);
@@ -87,7 +100,7 @@ mod tests {
     #[test]
     fn test_able_to_reset_base_delay() {
         let base = Duration::from_secs(10);
-        let mut retry = BackoffRetry::new(base, Duration::from_secs(30), 1);
+        let mut retry = BackoffRetry::new(base, Duration::from_secs(30), 1, false);
         let start = Instant::now();
 
         assert_eq!(retry.next_delay(start), base);
@@ -102,7 +115,7 @@ mod tests {
     fn test_with_backoff() {
         let base = Duration::from_secs(10);
         let max = Duration::from_secs(60);
-        let mut retry = BackoffRetry::new(base, max, 2);
+        let mut retry = BackoffRetry::new(base, max, 2, false);
         let start = Instant::now() - Duration::from_secs(60);
 
         assert_eq!(retry.next_delay(start), base);
@@ -117,12 +130,23 @@ mod tests {
         assert_eq!(retry.next_delay(start.add(Duration::from_secs(3))), max);
     }
 
+    #[test]
+    fn test_with_jitter() {
+        let base = Duration::from_secs(10);
+        let max = Duration::from_secs(60);
+        let mut retry = BackoffRetry::new(base, max, 1, true);
+        let start = Instant::now() - Duration::from_secs(60);
+
+        let delay = retry.next_delay(start);
+        assert!(base / 2 <= delay && delay <= base);
+    }
+
     #[test]
     fn test_retry_holds_at_max() {
         let base = Duration::from_secs(20);
         let max = Duration::from_secs(30);
 
-        let mut retry = BackoffRetry::new(base, max, 2);
+        let mut retry = BackoffRetry::new(base, max, 2, false);
         let start = Instant::now();
         retry.reset(start);
 
@@ -146,7 +170,7 @@ mod tests {
         let reset_interval = Duration::from_secs(45);
 
         // Prepare a retry strategy that has succeeded at a specific point.
-        let mut retry = BackoffRetry::new(base, max, 2);
+        let mut retry = BackoffRetry::new(base, max, 2, false);
         retry.reset_interval = reset_interval;
         let start = Instant::now() - Duration::from_secs(60);
         retry.reset(start);