Avoid UnobservedTaskExceptions from HttpConnection.PrepareForReuse (#104335)

* Add regression test

* Fix the bug

Author: MihaZupan

src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnection.cs

@@ -169,7 +169,15 @@ public bool PrepareForReuse(bool async)
                     _readAheadTask = _stream.ReadAsync(_readBuffer.AvailableMemory);
 #pragma warning restore CA2012
 
-                    return !_readAheadTask.IsCompleted;
+                    // If the read-ahead task already completed, we can't reuse the connection.
+                    // We're still responsible for observing potential exceptions thrown by the read-ahead task to avoid leaking unobserved exceptions.
+                    if (_readAheadTask.IsCompleted)
+                    {
+                        LogExceptions(_readAheadTask.AsTask());
+                        return false;
+                    }
+
+                    return true;
                 }
                 catch (Exception error)
                 {

src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/HttpConnectionBase.cs

@@ -181,24 +181,6 @@ internal static int ParseStatusCode(ReadOnlySpan<byte> value)
             return 100 * (status1 - '0') + 10 * (status2 - '0') + (status3 - '0');
         }
 
-        /// <summary>Awaits a task, ignoring any resulting exceptions.</summary>
-        internal static void IgnoreExceptions(ValueTask<int> task)
-        {
-            // Avoid TaskScheduler.UnobservedTaskException firing for any exceptions.
-            if (task.IsCompleted)
-            {
-                if (task.IsFaulted)
-                {
-                    _ = task.AsTask().Exception;
-                }
-            }
-            else
-            {
-                task.AsTask().ContinueWith(static t => _ = t.Exception,
-                    CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);
-            }
-        }
-
         /// <summary>Awaits a task, logging any resulting exceptions (which are otherwise ignored).</summary>
         internal void LogExceptions(Task task)
         {

src/libraries/System.Net.Http/tests/FunctionalTests/SocketsHttpHandlerTest.cs

@@ -99,6 +99,75 @@ static async Task MakeARequestWithoutDisposingTheHandlerAsync()
             }
         }
 
+        [Fact]
+        public async Task ReadAheadTaskOnConnectionReuse_ExceptionsAreObserved()
+        {
+            bool seenUnobservedExceptions = false;
+
+            EventHandler<UnobservedTaskExceptionEventArgs> eventHandler = (_, e) =>
+            {
+                if (e.Exception.InnerException?.Message == nameof(ReadAheadTaskOnConnectionReuse_ExceptionsAreObserved))
+                {
+                    _output.WriteLine(e.Exception.ToString());
+                    seenUnobservedExceptions = true;
+                }
+            };
+
+            TaskScheduler.UnobservedTaskException += eventHandler;
+            try
+            {
+                const string FirstResponse = "HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nfoo";
+                int firstResponseOffset = 0;
+
+                TaskCompletionSource firstWriteCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+                using var client = new HttpClient(new SocketsHttpHandler
+                {
+                    PooledConnectionIdleTimeout = Timeout.InfiniteTimeSpan,
+                    ConnectCallback = (_, _) =>
+                    {
+                        return ValueTask.FromResult<Stream>(new DelegateDelegatingStream(Stream.Null)
+                        {
+                            ReadAsyncMemoryFunc = async (buffer, ct) =>
+                            {
+                                if (firstResponseOffset < FirstResponse.Length)
+                                {
+                                    await firstWriteCompleted.Task.WaitAsync(ct);
+                                    int toWrite = Math.Min(buffer.Length, FirstResponse.Length - firstResponseOffset);
+                                    Assert.Equal(toWrite, Encoding.ASCII.GetBytes(FirstResponse.AsSpan(firstResponseOffset, toWrite), buffer.Span));
+                                    firstResponseOffset += toWrite;
+                                    return toWrite;
+                                }
+
+                                throw new Exception(nameof(ReadAheadTaskOnConnectionReuse_ExceptionsAreObserved));
+                            },
+                            WriteAsyncMemoryFunc = (_, _) =>
+                            {
+                                firstWriteCompleted.TrySetResult();
+                                return default;
+                            }
+                        });
+                    }
+                });
+
+                // The first request succeeds.
+                Assert.Equal("foo", await client.GetStringAsync("http://foo"));
+
+                // The connection throws an exception after the first request.
+                // No matter what happens with the second request, we should always be observing the connection exception.
+                await Assert.ThrowsAsync<Exception>(() => client.GetAsync("http://foo"));
+
+                GC.Collect();
+                GC.WaitForPendingFinalizers();
+            }
+            finally
+            {
+                TaskScheduler.UnobservedTaskException -= eventHandler;
+            }
+
+            Assert.False(seenUnobservedExceptions);
+        }
+
         [Fact]
         public async Task ExecutionContext_Suppressed_Success()
         {