Make reads that return canceled ReadResult throw (#7618)

src/Http/Http/src/ReadOnlyPipeStream.cs

@@ -162,6 +162,12 @@ namespace System.IO.Pipelines
             while (true)
             {
                 var result = await InnerPipeReader.ReadAsync(cancellationToken);
+
+                if (result.IsCanceled)
+                {
+                    ThrowHelper.ThrowOperationCanceledException_ReadCanceled();
+                }
+
                 var readableBuffer = result.Buffer;
                 var readableBufferLength = readableBuffer.Length;
 

src/Http/Http/src/ThrowHelper.cs

@@ -42,5 +42,9 @@ namespace System.IO.Pipelines
         public static void ThrowArgumentOutOfRangeException(string argument) => throw CreateArgumentOutOfRangeException(argument);
         [MethodImpl(MethodImplOptions.NoInlining)]
         public static Exception CreateArgumentOutOfRangeException(string argument) => new ArgumentOutOfRangeException(argument);
+
+        public static void ThrowOperationCanceledException_ReadCanceled() => throw CreateOperationCanceledException_ReadCanceled();
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        public static Exception CreateOperationCanceledException_ReadCanceled() => new OperationCanceledException("Read was canceled on underlying PipeReader.");
     }
 }

src/Http/Http/test/ReadOnlyPipeStreamTests.cs

@@ -157,6 +157,21 @@ namespace System.IO.Pipelines.Tests
             Assert.Equal(Reader, readOnlyPipeStream.InnerPipeReader);
         }
 
+        [Fact]
+        public async Task ThrowsOperationCanceledExceptionIfCancelPendingReadWasCalledOnInnerPipeReader()
+        {
+            var readOnlyPipeStream = new ReadOnlyPipeStream(Reader);
+            var readOperation = readOnlyPipeStream.ReadAsync(new byte[1]);
+
+            Assert.False(readOperation.IsCompleted);
+
+            Reader.CancelPendingRead();
+
+            var ex = await Assert.ThrowsAsync<OperationCanceledException>(async () => await readOperation);
+
+            Assert.Equal(ThrowHelper.CreateOperationCanceledException_ReadCanceled().Message, ex.Message);
+        }
+
         private async Task<Mock<PipeReader>> SetupMockPipeReader()
         {
             await WriteByteArrayToPipeAsync(new byte[1]);