Handle DefaultConnectionContext abort/dispose cancellation race

src/Servers/Connections.Abstractions/src/DefaultConnectionContext.cs

@@ -98,7 +98,17 @@ public DefaultConnectionContext(string id, IDuplexPipe transport, IDuplexPipe ap
     /// <inheritdoc />
     public override void Abort(ConnectionAbortedException abortReason)
     {
-        ThreadPool.UnsafeQueueUserWorkItem(cts => ((CancellationTokenSource)cts!).Cancel(), _connectionClosedTokenSource);
+        ThreadPool.UnsafeQueueUserWorkItem(static cts =>
+        {
+            try
+            {
+                ((CancellationTokenSource)cts!).Cancel();
+            }
+            catch (ObjectDisposedException)
+            {
+                // The connection can be disposed between scheduling and running the cancellation.
+            }
+        }, _connectionClosedTokenSource);
     }
 
     /// <inheritdoc />

src/Servers/Kestrel/Core/test/ConnectionContextTests.cs

@@ -1,7 +1,12 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System;
+using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.InternalTesting;
+using Microsoft.DotNet.RemoteExecutor;
 using Moq;
 using Xunit;
 
@@ -23,4 +28,46 @@ public void ParameterlessAbortCreateConnectionAbortedException()
         Assert.NotNull(ex);
         Assert.Equal("The connection was aborted by the application via ConnectionContext.Abort().", ex.Message);
     }
+
+    [ConditionalFact]
+    [RemoteExecutionSupported]
+    public void DefaultConnectionContextDisposeAsyncAfterAbortDoesNotCrashProcess()
+    {
+        using var remoteHandle = RemoteExecutor.Invoke(static async () =>
+        {
+            ThreadPool.GetMinThreads(out var originalMinWorkerThreads, out var originalMinCompletionPortThreads);
+            ThreadPool.GetMaxThreads(out var originalMaxWorkerThreads, out var originalMaxCompletionPortThreads);
+
+            Assert.True(ThreadPool.SetMinThreads(1, originalMinCompletionPortThreads));
+            Assert.True(ThreadPool.SetMaxThreads(1, originalMaxCompletionPortThreads));
+
+            using var blockerStarted = new ManualResetEventSlim();
+            using var releaseBlocker = new ManualResetEventSlim();
+
+            try
+            {
+                ThreadPool.QueueUserWorkItem(_ =>
+                {
+                    blockerStarted.Set();
+                    releaseBlocker.Wait();
+                });
+
+                Assert.True(blockerStarted.Wait(TimeSpan.FromSeconds(10)));
+
+                var connection = new DefaultConnectionContext();
+                connection.Abort();
+                await connection.DisposeAsync();
+
+                releaseBlocker.Set();
+
+                await Task.Delay(TimeSpan.FromSeconds(1));
+            }
+            finally
+            {
+                releaseBlocker.Set();
+                ThreadPool.SetMaxThreads(originalMaxWorkerThreads, originalMaxCompletionPortThreads);
+                ThreadPool.SetMinThreads(originalMinWorkerThreads, originalMinCompletionPortThreads);
+            }
+        });
+    }
 }