enhancement(mqtt source): support end-to-end acknowledgements by antaresindustries · Pull Request #25785 · vectordotdev/vector

antaresindustries · 2026-07-08T22:19:48Z

Summary

Makes the mqtt source participate in Vector's end-to-end acknowledgement
framework, so it can provide at-least-once delivery instead of best-effort.

Adds an acknowledgements option (SourceAcknowledgementsConfig, the standard
bool_or_struct form used by other sources) and flips can_acknowledge() to
true.
When acknowledgements are enabled, the connector sets
MqttOptions::set_manual_acks(true) so rumqttc stops auto-acking incoming
publishes.
Each incoming publish gets a BatchNotifier; the publish is registered with an
OrderedFinalizer, and the QoS 1 PUBACK is sent (client.ack(&publish))
only once the batch reaches BatchStatus::Delivered. On Errored/Rejected
the message is left unacked. Finalization is ordered (not just per-packet
independent) because MQTT requires PUBACKs to be sent in the order their
publishes were received (MQTT-4.6.0-2), the same reason the kafka source
uses ordered offset commits.
The event loop is polled and processed on a single task. Because that task
would otherwise risk deadlocking itself if it awaited client.ack/
client.subscribe directly (the same task also has to drain rumqttc's
outgoing request channel that those calls feed into), it uses the
non-blocking try_ack/try_subscribe variants plus a small retry queue
(PendingAcks) for anything that couldn't be sent immediately, and retries
resubscribing after a reconnect until it succeeds. This keeps polling and
processing coupled: when a downstream sink can't keep up, this task simply
stops polling, which throttles the broker via ordinary TCP backpressure —
no message is ever buffered or dropped locally, and no broker-specific
behavior is assumed anywhere.

Why

The source previously created a BatchNotifier but discarded the receiver and
auto-acked every message on arrival, so events could be lost if Vector or a
downstream sink failed before the write completed. Because the source already
uses clean_session = false and subscribes at QoS AtLeastOnce, deferring the
PUBACK lets the broker redeliver any unacknowledged message after a reconnect —
giving genuine at-least-once delivery. This is the work tracked in #21967.

A note on the design

An earlier attempt at this split polling and processing across two tasks, so
the main task could use blocking client.ack/client.subscribe calls without
needing the retry-queue machinery described above. That's a real simplification
in the ack/subscribe path, but it requires bounding a channel between the two
tasks that can't safely block without risking a deadlock — so it either drops
messages under sustained backpressure (relying on the broker to redelivered
them, which the MQTT spec only guarantees on reconnect, not on an open
connection) or grows memory unboundedly. Verified against a live EMQX broker
that forcing a reconnect to close that gap introduces a different, worse
failure mode (see the history on the linked PR for detail). Given the whole
point of acknowledgements is a delivery guarantee, this PR keeps the
single-task design: more code in the ack/subscribe path, but it never drops or
buffers a message locally under any downstream condition.

Behavior / compatibility

Default is unchanged (acknowledgements off → manual acks off → rumqttc
auto-acks exactly as before). Opt-in only.
At-least-once, not exactly-once: on redelivery, duplicate events are possible.
This matches the guarantee of other ack-capable sources.

Vector configuration

sources:
  mqtt_in:
    type: mqtt
    host: localhost
    port: 1883
    topic: my/topic
    acknowledgements: true

How did you test this PR?

