FIXED unobserved exceptions surfacing due to unobserved tcs.

src/NATS.Client.Core/Internal/NatsReadProtocolProcessor.cs

@@ -298,8 +298,8 @@ await _connection.PublishToClientHandlersAsync(subject, replyTo, sid, headerSlic
             catch (SocketClosedException e)
             {
                 _logger.LogDebug(NatsLogEvents.Protocol, e, "Socket closed during read loop");
-                _waitForInfoSignal.TrySetException(e);
-                _waitForPongOrErrorSignal.TrySetException(e);
+                _waitForInfoSignal.TrySetObservedException(e);
+                _waitForPongOrErrorSignal.TrySetObservedException(e);
                 return;
             }
             catch (Exception ex)
@@ -379,14 +379,14 @@ private async ValueTask<ReadOnlySequence<byte>> DispatchCommandAsync(int code, R
                 var newPosition = newBuffer.PositionOf((byte)'\n');
                 var error = ParseError(newBuffer.Slice(0, newBuffer.GetOffset(newPosition!.Value) - 1));
                 _logger.LogError(NatsLogEvents.Protocol, "Server error {Error}", error);
-                _waitForPongOrErrorSignal.TrySetException(new NatsServerException(error));
+                _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error));
                 return newBuffer.Slice(newBuffer.GetPosition(1, newPosition!.Value));
             }
             else
             {
                 var error = ParseError(buffer.Slice(0, buffer.GetOffset(position.Value) - 1));
                 _logger.LogError(NatsLogEvents.Protocol, "Server error {Error}", error);
-                _waitForPongOrErrorSignal.TrySetException(new NatsServerException(error));
+                _waitForPongOrErrorSignal.TrySetObservedException(new NatsServerException(error));
                 return buffer.Slice(buffer.GetPosition(1, position.Value));
             }
         }

src/NATS.Client.Core/Internal/SocketConnectionWrapper.cs

@@ -20,7 +20,7 @@ public void SignalDisconnected(Exception exception)
         _sem.Wait();
         try
         {
-            _waitForClosedSource.TrySetException(exception);
+            _waitForClosedSource.TrySetObservedException(exception);
         }
         finally
         {

src/NATS.Client.Core/Internal/TaskCompletionSourceExtensions.cs

@@ -0,0 +1,22 @@
+namespace NATS.Client.Core.Internal;
+
+internal static class TaskCompletionSourceExtensions
+{
+    /// <summary>
+    /// Sets an exception on the task completion source and immediately observes it to avoid unobserved task exceptions surfacing.
+    /// </summary>
+    /// <param name="source">The task completion source.</param>
+    /// <param name="ex">The exception to set and observe.</param>
+    /// <returns>True if the exception was successfully set, otherwise false</returns>
+    internal static bool TrySetObservedException(this TaskCompletionSource source, Exception ex)
+    {
+        var result = source.TrySetException(ex);
+
+        if (result)
+        {
+            _ = source.Task.Exception;
+        }
+
+        return result;
+    }
+}

src/NATS.Client.Core/NatsConnection.cs

@@ -407,13 +407,7 @@ private async ValueTask InitialConnectAsync()
                 ConnectionState = NatsConnectionState.Closed; // allow retry connect
 
                 // throw for the waiter
-                if (_waitForOpenConnection.TrySetException(exception))
-                {
-                    // Suppress unobserved exceptions as the exceptions will surface elsewhere,
-                    // the exception is thrown below as well.
-                    _ = _waitForOpenConnection.Task.Exception;
-                }
-
+                _waitForOpenConnection.TrySetObservedException(exception);
                 _waitForOpenConnection = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
             }
 
@@ -435,13 +429,7 @@ private async ValueTask InitialConnectAsync()
                 ConnectionState = NatsConnectionState.Closed; // allow retry connect
 
                 // throw for the waiter
-                if (_waitForOpenConnection.TrySetException(exception))
-                {
-                    // Suppress unobserved exceptions as the exceptions will surface elsewhere,
-                    // the exception is thrown below as well.
-                    _ = _waitForOpenConnection.Task.Exception;
-                }
-
+                _waitForOpenConnection.TrySetObservedException(exception);
                 _waitForOpenConnection = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
             }
 
