Add abort on content stream disposal, fix NRE, fix abort direction, a…

…dd test

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

@@ -82,8 +82,9 @@ public void Dispose()
             if (!_disposed)
             {
                 _disposed = true;
-                _stream.Dispose();
-                DisposeSyncHelper();
+                var stream = Interlocked.Exchange(ref _stream, null!);
+                stream.Dispose();
+                DisposeSyncHelper(stream);
             }
         }
 
@@ -92,16 +93,15 @@ public async ValueTask DisposeAsync()
             if (!_disposed)
             {
                 _disposed = true;
-                await _stream.DisposeAsync().ConfigureAwait(false);
-                DisposeSyncHelper();
+                var stream = Interlocked.Exchange(ref _stream, null!);
+                await stream.DisposeAsync().ConfigureAwait(false);
+                DisposeSyncHelper(stream);
             }
         }
 
-        private void DisposeSyncHelper()
+        private void DisposeSyncHelper(QuicStream stream)
         {
-            _connection.RemoveStream(_stream);
-            _connection = null!;
-            _stream = null!;
+            Interlocked.Exchange(ref _connection, null!).RemoveStream(stream);
 
             _sendBuffer.Dispose();
             _recvBuffer.Dispose();
@@ -1111,18 +1111,26 @@ private void HandleReadResponseContentException(Exception ex, CancellationToken
                     throw new IOException(SR.net_http_client_execution_error, new HttpRequestException(SR.net_http_client_execution_error, abortException));
                 case Http3ConnectionException _:
                     // A connection-level protocol error has occurred on our stream.
-                    _connection.Abort(ex);
+                    _connection?.Abort(ex);
                     throw new IOException(SR.net_http_client_execution_error, new HttpRequestException(SR.net_http_client_execution_error, ex));
                 case OperationCanceledException oce when oce.CancellationToken == cancellationToken:
-                    _stream.AbortWrite((long)Http3ErrorCode.RequestCancelled);
+                    _stream?.AbortRead((long)Http3ErrorCode.RequestCancelled);
                     ExceptionDispatchInfo.Throw(ex); // Rethrow.
                     return; // Never reached.
                 default:
-                    _stream.AbortWrite((long)Http3ErrorCode.InternalError);
+                    _stream?.AbortRead((long)Http3ErrorCode.InternalError);
                     throw new IOException(SR.net_http_client_execution_error, new HttpRequestException(SR.net_http_client_execution_error, ex));
             }
         }
 
+        private void CancelResponseContentRead()
+        {
+            if (_responseDataPayloadRemaining != -1) // -1 indicates EOS
+            {
+                _stream.AbortRead((long)Http3ErrorCode.RequestCancelled);
+            }
+        }
+
         private async ValueTask<bool> ReadNextDataFrameAsync(HttpResponseMessage response, CancellationToken cancellationToken)
         {
             if (_responseDataPayloadRemaining == -1)
@@ -1194,6 +1202,7 @@ protected override void Dispose(bool disposing)
                 {
                     if (disposing)
                     {
+                        _stream.CancelResponseContentRead();
                         // This will remove the stream from the connection properly.
                         _stream.Dispose();
                     }
@@ -1216,6 +1225,7 @@ public override async ValueTask DisposeAsync()
             {
                 if (_stream != null)
                 {
+                    _stream.CancelResponseContentRead();
                     await _stream.DisposeAsync().ConfigureAwait(false);
                     _stream = null!;
                 }

src/libraries/System.Net.Http/tests/FunctionalTests/HttpClientHandlerTest.Http3.cs

@@ -214,6 +214,72 @@ public async Task Public_Interop_Upgrade_Success(string uri)
             }
         }
 
+        [ConditionalFact(nameof(IsMsQuicSupported))]
+        public async Task ResponseCancellationViaBothDisposeAndCancellationToken_Success()
+        {
+            if (UseQuicImplementationProvider != QuicImplementationProviders.MsQuic)
+            {
+                return;
+            }
+
+            using Http3LoopbackServer server = CreateHttp3LoopbackServer();
+
+            var pauseServerTcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var pauseClientTcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+            Task serverTask = Task.Run(async () =>
+            {
+                using Http3LoopbackConnection connection = (Http3LoopbackConnection)await server.EstablishGenericConnectionAsync();
+                using Http3LoopbackStream stream = await connection.AcceptRequestStreamAsync();
+                HttpRequestData request = await stream.ReadRequestDataAsync().ConfigureAwait(false);
+
+                int contentLength = 2*1024;
+                var headers = new List<HttpHeaderData>();
+                headers.Append(new HttpHeaderData("Content-Length", contentLength.ToString(CultureInfo.InvariantCulture)));
+
+                await stream.SendResponseHeadersAsync(HttpStatusCode.OK, headers).ConfigureAwait(false);
+                await stream.SendDataFrameAsync(new byte[1024]).ConfigureAwait(false);
+
+                await pauseServerTcs.Task.WaitAsync(TimeSpan.FromSeconds(10));
+
+                var ex = await Assert.ThrowsAsync<QuicStreamAbortedException>(() => stream.SendDataFrameAsync(new byte[1024]));
+
+                await stream.ShutdownSendAsync().ConfigureAwait(false);
+                pauseClientTcs.SetResult(true);
+                });
+
+            Task clientTask = Task.Run(async () =>
+            {
+                using HttpClient client = CreateHttpClient();
+
+                using HttpRequestMessage request = new()
+                    {
+                        Method = HttpMethod.Get,
+                        RequestUri = server.Address,
+                        Version = HttpVersion30,
+                        VersionPolicy = HttpVersionPolicy.RequestVersionExact
+                    };
+                HttpResponseMessage response = await client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead).WaitAsync(TimeSpan.FromSeconds(10));
+
+                Stream stream = await response.Content.ReadAsStreamAsync();
+
+                int bytesRead = await stream.ReadAsync(new byte[1024]);
+                Assert.Equal(1024, bytesRead);
+
+                var cts = new CancellationTokenSource();
+
+                cts.Token.Register(() => response.Dispose());
+                cts.CancelAfter(200);
+
+                await Assert.ThrowsAsync<OperationCanceledException>(() => stream.ReadAsync(new byte[1024], cancellationToken: cts.Token).AsTask());
+
+                pauseServerTcs.SetResult(true);
+                await pauseClientTcs.Task.WaitAsync(TimeSpan.FromSeconds(3));
+            });
+
+            await new[] { clientTask, serverTask }.WhenAllOrAnyFailed(20_000);
+        }
+
         /// <summary>
         /// These are public interop test servers for various QUIC and HTTP/3 implementations,
         /// taken from https://github.com/quicwg/base-drafts/wiki/Implementations

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/MsQuic/MsQuicStream.cs

@@ -249,6 +249,10 @@ private async ValueTask<CancellationTokenRegistration> HandleWriteStartState(Can
                 if (_state.SendState == SendState.Aborted)
                 {
                     cancellationToken.ThrowIfCancellationRequested();
+                    if (_state.SendErrorCode != -1)
+                    {
+                        throw new QuicStreamAbortedException(_state.SendErrorCode);
+                    }
                     throw new OperationCanceledException(SR.net_quic_sending_aborted);
                 }
                 else if (_state.SendState == SendState.ConnectionClosed)