Avoid race condition that can cause RequestAborted not to fire

- With HTTP/1.1 there's a narrow race where if we cancel the _abortedCts
  on a background thread  due to the underlying connection closing
  between requests, it can be missed because CTS.TryReset is not thread safe with concurrent
  cancellation

https://github.com/dotnet/runtime/blob/2dbdff1a7aebd64b4b265d99b173cd77f088c36e/src/libraries/System.Private.CoreLib/src/System/Threading/CancellationTokenSource.cs#L477-L479

Author: halter73

src/Servers/Kestrel/Core/src/Internal/Http/HttpProtocol.cs

@@ -407,21 +407,19 @@ public void Reset()
 
         _manuallySetRequestAbortToken = null;
 
-        // Lock to prevent CancelRequestAbortedToken from attempting to cancel a disposed CTS.
-        CancellationTokenSource? localAbortCts = null;
-
         lock (_abortLock)
         {
             _preventRequestAbortedCancellation = false;
-            if (_abortedCts?.TryReset() == false)
+
+            // If the connection has already been aborted, allow that to be observed during the next request.
+            if (!_connectionAborted && _abortedCts is not null)
             {
-                localAbortCts = _abortedCts;
-                _abortedCts = null;
+                // _connectionAborted is terminal and only set inside the _abortLock, so if it isn't set here,
+                // it has not been canceled yet.
+                Trace.Assert(_abortedCts.TryReset());
             }
         }
 
-        localAbortCts?.Dispose();
-
         Output?.Reset();
 
         _requestHeadersParsed = 0;
@@ -760,7 +758,7 @@ private async Task ProcessRequests<TContext>(IHttpApplication<TContext> applicat
                 }
                 else if (!HasResponseStarted)
                 {
-                    // If the request was aborted and no response was sent, we use status code 499 for logging                    
+                    // If the request was aborted and no response was sent, we use status code 499 for logging
                     StatusCode = StatusCodes.Status499ClientClosedRequest;
                 }
             }

src/Servers/Kestrel/Core/src/Internal/Http2/Http2Stream.cs

@@ -129,7 +129,6 @@ public bool ReceivedEmptyRequestBody
     protected override void OnReset()
     {
         _keepAlive = true;
-        _connectionAborted = false;
         _userTrailers = null;
 
         // Reset Http2 Features

src/Servers/Kestrel/Core/src/Internal/Http3/Http3Connection.cs

@@ -602,15 +602,16 @@ private async Task CreateHttp3Stream<TContext>(ConnectionContext streamContext,
 
         // Check whether there is an existing HTTP/3 stream on the transport stream.
         // A stream will only be cached if the transport stream itself is reused.
-        if (!persistentStateFeature.State.TryGetValue(StreamPersistentStateKey, out var s))
+        if (!persistentStateFeature.State.TryGetValue(StreamPersistentStateKey, out var s) ||
+            s is not Http3Stream<TContext> { CanReuse: true } reusableStream)
         {
             stream = new Http3Stream<TContext>(application, CreateHttpStreamContext(streamContext));
             persistentStateFeature.State.Add(StreamPersistentStateKey, stream);
         }
         else
         {
-            stream = (Http3Stream<TContext>)s!;
-            stream.InitializeWithExistingContext(streamContext.Transport);
+            reusableStream.InitializeWithExistingContext(streamContext.Transport);
+            stream = reusableStream;
         }
 
         _streamLifetimeHandler.OnStreamCreated(stream);

src/Servers/Kestrel/Core/src/Internal/Http3/Http3Stream.cs

@@ -65,6 +65,8 @@ internal abstract partial class Http3Stream : HttpProtocol, IHttp3Stream, IHttpS
     private bool IsAbortedRead => (_completionState & StreamCompletionFlags.AbortedRead) == StreamCompletionFlags.AbortedRead;
     public bool IsCompleted => (_completionState & StreamCompletionFlags.Completed) == StreamCompletionFlags.Completed;
 
+    public bool CanReuse => !_connectionAborted && HasResponseCompleted;
+
     public Pipe RequestBodyPipe { get; private set; } = default!;
     public long? InputRemaining { get; internal set; }
     public QPackDecoder QPackDecoder { get; private set; } = default!;
@@ -941,7 +943,6 @@ private Task ProcessDataFrameAsync(in ReadOnlySequence<byte> payload)
     protected override void OnReset()
     {
         _keepAlive = true;
-        _connectionAborted = false;
         _userTrailers = null;
         _isWebTransportSessionAccepted = false;
         _isMethodConnect = false;