Fix NestedPipeReader.AdvanceTo after ReadAsync encounters end of underlying PipeReader

This includes a behavioral breaking change that brings it in line with the behavior of `DefaultPipeReader`, which is that we fully expect `AdvanceTo` to be called even when the end of the `NestedPipeReader` is reached. Otherwise the underlying `PipeReader.ReadAsync` will throw when next called.

Fixes #204

Author: AArnott

src/Nerdbank.Streams.Tests/NestedPipeReaderTests.cs

@@ -46,6 +46,7 @@ public void TryRead_AllAtOnce_ExamineEverything()
         Assert.True(sliceReader.TryRead(out readResult));
         Assert.True(readResult.IsCompleted);
         Assert.True(readResult.Buffer.IsEmpty);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
 
         // Verify that the original PipeReader can still produce bytes.
         Assert.True(this.pipe.Reader.TryRead(out readResult));
@@ -65,6 +66,7 @@ public void TryRead_AllAtOnce()
         Assert.True(sliceReader.TryRead(out readResult));
         Assert.True(readResult.IsCompleted);
         Assert.Equal(0, readResult.Buffer.Length);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
 
         // Verify that the original PipeReader can still produce bytes.
         Assert.True(this.pipe.Reader.TryRead(out readResult));
@@ -113,6 +115,7 @@ public async Task ReadAsync_AllAtOnce()
         readResult = await sliceReader.ReadAsync(this.TimeoutToken);
         Assert.True(readResult.IsCompleted);
         Assert.True(readResult.Buffer.IsEmpty);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
 
         // Verify that the original PipeReader can still produce bytes.
         Assert.True(this.pipe.Reader.TryRead(out readResult));
@@ -138,6 +141,7 @@ public async Task ReadAsync_ExamineEverything()
         readResult = await sliceReader.ReadAsync(this.TimeoutToken);
         Assert.True(readResult.IsCompleted);
         Assert.True(readResult.Buffer.IsEmpty);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
 
         // Verify that the original PipeReader can still produce bytes.
         Assert.True(this.pipe.Reader.TryRead(out readResult));
@@ -164,6 +168,7 @@ public async Task ReadAsync_MultipleReads()
         readResult = await sliceReader.ReadAsync(this.TimeoutToken);
         Assert.True(readResult.IsCompleted);
         Assert.Equal(0, readResult.Buffer.Length);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
 
         // Verify that the original PipeReader can still produce bytes.
         Assert.True(this.pipe.Reader.TryRead(out readResult));
@@ -186,6 +191,61 @@ public async Task ReadAsync_SliceExceedsUnderlyingLength()
         Assert.True(readResult.IsCompleted);
     }
 
+    [Fact]
+    public async Task ReadAsync_TwiceOnCompletion_Throws()
+    {
+        this.pipe.Writer.Complete();
+
+        var sliceReader = this.pipe.Reader.ReadSlice(OriginalBuffer.Length);
+        var readResult = await sliceReader.ReadAsync(this.TimeoutToken);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
+        await sliceReader.ReadAsync(this.TimeoutToken);
+        await Assert.ThrowsAsync<InvalidOperationException>(async () => await sliceReader.ReadAsync(this.TimeoutToken));
+    }
+
+    [Fact]
+    public async Task ReadAsync_ToEnd_AdvanceTo_Partial_ThenReadAsyncAgain()
+    {
+        this.pipe.Writer.Complete();
+
+        var sliceReader = this.pipe.Reader.ReadSlice(OriginalBuffer.Length);
+        var readResult = await sliceReader.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.IsCompleted);
+        sliceReader.AdvanceTo(readResult.Buffer.GetPosition(3));
+
+        readResult = await sliceReader.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.IsCompleted);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
+
+        readResult = await sliceReader.ReadAsync(this.TimeoutToken);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
+
+        readResult = await this.pipe.Reader.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.IsCompleted);
+        this.pipe.Reader.AdvanceTo(readResult.Buffer.End);
+    }
+
+    [Fact]
+    public async Task ReadAsync_AfterReadingExactBytes()
+    {
+        var slice = this.pipe.Reader.ReadSlice(OriginalBuffer.Length);
+        var readResult = await slice.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.IsCompleted);
+        slice.AdvanceTo(readResult.Buffer.End);
+
+        // Try to read again...
+        readResult = await slice.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.Buffer.IsEmpty);
+        Assert.True(readResult.IsCompleted);
+        slice.AdvanceTo(readResult.Buffer.End);
+
+        // And again...
+        readResult = await slice.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.Buffer.IsEmpty);
+        Assert.True(readResult.IsCompleted);
+        slice.AdvanceTo(readResult.Buffer.End);
+    }
+
     [Fact]
     public void OnWriterCompleted_NoOps()
     {
@@ -199,30 +259,51 @@ public void OnWriterCompleted_NoOps()
     }
 
     [Fact]
