Improve Http1ContentLengthMessageBody's reset logic (#25799)

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

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
@@ -9,6 +10,8 @@ using Microsoft.AspNetCore.Connections;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 {
+    using BadHttpRequestException = Microsoft.AspNetCore.Http.BadHttpRequestException;
+
     internal sealed class Http1ContentLengthMessageBody : Http1MessageBody
     {
         private ReadResult _readResult;
@@ -18,6 +21,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private bool _isReading;
         private int _userCanceled;
         private bool _finalAdvanceCalled;
+        private bool _cannotResetInputPipe;
 
         public Http1ContentLengthMessageBody(bool keepAlive, long contentLength, Http1Connection context)
             : base(context)
@@ -35,15 +39,19 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         public override async ValueTask<ReadResult> ReadAsyncInternal(CancellationToken cancellationToken = default)
         {
-            if (_isReading)
+            VerifyIsNotReading();
-            {
-                throw new InvalidOperationException("Reading is already in progress.");
-            }
 
             if (_readCompleted)
             {
                 _isReading = true;
-                return new ReadResult(_readResult.Buffer, Interlocked.Exchange(ref _userCanceled, 0) == 1, _readResult.IsCompleted);
+                return new ReadResult(_readResult.Buffer, Interlocked.Exchange(ref _userCanceled, 0) == 1, isCompleted: true);
+            }
+
+            // The issue is that TryRead can get a canceled read result
+            // which is unknown to StartTimingReadAsync.
+            if (_context.RequestTimedOut)
+            {
+                KestrelBadHttpRequestException.Throw(RequestRejectionReason.RequestBodyTimeout);
             }
 
             TryStart();
@@ -54,12 +62,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             // We internally track an int for that.
             while (true)
             {
-                // The issue is that TryRead can get a canceled read result
-                // which is unknown to StartTimingReadAsync. 
-                if (_context.RequestTimedOut)
-                {
-                    KestrelBadHttpRequestException.Throw(RequestRejectionReason.RequestBodyTimeout);
-                }
 
                 try
                 {
@@ -76,10 +78,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
                 void ResetReadingState()
                 {
-                    _isReading = false;
                     // Reset the timing read here for the next call to read.
                     StopTimingRead(0);
-                    _context.Input.AdvanceTo(_readResult.Buffer.Start);
+
+                    if (!_cannotResetInputPipe)
+                    {
+                        _isReading = false;
+                        _context.Input.AdvanceTo(_readResult.Buffer.Start);
+                    }
                 }
 
                 if (_context.RequestTimedOut)
@@ -95,7 +101,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 }
 
                 // Ignore the canceled readResult if it wasn't canceled by the user.
-                if (!_readResult.IsCanceled || Interlocked.Exchange(ref _userCanceled, 0) == 1)
+                // Normally we do not return a canceled ReadResult unless CancelPendingRead was called on the request body PipeReader itself,
+                // but if the last call to AdvanceTo examined data it did not consume, we cannot reset the state of the Input pipe.
+                // https://github.com/dotnet/aspnetcore/issues/19476
+                if (!_readResult.IsCanceled || Interlocked.Exchange(ref _userCanceled, 0) == 1 || _cannotResetInputPipe)
                 {
                     var returnedReadResultLength = CreateReadResultFromConnectionReadResult();
 
@@ -124,18 +133,20 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         public override bool TryReadInternal(out ReadResult readResult)
         {
-            if (_isReading)
+            VerifyIsNotReading();
-            {
-                throw new InvalidOperationException("Reading is already in progress.");
-            }
 
             if (_readCompleted)
             {
                 _isReading = true;
-                readResult = new ReadResult(_readResult.Buffer, Interlocked.Exchange(ref _userCanceled, 0) == 1, _readResult.IsCompleted);
+                readResult = new ReadResult(_readResult.Buffer, Interlocked.Exchange(ref _userCanceled, 0) == 1, isCompleted: true);
                 return true;
             }
 
+            if (_context.RequestTimedOut)
+            {
+                KestrelBadHttpRequestException.Throw(RequestRejectionReason.RequestBodyTimeout);
+            }
+
             TryStart();
 
             // The while(true) because we don't want to return a canceled ReadResult if the user themselves didn't cancel it.
@@ -147,7 +158,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                     return false;
                 }
 
-                if (!_readResult.IsCanceled || Interlocked.Exchange(ref _userCanceled, 0) == 1)
+                if (!_readResult.IsCanceled || Interlocked.Exchange(ref _userCanceled, 0) == 1 || _cannotResetInputPipe)
                 {
                     break;
                 }
@@ -157,7 +168,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
             if (_readResult.IsCompleted)
             {
-                _context.Input.AdvanceTo(_readResult.Buffer.Start);
+                if (_cannotResetInputPipe)
+                {
+                    _isReading = true;
+                }
+                else
+                {
+                    _context.Input.AdvanceTo(_readResult.Buffer.Start);
+                }
+
                 ThrowUnexpectedEndOfRequestContent();
             }
 
@@ -214,7 +233,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             if (_readCompleted)
             {
                 // If the old stored _readResult was canceled, it's already been observed. Do not store a canceled read result permanently.
-                _readResult = new ReadResult(_readResult.Buffer.Slice(consumed, _readResult.Buffer.End), isCanceled: false, _readCompleted);
+                _readResult = new ReadResult(_readResult.Buffer.Slice(consumed, _readResult.Buffer.End), isCanceled: false, isCompleted: true);
 
                 if (!_finalAdvanceCalled && _readResult.Buffer.Length == 0)
                 {
@@ -226,6 +245,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 return;
             }
 
+            // If consumed != examined, we cannot reset _context.Input back to a non-reading state after the next call to ReadAsync
+            // simply by calling _context.Input.AdvanceTo(_readResult.Buffer.Start) because the DefaultPipeReader will complain that
+            // "The examined position cannot be less than the previously examined position."
+            _cannotResetInputPipe = !consumed.Equals(examined);
             _unexaminedInputLength -= TrackConsumedAndExaminedBytes(_readResult, consumed, examined);
             _context.Input.AdvanceTo(consumed, examined);
         }
@@ -255,5 +278,29 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             Complete(null);
             return Task.CompletedTask;
         }
+
+        [StackTraceHidden]
+        private void VerifyIsNotReading()
+        {
+            if (!_isReading)
+            {
+                return;
+            }
+
+            if (_cannotResetInputPipe)
+            {
+                if (_readResult.IsCompleted)
+                {
+                    KestrelBadHttpRequestException.Throw(RequestRejectionReason.UnexpectedEndOfRequestContent);
+                }
+
+                if (_context.RequestTimedOut)
+                {
+                    KestrelBadHttpRequestException.Throw(RequestRejectionReason.RequestBodyTimeout);
+                }
+            }
+
+            throw new InvalidOperationException("Reading is already in progress.");
+        }
     }
 }

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

@@ -89,7 +89,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                     AdvanceTo(result.Buffer.End);
                 } while (!result.IsCompleted);
             }
