[BUG] MQTT inbound inflight cleanup can remove concurrently registered QoS 2 packet IDs #8021

Is there an existing issue for this?

• I have searched the existing issues.

Current Behavior

InflightRepository.release removes a packet ID, checks whether the mutable set is empty, and then removes the map entry with remove(clientId, packetIds). A concurrent register can add a new ID to that same set after isEmpty() succeeds but before remove(). The conditional map removal still matches the same set and deletes it. Its comment claims two-argument remove protects concurrent registration, but it protects replacement of the map value, not mutation of that value. A second interleaving exists because register performs add outside computeIfAbsent's map operation.

Suggested severity: Medium: a lost deduplication entry permits a QoS 2 retransmission to be delivered again.

Expected Behavior

Registration and release/removal must be atomic per client ID. A successfully registered packet ID must remain discoverable until its own PUBREL or an intentional session reset; empty-set cleanup must not detach a set that an overlapping register is using.

Steps To Reproduce

Deterministic proposed interleaving for the repository:

1. Register packet 1 for client X.
2. Thread A releases packet 1 and observes that its set is empty. Pause A before map removal.
3. Thread B registers packet 2 for X; computeIfAbsent returns the existing set and add(2) succeeds.
4. Resume A: remove(X, sameSet) removes the entire entry, including packet 2.
5. Register retransmitted packet 2. It now returns true, incorrectly identifying it as new.

Use latches/a controllable Set or a test hook to make the race deterministic. Concurrent processing is relevant when old/new channels overlap during duplicate-client-ID takeover, since state is keyed by client ID. This interleaving was identified by source inspection, not a stress-test run.

Environment

• ShenYu: master / 2.7.2-SNAPSHOT, commit 34f2ed036747936a5e28aea2c94cd9b3cddcce47.
• Evidence: source inspection. The reporting environment has no available Java runtime, so no Maven/JUnit execution or live reproduction is claimed.
• These issues concern existing runtime behavior, not the proposed plugin DAG implementation.

Debug logs

No runtime logs captured; this report is based on the source paths below.

Code evidence

• Registration and non-atomic release
• Session identity shared across connection takeover
• QoS 2 delivery depends on register returning true
• PUBREL releases client-scoped packet state

Suggested fix direction

Coordinate both register's add and release's remove/empty cleanup under the same per-key atomic compute or session lock. Merely changing release to compute is insufficient if register still mutates the obtained set outside the per-key boundary. Preserve explicit session-reset semantics and fence stale-channel actions as needed.

Acceptance criteria

• A deterministic concurrent register/release test retains the newly registered ID and suppresses its retransmission.
• Cover the computeIfAbsent-return-before-add interleaving as well as empty-check-before-remove.
• Last-ID cleanup, packet-ID reuse after PUBREL, clean-session reset and persistent-session takeover still work.
• No unbounded retention of empty client entries.

Related work

Follow-up to inbound QoS 2 deduplication #7919 / #7938. Distinct from the previously fixed subscriber-list race #6746.