-    public void TryRead_ThrowsAfterCompleting()
+    public void TryRead_ThrowsAfterCompleting_Prematurely()
     {
         var sliceReader = this.pipe.Reader.ReadSlice(5);
         sliceReader.Complete();
         Assert.Throws<InvalidOperationException>(() => sliceReader.TryRead(out ReadResult result));
     }
 
     [Fact]
-    public async Task ReadAsync_ThrowsAfterCompleting()
+    public async Task ReadAsync_ThrowsAfterCompleting_Prematurely()
     {
         var sliceReader = this.pipe.Reader.ReadSlice(5);
         sliceReader.Complete();
         await Assert.ThrowsAsync<InvalidOperationException>(() => sliceReader.ReadAsync(this.TimeoutToken).AsTask());
     }
 
     [Fact]
-    public void Complete_DoesNotCompleteUnderlyingReader()
+    public void TryRead_ThrowsAfterCompleting_AfterFullRead()
     {
         var sliceReader = this.pipe.Reader.ReadSlice(5);
+        Assert.True(sliceReader.TryRead(out ReadResult readResult));
+        Assert.True(readResult.IsCompleted);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
         sliceReader.Complete();
-        Assert.True(this.pipe.Reader.TryRead(out ReadResult result));
-        Assert.Equal(OriginalBuffer.Length, result.Buffer.Length);
+        Assert.Throws<InvalidOperationException>(() => sliceReader.TryRead(out ReadResult result));
+
+        Assert.True(this.pipe.Reader.TryRead(out readResult));
+        Assert.False(readResult.Buffer.IsEmpty);
         this.pipe.Reader.Complete();
-        Assert.Throws<InvalidOperationException>(() => this.pipe.Reader.TryRead(out result));
+        Assert.Throws<InvalidOperationException>(() => this.pipe.Reader.TryRead(out ReadResult result));
+    }
+
+    [Fact]
+    public async Task ReadAsync_ThrowsAfterCompleting_AfterFullRead()
+    {
+        var sliceReader = this.pipe.Reader.ReadSlice(5);
+        var readResult = await sliceReader.ReadAsync(this.TimeoutToken);
+        Assert.True(readResult.IsCompleted);
+        sliceReader.AdvanceTo(readResult.Buffer.End);
+        sliceReader.Complete();
+        await Assert.ThrowsAsync<InvalidOperationException>(() => sliceReader.ReadAsync(this.TimeoutToken).AsTask());
+
+        readResult = await this.pipe.Reader.ReadAsync(this.TimeoutToken);
+        Assert.False(readResult.Buffer.IsEmpty);
+        this.pipe.Reader.Complete();
+        await Assert.ThrowsAsync<InvalidOperationException>(() => this.pipe.Reader.ReadAsync(this.TimeoutToken).AsTask());
     }
 
     [Fact]
@@ -233,6 +314,25 @@ public void Complete_WithException_DoesCompleteUnderlyingReader()
         Assert.Throws<InvalidOperationException>(() => this.pipe.Reader.TryRead(out ReadResult result));
     }
 
