From 676f642635c5f2a970d6f491d1c6a8cdeee7b542 Mon Sep 17 00:00:00 2001 From: meisa Date: Wed, 6 May 2026 20:12:04 +0200 Subject: [PATCH] Client: Tolerate late acknowledgement packets (#2078, #2079) 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. --- .../Clients/MqttClient/MqttClient_Tests.cs | 105 ++++++++++++++++++ Source/MQTTnet/IMqttClient.cs | 15 +++ Source/MQTTnet/MqttClient.cs | 15 +++ Source/ReleaseNotes.md | 1 + 4 files changed, 136 insertions(+) diff --git a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs index bf3429358..4b478427f 100644 --- a/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs +++ b/Source/MQTTnet.Tests/Clients/MqttClient/MqttClient_Tests.cs @@ -870,4 +870,109 @@ public async Task Subscribe_With_QoS2() 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(() => 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); + } } \ No newline at end of file diff --git a/Source/MQTTnet/IMqttClient.cs b/Source/MQTTnet/IMqttClient.cs index 326202d60..19114570a 100644 --- a/Source/MQTTnet/IMqttClient.cs +++ b/Source/MQTTnet/IMqttClient.cs @@ -2,6 +2,21 @@ namespace MQTTnet; +/// +/// Represents an MQTT client. +/// +/// Instances of are safe for concurrent use: , +/// , and 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. +/// +/// +/// and must not be called concurrently +/// with each other or with publish/subscribe operations. Coordinate connection lifecycle changes on +/// a single thread. +/// +/// public interface IMqttClient : IDisposable { event Func ApplicationMessageReceivedAsync; diff --git a/Source/MQTTnet/MqttClient.cs b/Source/MQTTnet/MqttClient.cs index 88a0a143c..a9231ddb9 100644 --- a/Source/MQTTnet/MqttClient.cs +++ b/Source/MQTTnet/MqttClient.cs @@ -953,6 +953,21 @@ async Task TryProcessReceivedPacket(MqttPacket packet, CancellationToken cancell { 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); diff --git a/Source/ReleaseNotes.md b/Source/ReleaseNotes.md index 95c9709d3..1a80619fe 100644 --- a/Source/ReleaseNotes.md +++ b/Source/ReleaseNotes.md @@ -2,6 +2,7 @@ * Core: Performance improvements * Core: Added validations for variable byte integers which do not match the .NET uint perfectly * Client: Added support for pre-encoded UTF-8 binary buffers for user properties (#2228, thanks to @koepalex) +* 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: Improved performance of retained messages when no event handler is attached (#2093, thanks to @zhaowgit) * Server: The event `InterceptingClientEnqueue` is now also called for retained messages (BREAKING CHANGE!) * Server: The local end point is now also exposed in the channel adapter (#2179)