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

When a publish, subscribe or unsubscribe request was cancelled or
timed out client-side, the corresponding awaitable was removed from
the packet dispatcher. A subsequent PUBACK / PUBCOMP / SUBACK /
UNSUBACK arriving from the broker then fell through to the default
case and raised an MqttProtocolViolationException, which forced a
disconnect. Under high publish load (issue #2079) and from multiple
threads (issue #2078) this manifested as sporadic connection drops.

Match these acknowledgement packets explicitly and log a warning
when no awaiter is registered, instead of treating it as a protocol
violation. The connection is now preserved.

Also documents the thread-safety guarantees of IMqttClient: publish,
subscribe, unsubscribe and ping operations are safe to invoke
concurrently. Connect / disconnect must still be coordinated.

Co-authored-by: meisa <meisa@MAFDEL>
Co-authored-by: christian <6939810+chkr1011@users.noreply.github.com>

Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs

@@ -870,4 +870,109 @@ public sealed class MqttClient_Tests : BaseTestClass
         Assert.IsTrue(client1.TryPingAsync().GetAwaiter().GetResult());
         Assert.IsFalse(disconnectedFired);
     }
+
+    // Regression test for issue #2079.
+    // A late PUBACK (one whose matching publish request was already cancelled or timed
+    // out client-side) must NOT raise a protocol violation and must NOT disconnect the
+    // client. Before the fix, the client received the late PUBACK, failed to find a
+    // waiter and threw a MqttProtocolViolationException, tearing down the connection.
+    [TestMethod]
+    [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
+    [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
+    public async Task Cancelled_Publish_Does_Not_Disconnect_Client(MqttQualityOfServiceLevel qos)
+    {
+        using var testEnvironment = new TestEnvironment(TestContext);
+        var server = await testEnvironment.StartServer();
+
+        // Delay the broker's acknowledgement long enough for the publish caller to give
+        // up. When the caller's cancellation token fires the awaitable is removed; the
+        // (slightly later) PUBACK / PUBREC then arrives without a matching waiter.
+        server.InterceptingPublishAsync += async e =>
+        {
+            await Task.Delay(TimeSpan.FromSeconds(2), CancellationToken.None);
+        };
+
+        var client = await testEnvironment.ConnectClient();
+
+        var disconnected = false;
+        client.DisconnectedAsync += _ =>
+        {
+            disconnected = true;
+            return CompletedTask.Instance;
+        };
+
+        var message = new MqttApplicationMessageBuilder().WithTopic("late-ack").WithQualityOfServiceLevel(qos).Build();
+
+        using (var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(250)))
+        {
+            await Assert.ThrowsExactlyAsync<MqttCommunicationTimedOutException>(() => client.PublishAsync(message, cts.Token));
+        }
+
+        // Wait long enough for the delayed broker acknowledgement to come back.
+        await Task.Delay(TimeSpan.FromSeconds(3));
+
+        Assert.IsFalse(disconnected, "Client must not disconnect when a late acknowledgement arrives.");
+        Assert.IsTrue(client.IsConnected, "Client must still be connected after a late acknowledgement.");
+
+        // The connection must remain usable for further publishes.
+        var followUp = new MqttApplicationMessageBuilder().WithTopic("late-ack").WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce).Build();
+        await client.PublishAsync(followUp);
+    }
+
+    // Regression test for issue #2078.
+    // The MqttClient must support concurrent publishes from multiple threads without
+    // dropping messages, mismatching acknowledgements or disconnecting the client.
+    [TestMethod]
+    [DataRow(MqttQualityOfServiceLevel.AtMostOnce)]
+    [DataRow(MqttQualityOfServiceLevel.AtLeastOnce)]
+    [DataRow(MqttQualityOfServiceLevel.ExactlyOnce)]
+    public async Task Concurrent_Publish_From_Multiple_Threads(MqttQualityOfServiceLevel qos)
+    {
+        const int Threads = 8;
+        const int MessagesPerThread = 250;
+
+        using var testEnvironment = new TestEnvironment(TestContext);
+        await testEnvironment.StartServer();
+
+        var publisher = await testEnvironment.ConnectClient();
+
+        var subscriber = await testEnvironment.ConnectClient();
+        var receivedCount = 0;
+        subscriber.ApplicationMessageReceivedAsync += _ =>
+        {
+            Interlocked.Increment(ref receivedCount);
+            return CompletedTask.Instance;
+        };
+        await subscriber.SubscribeAsync("concurrent/#", qos);
+
+        var disconnected = false;
+        publisher.DisconnectedAsync += _ =>
+        {
+            disconnected = true;
+            return CompletedTask.Instance;
+        };
+
+        var tasks = Enumerable.Range(0, Threads).Select(threadIndex => Task.Run(async () =>
+        {
+            for (var i = 0; i < MessagesPerThread; i++)
+            {
+                var message = new MqttApplicationMessageBuilder()
+                    .WithTopic($"concurrent/{threadIndex}/{i}")
+                    .WithPayload(BitConverter.GetBytes(i))
+                    .WithQualityOfServiceLevel(qos)
+                    .Build();
+
+                await publisher.PublishAsync(message);
+            }
+        })).ToArray();
+
+        await Task.WhenAll(tasks);
+
+        // Allow the broker to deliver any in-flight messages to the subscriber.
+        SpinWait.SpinUntil(() => receivedCount >= Threads * MessagesPerThread, TimeSpan.FromSeconds(30));
+
+        Assert.IsFalse(disconnected, "Publisher must not disconnect during concurrent publishes.");
+        Assert.IsTrue(publisher.IsConnected);
+        Assert.AreEqual(Threads * MessagesPerThread, receivedCount);
+    }
 }

