From 0f74ff00d0682a6f3e6063b1bcda14c3eb375d3f Mon Sep 17 00:00:00 2001 From: ashaffah Date: Mon, 29 Jun 2026 16:53:38 +0700 Subject: [PATCH] fix(mqtt): avoid driver deadlock when re-subscribing many filters On (re)connect the shared-connection driver re-subscribed every mqtt-in filter inline with `client.subscribe().await`. Each call only enqueues a request into a bounded channel (capacity 10) that is drained solely by the driver's own `poller.poll()`. Awaiting the subscribes inline meant the driver was not polling, so once the queue filled the next subscribe blocked forever, deadlocking the whole connection: no further subscribes completed and no publishes were sent. A flow with more than ~10 mqtt-in nodes on one shared broker therefore went completely silent on connect. Move the re-subscribe + birth publish into a detached task so the event loop keeps polling and the request queue drains. --- src/runtime/nodes/mqtt_shared.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/src/runtime/nodes/mqtt_shared.rs b/src/runtime/nodes/mqtt_shared.rs index a5b0e3e..ca7d5e0 100644 --- a/src/runtime/nodes/mqtt_shared.rs +++ b/src/runtime/nodes/mqtt_shared.rs @@ -164,16 +164,26 @@ async fn drive( loop { match poller.poll().await { Ok(PollEvent::Connected) => { + // Re-subscribe and publish birth, but NOT inline: each client + // request enqueues into a bounded channel that only this driver + // drains by polling. Awaiting many requests here (e.g. dozens of + // mqtt-in filters) fills the queue while the loop is not polling, + // which deadlocks the connection. Hand the work to a detached + // task so the event loop keeps being polled and the queue drains. let filters: Vec<(String, u8)> = subs .lock() .unwrap() .iter() .map(|s| (s.filter.clone(), s.qos)) .collect(); - for (filter, qos) in filters { - let _ = client.subscribe(&filter, qos).await; - } - birth.publish(client.as_ref()).await; + let client = client.clone(); + let birth = birth.clone(); + tokio::spawn(async move { + for (filter, qos) in filters { + let _ = client.subscribe(&filter, qos).await; + } + birth.publish(client.as_ref()).await; + }); } Ok(PollEvent::Message(m)) => { let subs = subs.lock().unwrap();