-            catch (Microsoft.AspNetCore.Http.BadHttpRequestException ex)
+            catch (BadHttpRequestException ex)
             {
                 _context.SetBadRequestState(ex);
             }

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

@@ -1226,6 +1226,39 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
                 input.Application.Output.Complete();
 
+#pragma warning disable CS0618 // Type or member is obsolete
+                var ex0 = Assert.Throws<BadHttpRequestException>(() => reader.TryRead(out var readResult));
+                var ex1 = Assert.Throws<BadHttpRequestException>(() => reader.TryRead(out var readResult));
+                var ex2 = await Assert.ThrowsAsync<BadHttpRequestException>(() => reader.ReadAsync().AsTask());
+                var ex3 = await Assert.ThrowsAsync<BadHttpRequestException>(() => reader.ReadAsync().AsTask());
+#pragma warning restore CS0618 // Type or member is obsolete
+
+                Assert.Equal(RequestRejectionReason.UnexpectedEndOfRequestContent, ex0.Reason);
+                Assert.Equal(RequestRejectionReason.UnexpectedEndOfRequestContent, ex1.Reason);
+                Assert.Equal(RequestRejectionReason.UnexpectedEndOfRequestContent, ex2.Reason);
+                Assert.Equal(RequestRejectionReason.UnexpectedEndOfRequestContent, ex3.Reason);
+
+                await body.StopAsync();
+            }
+        }
+
+        [Fact]
+        public async Task UnexpectedEndOfRequestContentIsRepeatedlyThrownForContentLengthBodyAfterExaminingButNotConsumingBytes()
+        {
+            using (var input = new TestInput())
+            {
+                var body = Http1MessageBody.For(HttpVersion.Http11, new HttpRequestHeaders { HeaderContentLength = "5" }, input.Http1Connection);
+                var reader = new HttpRequestPipeReader();
+                reader.StartAcceptingReads(body);
+
+                await input.Application.Output.WriteAsync(new byte[] { 0 });
+
+                var readResult = await reader.ReadAsync();
+
+                reader.AdvanceTo(readResult.Buffer.Start, readResult.Buffer.End);
+
+                input.Application.Output.Complete();
+
 #pragma warning disable CS0618 // Type or member is obsolete
                 var ex0 = Assert.Throws<BadHttpRequestException>(() => reader.TryRead(out var readResult));
                 var ex1 = Assert.Throws<BadHttpRequestException>(() => reader.TryRead(out var readResult));