enhancement(mqtt source): support end-to-end acknowledgements#25770
enhancement(mqtt source): support end-to-end acknowledgements#25770antaresindustries wants to merge 5 commits into
Conversation
The mqtt source now participates in Vector's end-to-end acknowledgement framework. When acknowledgements are enabled, the QoS 1 PUBACK is deferred via an OrderedFinalizer until the events are delivered to all sinks; with the source's existing clean_session=false + QoS AtLeastOnce subscription, unacknowledged messages are redelivered after a reconnect (at-least-once). Closes vectordotdev#21967
trying to get ahead of lots of smaller comments on how this works by moving to more state machine like helpers into more dedicated structs
this removes much of the tracing by storing state within the task
…void unbounded growth The poller task forwarded events to the main task over an unbounded channel, so a stalled downstream sink let the backlog grow without limit. Switch to a bounded channel and drop events on saturation instead: a dropped QoS 1/2 publish was never acknowledged, so the broker's own retry timer redelivers it independently of any reconnect. Forcing a reconnect on saturation was tried first but causes a reconnect storm, since the broker immediately refloods the freshly reconnected client from the same backlog. Also raises the mqtt integration test broker's max_mqueue_len/max_inflight and shortens retry_interval so the new saturation-recovery test can actually reach and exercise this path against a real broker.
github-actions
Bot
added
domain: sources
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e18dc1000a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Err(mpsc::error::TrySendError::Full(_)) => warn_event_backlog_saturated(), | ||
| Err(mpsc::error::TrySendError::Closed(_)) => break, | ||
| } | ||
| } else if guaranteed_tx.send(event).is_err() { |
There was a problem hiding this comment.
When acknowledgements are disabled (the default) or the broker delivers QoS 0, safe_to_drop_under_backlog is false, so every publish goes through this unbounded sender. Because the poller is now independent of out.send_batch, a slow or stalled downstream lets the poller continue reading full payloads while guaranteed_rx grows without limit; the previous single loop stopped polling while send_batch was blocked. Under normal MQTT traffic this can OOM the process, so non-droppable publishes still need backpressure rather than an unbounded queue.
Useful? React with 👍 / 👎.
| client.ack(&entry.publish).await.map_err(|_| ())?; | ||
| } | ||
| }, | ||
| mqtt_event = droppable_rx.recv() => { |
There was a problem hiding this comment.
When a persistent session already has queued QoS 1 messages, the poller can enqueue the ConnAck on guaranteed_rx and publishes on droppable_rx before this select runs. Since the select may take the droppable branch first, those publishes are processed with connection_generation == 0 and connected == false; their finalizers are then skipped by should_ack_finalized_publish, leaving delivered messages unacked and forcing redelivery or broker inflight stalls even though the ConnAck was available. The connection-state events need to be processed before publish events, or the split queues need to preserve protocol order.
Useful? React with 👍 / 👎.
| if safe_to_drop_under_backlog(&event, acknowledgements) { | ||
| match droppable_tx.try_send(event) { | ||
| Ok(()) => {} | ||
| Err(mpsc::error::TrySendError::Full(_)) => warn_event_backlog_saturated(), |
There was a problem hiding this comment.
When acknowledgements are enabled and the downstream backlog fills, this path drops the only Vector-side copy but keeps the MQTT connection open. Brokers are only required to resend unacknowledged QoS > 0 publishes after a reconnect (and MQTT 5 forbids same-connection resends), so brokers without EMQX-style retry timers will keep waiting for a PUBACK/PUBREC that Vector can no longer send; once the broker's in-flight window fills, delivery can stall indefinitely. The saturation path needs to force a reconnect or retain the publish until it can be processed.
Useful? React with 👍 / 👎.
| // disabled (rumqttc auto-acks in that mode). MQTT PUBACKs are independent | ||
| // per packet id (unlike Kafka offsets), so finalization is unordered — a | ||
| // slow/stuck batch must not hold back acks for already-delivered publishes. | ||
| let (finalizer, mut ack_stream) = UnorderedFinalizer::<FinalizerEntry>::maybe_new( |
There was a problem hiding this comment.
With acknowledgements enabled, using UnorderedFinalizer lets a later batch send its PUBACK/PUBREC before an earlier publish whose downstream delivery is still pending. MQTT requires clients to acknowledge QoS 1/2 publishes in the order they were received, so a slow earlier sink batch can make this source emit out-of-order protocol acknowledgements; use an ordered finalizer or otherwise hold later MQTT acks until prior publishes have finalized.
Useful? React with 👍 / 👎.
| if let Some((status, entry)) = entry | ||
| && protocol_state.should_ack_finalized_publish( | ||
| status, | ||
| entry.connection_generation, | ||
| ) | ||
| { | ||
| client.ack(&entry.publish).await.map_err(|_| ())?; |
There was a problem hiding this comment.
If the poller has already observed a connection error and queued Disconnect, this branch can still win the select first while protocol_state.connected and connection_generation still describe the old connection. A batch finalized in that window queues a PUBACK/PUBREC for a packet id from the dead connection, and rumqttc can send it after reconnecting, including after a fresh session where that acknowledgement is unsolicited. Process pending connection-state events before finalizer completions, or mark the state disconnected as soon as the poller reports the error.
Useful? React with 👍 / 👎.
…reconnect escalation Addressed PR review findings: - Merge the droppable/guaranteed event channels back into one, preserving the order events were polled in. Splitting into two channels (from the previous commit) broke this: tokio::select! doesn't guarantee relative order between two channels when both have items ready, so a backlogged publish could reach the main task before a ConnAck that was actually polled first, letting ProtocolState be acted on while stale. - Switch UnorderedFinalizer to OrderedFinalizer: PUBACKs must be sent in the order their publishes were received (MQTT-4.6.0-2), same as the kafka source's ordered offset commits. - Drop `biased` on the main select: it fixed a narrow stale-ack race but starves the ack-stream branch entirely under sustained publish volume, since the events channel being ready again next iteration is the common case, not the exception. Fair random selection keeps that race narrow instead of guaranteeing worse behavior under load. Also attempted, then reverted, a fix for the remaining gap (dropped messages rely on broker-specific retry-while-connected behavior, which isn't spec- guaranteed): forcing a reconnect once the backlog stays saturated long enough. Verified against a live EMQX broker that this triggers a broker-side `unexpected_sock_close`/`einval` error after which the session stops delivering anything further at all -- worse than the problem it was meant to solve. Reverted to plain dropping and documented the limitation in code, since forcing a *correct* reconnect needs more investigation than fits here.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 295993ef7e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| entry.connection_generation, | ||
| ) | ||
| { | ||
| client.ack(&entry.publish).await.map_err(|_| ())?; |
There was a problem hiding this comment.
Even with the new OrderedFinalizer, this branch only filters for BatchStatus::Delivered: if publish N finalizes as Errored/Rejected and publish N+1 is delivered while the connection stays up, N is skipped and this line still sends the PUBACK/PUBREC for N+1. MQTT requires acknowledgements to follow receive order (MQTT-4.6.0-2), so a normal downstream rejection can leave an earlier packet unacknowledged while later packets are acknowledged on the same session; brokers may then stall or treat the client as non-conformant. Once an earlier finalizer is not acked, later acks should be held until reconnect or the connection should be closed.
Useful? React with 👍 / 👎.
| if actions.flush_finalizer | ||
| && let Some(finalizer) = finalizer | ||
| { | ||
| finalizer.flush(); |
There was a problem hiding this comment.
This flush() only drops entries that the FinalizerSet task has already moved into its internal ordered future set; entries that finalizer.add sent on the unbounded pending-entry channel but the task has not yet pulled survive the reconnect flush. If a connection drops right after forwarding a publish, that stale receiver can later be inserted ahead of publishes from the new connection, and because OrderedFinalizer waits in insertion order, a slow or never-finalized old batch can hold all later PUBACK/PUBRECs even though the old packet IDs were meant to be discarded.
Useful? React with 👍 / 👎.
| // entry to avoid pinning payloads in memory while sinks process. | ||
| if let Some((finalizer, receiver)) = finalizer.zip(batch_receiver) { | ||
| publish.payload = Default::default(); | ||
| finalizer.add( |
There was a problem hiding this comment.
When the broker retries the same QoS 1/2 PUBLISH before the first copy finalizes, this registers another finalizer with the same packet id and connection generation. Both copies can later satisfy should_ack_finalized_publish, so after the first PUBACK/PUBREC completes the broker flow, a later stale ack for the duplicate may collide with that packet id if the broker has already reused it for another in-flight publish. Track pending packet ids and avoid adding a second deferred ack for retransmissions of the same broker message.
Useful? React with 👍 / 👎.
|
Thanks for the review — the codex findings here were valuable and caught real bugs (ConnAck/publish reordering across the split channels, the stale-ack race, and PUBACKs not being sent in receipt order per [MQTT-4.6.0-2]). All three are fixed in the latest commit here. The remaining finding (dropping a message under backlog saturation relies on broker-specific retry-while-connected behavior, since the MQTT spec only guarantees redelivery on reconnect, not on an already-open connection) led somewhere I want to flag directly rather than paper over. I attempted the "obvious" fix: force a reconnect once the backlog stays saturated for too long, rate-limited to avoid a reconnect storm. Verified against a live EMQX broker, this actually made things worse — the non-graceful reconnect ( Sitting back from that, I don't think this branch's core trade-off is the right one for a feature whose whole point is a delivery guarantee. The single-task design (#25666's approach, which this branch replaced) never drops an incoming publish at all — when downstream can't keep up, it just stops polling, which throttles the broker via ordinary TCP backpressure, with no broker-specific assumptions anywhere. This branch's poller/main-task split enables blocking I'm going back to the #25666 approach as the final implementation (carrying forward the |
|
Closing in favor of #25785, which goes back to the single-task design (see the discussion on this PR for why: the two-task split here needs to either drop messages under sustained backlog or grow memory unboundedly, and closing that gap properly turned out to be harder than expected — a forced-reconnect attempt broke against a live EMQX broker). #25785 has since had its own review pass and fixes for the ack-ordering and finalizer-flush issues found there. |
1 participant
Summary
Supersedes #25666 with a revised implementation of the same feature (end-to-end
acknowledgements for the
mqttsource, closing #21967). The coreacknowledgementsoption,manual_acks/deferred-PUBACK mechanism, and at-least-onceguarantee are unchanged from #25666. What's different is how the MQTT event loop
is polled and how backpressure from a slow downstream sink is handled:
connection.poll()and process events (including awaiting
out.send_batch()). Because that taskalso has to drain rumqttc's own outgoing request channel (used by
client.ack()/client.subscribe()), enhancement(mqtt source): support end-to-end acknowledgements #25666 needed non-blockingtry_ack/try_subscribeplus a hand-rolled retry queue (PendingAcks) andpending_resubscribestate to avoid deadlocking itself. This PR splitspolling onto its own task, so the main task can just
client.ack().await/client.subscribe().awaitdirectly — no retry-queue machinery needed.Decoupling polling from processing means a slow downstream sink could let the
poller-to-main-task backlog grow without bound. Events are split by whether
dropping them is protocol-safe: a QoS 1/2 publish that will actually be
deferred-acked goes through a bounded channel and is dropped (with a warning)
if the backlog saturates — safe, because the broker still considers it
unacknowledged and redelivers it via its own retry timer. Everything else
(ConnAck/SubAck/Disconnect, and any publish that wouldn't get broker
redelivery — acknowledgements disabled, or QoS 0) goes through a separate
unbounded channel and is never dropped, since losing one of those would
either desync the connection's protocol state or be true unrecoverable data
loss.
instead of dropping. That reintroduces the same problem immediately: the
broker refloods the freshly reconnected client from the same backlog, so
under sustained backpressure it becomes a reconnect storm rather than
actually recovering. Verified against a real broker with a large inflight
window before settling on drop-and-redeliver.
Net effect: fewer lines than #25666's single-task approach despite the extra
task and channel-splitting, no retry-queue state machine, and it's now backed
by a dedicated test (
mqtt_recovers_from_downstream_saturation) that provesthe backlog-saturation path actually recovers instead of deadlocking or losing
data outside the one case where loss is recoverable.
Vector configuration
How did you test this PR?
cargo check/cargo clippy -D warnings/cargo fmt --checkonsrc/sources/mqtt/*.rs— clean.cargo test -p vector --lib --no-default-features --features sources-mqtt sources::mqtt— all pass, including a newprotocol_contract_matrix_for_backlog_drop_safetytest covering everyacknowledgements/QoS combination's drop-safety.
cargo vdev int start mqtt/cargo vdev int test mqtt): all 4 tests pass, including the newmqtt_recovers_from_downstream_saturation, which stalls the sink, floods thebacklog past capacity, and asserts (a) the source recovers instead of
hanging and (b) at least one duplicate was observed, so the pass is pinned
on the drop-and-redeliver path actually running rather than incidental
buffering. Reran the new test repeatedly (stable, ~5s/run) and the full
suite together.
max_mqueue_len/max_inflight/retry_intervalare raisedin
tests/integration/mqtt/config/compose.yaml(test env only) so the newtest can actually reach the source's own backlog capacity instead of being
bottlenecked by the broker's defaults first.
Change Type
Is this a breaking change?
Default is unchanged (
acknowledgementsoff → manual acks off → rumqttcauto-acks exactly as before). Opt-in only.
Does this PR include user facing changes?
changelog.d/21967_mqtt_source_acknowledgements.enhancement.md(carried over from enhancement(mqtt source): support end-to-end acknowledgements #25666; describes the shipped behavior, which is
unchanged by this PR's internal rework).
References
Incoming::PubAckin themqttsource #21967Notes
This branch was rebased onto current
masterand force-pushed before openingthis PR (no review had started on it yet, so this is the one time that's
appropriate per the contributor notes below).