fix(rust): bound queued incoming QoS 0 publishes - #1437
fix(rust): bound queued incoming QoS 0 publishes#1437hugues bouvier (huguesBouvier) wants to merge 1 commit into
Conversation
Mirror microsoft/rust-mqtt-client#96 so excess incoming QoS 0 publishes are dropped at a configurable queue limit without blocking MQTT connection progress. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
This PR mirrors the upstream fix from the vendored azure_mqtt client into this repo’s Rust MQTT SDK to prevent unbounded memory growth when applications stop draining incoming QoS 0 publishes. It introduces a configurable permit budget to cap queued incoming QoS 0 messages while preserving existing QoS 1/2 behavior.
Changes:
- Added a new
ClientOptions::incoming_qos0_queue_sizesetting and used atokio::sync::Semaphoreto bound queued QoS 0 publishes. - Updated the internal incoming-publish channel payload to carry a permit for QoS 0 so capacity is recovered when the queued item is dropped.
- Added integration tests covering QoS 0 overflow/drop + capacity recovery and QoS 1 non-regression.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| rust/azure_iot_operations_mqtt/tests/low_level_receive_queue.rs | Adds tests validating QoS 0 overflow dropping/recovery and QoS 1 behavior. |
| rust/azure_iot_operations_mqtt/src/azure_mqtt/client/session.rs | Enforces the QoS 0 permit budget when handling incoming publishes. |
| rust/azure_iot_operations_mqtt/src/azure_mqtt/client/channel_data.rs | Extends QoS 0 incoming publish variant to hold the semaphore permit. |
| rust/azure_iot_operations_mqtt/src/azure_mqtt/client.rs | Adds the new option, initializes the semaphore, and updates receiver conversion. |
| rust/azure_iot_operations_mqtt/src/azure_mqtt_adapter.rs | Uses ..Default::default() to populate the newly added client option field. |
| // Copyright (c) Microsoft Corporation. | ||
| // Licensed under the MIT License. | ||
|
|
||
| #![cfg(feature = "test-utils")] |
|
CI note: this PR originates from a fork, so GitHub does not expose Validation completed outside those secret-dependent jobs:
A maintainer-owned branch or trusted rerun with repository secrets is needed to execute the remaining integration matrix. |
|
Discussion on this issue has moved to the dedicated MQTT client repo |
Summary
azure_mqttmoduleContext
The low-level incoming application-message channel is unbounded. QoS 1/2 traffic is constrained by MQTT Receive Maximum and packet identifiers, but QoS 0 has no protocol-level flow control. If an application stops draining the low-level receiver, QoS 0 publishes can accumulate until the process runs out of memory.
This was found while investigating Azure-NBC PR 16518887. A physically bounded shared channel would either block keepalive/PUBACK/disconnect processing or risk dropping QoS 1. Instead, each queued QoS 0 publish owns a permit; overflow QoS 0 is discarded while the event loop continues running.
Validation
make checkMANIFEST=azure_iot_operations_mqtt/Cargo.toml make test