From b5c778b5c629f8ef1379ffc2431491bdf888b63c Mon Sep 17 00:00:00 2001 From: ashaffah Date: Mon, 29 Jun 2026 15:32:00 +0700 Subject: [PATCH] fix(mqtt-out): keep Stop responsive when a publish blocks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the broker is unreachable the shared client's request queue fills and `publish().await` blocks, so the node never returned to its select to see the shutdown signal — it was aborted before releasing the shared connection, leaking the driver task (errors kept coming after Stop). Race the publish against the shutdown signal so Stop interrupts a blocked publish, lets the node release the connection, and the driver is aborted. Verified: Stop returns in ~0.5s and mqtt errors cease (previously kept erroring after Stop). --- src/runtime/nodes/mqtt.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/src/runtime/nodes/mqtt.rs b/src/runtime/nodes/mqtt.rs index 9607c51..96a88d7 100644 --- a/src/runtime/nodes/mqtt.rs +++ b/src/runtime/nodes/mqtt.rs @@ -471,11 +471,17 @@ impl Node for MqttOutNode { status("error", "no_topic", Value::Null); continue; }; - if let Err(e) = shared - .publish(&topic, self.qos, self.retain, value_to_bytes(&msg.payload)) - .await - { - status("error", "publish_error", json!({ "error": e })); + // A publish blocks when the broker is down (the client's + // request queue fills). Race it against shutdown so Stop + // still works — otherwise the node is aborted before it + // can release the shared connection (driver leak). + tokio::select! { + _ = NodeCtx::shutdown_requested(&mut shutdown) => break, + r = shared.publish(&topic, self.qos, self.retain, value_to_bytes(&msg.payload)) => { + if let Err(e) = r { + status("error", "publish_error", json!({ "error": e })); + } + } } } None => break,