Repository navigation
fix(mqtt): make inbound packet id register/release atomic per client (#8021) - #8075
Conversation
454e9b2 to
b04b85b
Compare
Aias00
left a comment
Aias00
left a comment
There was a problem hiding this comment.
Verified the fix closes both windows the issue describes.
Correctness
registerandreleasenow both run inside a single per-keycompute, so the add and the remove+empty-eviction are mutually exclusive per client id by construction. ThecomputeIfAbsent-returns-then-addwindow is gone as well, since theaddno longer happens outside the map operation.- The
computelambda never returnsnullinregisterand returnsnullonly when the set is empty inrelease, so no empty client entry can be left behind. - Moving the map from
staticto an instance field does not split state in production:MqttBootstrapServer.initRepositories()reflects overBaseRepositorysubtypes and registers exactly one instance per class inSingleton, which is the only path the callers use (Publish,PubRel,Connect,Disconnect). It also makes a listener restart drop stale in-flight ids, consistent with the change already made toChannelRepository.
Verification I ran locally
- Reverting only
InflightRepository.java(new tests kept):InflightRepositoryTestfails 2/6 —concurrentRegisterAndReleaseMustNotLoseRegisteredIdsat iteration 2 ("a freshly registered id was evicted by a concurrent release") andreleaseOfUnknownIdOrClientIsNoOp(expected: <null> but was: <[1]>, the static map leaking state across methods). The new tests are not vacuous. - Reverting only the two
Singleton.INST.single(...)lines inSubscribeTest: 2 NPEs (testSubAckGrantsRequestedQoS:85,testSubAckExcludesFailureSubscriptions:100). That hunk is required, not cosmetic. checkstyle:checkon the module: clean.
One follow-up worth a separate PR (not blocking)
The module still has a sibling of the exact fixture bug this PR fixes: MqttTransportHandlerTest registers Will/Channel/Session/Subscribe repositories but not InflightRepository, while Connect/Disconnect call Singleton.INST.get(InflightRepository.class).remove(...). Run standalone it produces 4 NPEs; in a full-module run it fails 3/9 whenever it is ordered before any class that registers InflightRepository. That is why the "146/146" figure is not reproducible on every filesystem ordering — here I get 146 run / 3 errors, and the same 3 errors appear with only the production change applied, i.e. they pre-date this PR. Since the PR already establishes the pattern in SubscribeTest, fixing that fixture in the same pass would make the module genuinely order-independent.
| @Test | ||
| public void concurrentRegisterAndReleaseMustNotLoseRegisteredIds() throws Exception { | ||
| final InflightRepository shared = new InflightRepository(); | ||
| final int iterations = 100_000; |
There was a problem hiding this comment.
iterations is never read — publishLoop hardcodes 100_000 in its loop bound (line 126). Either pass it through as a parameter or drop the local, otherwise the two values can drift.
| second.start(); | ||
| start.countDown(); | ||
| first.join(TimeUnit.MINUTES.toMillis(1)); | ||
| second.join(TimeUnit.MINUTES.toMillis(1)); |
There was a problem hiding this comment.
The return value of join(timeout) is ignored, and these are non-daemon threads: if the loop ever exceeds the timeout the test still passes and the threads keep running into the next test / fork shutdown. Consider asserting assertFalse(first.isAlive()) after the joins, or marking the threads daemon.
Fixes #8021.
Summary
The follow-up to #7919 is correct:
release's empty-entry cleanup did not protect against concurrent registration. The two-argumentremove(clientId, packetIds)guards against the map value being replaced, not against the referenced set being mutated — a packet id registered betweenisEmpty()and the cleanup was evicted with the entry, so its retransmission registered as first-seen and was delivered twice. Symmetrically,register'saddran outside thecomputeIfAbsentboundary, so a concurrent cleanup could leave the id in a set the map had already dropped.Modifications
registerandreleaseeach run inside a single per-keycompute: the add, the remove and the empty-entry eviction are one atomic step, andConcurrentHashMapserializes computes on the same key — both windows are closed by construction. This is the coordination the issue suggests ("under the same per-key atomic compute"); the session-reset (remove) and PUBREL id-reuse semantics are unchanged.ChannelRepositoryalready made.SubscribeTestthe repositoriessubscribe()actually touches (channel/session), which the new test class's alphabetical position surfaced as an order-dependent fixture onmaster.Verifying this change
InflightRepositoryTest: last-id cleanup evicts the client entry; unknown id/client release is a no-op; a released id becomes reusable; session reset re-registers as first-seen;BaseRepositoryaccessors delegate.checkstyle:checkclean.