@@ -788,7 +776,7 @@ private async void ReconnectLoop()
         }
         catch (Exception ex)
         {
-            _waitForOpenConnection.TrySetException(ex);
+            _waitForOpenConnection.TrySetObservedException(ex);
             try
             {
                 if (!IsDisposed)

tests/NATS.Client.Core.Tests/AuthErrorTest.cs

@@ -1,4 +1,3 @@
-using System.Diagnostics;
 using Microsoft.Extensions.Logging;
 using NATS.Client.Platform.Windows.Tests;
 using NATS.Client.TestUtilities;
@@ -158,4 +157,61 @@ public async Task Auth_err_can_be_ignored_for_retires()
 
         await server.DisposeAsync();
     }
+
+    [Fact]
+    public async Task Auth_err_then_connection_recreation_does_not_cause_unobserved_exception()
+    {
+        // Arrange
+        var exceptionThrown = false;
+
+        var confFile = $"{nameof(Auth_err_then_connection_recreation_does_not_cause_unobserved_exception)}_server.conf";
+        var confContents = """
+                           authorization: {
+                               users: [
+                                   {user: a, password: b}
+                               ]
+                           }
+                           """;
+
+        // Act
+        TaskScheduler.UnobservedTaskException += (_, _) =>
+        {
+            exceptionThrown = true;
+        };
+
+        File.WriteAllText(path: confFile, contents: confContents);
+
+        var server = await NatsServerProcess.StartAsync(config: confFile);
+
+        await Task.Run(async () =>
+        {
+            for (var i = 0; i < 2; i++)
+            {
+                var conn = new NatsConnection(new NatsOpts
+                {
+                    Url = server.Url,
+                });
+
+                try
+                {
+                    await conn.ConnectAsync();
+                }
+                catch (NatsException)
+                {
+                }
+                finally
+                {
+                    await conn.DisposeAsync();
+                }
+            }
+        });
+
+        await Task.Delay(100);
+
+        GC.Collect();
+        GC.WaitForPendingFinalizers();
+
+        // Assert
+        Assert.False(exceptionThrown);
+    }
 }

tests/NATS.Client.Core.Tests/ProtocolTest.cs

@@ -21,7 +21,7 @@ public async Task Protocol_parser_under_load(int size)
 
         var signal = new WaitSignal();
         var counts = 0;
-        _ = Task.Run(
+        var subscribeTask = Task.Run(
             async () =>
             {
                 var count = 0;
@@ -52,7 +52,7 @@ public async Task Protocol_parser_under_load(int size)
 
         var r = 0;
         var payload = new byte[size];
-        _ = Task.Run(
+        var publishTask = Task.Run(
             async () =>
             {
                 while (!cts.Token.IsCancellationRequested)
@@ -95,6 +95,10 @@ public async Task Protocol_parser_under_load(int size)
             await Retry.Until("subject count goes up", () => Volatile.Read(ref counts) > subjectCount, timeout: TimeSpan.FromSeconds(60));
         }
 
+        cts.Cancel();
+        await subscribeTask;
+        await publishTask;
+
         foreach (var log in logger.Logs.Where(x => x.EventId == NatsLogEvents.Protocol && x.LogLevel == LogLevel.Error))
         {
             Assert.DoesNotContain("Unknown Protocol Operation", log.Message);

tests/NATS.Client.CoreUnit.Tests/SocketConnectionWrapperTests.cs

@@ -0,0 +1,40 @@
+namespace NATS.Client.CoreUnit.Tests;
+
+public class SocketConnectionWrapperTests
+{
+    [Fact]
+    public async Task SignalDisconnected_DoesNotCause_UnobservedException()
+    {
+        // Arrange
+        var exceptionThrown = false;
+        var socketConnection = new FakeSocketConnection();
+        var socket = new SocketConnectionWrapper(socketConnection);
+
+        // Act
+        TaskScheduler.UnobservedTaskException += (_, _) =>
+        {
+            exceptionThrown = true;
+        };
+
+        socket.SignalDisconnected(new Exception("test"));
+        await socket.DisposeAsync();
+        socket = null;
+
+        await Task.Delay(100);
+
+        GC.Collect();
+        GC.WaitForPendingFinalizers();
+
+        // Assert
+        Assert.False(exceptionThrown);
+    }
+
+    internal class FakeSocketConnection : INatsSocketConnection
+    {
+        public ValueTask DisposeAsync() => default;
+
+        public ValueTask<int> ReceiveAsync(Memory<byte> buffer) => throw new NotImplementedException();
+
+        public ValueTask<int> SendAsync(ReadOnlyMemory<byte> buffer) => throw new NotImplementedException();
+    }
+}