fix(mqtt): Surface clean-session reconnect on Subscription::next_message

Cargo.toml

@@ -43,7 +43,7 @@ sequential-storage = { version = "7.0", features = [
 serde-json-core = { version = "0.6" }
 rustot-derive = { path = "rustot_derive", version = "0.2.1" }
 embedded-storage-async = { version = "0.4", optional = true }
-mqttrust = { git = "https://github.com/FactbirdHQ/mqttrust", rev = "d4d0cd8", optional = true }
+mqttrust = { git = "https://github.com/FactbirdHQ/mqttrust", rev = "04e4a0c", optional = true }
 
 embassy-time = { version = "0.5.0" }
 embassy-sync = "0.8"

src/mqtt/mod.rs

@@ -125,7 +125,12 @@ pub trait MqttSubscription {
 
     /// Wait for and return the next message.
     ///
-    /// Returns `None` if the subscription has been closed.
+    /// Returns `None` if the subscription has been closed, or if the broker
+    /// has dropped subscriptions for this client (a CONNACK with
+    /// `session_present=false`). Transient disconnects and session-resume
+    /// reconnects keep the subscription valid. Callers should treat `None`
+    /// as a recoverable signal: drop the subscription and resubscribe on
+    /// the next operation.
     fn next_message(&mut self) -> impl Future<Output = Option<Self::Message<'_>>>;
 
     /// Unsubscribe and close this subscription.

src/mqtt/rumqttc.rs

@@ -7,7 +7,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::{Mutex, mpsc, watch};
 
 use crate::mqtt::{MqttMessage, MqttSubscription, PublishOptions, QoS, ToPayload};
 
@@ -22,8 +22,16 @@ pub struct RumqttcClient {
     client_id: String,
     /// Routes incoming publishes to subscriptions by subscription ID
     router: Arc<Mutex<MessageRouter>>,
-    /// Tracks connection state
-    connected: Arc<std::sync::atomic::AtomicBool>,
+    /// `true` when CONNACK received, `false` after disconnect or error.
+    /// Used by `wait_connected` only.
+    connection_state: watch::Sender<bool>,
+    /// Bumped on each CONNACK with `session_present=false` (broker dropped
+    /// our subs). Subscriptions snapshot this at creation; `next_message`
+    /// races against `changed()` so a parked waiter wakes with `None` when
+    /// the broker has actually dropped our routing. Transient disconnects
+    /// and session-resume reconnects do not bump this counter — broker
+    /// state is intact in those cases, so the subscription stays valid.
+    clean_session_count: watch::Sender<u8>,
 }
 
 struct MessageRouter {
@@ -44,20 +52,29 @@ impl RumqttcClient {
             next_sub_id: 0,
             subscriptions: HashMap::new(),
         }));
-        let connected = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let (connection_state, _) = watch::channel(false);
+        let (clean_session_count, _) = watch::channel(0u8);
 
         let router_clone = router.clone();
-        let connected_clone = connected.clone();
+        let connection_state_clone = connection_state.clone();
+        let clean_session_count_clone = clean_session_count.clone();
         let handle = tokio::spawn(async move {
-            Self::run_eventloop(eventloop, router_clone, connected_clone).await;
+            Self::run_eventloop(
+                eventloop,
+                router_clone,
+                connection_state_clone,
+                clean_session_count_clone,
+            )
+            .await;
         });
 
         (
             Self {
                 client,
                 client_id,
                 router,
-                connected,
+                connection_state,
+                clean_session_count,
             },
             handle,
         )
@@ -66,12 +83,17 @@ impl RumqttcClient {
     async fn run_eventloop(
         mut eventloop: rumqttc::EventLoop,
         router: Arc<Mutex<MessageRouter>>,
-        connected: Arc<std::sync::atomic::AtomicBool>,
+        connection_state: watch::Sender<bool>,
+        clean_session_count: watch::Sender<u8>,
     ) {
         loop {
             match eventloop.poll().await {
-                Ok(rumqttc::Event::Incoming(rumqttc::Packet::ConnAck(_))) => {
-                    connected.store(true, std::sync::atomic::Ordering::SeqCst);
+                Ok(rumqttc::Event::Incoming(rumqttc::Packet::ConnAck(connack))) => {
+                    let _ = connection_state.send(true);
+                    if !connack.session_present {
+                        let next = clean_session_count.borrow().wrapping_add(1);
+                        let _ = clean_session_count.send(next);
+                    }
                 }
                 Ok(rumqttc::Event::Incoming(rumqttc::Packet::Publish(publish))) => {
                     // Route to matching subscriptions
@@ -90,12 +112,12 @@ impl RumqttcClient {
                     }
                 }
                 Ok(rumqttc::Event::Incoming(rumqttc::Packet::Disconnect)) => {
-                    connected.store(false, std::sync::atomic::Ordering::SeqCst);
+                    let _ = connection_state.send(false);
                 }
                 Ok(_) => {} // Other events (SubAck, PubAck, etc.)
                 Err(_e) => {
                     // Connection error - rumqttc will auto-reconnect on next poll
-                    connected.store(false, std::sync::atomic::Ordering::SeqCst);
+                    let _ = connection_state.send(false);
                     #[cfg(feature = "log")]
                     log::warn!("EventLoop error: {:?}", _e);
                 }
@@ -116,10 +138,16 @@ impl crate::mqtt::MqttClient for RumqttcClient {
     }
 
     fn wait_connected(&self) -> impl core::future::Future<Output = ()> {
-        let connected = self.connected.clone();
+        let mut rx = self.connection_state.subscribe();
         async move {
-            while !connected.load(std::sync::atomic::Ordering::SeqCst) {
-                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+            loop {
+                if *rx.borrow_and_update() {
+                    return;
+                }
+                if rx.changed().await.is_err() {
+                    // Sender dropped — eventloop task is gone, will never connect.
+                    return;
+                }
             }
         }
     }
@@ -155,6 +183,11 @@ impl crate::mqtt::MqttClient for RumqttcClient {
     ) -> impl core::future::Future<Output = Result<Self::Subscription<'_, N>, Self::Error>> {
         let client = self.client.clone();
         let router = self.router.clone();
+        let mut clean_session_count = self.clean_session_count.subscribe();
+        // Mark current value as seen so `changed()` only fires when the
+        // counter is bumped (i.e. CONNACK with session_present=false) after
+        // this subscribe.
+        let _ = clean_session_count.borrow_and_update();
         let topics_owned: [(&str, QoS); N] = *topics;
         async move {
             // Subscribe to all topics
@@ -183,6 +216,7 @@ impl crate::mqtt::MqttClient for RumqttcClient {
                 router,
                 sub_id,
                 topics: topic_strings,
+                clean_session_count,
             })
         }
     }
@@ -195,6 +229,7 @@ pub struct RumqttcSubscription<const N: usize> {
     router: Arc<Mutex<MessageRouter>>,
     sub_id: u64,
     topics: [String; N],
+    clean_session_count: watch::Receiver<u8>,
 }
 
 impl<const N: usize> MqttSubscription for RumqttcSubscription<N> {
@@ -205,7 +240,14 @@ impl<const N: usize> MqttSubscription for RumqttcSubscription<N> {
     type Error = rumqttc::ClientError;
 
     async fn next_message(&mut self) -> Option<Self::Message<'_>> {
-        self.receiver.recv().await
+        tokio::select! {
+            msg = self.receiver.recv() => msg,
+            // Bumped only on CONNACK with session_present=false (broker
+            // dropped our subs). Transient disconnects and session-resume
+            // reconnects don't trigger this. Caller treats `None` as
+            // recoverable: drop the subscription and resubscribe.
+            _ = self.clean_session_count.changed() => None,
+        }
     }
 
     async fn unsubscribe(self) -> Result<(), Self::Error> {