Merge pull request astarte-platform#446 from joshuachp/refactor/retry-logic

refactor(retry): improve the retry logic

Author: harlem88

src/connection/mod.rs

@@ -26,6 +26,7 @@ use tokio::task::JoinHandle;
 use tracing::{debug, error, info, trace, warn};
 
 use crate::error::Report;
+use crate::retry::ExponentialIter;
 use crate::state::{SharedState, Status};
 use crate::transport::TransportError;
 use crate::Timestamp;
@@ -87,6 +88,7 @@ where
     sender: C::Sender,
     state: Arc<SharedState>,
     resend: Option<JoinHandle<()>>,
+    backoff: ExponentialIter,
 }
 
 impl<C> DeviceConnection<C>
@@ -107,6 +109,7 @@ where
             connection,
             sender,
             resend: None,
+            backoff: ExponentialIter::default(),
         }
     }
 

src/connection/resend.rs

@@ -25,7 +25,6 @@ use crate::builder::DEFAULT_CHANNEL_SIZE;
 use crate::error::Report;
 use crate::retention::memory::ItemValue;
 use crate::retention::{RetentionId, StoredRetention, StoredRetentionExt};
-use crate::retry::ExponentialIter;
 use crate::state::SharedState;
 use crate::store::wrapper::StoreWrapper;
 use crate::store::StoreCapabilities;
@@ -115,7 +114,7 @@ where
 
             match resend.await {
                 Ok(()) => {
-                    trace!("resend task joined")
+                    trace!("task already exited")
                 }
                 Err(err) if err.is_cancelled() => {
                     debug!("resend task was cancelled");
@@ -191,10 +190,32 @@ where
     {
         let interfaces = self.state.interfaces.read().await;
         debug!("reconnecting");
-        let mut exp = ExponentialIter::default();
+
+        // Wait before trying to reconnect the first time, this will prevent cases where the
+        // connection will loop that will keep throwing errors.
+        //
+        // The back-off will start with a wait time of 0s and increase exponentially until it
+        // reaches a max. We also keep track of when the last disconnection happened, to reset the
+        // wait time only if the connection has been stable for a certain duration of time.
+        //
+        // An example of an error loop is when the MQTT connection Interfaces (the device
+        // introspection) contains interfaces not installed on Astarte:
+        //
+        // - Device: publishes on that interface
+        // - Astarte: disconnects us since we don't have that interface in the introspection
+        // - Device: reconnects and succeeds on the first try (the while loop bellow)
+        // - Device: publishes on the same interface and gets disconnected again.
+        //
+        // If we didn't keep track of the last disconnection, the error loop above would continue to
+        // happen without timeouts, wasting device bandwidth and resources.
+        let timeout = self.backoff.next();
+
+        debug!("waiting {timeout} seconds before retrying");
+
+        tokio::time::sleep(Duration::from_secs(timeout)).await;
 
         while !self.connection.reconnect(&interfaces).await? {
-            let timeout = exp.next();
+            let timeout = self.backoff.next();
 
             debug!("waiting {timeout} seconds before retrying");
 

src/retention/mod.rs

@@ -517,8 +517,9 @@ impl Context {
         let timestamp = TimestampMillis::now();
 
         // We want the values to be unique, this will wrap around, but it will never wrap on the
-        // same ms
-        let counter = self.counter.fetch_add(1, Ordering::AcqRel);
+        // same ms, the ordering can be relaxed since the only guarantee we need is for the counter
+        // to yield unique values.
+        let counter = self.counter.fetch_add(1, Ordering::Relaxed);
 
         Id { timestamp, counter }
     }
@@ -540,35 +541,40 @@ mod tests {
     #[test]
     fn id_should_be_unique() {
         const NUM: usize = 5;
+        const CAP: usize = 1000;
         let ctx = Arc::new(Context::new());
 
-        let (tx, rx) = std::sync::mpsc::sync_channel::<Id>(NUM);
+        let (tx, rx) = std::sync::mpsc::sync_channel::<Vec<Id>>(NUM);
 
         for _ in 0..NUM {
             std::thread::spawn({
                 let ctx = Arc::clone(&ctx);
                 let tx = tx.clone();
 
                 move || {
+                    let mut gen = Vec::with_capacity(CAP);
                     let mut prev = ctx.next();
-                    for _i in 0..1000 {
+                    for _i in 0..CAP {
                         let new = ctx.next();
 
-                        tx.send(new).expect("channel closed");
-
                         assert!(new > prev);
 
+                        gen.push(prev);
                         prev = new;
                     }
+
+                    tx.send(gen).expect("channel closed");
                 }
             });
         }
 
         drop(tx);
 
         let mut recvd = HashSet::new();
-        while let Ok(new) = rx.recv() {
-            assert!(recvd.insert(new));
+        while let Ok(gen) = rx.recv() {
+            for i in gen {
+                assert!(recvd.insert(i));
+            }
         }
     }
 }

src/retry.rs

@@ -18,15 +18,33 @@
 
 //! Module to handle the retry of the MQTT connection
 
+use std::time::{Duration, Instant};
+
+use tracing::trace;
+
 /// Iterator that yields a delay that will increase exponentially till the max,
 #[derive(Debug, Clone, Copy)]
 pub(crate) struct ExponentialIter {
     n: u32,
     max: u32,
+    reset_after: Duration,
+    last: Option<Instant>,
 }
 
 impl ExponentialIter {
     pub(crate) fn next(&mut self) -> u64 {
+        if self
+            .last
+            .is_some_and(|instant| instant.elapsed() > self.reset_after)
+        {
+            trace!("resetting timeout");
+
+            // Start from the beginning
+            self.n = 0;
+        }
+
+        self.last = Some(Instant::now());
+
         let v = ((self.n > 0) as u64).wrapping_shl(self.n.saturating_sub(1));
 
         self.n = self.n.saturating_add(1).min(self.max);
@@ -37,7 +55,12 @@ impl ExponentialIter {
 
 impl Default for ExponentialIter {
     fn default() -> Self {
-        Self { n: 0, max: 9 }
+        Self {
+            n: 0,
+            max: 9,
+            reset_after: Duration::from_secs(256 * 4),
+            last: None,
+        }
     }
 }