[3.1] Don't call AdvanceTo(default) in PipeReader.CopyToAsync

pkg/Microsoft.Private.PackageBaseline/packageIndex.json

@@ -2567,15 +2567,17 @@
         "4.5.2",
         "4.5.3",
         "4.6.0",
-        "4.7.0"
+        "4.7.0",
+        "4.7.1"
       ],
       "BaselineVersion": "4.7.0",
       "InboxOn": {},
       "AssemblyVersionInPackageVersion": {
         "4.0.0.0": "4.5.0",
         "4.0.0.1": "4.5.2",
         "4.0.1.0": "4.6.0",
-        "4.0.2.0": "4.7.0"
+        "4.0.2.0": "4.7.0",
+        "4.0.2.1": "4.7.1"
       }
     },
     "System.IO.Pipes": {

src/System.IO.Pipelines/Directory.Build.props

@@ -1,7 +1,8 @@
 <Project>
   <Import Project="..\Directory.Build.props" />
   <PropertyGroup>
-    <AssemblyVersion>4.0.2.0</AssemblyVersion>
+    <AssemblyVersion>4.0.2.1</AssemblyVersion>
+    <PackageVersion>4.7.1</PackageVersion>
     <StrongNameKeyId>Open</StrongNameKeyId>
   </PropertyGroup>
 </Project>

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeReader.cs

@@ -176,14 +176,13 @@ private async Task CopyToAsyncCore<TStream>(TStream destination, Func<TStream, R
         {
             while (true)
             {
-                SequencePosition consumed = default;
-
                 ReadResult result = await ReadAsync(cancellationToken).ConfigureAwait(false);
+                ReadOnlySequence<byte> buffer = result.Buffer;
+                SequencePosition position = buffer.Start;
+                SequencePosition consumed = position;
+
                 try
                 {
-                    ReadOnlySequence<byte> buffer = result.Buffer;
-                    SequencePosition position = buffer.Start;
-
                     if (result.IsCanceled)
                     {
                         ThrowHelper.ThrowOperationCanceledException_ReadCanceled();
@@ -196,11 +195,9 @@ private async Task CopyToAsyncCore<TStream>(TStream destination, Func<TStream, R
                         consumed = position;
                     }
 
-                    if (consumed.Equals(default))
-                    {
-                        consumed = buffer.End;
-                    }
-
+                    // The while loop completed successfully, so we've consumed the entire buffer.
+                    consumed = buffer.End;
+
                     if (result.IsCompleted)
                     {
                         break;

src/System.IO.Pipelines/tests/PipeReaderCopyToAsyncTests.cs

@@ -265,11 +265,95 @@ public async Task ThrowingFromStreamDoesNotLeavePipeReaderInBrokenState()
             Assert.Equal(20, result.Buffer.Length);
             pipe.Reader.Complete();
         }
+
+        [Theory]
+        [InlineData(0)]
+        [InlineData(1)]
+        public async Task ThrowingFromStreamCallsAdvanceToWithStartOfLastReadResult(int throwAfterNWrites)
+        {
+            var pipe = new Pipe(s_testOptions);
+            var wrappedPipeReader = new TestPipeReader(pipe.Reader);
+
+            var stream = new ThrowAfterNWritesStream(throwAfterNWrites);
+            Task task = wrappedPipeReader.CopyToAsync(stream);
+
+            pipe.Writer.WriteEmpty(10);
+            await pipe.Writer.FlushAsync();
+
+            // Write twice for the test case where the stream throws on the second write.
+            pipe.Writer.WriteEmpty(10);
+            await pipe.Writer.FlushAsync();
+
+            await Assert.ThrowsAsync<InvalidOperationException>(() => task);
+
+            SequencePosition startPosition = wrappedPipeReader.LastReadResult.Buffer.Start;
+
+            Assert.NotNull(startPosition.GetObject());
+            Assert.True(startPosition.Equals(wrappedPipeReader.LastConsumed));
+            Assert.True(startPosition.Equals(wrappedPipeReader.LastExamined));
+        }
+
         private class ThrowingStream : ThrowAfterNWritesStream
         {
             public ThrowingStream() : base(0)
             {
             }
         }
+
+        private class TestPipeReader : PipeReader
+        {
+            private readonly PipeReader _inner;
+
+
+            public TestPipeReader(PipeReader inner)
+            {
+                _inner = inner;
+            }
+
+            public ReadResult LastReadResult { get; private set; }
+            public SequencePosition LastConsumed { get; private set; }
+            public SequencePosition LastExamined { get; private set; }
+
+            public override void AdvanceTo(SequencePosition consumed)
+            {
+                LastConsumed = consumed;
+                LastExamined = consumed;
+                _inner.AdvanceTo(consumed);
+            }
+
+            public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
+            {
+                LastConsumed = consumed;
+                LastExamined = examined;
+                _inner.AdvanceTo(consumed);
+            }
+
+            public override void CancelPendingRead()
+            {
+                _inner.CancelPendingRead();
+            }
+
+            public override void Complete(Exception exception = null)
+            {
+                _inner.Complete(exception);
+            }
+
+            public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
+            {
+                LastReadResult = await _inner.ReadAsync(cancellationToken);
+                return LastReadResult;
+            }
+
+            public override bool TryRead(out ReadResult result)
+            {
+                if (_inner.TryRead(out result))
+                {
+                    LastReadResult = result;
+                    return true;
+                }
+
+                return false;
+            }
+        }
     }
 }

src/packages.builds

@@ -29,6 +29,9 @@
     <Project Include="$(MSBuildThisFileDirectory)System.Runtime.CompilerServices.Unsafe\pkg\System.Runtime.CompilerServices.Unsafe.pkgproj">
       <AdditionalProperties>$(AdditionalProperties)</AdditionalProperties>
     </Project>
+    <Project Include="$(MSBuildThisFileDirectory)System.IO.Pipelines\pkg\System.IO.Pipelines.pkgproj">
+      <AdditionalProperties>$(AdditionalProperties)</AdditionalProperties>
+    </Project>
   </ItemGroup>
 
   <ItemGroup>