attempt handle mqtt connection error & exponential backoff

src-tauri/Cargo.lock

···
  "tauri-plugin-autostart",
  "tauri-plugin-opener",
  "tokio",
+ "tokio-retry2",
  "tokio-util",
  "tracing",
  "tracing-subscriber",
···
 ]
 
 [[package]]
+name = "pin-project"
+version = "1.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "677f1add503faace112b9f1373e43e9e054bfdd22ff1a63c1bc485eaec6a6a8a"
+dependencies = [
+ "pin-project-internal",
+]
+
+[[package]]
+name = "pin-project-internal"
+version = "1.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.108",
+]
+
+[[package]]
 name = "pin-project-lite"
 version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "rand"
+version = "0.9.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1"
+dependencies = [
+ "rand_chacha 0.9.0",
+ "rand_core 0.9.3",
+]
+
+[[package]]
 name = "rand_chacha"
 version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.3",
+]
+
+[[package]]
 name = "rand_core"
 version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
 dependencies = [
  "getrandom 0.2.16",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "99d9a13982dcf210057a8a78572b2217b667c3beacbf3a0d8b454f6f82837d38"
+dependencies = [
+ "getrandom 0.3.4",
 ]
 
 [[package]]
···
 checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2"
 dependencies = [
  "native-tls",
+ "tokio",
+]
+
+[[package]]
+name = "tokio-retry2"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae5ed96d30859f64c721d9155b66c53c39867e2ced2f0614ba61da5665440b0a"
+dependencies = [
+ "pin-project",
+ "rand 0.9.2",
  "tokio",
 ]
 

src-tauri/Cargo.toml

···
 objc2-io-kit = "0.3.2"
 objc2-core-foundation = "0.3.2"
 objc2-foundation = "0.3.2"
+tokio-retry2 = { version = "0.7.0", features = ["jitter"] }
 
 [target.'cfg(not(any(target_os = "android", target_os = "ios")))'.dependencies]
 tauri-plugin-autostart = "2"

src-tauri/src/lib.rs

···
 #![feature(gethostname)]
 #![feature(result_option_map_or_default)]
 
-use std::str::FromStr;
-
 pub mod app_config;
 pub mod constants;
 pub mod hex_slice;

src-tauri/src/main.rs

···
     tauri_app,
 };
 use tokio::{sync::watch, task::JoinHandle};
+use tokio_util::sync::CancellationToken;
 use tracing::error;
 
 #[tokio::main]
···
     tokio::spawn(async move {
         let mut mqtt_loop_handle: Option<JoinHandle<()>> = None;
         let mut smc_sender_handle: Option<JoinHandle<()>> = None;
+        let ct = CancellationToken::new();
 
         loop {
             // Drop previous handles first
···
             };
 
             let (mqttc, eventloop) = mqtt::create(host.unwrap(), port, username, password);
-            mqtt_loop_handle = Some(mqtt::run_loop(eventloop));
+            mqtt_loop_handle = Some(mqtt::run_loop(eventloop, ct.clone()));
 
             mqtt::interview(&mqttc, None, Some("MacBook Pro"))
                 .await

src-tauri/src/mqtt.rs

···
 use rootcause::{Report, prelude::ResultExt};
 use rumqttc::{AsyncClient, EventLoop, MqttOptions};
 use serde_json::{Value, json};
-use tracing::{info, instrument, trace};
+use tokio::select;
+use tokio_retry2::strategy::{ExponentialBackoff, ExponentialFactorBackoff, jitter};
+use tokio_util::sync::CancellationToken;
+use tracing::{error, info, instrument, trace, warn};
 
 use crate::{hex_slice::HexSlice, smc::PowerData};
 
···
 }
 
 #[instrument(skip(e))]
-pub fn run_loop(mut e: EventLoop) -> tokio::task::JoinHandle<()> {
-    tokio::spawn(async move {
-        trace!("Polling MQTT event loop");
-        // select! {}
-        while let Ok(notif) = e.poll().await {
-            trace!("Received: {notif:?}");
+pub fn run_loop(e: EventLoop, ct: CancellationToken) -> tokio::task::JoinHandle<()> {
+    trace!("Polling MQTT event loop");
+    tokio::spawn(run_loop_inner(e, ct))
+}
+
+async fn run_loop_inner(mut e: EventLoop, ct: CancellationToken) {
+    let mut retry_strategy = generate_default_backoff();
+
+    loop {
+        select! {
+            e = e.poll() => {
+                if e.is_ok() {
+                    retry_strategy = generate_default_backoff();
+                    continue;
+                }
+
+                let err = e.unwrap_err();
+                match err {
+                    rumqttc::ConnectionError::ConnectionRefused(connect_return_code) => {
+                        error!("MQTT Connection Refused - Error code {connect_return_code:?}");
+                        break
+                    },
+                    e => error!("MQTT connection error occurred - {e:?}")
+                }
+
+                let sleep_time = retry_strategy.next().unwrap();
+                error!("Waiting for {} to retry poll", sleep_time.as_secs());
+                tokio::time::sleep(sleep_time).await;
+            }
+            _ = ct.cancelled() => {
+                warn!("Caught cancellation token!");
+                break
+            }
         }
-    })
+    }
+}
+
+fn generate_default_backoff() -> ExponentialFactorBackoff {
+    ExponentialFactorBackoff::from_millis(5_000, 1.5).max_delay(Duration::from_secs(60))
 }