fix(mqtt): do not advance the inbox watermark on out-of-order PUBACK — fixes #286

[ImDanXie]

Baseline: main @ eef5e3af (the PR's base commit).

Summary

confirm() unconditionally called onConfirm(confirmingMsg.seq) with the passed-in message, even when the drain loop removed nothing because the head of unconfirmedPacketIds was still un-acked. A client acknowledging packetId 2 while packetId 1 is in flight (the spec places no ordering requirement on PUBACK) would then:

1. break the drain loop on its first iteration with zero removals (confirmingMsg still refers to the parameter),
2. call onConfirm(secondMsg.seq) anyway,
3. in MQTTPersistentSessionHandler, clear stagingBuffer.headMap(secondSeq, true) — including the never-acknowledged first message — advance inboxConfirmedUpToSeq, and commit sendBufferUpToSeq to the inbox store, which deletes the corresponding chunk range.

On session resume the first message is not redelivered: permanent loss of a never-acknowledged message.

Root cause

private void confirm(ConfirmingMessage confirmingMsg, boolean delivered) {
    confirmingMsg.setAcked();                       // marks the PASSED-IN entry
    Iterator<Integer> packetIdItr = unconfirmedPacketIds.keySet().iterator();
    while (packetIdItr.hasNext()) {
        ConfirmingMessage head = unconfirmedPacketIds.get(packetIdItr.next());
        if (head.acked) {
            packetIdItr.remove();
            confirmingMsg = head;                  // only reassigned when something is removed
        } else {
            break;                                  // zero removals -> confirmingMsg unchanged
        }
    }
    onConfirm(confirmingMsg.seq);                   // uses the PASSED-IN seq regardless
}

The same over-confirmation is reachable without any client misbehaviour: the five delayed-failure paths (:1115, :1122, :1152, :1161, :1166) run ctx.executor().execute(() -> confirm(packetId, false)) after the fact, by which time an older entry may be sitting un-acked at the head.

Change

Advance the watermark only when the drain loop actually removed a contiguous prefix: track whether any entry was removed and skip the onConfirm call entirely when nothing was.

boolean anyConfirmed = false;
while (...) { if (head.acked) { packetIdItr.remove(); confirmingMsg = head; anyConfirmed = true; ... } else break; }
if (anyConfirmed) {
    onConfirm(confirmingMsg.seq);
}

Testing

Regression test MQTT3PersistentSessionHandlerTest.outOfOrderPubAckMustNotAdvanceInboxWatermark: fetch two QoS1 messages, PUBACK the second only, verify inboxClient never receives a commit with sendBufferUpToSeq.

Control experiment on unfixed code: the test fails (a commit IS issued past the un-acked message). With the fix: passes. Full MQTT3PersistentSessionHandlerTest (24 tests) passes.

Happy to adjust the semantics if the maintainers prefer a different watermark policy.

Full report: #286.

Fixes #286