fix: prevent polling loop hang when MQTT broker is unreachable

Add three layers of defense against the polling loop blocking indefinitely
during extended broker outages:

1. Shared mqtt_connected flag (Arc<AtomicBool>) set by the event loop on
   ConnAck/Disconnect/Error, checked at the top of each polling cycle to
   skip vcontrold queries and publishes entirely when the broker is down.

2. Publish timeout (5s) around publish_retained calls so individual
   publishes never block the polling loop if rumqttc's internal channel
   fills up.

3. MissedTickBehavior::Skip on the polling interval to prevent burst-
   firing all missed ticks after a stall.

Author: tma

Cargo.toml

@@ -18,6 +18,9 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 thiserror = "2"
 hostname = "0.4"
 
+[dev-dependencies]
+tokio = { version = "1", features = ["test-util"] }
+
 [profile.release]
 lto = true
 strip = true

src/main.rs

@@ -13,6 +13,7 @@ mod polling;
 mod process;
 mod vcontrold;
 
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use tokio::sync::mpsc;
 use tracing::{error, info};
@@ -91,6 +92,10 @@ async fn run() -> Result<()> {
     let (mqtt_client, eventloop) = MqttClient::new(&config.mqtt, &publisher_client_id)?;
     let mqtt_client = Arc::new(mqtt_client);
 
+    // Shared flag: tracks whether the MQTT broker is currently reachable.
+    // Written by run_event_loop, read by run_polling_loop.
+    let mqtt_connected = Arc::new(AtomicBool::new(false));
+
     // Channel for subscriber messages (if enabled)
     let (message_tx, message_rx) = if config.mqtt_subscribe {
         let (tx, rx) = mpsc::channel(100);
@@ -114,15 +119,17 @@ async fn run() -> Result<()> {
         mqtt_client.clone_client(),
         subscribe_topics,
         message_tx,
+        Arc::clone(&mqtt_connected),
     ));
 
     // Spawn polling loop (if commands are configured)
     let polling_handle = if !config.commands.is_empty() {
         let config_clone = config.clone();
         let vcontrold_clone = Arc::clone(&vcontrold_client);
         let mqtt_clone = Arc::clone(&mqtt_client);
+        let connected = Arc::clone(&mqtt_connected);
         Some(tokio::spawn(async move {
-            run_polling_loop(&config_clone, vcontrold_clone, mqtt_clone).await;
+            run_polling_loop(&config_clone, vcontrold_clone, mqtt_clone, connected).await;
         }))
     } else {
         info!("No commands configured, polling disabled");

src/mqtt/client.rs

@@ -10,6 +10,7 @@ use rustls::ClientConfig;
 use std::fs::File;
 use std::io::BufReader;
 use std::path::Path;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::mpsc;
@@ -269,6 +270,7 @@ pub async fn run_event_loop(
     client: AsyncClient,
     subscribe_topics: Vec<String>,
     message_tx: Option<mpsc::Sender<IncomingMessage>>,
+    mqtt_connected: Arc<AtomicBool>,
 ) {
     loop {
         match eventloop.poll().await {
@@ -289,6 +291,7 @@ pub async fn run_event_loop(
                         }
                         rumqttc::v5::Incoming::ConnAck(_) => {
                             info!("Connected to MQTT broker");
+                            mqtt_connected.store(true, Ordering::Relaxed);
 
                             // Re-subscribe to all topics on every (re)connection
                             for topic in &subscribe_topics {
@@ -306,13 +309,15 @@ pub async fn run_event_loop(
                         }
                         rumqttc::v5::Incoming::Disconnect(_) => {
                             warn!("Disconnected from MQTT broker");
+                            mqtt_connected.store(false, Ordering::Relaxed);
                         }
                         _ => {}
                     }
                 }
             }
             Err(e) => {
                 error!("MQTT event loop error: {}", e);
+                mqtt_connected.store(false, Ordering::Relaxed);
                 // Wait before retrying
                 tokio::time::sleep(Duration::from_secs(10)).await;
             }

src/mqtt/publisher.rs

@@ -2,11 +2,20 @@
 //!
 //! Publishes vcontrold command results to MQTT topics.
 
+use std::time::Duration;
+
+use tokio::time::timeout;
 use tracing::{debug, error, warn};
 
 use crate::error::MqttError;
 use crate::vcontrold::{CommandResult, Value};
 
+/// Timeout for individual MQTT publish operations.
+///
+/// Prevents the polling loop from blocking indefinitely when the MQTT
+/// client's internal channel is full (e.g. during a broker outage).
+const PUBLISH_TIMEOUT: Duration = Duration::from_secs(5);
+
 use super::client::MqttClient;
 
 /// Publisher for vcontrold polling results
@@ -48,7 +57,17 @@ impl<'a> Publisher<'a> {
         let topic = self.client.topic(&format!("command/{}", result.command));
         debug!("Publishing to {}: {}", topic, payload);
 
-        self.client.publish_retained(&topic, &payload).await
+        match timeout(PUBLISH_TIMEOUT, self.client.publish_retained(&topic, &payload)).await {
+            Ok(result) => result,
+            Err(_) => {
+                warn!(
+                    "Publish timeout for {} after {}s - MQTT client may be stalled",
+                    topic,
+                    PUBLISH_TIMEOUT.as_secs()
+                );
+                Ok(())
+            }
+        }
     }
 
     /// Publish multiple command results
@@ -93,4 +112,23 @@ mod tests {
         assert_eq!(format_number(3.14159), "3.14159");
         assert_eq!(format_number(0.5), "0.5");
     }
+
+    #[test]
+    fn test_publish_timeout_is_5_seconds() {
+        assert_eq!(PUBLISH_TIMEOUT, Duration::from_secs(5));
+    }
+
+    /// Verify that `tokio::time::timeout` with `PUBLISH_TIMEOUT` fires
+    /// instead of blocking forever when the inner future never resolves.
+    /// This simulates the scenario where `publish_retained` blocks because
+    /// rumqttc's internal channel is full during a broker outage.
+    #[tokio::test]
+    async fn test_publish_timeout_fires_on_stalled_future() {
+        tokio::time::pause();
+
+        let stalled = std::future::pending::<Result<(), MqttError>>();
+        let result = timeout(PUBLISH_TIMEOUT, stalled).await;
+
+        assert!(result.is_err(), "timeout should fire on a stalled future");
+    }
 }

src/polling.rs

@@ -2,6 +2,7 @@
 //!
 //! Handles command batching and periodic execution.
 
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use tokio::time::interval;
 use tracing::{debug, error, info, warn};
@@ -66,6 +67,7 @@ pub async fn run_polling_loop(
     config: &Config,
     vcontrold: Arc<VcontroldClient>,
     mqtt_client: Arc<MqttClient>,
+    mqtt_connected: Arc<AtomicBool>,
 ) {
     if config.commands.is_empty() {
         warn!("No commands configured for polling");
@@ -88,11 +90,32 @@ pub async fn run_polling_loop(
     }
 
     let mut poll_interval = interval(config.interval);
+    // Skip missed ticks instead of bursting them all at once. This prevents
+    // overwhelming the MQTT client after a stall (e.g. broker outage where
+    // publishes hit the timeout and the interval falls behind).
+    poll_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
     let publisher = Publisher::new(&mqtt_client);
 
+    let mut was_disconnected = false;
+
     loop {
         poll_interval.tick().await;
 
+        // Skip entire cycle when the MQTT broker is unreachable. This avoids
+        // unnecessary vcontrold/Optolink traffic and prevents filling the
+        // rumqttc internal channel (which would block the polling loop).
+        if !mqtt_connected.load(Ordering::Relaxed) {
+            if !was_disconnected {
+                warn!("MQTT broker disconnected, skipping polling cycles");
+                was_disconnected = true;
+            }
+            continue;
+        }
+        if was_disconnected {
+            info!("MQTT broker reconnected, resuming polling");
+            was_disconnected = false;
+        }
+
         debug!("Starting polling cycle");
 
         for (batch_idx, batch) in batches.iter().enumerate() {
@@ -183,4 +206,66 @@ mod tests {
         assert_eq!(batches.len(), 1);
         assert_eq!(batches[0], vec!["veryLongCommandName"]);
     }
+
+    /// Verify that the polling interval uses Skip behavior: after a long stall
+    /// only one tick fires rather than a burst of all missed ticks.
+    ///
+    /// With the default `Burst` behavior, advancing time by 5x the interval
+    /// would yield 5 immediately-ready ticks. With `Skip`, only the next
+    /// natural tick fires, so we get exactly 1.
+    #[tokio::test]
+    async fn test_interval_skips_missed_ticks() {
+        use std::time::Duration;
+        use tokio::time::{interval, MissedTickBehavior};
+
+        tokio::time::pause();
+
+        let period = Duration::from_secs(10);
+        let mut ivl = interval(period);
+        ivl.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+        // First tick fires immediately (interval semantics)
+        ivl.tick().await;
+
+        // Simulate a stall: advance time by 5 periods without ticking
+        tokio::time::advance(period * 5).await;
+
+        // After the stall, exactly one tick should be ready (Skip discards
+        // missed ticks and resets to the next future deadline).
+        ivl.tick().await;
+
+        // The next tick should NOT be immediately available — it should
+        // require waiting another full period.
+        let next = tokio::time::timeout(period / 2, ivl.tick()).await;
+        assert!(
+            next.is_err(),
+            "no burst tick should be available; Skip must discard missed ticks"
+        );
+    }
+
+    #[test]
+    fn test_mqtt_connected_flag_state_transitions() {
+        // Verify the AtomicBool flag behaves correctly across the
+        // state transitions that run_event_loop and run_polling_loop rely on.
+        let connected = Arc::new(AtomicBool::new(false));
+
+        // Initial state: disconnected (same as main.rs)
+        assert!(!connected.load(Ordering::Relaxed));
+
+        // Simulate ConnAck in event loop
+        connected.store(true, Ordering::Relaxed);
+        assert!(connected.load(Ordering::Relaxed));
+
+        // Simulate Disconnect
+        connected.store(false, Ordering::Relaxed);
+        assert!(!connected.load(Ordering::Relaxed));
+
+        // Simulate reconnect
+        connected.store(true, Ordering::Relaxed);
+        assert!(connected.load(Ordering::Relaxed));
+
+        // Simulate event loop error
+        connected.store(false, Ordering::Relaxed);
+        assert!(!connected.load(Ordering::Relaxed));
+    }
 }