Throttle MQTT errors in log

common/Cargo.toml

@@ -12,12 +12,12 @@ rumqttc.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_with.workspace = true
+tokio.workspace = true
 toml.workspace = true
 tracing.workspace = true
 url.workspace = true
 
 [dev-dependencies]
 ctor.workspace = true
 satori-testing-utils.workspace = true
-tokio.workspace = true
 tracing-subscriber.workspace = true

common/src/lib.rs

@@ -16,4 +16,4 @@ pub const SEGMENT_FILENAME_FORMAT: &str = "%Y-%m-%dT%H_%M_%S%z.ts";
 mod version;
 
 mod utils;
-pub use self::utils::load_config_file;
+pub use self::utils::{load_config_file, ThrottledErrorLogger};

common/src/mqtt.rs

@@ -1,3 +1,4 @@
+use crate::utils::ThrottledErrorLogger;
 use rumqttc::{AsyncClient, Event, EventLoop, Incoming, MqttOptions, Outgoing, Publish, QoS};
 use serde::Deserialize;
 use std::time::Duration;
@@ -19,6 +20,7 @@ pub struct MqttConfig {
 pub struct MqttClient {
     client: AsyncClient,
     event_loop: EventLoop,
+    poll_error_logger: ThrottledErrorLogger<String>,
 
     topic: String,
 }
@@ -34,6 +36,7 @@ impl From<MqttConfig> for MqttClient {
         Self {
             client,
             event_loop,
+            poll_error_logger: ThrottledErrorLogger::new(Duration::from_secs(5)),
             topic: config.topic,
         }
     }
@@ -64,7 +67,9 @@ impl MqttClient {
             }
             Ok(_) => None,
             Err(e) => {
-                warn!("rumqttc error: {:?}", e);
+                if let Some(e) = self.poll_error_logger.log(format!("{:?}", e)) {
+                    warn!("rumqttc error: {}", e);
+                }
                 None
             }
         }
@@ -81,7 +86,9 @@ impl MqttClient {
                 }
                 Ok(_) => {}
                 Err(e) => {
-                    warn!("rumqttc error: {:?}", e);
+                    if let Some(e) = self.poll_error_logger.log(format!("{:?}", e)) {
+                        warn!("rumqttc error: {}", e);
+                    }
                 }
             }
         }
@@ -114,7 +121,9 @@ impl MqttClient {
                 }
                 Ok(_) => {}
                 Err(e) => {
-                    warn!("rumqttc error: {:?}", e);
+                    if let Some(e) = self.poll_error_logger.log(format!("{:?}", e)) {
+                        warn!("rumqttc error: {}", e);
+                    }
                 }
             }
         }

common/src/utils/mod.rs

@@ -0,0 +1,4 @@
+mod config_file;
+mod throttled_error;
+
+pub use self::{config_file::load_config_file, throttled_error::ThrottledErrorLogger};

common/src/utils/throttled_error.rs

@@ -0,0 +1,115 @@
+use tokio::time::{Duration, Instant};
+use tracing::warn;
+
+struct ThrottledError<E> {
+    error: E,
+    first_seen: Instant,
+    times_seen: usize,
+}
+
+pub struct ThrottledErrorLogger<E> {
+    timeout: Duration,
+    last_error: Option<ThrottledError<E>>,
+}
+
+impl<E: PartialEq + std::fmt::Display> ThrottledErrorLogger<E> {
+    pub fn new(timeout: Duration) -> Self {
+        Self {
+            timeout,
+            last_error: None,
+        }
+    }
+
+    pub fn log(&mut self, error: E) -> Option<&E> {
+        let now = Instant::now();
+        let last_error = self.last_error.as_mut();
+
+        match last_error {
+            Some(last_error)
+                if last_error.error == error && last_error.first_seen.elapsed() > self.timeout =>
+            {
+                warn!(
+                    "Last error ({}) was (seen {} times)",
+                    last_error.error, last_error.times_seen
+                );
+                last_error.first_seen = now;
+                last_error.times_seen = 1;
+                Some(&self.last_error.as_ref().unwrap().error)
+            }
+            Some(last_error)
+                if last_error.error == error && last_error.first_seen.elapsed() <= self.timeout =>
+            {
+                last_error.times_seen += 1;
+                None
+            }
+            Some(last_error) => {
+                warn!(
+                    "Last error ({}) was (seen {} times)",
+                    last_error.error, last_error.times_seen
+                );
+                self.last_error = Some(ThrottledError {
+                    error,
+                    first_seen: now,
+                    times_seen: 1,
+                });
+                Some(&self.last_error.as_ref().unwrap().error)
+            }
+            None => {
+                self.last_error = Some(ThrottledError {
+                    error,
+                    first_seen: now,
+                    times_seen: 1,
+                });
+                Some(&self.last_error.as_ref().unwrap().error)
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[tokio::test]
+    async fn new() {
+        let te = ThrottledErrorLogger::<String>::new(Duration::from_millis(100));
+        assert!(te.last_error.is_none());
+    }
+
+    #[tokio::test]
+    async fn log_once() {
+        let mut te = ThrottledErrorLogger::<String>::new(Duration::from_millis(100));
+        assert_eq!(te.log("test".to_string()).unwrap(), "test");
+    }
+
+    #[tokio::test]
+    async fn log_duplicates_fast() {
+        let mut te = ThrottledErrorLogger::<String>::new(Duration::from_millis(100));
+        assert_eq!(te.log("test".to_string()).unwrap(), "test");
+        assert!(te.log("test".to_string()).is_none());
+        assert!(te.log("test".to_string()).is_none());
+        tokio::time::sleep(Duration::from_millis(95)).await;
+        assert!(te.log("test".to_string()).is_none());
+    }
+
+    #[tokio::test]
+    async fn log_duplicates_slow() {
+        let mut te = ThrottledErrorLogger::<String>::new(Duration::from_millis(100));
+        assert_eq!(te.log("test".to_string()).unwrap(), "test");
+        assert!(te.log("test".to_string()).is_none());
+        assert!(te.log("test".to_string()).is_none());
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        assert_eq!(te.log("test".to_string()).unwrap(), "test");
+        assert!(te.log("test".to_string()).is_none());
+    }
+
+    #[tokio::test]
+    async fn log_unique_fast() {
+        let mut te = ThrottledErrorLogger::<String>::new(Duration::from_millis(100));
+        assert_eq!(te.log("test a".to_string()).unwrap(), "test a");
+        assert_eq!(te.log("test b".to_string()).unwrap(), "test b");
+        assert_eq!(te.log("test a".to_string()).unwrap(), "test a");
+        tokio::time::sleep(Duration::from_millis(95)).await;
+        assert_eq!(te.log("test b".to_string()).unwrap(), "test b");
+    }
+}