Close transport on broker server disconnect

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add request problem information support
 - Add reason string to incoming PUBACK, PUBREC, PUBREL, PUBCOMP, SUBACK, UNSUBACK
 - Detect protocol error when server sends user properties (only when `MAX_USER_PROPERTIES` > 0) or a reason string in the PUBACK, PUBREC, PUBREL, PUBCOMP, SUBACK and UNSUBACK packets
+- Fix client not dropping network connection on receival of a DISCONNECT packet leading to a panic on subsequent call to `Client::abort` in debug builds
 
 ## 0.5.1 - 2026-04-10
 

src/client/mod.rs

@@ -1598,6 +1598,10 @@ impl<
                     .recv_body::<DisconnectPacket<MAX_USER_PROPERTIES>>(&header)
                     .await?;
 
+                // The server initiated the disconnect. We must close the transport on our side
+                // as well so that subsequent error handling (e.g. `abort`) sees a non-Ok network state.
+                self.raw.close_with(None);
+
                 return Err(MqttError::Disconnect {
                     reason: disconnect.reason_code,
                     reason_string: disconnect.reason_string.map(Property::into_inner),

tests/integration/connect_options.rs

@@ -595,6 +595,9 @@ async fn keep_alive_not_kept_alive_idle_network() {
             server_reference: _,
         } | MqttError::Network(_)
     ));
+
+    // A non-recoverable poll error should always allow calling abort without panicking.
+    c.abort().await;
 }
 
 #[tokio::test]
@@ -635,15 +638,21 @@ async fn keep_alive_not_kept_alive_incoming_qos0() {
 
         assert_ok!(
             timeout(Duration::from_secs(4), async {
-                loop {
-                    if let Err(MqttError::Network(_)) = rx.poll().await {
-                        break;
-                    }
-                }
+                while !matches!(
+                    rx.poll().await,
+                    Err(MqttError::Disconnect {
+                        reason: ReasonCode::KeepAliveTimeout,
+                        reason_string: _,
+                        user_properties: _,
+                        server_reference: _,
+                    }) | Err(MqttError::Network(_))
+                ) {}
             })
             .await,
             "expected to be disconnected"
         );
+        // A non-recoverable poll error should always allow calling abort without panicking.
+        rx.abort().await;
     };
 
     join!(receiver, publisher);
@@ -695,6 +704,8 @@ async fn keep_alive_not_kept_alive_will_timing() {
             server_reference: _,
         } | MqttError::Network(_)
     ));
+    // A non-recoverable poll error should always allow calling abort without panicking.
+    tx.abort().await;
 
     disconnect(&mut rx, DEFAULT_DC_OPTIONS).await;
 }