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

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<MqttPendingMessagesOverflowException>(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());
+    }
 }

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

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<MqttPacket> WaitAsync()