Source/MQTTnet/IMqttClient.cs

@@ -2,6 +2,21 @@ using MQTTnet.Diagnostics.PacketInspection;
 
 namespace MQTTnet;
 
+/// <summary>
+/// Represents an MQTT client.
+/// <para>
+/// Instances of <see cref="IMqttClient" /> are safe for concurrent use: <see cref="PublishAsync" />,
+/// <see cref="SubscribeAsync" />, <see cref="UnsubscribeAsync" /> and <see cref="PingAsync" /> may be
+/// invoked from multiple threads at the same time. Outgoing packets are serialized internally so
+/// that they reach the network in well-formed order, and incoming acknowledgements are routed back
+/// to the corresponding caller via the packet identifier.
+/// </para>
+/// <para>
+/// <see cref="ConnectAsync" /> and <see cref="DisconnectAsync" /> must not be called concurrently
+/// with each other or with publish/subscribe operations. Coordinate connection lifecycle changes on
+/// a single thread.
+/// </para>
+/// </summary>
 public interface IMqttClient : IDisposable
 {
     event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;

Source/MQTTnet/MqttClient.cs

@@ -953,6 +953,21 @@ public sealed class MqttClient : Disposable, IMqttClient
             {
                 case MqttPublishPacket publishPacket:
                     EnqueueReceivedPublishPacket(publishPacket);
+                    break;
+                case MqttPubAckPacket _:
+                case MqttPubCompPacket _:
+                case MqttSubAckPacket _:
+                case MqttUnsubAckPacket _:
+                    // Acknowledgement packets are dispatched to the awaiter that issued the
+                    // matching request. If no awaiter is registered (for example because the
+                    // request was cancelled or timed out client-side before the broker replied)
+                    // the late acknowledgement is harmless and must NOT terminate the connection.
+                    // See issues #2078 and #2079.
+                    if (!_packetDispatcher.TryDispatch(packet))
+                    {
+                        _logger.Warning("Received {0} for an unknown packet identifier. Probably the matching request was cancelled or timed out.", packet.GetType().Name);
+                    }
+
                     break;
                 case MqttPubRecPacket pubRecPacket:
                     await ProcessReceivedPubRecPacket(pubRecPacket, cancellationToken).ConfigureAwait(false);

Source/ReleaseNotes.md

@@ -1 +1,2 @@
-- Server: Fixed exposinbg Topic Alias to clients (#2250, thanks to @suhashollakc)
+* Client: Late acknowledgement packets (PUBACK, PUBCOMP, SUBACK, UNSUBACK) for cancelled or timed-out requests no longer trigger a protocol violation and tear down the connection (#2078, #2079)
+* Server: Fixed exposing Topic Alias to clients (#2250, thanks to @suhashollakc)