From 4f20bd41784d75d3ed97a8e9b8be3ef48305d9e0 Mon Sep 17 00:00:00 2001 From: ashaffah Date: Fri, 26 Jun 2026 12:17:07 +0700 Subject: [PATCH] fix(mqtt): stop no longer leaks the shared connection driver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a shared MQTT connection was torn down while the broker was unreachable, the client's request queue could be full (inject kept publishing), so `disconnect().await` blocked. The node then missed the graceful window and was aborted before stopping the connection's driver task — which kept reconnecting and emitting errors, so Stop appeared to do nothing. Time-bound the graceful close (500ms) and always abort the driver after, so Stop reliably tears the connection down. Verified: Stop returns in ~0.5s and errors cease (previously hung 2s and kept erroring). --- src/runtime/nodes/mqtt_shared.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/runtime/nodes/mqtt_shared.rs b/src/runtime/nodes/mqtt_shared.rs index 1ec1ff6..355ea34 100644 --- a/src/runtime/nodes/mqtt_shared.rs +++ b/src/runtime/nodes/mqtt_shared.rs @@ -230,8 +230,15 @@ impl Shared { } }; if teardown { - self.close.publish(self.client.as_ref()).await; - self.client.disconnect().await; + // Best-effort graceful close, time-bounded: if the broker is + // unreachable the client's request queue can be full, making + // `disconnect()` block — never let that stall teardown. + let _ = tokio::time::timeout(Duration::from_millis(500), async { + self.close.publish(self.client.as_ref()).await; + self.client.disconnect().await; + }) + .await; + // Always stop the driver so it can no longer reconnect or emit errors. if let Some(handle) = self.driver.lock().unwrap().take() { handle.abort(); }