fix(mqtt): drop max-retried entries after the resend iteration completes — fixes #289

[ImDanXie]

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

Summary

resend() iterated unconfirmedPacketIds.values() with a for-each loop and called confirm(...) inside the loop for entries past maxResendTimes. confirm() removes entries from the same map through a different iterator, so the outer iterator throws ConcurrentModificationException once another entry remains after the dropped head.

The exception escapes the scheduled task, so flush(true) and scheduleResend() never run. The retransmission chain dies permanently: remaining in-flight entries are neither retransmitted nor dropped, unconfirmedPacketIds stays non-empty, clientReceiveQuota() stays at 0, and the session silently stops delivering messages.

Root cause

for (ConfirmingMessage confirmingMsg : unconfirmedPacketIds.values()) {
    if (confirmingMsg.sentCount <= settings.maxResendTimes) {
        ... // retransmit
    } else {
        reportDropConfirmableMsgEvent(confirmingMsg.message, DropReason.MaxRetried);
        confirm(confirmingMsg, false);   // -> removes from the map via another iterator
        receiveQuota.onErrorSignal(now);
    }
}

All five other confirm() call sites (:1115, :1122, :1152, :1161, :1166) already defer through ctx.executor().execute(...) to avoid exactly this — only the one inside resend() is a direct call.

Change

Collect the entries to drop into a local list and process them after the iteration completes.

Testing

Regression test MQTT3TransientSessionHandlerTest.resendDroppingHeadAbortsDropOfRemaining: MaxResendTimes=1, ResendTimeoutSeconds=1, ReceivingMaximum=2; publish two QoS1 messages and acknowledge neither — the crucial difference from the existing qoS1ResendKeepsRunningAfterPartialAck, which acks the first message and therefore leaves a single in-flight entry whose iterator never has a next element, so it can never trigger the CME.

Control experiment on the unfixed code: the assertion fails with Wanted 2 times, But was 1 time (the loop aborts right after dropping the head). With the fix: passes. Full MQTT3TransientSessionHandlerTest (64 tests) and MQTT3PersistentSessionHandlerTest (23 tests) pass.

Full report with reproduction details: #289.

Fixes #289

[ImDanXie]

Noted while preparing the regression test — pre-existing behaviour, not introduced by this PR:

resend()'s retransmit branch has no acked guard: an entry that was acknowledged out-of-order (so acked=true is set) but is still sitting in the map behind an un-acked head may be retransmitted once more (with DUP=1) if its resend timeout elapses before the head is resolved. Harmless per QoS 1 semantics (duplicates are permitted; the client will simply re-ack), but a skip if acked check in the retransmit branch would avoid the wasted send. Left as a follow-up; does not affect the correctness of this PR.