cargo check/cargo clippy -D warnings/cargo fmt --check on
src/sources/mqtt/*.rs — clean.
Unit tests: cargo test -p vector --lib --no-default-features --features sources-mqtt sources::mqtt — all pass, including the PendingAcks
retry/backpressure matrix and protocol-state contract tests.
Integration tests against a real broker (EMQX, via cargo vdev int start mqtt / cargo vdev int test mqtt): all 3 tests pass, including
mqtt_redelivers_unacknowledged_messages, which forces a sink rejection to
prove a message is redelivered on reconnect. Reran the full suite 3 times —
stable, ~1s/run.
Generated component docs updated
(website/cue/reference/components/sources/generated/mqtt.cue gains the
standard acknowledgements block).
Changelog fragment added
(changelog.d/21967_mqtt_source_acknowledgements.enhancement.md).

Change Type

Is this a breaking change?

Yes
No

Does this PR include user facing changes?

Yes. Changelog fragment: changelog.d/21967_mqtt_source_acknowledgements.enhancement.md
No.

References

Closes: Handle Incoming::PubAck in the mqtt source #21967
Related: enhancement(mqtt source): support end-to-end acknowledgements #25770 (an alternative two-task design; superseded by this one for
the reasons above)

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 03cb2a37bf

ℹ️ 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".

antaresindustries · 2026-07-08T23:25:59Z

Thanks for the review — both findings were real and are fixed in 8e07a86 (replies inline on each thread with detail):

Ack ordering: after a non-Delivered finalization in the current connection generation, later publishes in that same generation are now also withheld from acking, not just the failed one, so PUBACK order is never violated.
finalizer.flush() race: replaced with dropping and recreating the finalizer/ack-stream pair on (re)connect, which fully closes the race instead of narrowing it.

Both covered by unit tests, and the full suite (unit + live integration against EMQX, 3 repeated runs) is green.

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 986ce0c24a

ℹ️ 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".

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

antaresindustries · 2026-07-09T23:22:10Z

Fixed all three new review findings in 4dcc85b:

Stale rumqttc-queued acks after disconnect — filtered Request::PubAck/Request::PubRec out of connection.pending alongside clearing Vector's own retry queue.
Resubscribe confirmed only by SUBACK — pending_resubscribe now survives a session_present = true reconnect if the previous attempt was never confirmed by a SUBACK.
Reconnect after withholding an ack — the first withheld ack in a generation now forces a graceful reconnect (try_disconnect), rate-limited to once per 30s so a persistent downstream failure can't thrash the connection. This one needed care: an earlier attempt at forcing a reconnect (on the now-closed enhancement(mqtt source): support end-to-end acknowledgements #25770) used an abrupt EventLoop::clean() and broke against a live EMQX broker. The graceful Disconnect approach here was verified against a live EMQX broker instead (new mqtt_forces_reconnect_after_withheld_ack test, rerun 4x stable) and shows no such issue.

All existing + new unit tests, clippy, and the full live-broker integration suite pass.

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 4dcc85b8f2

ℹ️ 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".

antaresindustries · 2026-07-10T17:33:56Z

03d857e fixes the latest review finding (a forced reconnect suppressed inside the cooldown window was skipped with no rescheduling path — now scheduled for cooldown expiry via a timer branch) plus six more issues found by a deeper audit targeting the same failure classes. All were verified against the rumqttc 0.24 source and, where applicable, a live EMQX broker:

Stale buffered events replayed across reconnects — rumqttc batches up to 10 incoming packets into state.events and clean() never clears it, so packets from a dead connection were yielded after the next ConnAck: a stale Publish got tagged with the new generation (its stale packet id then PUBACKed on the new connection — potential wrong-message ack on a fresh session), and a stale SubAck falsely confirmed the new connection's subscribe. Now cleared alongside the existing connection.pending scrub, which additionally drops a queued Request::Disconnect (previously replayed onto, and immediately killing, the next connection).
SUBACK failure return codes (0x80) treated as success — an ACL-denied subscription silently cleared pending_resubscribe, leaving the source connected and receiving nothing with no log and no retry. Return codes are now checked per topic ([MQTT-3.9.3-1] order), rejections logged at error level, and the (re)subscribe retried on a 30s pace.
Initial SUBSCRIBE untracked — the SUBACK-confirmation machinery only covered resubscribes; a lost initial SUBSCRIBE + session_present=true reconnect meant permanently subscribed-to-nothing. The initial subscribe is now issued on the first ConnAck through the same tracking (pending_resubscribe starts true).
Forced reconnect assumed the broker closes the socket after DISCONNECT — rumqttc never closes its own side and the server only SHOULD; a non-closing broker left ack_suppressed pinned forever. If no disconnect is observed within 10s of sending DISCONNECT, the connection is now dropped locally ([MQTT-3.14.4-2] puts that obligation on the client anyway). While a DISCONNECT is unanswered, nothing further is written ([MQTT-3.14.4-1]).
PUBACK order inversion — a newly finalized ack could be sent while earlier acks were still parked for retry; it now queues behind them ([MQTT-4.6.0-2]).
keep_alive = 0 with acknowledgements — now rejected at config build: with the heartbeat disabled, a silently dead connection is never detected on a quiet topic, so unacked messages would never be redelivered.

Verified: 20 unit tests (including new coverage for each state-machine change), clippy clean, and the full live-broker integration suite (4 tests including forced-reconnect redelivery) rerun 4x stable against EMQX 5.0.15.

Known limitations documented rather than changed: prolonged sink backpressure still blocks the poll loop past the keepalive deadline (self-healing via redelivery — inherent to the single-task backpressure design this PR deliberately keeps), and rumqttc's flush is not cancellation-safe under a full socket send buffer (upstream issue; worst case is a spurious reconnect).

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 03d857e8d8

ℹ️ 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".

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.

antaresindustries · 2026-07-10T20:06:58Z

3e45f16 adds two structural test layers aimed at ending the find-fix-review cycle, rather than waiting for the next round of findings:

Exhaustive state-machine model check (exhaustive_protocol_state_model_check): the protocol machinery is pure value types, so the test BFS-explores every reachable abstract state under every legal event interleaving (ConnAck × session_present, publishes, finalizations, SUBACK grant/reject, poll errors, all three timers), asserting on each transition the invariants that past review findings violated — a suppressed generation always has a recovery action armed, per-connection timers die with the connection, an unconfirmed subscription always has a live retry path, pending_resubscribe only clears on a fully granted SUBACK, and nothing is written after an unanswered DISCONNECT. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug from earlier rounds makes it fail immediately with a minimal counterexample trace.

Fault-injection integration tests: all prior integration coverage used client-side aborts with acknowledgements on. Three new tests force server-side disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2] — portable, no broker admin API): recovery with the session preserved, resubscribing after the broker discards the session entirely (the session_present = false path, confirmed by SUBACK against a real broker), and no message loss across a kick in the default auto-ack mode — the mode and disconnect direction that previously had zero coverage.

All 29 tests pass against live EMQX 5.0.15, rerun 5x stable.

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 507e548ba2

ℹ️ 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".

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 828844cc3f

ℹ️ 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".

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.

chatgpt-codex-connector

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: dd8f55f298

ℹ️ 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".

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

MQTT requires a client to send PUBACK (QoS 1) and PUBREC (QoS 2) packets in the order the corresponding PUBLISH packets were received (MQTT-4.6.0-2). UnorderedFinalizer lets a later batch's ack race ahead of an earlier one still pending delivery, violating this. Switch to OrderedFinalizer, same as the kafka source's ordered offset commits: a slow/stuck earlier batch now holds back acks for publishes received after it, trading a little head-of-line latency for protocol compliance.

…izer on reconnect Addressed PR review findings: - Once a publish in the current connection generation finalizes as anything other than Delivered, withhold acks for every later publish in that same generation too, not just the failed one. MQTT requires PUBACKs to be sent in the order their publishes were received (MQTT-4.6.0-2); acking a later packet while an earlier one was never acked would violate that order. The withheld ones are redelivered the same way the one that failed already is. - Replace finalizer.flush() with recreating the finalizer/ack-stream pair on (re)connect. flush() only clears entries already pulled into its internal ordered set; an entry added concurrently (via finalizer.add) that the finalizer's background task hadn't yet picked up survives the flush and gets pushed into the fresh set anyway. Since OrderedFinalizer won't yield anything newer until that stale entry resolves, it could hold back acks for the new generation. Dropping the old finalizer outright removes the race: its background task sees the channel closed and exits, taking any not-yet-picked-up entries with it.

…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.

… holes and cross-layer staleness One fix for the latest review finding, plus six more from a deeper audit of the same failure classes (states whose required action has no wake-up path, Vector state diverging from rumqttc's internals, and locally-queued mistaken for peer-confirmed): - A withheld-ack forced reconnect landing inside the cooldown window is now scheduled for cooldown expiry instead of skipped: `just_suppressed` fires only once per generation, so a skipped reconnect never got another trigger and the source stayed connected (and stuck) indefinitely. - On disconnect, clear rumqttc's buffered `state.events`: `clean()` doesn't, so packets batched from the dead connection were yielded after the next ConnAck — a stale Publish got tagged with the new generation (its stale packet id then acked on the new connection), and a stale SubAck falsely confirmed the new connection's subscribe. Also drop a queued `Request::Disconnect` from `connection.pending`, which would otherwise be replayed onto (and immediately kill) the next connection. - SUBACK failure return codes (0x80, e.g. an ACL denial) no longer count as subscribed: the rejected topic is logged at error level (return codes map to topics by order, [MQTT-3.9.3-1]) and the (re)subscribe is retried on a 30s pace instead of being silently abandoned. - The initial SUBSCRIBE now goes through the same SUBACK-confirmation tracking as re-subscribes (issued on the first ConnAck, driven by `pending_resubscribe` starting true). Previously a lost initial SUBSCRIBE followed by a session_present reconnect left the source permanently subscribed to nothing. - The forced reconnect no longer assumes the broker closes the socket after our DISCONNECT (rumqttc never closes its side; the server only SHOULD): if no disconnect is observed within 10s, drop the connection locally, which [MQTT-3.14.4-2] requires of the client anyway. While a DISCONNECT is unanswered, nothing further is written to the connection ([MQTT-3.14.4-1]). - A newly finalized PUBACK now queues behind earlier acks still parked for retry instead of potentially jumping ahead of them ([MQTT-4.6.0-2]). - `keep_alive = 0` is rejected when acknowledgements are enabled: with the heartbeat disabled, a silently dead connection is never detected on a quiet topic, so unacked messages would never be redelivered. Verified against a live EMQX broker: all 4 integration tests (including the forced-reconnect redelivery test) rerun 4x stable; 20 unit tests cover the new state transitions.

…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.

…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.

… ack mode The disconnect-time scrub of rumqttc's buffered `state.events` decided per-mode: manual-ack mode dropped every buffered event, auto-ack mode kept publishes. But acks-on traffic isn't uniformly redeliverable: a publisher sending QoS 0 (explicitly forwarded, with a warning) has no packet id, no acknowledgement, and no session redelivery, so a buffered QoS 0 publish hit by the manual-ack clear was lost permanently. Replace the per-mode branches with a single per-event rule (`keep_buffered_event_after_disconnect`): keep exactly what would be lost outright if dropped. QoS 0 publishes are kept in both modes (never redelivered, and inert to the ack machinery: they are never registered with the finalizer, so retaining them across the generation boundary can't leak a stale packet id). QoS 1/2 publishes are kept only in auto-ack mode (their ack may already be on the wire, in which case the broker never redelivers) and dropped in manual-ack mode (guaranteed unacked, so the resumed session redelivers them). All non-publish events are dropped in both modes (a stale SubAck would falsely confirm the new connection's subscribe).

…ess of how it's enabled The generated docs for MQTT's (deprecated) source-level `acknowledgements` setting inherited the generic claim that "enabling or disabling acknowledgements at the source level has no effect on acknowledgement behavior." That's false here: `MqttSourceConfig::build` passes the resolved value through to `set_manual_acks` and the finalizer path, and `SourceContext::do_acknowledgements` resolves source-level > global > sink-driven, in that precedence, so an explicit source-level value can silently override (including disable) acknowledgements a connected sink requires. Following the precedent already used by `windows_event_log` (a field-level doc comment overriding the shared type's generic description rather than editing the shared description used by every source), added a doc comment on MQTT's `acknowledgements` field: it now correctly states that the setting takes effect however it ends up enabled, and restates the stable `client_id` requirement so it's visible to users relying on global/sink-level enablement too, not just those who set the field directly. Docs regenerated with cue v0.16.1 (the version pinned in CI's setup action) to avoid unrelated cue-fmt drift from a newer local install.

…connection.pending Review flagged a queued PUBCOMP surviving a disconnect into the next connection's `connection.pending`, where it could complete the wrong QoS 2 exchange. Verified against the vendored rumqttc 0.24 source: `Request::PubComp` is never actually constructed by this rumqttc version (incoming PUBREL is answered by writing PubComp bytes directly into `state.write`, bypassing the Request channel entirely), so the specific mechanism described can't occur here today. But the underlying gap is real and bigger than PubComp: `rumqttc::Request` has 12 variants, and the scrub only named 3 (PubAck, PubRec, Disconnect) via `!matches!(request, A | B | C)` -- a block-list where every other variant, including PubComp, silently passed through and was kept. That's the opposite default from the sibling `state.events` scrub a few lines below, which already drops anything not explicitly allow-listed. Replaced the block-list with `keep_pending_request_after_disconnect`, an exhaustive match with no wildcard arm covering all 12 variants, each with its own rationale (drop: PubAck/PubRec/PubComp/Disconnect/Publish/PubRel/ Unsubscribe/SubAck/UnsubAck; keep: Subscribe/PingReq/PingResp -- harmless to replay). No wildcard means a future rumqttc upgrade adding a 13th variant fails to compile here until someone decides where it belongs, rather than a wildcard silently picking a default -- verified by temporarily deleting one arm and confirming the exact "pattern not covered" compile error. Added a 12-row unit test matrix pinning down today's bucketing as living documentation alongside the exhaustive match.

…t and test the rumqttc reconnect seam Review flagged that the disconnect-time scrub of rumqttc's buffered `state.events` removed manual-ack QoS 1/2 publishes before the next reconnect's session outcome was known: if that ConnAck reports `session_present = false`, the broker has forgotten those publishes too, so the scrub was discarding the only remaining copy. Fix: `partition_buffered_events` pulls unprocessed manual-ack QoS 1/2 publishes out of the buffer at disconnect, tagged with the (now permanently stale) generation they arrived under. The next ConnAck resolves them: on a resumed session they're dropped (the broker redelivers them under the same packet ids -- forwarding ours too would duplicate); on a fresh session they're forwarded as the only copy left, still under their old generation, so `finalize_publish`'s existing generation check guarantees no ack is ever attempted with the stale packet id. Orphans accumulate correctly across repeated failed reconnect attempts and resolve exactly once. Investigated live before landing: the extraction path never fired across any run against EMQX, including bursts fired immediately before a session-discarding kick. Traced why in rumqttc 0.24: `select()` drains `state.events` before any network read, `readb()` cannot fail once it has parsed a packet in a call, and `Network::flush` is a no-op on an empty write buffer -- so in manual-ack mode (no ack written at receipt) the buffered-at-error state needs an incoming PUBREL, i.e. a nonconformant broker delivering QoS 2 to our QoS 1 subscription, or our own escalation teardown of a live connection. The fix is kept as defensive depth for those paths; its trigger is far narrower than general QoS 1 traffic. Since every recent review finding probed this same seam one field at a time, close the whole seam instead of the one field: - Documented the complete inventory of state rumqttc carries across a reconnect on `handle_connection_lost` -- what rumqttc resets, what we scrub, what persists but is safe by construction (e.g. `state.collision`, a PUBACK-starvation tripwire if a publish path is ever added), and the residual risks that cannot be closed at this layer (the private `Network.read` buffer on session loss, upstream flush cancellation-unsafety, the auto-ack duplicate window). - Added a seam unit test (`connection_lost_scrubs_the_rumqttc_seam`) that builds a real `rumqttc::EventLoop` with no network, stuffs `state.events` and `pending` with exactly the states past findings probed, runs the real handler, and asserts every postcondition in both ack modes -- making states deterministically testable that live-broker tests cannot reliably reach. Plus a mutation-verified unit matrix for the partition function.

mrmarbury · 2026-09-24T10:21:34Z

Is there a timeline when this will land on main? we have big production issues due to the subscriptions not being renewed.

antaresindustries · 2026-09-25T04:12:49Z

I've been running a fork of this for the past few months and haven't found any issues yet, so let me resolve conflicts so this will be ready for merge

…nowledgements # Conflicts: # src/sources/mqtt/config.rs # src/sources/mqtt/integration_tests.rs

…nowledgements

These hunks were previously reverted by hand as suspected nightly-rustfmt drift, but `vdev check fmt` (`cargo fmt -- --check` at the workspace root, edition 2024) flags them, so CI's format check would fail without them. Pure `cargo fmt` output; no code changes.

…ence the shared schema Upstream (vectordotdev#26669) changed the docs generator to recognize boolean-or-object fields by the fully qualified `crate::serde::bool_or_struct` path and converted every existing source; this branch's new `acknowledgements` field still used the bare import, so the merge kept a generated reference to the pre-rename schema key (`vector_core::config::SourceAcknowledgementsConfig`), which no longer exists and fails the website CUE build. Matches the form kafka and windows_event_log now use.

chatgpt-codex-connector · 2026-10-16T06:46:47Z

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-19T18:25:13.956885Z	`c5dd596`	New commits

ℹ️ 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" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

…nowledgements Conflict in src/sources/mqtt/source.rs: upstream vectordotdev#26549 changed `MqttSource::run` to return `Result<(), SourceError>` and reworked the original subscribe-then-poll loop body (vectordotdev#26705). This branch replaces that body entirely (subscribe is issued on ConnAck with SUBACK confirmation), so both hunks keep this branch's side; the new signature and `SourceError` import from upstream are kept. Workspace-wide pedantic Clippy (vectordotdev#26704) flagged this branch's new code: mechanical fixes applied (doc backticks, unit patterns in `select!`, trailing semicolons, `Bytes::default()`, `Duration::from_mins`, a renamed test binding), and the structural lints are allowed with reasons following upstream's convention (`struct_excessive_bools` on two flag structs, `match_same_arms` on the rationale-grouped request scrub, `too_many_lines` on `run` and the end-to-end seam test).

antaresindustries requested review from a team as code owners July 8, 2026 22:19

github-actions Bot added docs review on hold The documentation team reviews PRs only after a PR is approved by the COSE team. domain: sources Anything related to Vector's sources domain: external docs Anything related to Vector's external, public documentation labels Jul 8, 2026

chatgpt-codex-connector Bot reviewed Jul 8, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs Outdated

Comment thread src/sources/mqtt/source.rs Outdated

antaresindustries mentioned this pull request Jul 9, 2026

enhancement(mqtt source): support end-to-end acknowledgements #25770

Closed

4 of 9 tasks

antaresindustries force-pushed the mqtt-source-e2e-acknowledgements branch from 8e07a86 to 986ce0c Compare July 9, 2026 16:50

chatgpt-codex-connector Bot reviewed Jul 9, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs Outdated

Comment thread src/sources/mqtt/source.rs

Comment thread src/sources/mqtt/source.rs Outdated

chatgpt-codex-connector Bot reviewed Jul 9, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs Outdated

chatgpt-codex-connector Bot reviewed Jul 10, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs Outdated

Comment thread src/sources/mqtt/source.rs

antaresindustries force-pushed the mqtt-source-e2e-acknowledgements branch from 3e45f16 to 69760e6 Compare July 13, 2026 17:58

antaresindustries force-pushed the mqtt-source-e2e-acknowledgements branch from 69760e6 to 507e548 Compare July 14, 2026 23:44

chatgpt-codex-connector Bot reviewed Jul 14, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs Outdated

antaresindustries force-pushed the mqtt-source-e2e-acknowledgements branch from 18ca0f1 to 459e749 Compare July 24, 2026 00:18

chatgpt-codex-connector Bot reviewed Aug 20, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs Outdated

antaresindustries force-pushed the mqtt-source-e2e-acknowledgements branch from 0f51e11 to dd8f55f Compare August 24, 2026 17:13

chatgpt-codex-connector Bot reviewed Aug 24, 2026

View reviewed changes

Comment thread src/sources/mqtt/source.rs

antaresindustries added 12 commits August 25, 2026 20:38

add more tests to all possible state transitions

7daf3f5

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

antaresindustries force-pushed the mqtt-source-e2e-acknowledgements branch from dd8f55f to d18c59c Compare August 26, 2026 03:40

antaresindustries added 4 commits September 24, 2026 21:34

Merge remote-tracking branch 'origin/master' into mqtt-source-e2e-ack…

2ee7026

…nowledgements # Conflicts: # src/sources/mqtt/config.rs # src/sources/mqtt/integration_tests.rs

Merge remote-tracking branch 'origin/master' into mqtt-source-e2e-ack…

39ce060

…nowledgements

Conversation

antaresindustries commented Jul 8, 2026

Summary

Why

A note on the design

Behavior / compatibility

Vector configuration

How did you test this PR?

Change Type

Is this a breaking change?

Does this PR include user facing changes?

References

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

Uh oh!

antaresindustries commented Jul 8, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

antaresindustries commented Jul 9, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

antaresindustries commented Jul 10, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

Uh oh!

antaresindustries commented Jul 10, 2026

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

chatgpt-codex-connector Bot left a comment

Choose a reason for hiding this comment

💡 Codex Review

Uh oh!

Uh oh!

mrmarbury commented Sep 24, 2026

Uh oh!

antaresindustries commented Sep 25, 2026

Uh oh!

chatgpt-codex-connector Bot commented Oct 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codex Review Summary

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

chatgpt-codex-connector Bot commented Oct 16, 2026 •

edited

Loading