From 483447e78347b4420b32cdf536dcd826c538dc4c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jochen=20K=C3=BChner?= Date: Wed, 30 Sep 2026 20:23:57 +0200 Subject: [PATCH] Fix unobserved task exceptions when pending messages overflow (#2272) Observe packet task failures to prevent unobserved overflow exceptions Server dispatch and session enqueue operations create packet bus items without waiting for delivery. When the pending-message queue overflows, Fail faults the item's completion task, which can later raise UnobservedTaskException even if the application handles the dropped-message notification. Observe the completion task's exception inside MqttPacketBusItem.Fail. This keeps the task faulted and preserves the original exception for existing and future WaitAsync callers, including DeliverApplicationMessageAsync. Add a garbage-collection regression test for failures without a waiter and verify that requesting the task before or after failure preserves the original overflow exception. Re-enable unobserved-exception tracking in the existing overflow integration tests for both drop strategies. Before the fix, the focused run failed the new unobserved-exception test and both enqueue overflow tests. After the fix, all nine focused cases and all 489 tests in the .NET 10 Release test suite pass. Fixes https://github.com/dotnet/MQTTnet/issues/2271 --- .../Internal/MqttPacketBusItem_Tests.cs | 63 +++++++++++++++++++ .../MQTTnet.Tests/Server/Injection_Tests.cs | 6 +- Source/MQTTnet/Internal/MqttPacketBusItem.cs | 4 ++ 3 files changed, 70 insertions(+), 3 deletions(-) diff --git a/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs b/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs index 0f670be5..57d85788 100644 --- a/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs +++ b/Source/MQTTnet.Tests/Internal/MqttPacketBusItem_Tests.cs @@ -2,8 +2,11 @@ // The .NET Foundation licenses this file to you under the MIT license. // See the LICENSE file in the project root for more information. +using System.Runtime.CompilerServices; using MQTTnet.Internal; using MQTTnet.Packets; +using MQTTnet.Server; +using MQTTnet.Server.Exceptions; namespace MQTTnet.Tests.Internal; @@ -11,6 +14,57 @@ namespace MQTTnet.Tests.Internal; [TestClass] public sealed class MqttPacketBusItem_Tests { + [TestMethod] + public void Fail_Without_Waiter_Does_Not_Raise_Unobserved_Exception() + { + var exception = new MqttPendingMessagesOverflowException("test", MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage); + var unobservedExceptions = 0; + + void OnUnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs args) + { + if (args.Exception.Flatten().InnerExceptions.Contains(exception)) + { + Interlocked.Increment(ref unobservedExceptions); + args.SetObserved(); + } + } + + TaskScheduler.UnobservedTaskException += OnUnobservedTaskException; + try + { + var taskReference = CreateFailedPacketBusItem(exception); + + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + + Assert.IsFalse(taskReference.IsAlive, "The faulted task must be collected to exercise unobserved exception reporting."); + Assert.AreEqual(0, unobservedExceptions); + } + finally + { + TaskScheduler.UnobservedTaskException -= OnUnobservedTaskException; + } + } + + [TestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task Wait_Failed_Packet_Bus_Item_Preserves_Exception(bool waitBeforeFailure) + { + var item = new MqttPacketBusItem(new MqttPublishPacket()); + var exception = new MqttPendingMessagesOverflowException("test", MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage); + var task = waitBeforeFailure ? item.WaitAsync() : null; + + item.Fail(exception); + + task ??= item.WaitAsync(); + var thrownException = await Assert.ThrowsExactlyAsync(async () => await task); + + Assert.AreSame(exception, thrownException); + Assert.IsTrue(task.IsFaulted); + } + [TestMethod] public void Fire_Completed_Event() { @@ -40,4 +94,13 @@ public sealed class MqttPacketBusItem_Tests await item.WaitAsync(); }); } + + [MethodImpl(MethodImplOptions.NoInlining)] + static WeakReference CreateFailedPacketBusItem(Exception exception) + { + // Keep strong references out of the collecting method, including in debug builds. + var item = new MqttPacketBusItem(new MqttPublishPacket()); + item.Fail(exception); + return new WeakReference(item.WaitAsync()); + } } \ No newline at end of file diff --git a/Source/MQTTnet.Tests/Server/Injection_Tests.cs b/Source/MQTTnet.Tests/Server/Injection_Tests.cs index c3c401e6..ed224353 100644 --- a/Source/MQTTnet.Tests/Server/Injection_Tests.cs +++ b/Source/MQTTnet.Tests/Server/Injection_Tests.cs @@ -48,7 +48,7 @@ public sealed class Injection_Tests : BaseTestClass [TestMethod] public async Task Enqueue_Application_Message_At_Session_Level_QueueOverflow_DropNewMessageStrategy() { - using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false); + using var testEnvironment = CreateTestEnvironment(); var server = await testEnvironment.StartServer( builder => builder @@ -126,7 +126,7 @@ public sealed class Injection_Tests : BaseTestClass [TestMethod] public async Task Enqueue_Application_Message_At_Session_Level_QueueOverflow_DropOldestQueuedMessageStrategy() { - using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false); + using var testEnvironment = CreateTestEnvironment(); var server = await testEnvironment.StartServer( builder => builder @@ -322,7 +322,7 @@ public sealed class Injection_Tests : BaseTestClass [TestMethod] public async Task Deliver_Application_Message_At_Session_Level_QueueOverflow_DropOldestQueuedMessageStrategy() { - using var testEnvironment = CreateTestEnvironment(trackUnobservedTaskException: false); + using var testEnvironment = CreateTestEnvironment(); var server = await testEnvironment.StartServer( builder => builder diff --git a/Source/MQTTnet/Internal/MqttPacketBusItem.cs b/Source/MQTTnet/Internal/MqttPacketBusItem.cs index 5f494473..f51b1955 100644 --- a/Source/MQTTnet/Internal/MqttPacketBusItem.cs +++ b/Source/MQTTnet/Internal/MqttPacketBusItem.cs @@ -33,6 +33,10 @@ public sealed class MqttPacketBusItem public void Fail(Exception exception) { _promise.TrySetException(exception); + + // Queued packets do not always have a waiter (for example, normal server dispatch). + // Observe the failure here while preserving it for callers of WaitAsync. + _ = _promise.Task.Exception; } public Task WaitAsync()