Client: Tolerate late acknowledgement packets (#2078, #2079)

[araxis]

Fixes #2078
Fixes #2079

Problem

When a PublishAsync / SubscribeAsync / UnsubscribeAsync call is cancelled or times out client-side, its awaitable is removed from MqttPacketDispatcher. If the broker''s matching PUBACK / PUBCOMP / SUBACK / UNSUBACK arrives slightly later it falls through the default case in MqttClient.TryProcessReceivedPacket, raises MqttProtocolViolationException("Received packet ''...'' at an unexpected time."), and tears down the connection.

This surfaced as:

• Received packet 'PubAck: [PacketIdentifier=6538] [ReasonCode=Success]' at an unexpected time. #2079 — sporadic MqttProtocolViolationException on a tiny fraction of QoS 1 publishes when sending high volumes (~100k messages).
• MQTTClient Thread-safety: Publishing #2078 — MqttClient appearing unsafe under concurrent publishes from multiple threads.

Why #2078 is the same root cause (not a missing lock)

The two issues look different but share one bug. The client was already correctly synchronised at every relevant layer; concurrency just made the late-ack race far more likely to be hit.

What was already protected:

Step	Protected	How
MqttPacketIdentifierProvider.GetNextPacketIdentifier	yes	lock (_syncRoot) — each caller gets a unique ID
MqttPacketDispatcher.AddAwaitable	yes	lock (_waiters)
MqttChannelAdapter.SendPacketAsync	yes	await _syncRoot.EnterAsync(...) (AsyncLock) — wire bytes never interleave
Each caller awaits its own MqttPacketAwaitable<T>	yes	Per-call instance
Incoming acks routed by packet identifier	yes	MqttPacketDispatcher.TryDispatch matches by ID

So two threads calling PublishAsync simultaneously was fine at the locking level. What broke under load was this sequence:

1. Thread A calls PublishAsync(message, ctsA.Token) with a tight timeout.
2. The awaitable for packet id N is registered, the PUBLISH goes on the wire.
3. ctsA fires before the broker''s PUBACK arrives - the awaitable is removed.
4. PUBACK for N arrives.
5. Old behaviour: dispatcher finds no waiter → MqttProtocolViolationException → connection torn down.
6. Thread B''s unrelated in-flight publish, sharing the same connection, now also fails.

That is why users perceived "the client breaks when used from multiple threads": one caller''s cancellation killed everyone else''s connection. The same race produced #2079 in a single-threaded high-volume scenario, where a small fraction of publishes get cancelled by their own timeout.

Adding a SemaphoreSlim around PublishAsync would have been the wrong fix - it would serialise publishes (hurting throughput) without addressing the actual bug, since the late-ack race exists regardless of how many threads are involved.

Fix

MqttClient.TryProcessReceivedPacket now matches PUBACK / PUBCOMP / SUBACK / UNSUBACK explicitly. If no awaiter is registered the packet is logged as a warning and dropped, rather than treated as a protocol violation. The connection is preserved. PUBREC / PUBREL already had spec-compliant handling for unknown identifiers and are unchanged.

IMqttClient now documents its concurrency contract: PublishAsync, SubscribeAsync, UnsubscribeAsync and PingAsync are safe to invoke concurrently from multiple threads. ConnectAsync / DisconnectAsync must still be coordinated on a single thread.

Tests

Two new tests in MqttClient_Tests:

• Cancelled_Publish_Does_Not_Disconnect_Client (AtLeastOnce, ExactlyOnce) - uses InterceptingPublishAsync to delay the broker''s ack, cancels the publish, and asserts the client stays connected and remains usable. Verified to fail on the unfixed code with MqttProtocolViolationException.
• Concurrent_Publish_From_Multiple_Threads (AtMostOnce, AtLeastOnce, ExactlyOnce) - 8 threads x 250 messages, asserts no disconnect and that every message is delivered to a subscriber.

Full suite: 487 / 487 passed locally on net8.0.

Release notes

Entry added to Source/ReleaseNotes.md.

[araxis]

@dotnet-policy-service agree