Skip to content

Commit d18c59c

Browse files
fix(mqtt source): forward publishes orphaned by session loss, document 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.
1 parent fd781c3 commit d18c59c

1 file changed

Lines changed: 385 additions & 22 deletions

File tree

0 commit comments

Comments
 (0)