+    [Fact]
+    public void Complete_Twice_WithoutReading()
+    {
+        var sliceReader = this.pipe.Reader.ReadSlice(5);
+        sliceReader.Complete();
+        sliceReader.Complete();
+    }
+
+    [Fact]
+    public void Complete_Twice_AfterReading()
+    {
+        var sliceReader = this.pipe.Reader.ReadSlice(5);
+        Assert.True(sliceReader.TryRead(out ReadResult result));
+        Assert.True(result.IsCompleted);
+        sliceReader.AdvanceTo(result.Buffer.End);
+        sliceReader.Complete();
+        sliceReader.Complete();
+    }
+
     [Fact]
     public void CancelPendingRead_UnderlyingReader_TryRead()
     {

src/Nerdbank.Streams/NestedPipeReader.cs

@@ -18,10 +18,8 @@ internal class NestedPipeReader : PipeReader
         private readonly PipeReader pipeReader;
         private long length;
         private long consumedLength;
-        private long examinedLength;
         private ReadResult resultOfPriorRead;
-        private bool nextReadCanceled;
-        private bool readerCompleted;
+        private bool completed;
 
         public NestedPipeReader(PipeReader pipeReader, long length)
         {
@@ -35,52 +33,33 @@ public NestedPipeReader(PipeReader pipeReader, long length)
         private long RemainingLength => this.length - this.consumedLength;
 
         /// <inheritdoc/>
-        public override void AdvanceTo(SequencePosition consumed)
-        {
-            if (this.Consumed(consumed, consumed))
-            {
-                this.pipeReader.AdvanceTo(consumed);
-
-                // When we call AdvanceTo on the underlying reader, we're not allowed to reference their buffer any more, so clear it to be safe.
-                this.resultOfPriorRead = new ReadResult(default, isCanceled: false, isCompleted: this.resultOfPriorRead.IsCompleted);
-            }
-        }
+        public override void AdvanceTo(SequencePosition consumed) => this.AdvanceTo(consumed, consumed);
 
         /// <inheritdoc/>
         public override void AdvanceTo(SequencePosition consumed, SequencePosition examined)
         {
-            if (this.Consumed(consumed, examined))
-            {
-                this.pipeReader.AdvanceTo(consumed, examined);
+            this.consumedLength += this.resultOfPriorRead.Buffer.Slice(0, consumed).Length;
+            this.pipeReader.AdvanceTo(consumed, examined);
 
-                // When we call AdvanceTo on the underlying reader, we're not allowed to reference their buffer any more, so clear it to be safe.
-                this.resultOfPriorRead = new ReadResult(default, isCanceled: false, isCompleted: this.resultOfPriorRead.IsCompleted);
-            }
+            // When we call AdvanceTo on the underlying reader, we're not allowed to reference their buffer any more, so clear it to be safe.
+            this.resultOfPriorRead = new ReadResult(default, isCanceled: false, isCompleted: this.resultOfPriorRead.IsCompleted);
         }
 
         /// <inheritdoc/>
-        public override void CancelPendingRead()
-        {
-            if (this.resultOfPriorRead.IsCompleted)
-            {
-                this.nextReadCanceled = true;
-            }
-            else
-            {
-                this.pipeReader.CancelPendingRead();
-            }
-        }
+        public override void CancelPendingRead() => this.pipeReader.CancelPendingRead();
 
         /// <inheritdoc/>
+        /// <remarks>
+        /// If the slice has not been fully read or if <paramref name="exception"/> is non-null, this call propagates to the underlying <see cref="PipeReader"/>.
+        /// But if the slice is fully read without errors, the call is suppressed so the underlying <see cref="PipeReader"/> can continue to function.
+        /// </remarks>
         public override void Complete(Exception? exception = null)
         {
-            // We don't want a nested PipeReader to complete the underlying one unless there was an exception.
-            if (exception != null)
+            this.completed = true;
+            if (exception is object || this.RemainingLength > 0)
             {
                 this.pipeReader.Complete(exception);
             }
-
-            this.readerCompleted = true;
         }
 
         /// <inheritdoc/>
@@ -92,18 +71,25 @@ public override void OnWriterCompleted(Action<Exception, object> callback, objec
         /// <inheritdoc/>
         public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancellationToken = default)
         {
-            Verify.Operation(!this.readerCompleted, Strings.ReadingAfterCompletionNotAllowed);
-            if (this.resultOfPriorRead.IsCompleted)
+            Verify.Operation(!this.completed, Strings.ReadingAfterCompletionNotAllowed);
+
+            ReadResult result;
+            if (this.RemainingLength == 0)
+            {
+                // We do NOT want to block on reading more bytes since we don't expect or want any.
+                // But we DO want to put the underlying reader back into a reading mode so we don't
+                // have to emulate that in our class.
+                this.pipeReader.TryRead(out result);
+            }
+            else
             {
-                var cachedResult = this.resultOfPriorRead = new ReadResult(this.resultOfPriorRead.Buffer, isCanceled: this.nextReadCanceled, isCompleted: true);
-                this.nextReadCanceled = false;
-                return cachedResult;
+                result = await this.pipeReader.ReadAsync(cancellationToken).ConfigureAwait(false);
             }
 
-            var result = await this.pipeReader.ReadAsync(cancellationToken).ConfigureAwait(false);
+            // Do not allow the reader to exceed the length of this slice.
             if (result.Buffer.Length >= this.RemainingLength)
             {
-                result = new ReadResult(result.Buffer.Slice(0, this.RemainingLength), isCanceled: false, isCompleted: true);
+                result = new ReadResult(result.Buffer.Slice(0, this.RemainingLength), isCanceled: result.IsCanceled, isCompleted: true);
             }
 
             return this.resultOfPriorRead = result;
@@ -112,19 +98,13 @@ public override async ValueTask<ReadResult> ReadAsync(CancellationToken cancella
         /// <inheritdoc/>
         public override bool TryRead(out ReadResult result)
         {
-            Verify.Operation(!this.readerCompleted, Strings.ReadingAfterCompletionNotAllowed);
-            if (this.resultOfPriorRead.IsCompleted)
-            {
-                result = this.resultOfPriorRead = new ReadResult(this.resultOfPriorRead.Buffer, isCanceled: this.nextReadCanceled, isCompleted: true);
-                this.nextReadCanceled = false;
-                return true;
-            }
-
+            Verify.Operation(!this.completed, Strings.ReadingAfterCompletionNotAllowed);
             if (this.pipeReader.TryRead(out result))
             {
+                // Do not allow the reader to exceed the length of this slice.
                 if (result.Buffer.Length > this.RemainingLength)
                 {
-                    result = new ReadResult(result.Buffer.Slice(0, this.RemainingLength), isCanceled: false, isCompleted: true);
+                    result = new ReadResult(result.Buffer.Slice(0, this.RemainingLength), isCanceled: result.IsCanceled, isCompleted: true);
                 }
 
                 this.resultOfPriorRead = result;
@@ -133,21 +113,5 @@ public override bool TryRead(out ReadResult result)
 
             return false;
         }
-
-        private bool Consumed(SequencePosition consumed, SequencePosition examined)
-        {
-            this.consumedLength += this.resultOfPriorRead.Buffer.Slice(0, consumed).Length;
-            this.examinedLength += this.resultOfPriorRead.Buffer.Slice(0, examined).Length;
-
-            if (this.resultOfPriorRead.IsCompleted)
-            {
-                // We have to keep our own ReadResult current since we can't call the inner read methods any more.
-                this.resultOfPriorRead = new ReadResult(this.resultOfPriorRead.Buffer.Slice(consumed), isCanceled: false, isCompleted: true);
-            }
-
-            // The caller is allowed to propagate the AdvanceTo call to the underlying reader iff
-            // we will be reading again from it, or our own final buffer is empty.
-            return !this.resultOfPriorRead.IsCompleted || this.resultOfPriorRead.Buffer.IsEmpty;
-        }
     }
 }