Detect when a read includes FIN

src/libraries/System.Net.Quic/ref/System.Net.Quic.cs

@@ -87,6 +87,7 @@ internal QuicStream() { }
         public override bool CanWrite { get { throw null; } }
         public override long Length { get { throw null; } }
         public override long Position { get { throw null; } set { } }
+        public bool ReadsCompleted { get { throw null; } }
         public long StreamId { get { throw null; } }
         public void AbortRead(long errorCode) { }
         public void AbortWrite(long errorCode) { }

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/Mock/MockStream.cs

@@ -44,6 +44,8 @@ internal override long StreamId
 
         internal override bool CanRead => !_disposed && ReadStreamBuffer is not null;
 
+        internal override bool ReadsCompleted => ReadStreamBuffer?.IsComplete ?? true;
+
         internal override int Read(Span<byte> buffer)
         {
             CheckDisposed();

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

@@ -1,6 +1,7 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System;
 using System.Buffers;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -10,6 +11,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using static System.Net.Quic.Implementations.MsQuic.Internal.MsQuicNativeMethods;
+using static System.Net.Quic.Implementations.MsQuic.MsQuicConnection;
 
 namespace System.Net.Quic.Implementations.MsQuic
 {
@@ -42,6 +44,8 @@ private sealed class State
 
             public ReadState ReadState;
 
+            public bool FinReceived;
+
             // set when ReadState.Aborted:
             public long ReadErrorCode = -1;
 
@@ -192,6 +196,8 @@ internal MsQuicStream(MsQuicConnection.State connectionState, QUIC_STREAM_OPEN_F
 
         internal override bool CanWrite => _disposed == 0 && _canWrite;
 
+        internal override bool ReadsCompleted => _state.ReadState >= ReadState.ReadsCompleted;
+
         internal override long StreamId
         {
             get
@@ -842,6 +848,7 @@ private static unsafe uint HandleEventRecv(State state, ref StreamEvent evt)
                         state.ReceiveQuicBuffersCount = (int)receiveEvent.BufferCount;
                         state.ReceiveQuicBuffersTotalBytes = checked((int)receiveEvent.TotalBufferLength);
                         state.ReadState = ReadState.IndividualReadComplete;
+                        state.FinReceived = receiveEvent.Flags.HasFlag(QUIC_RECEIVE_FLAGS.FIN);
                         return MsQuicStatusCodes.Pending;
                     case ReadState.PendingRead:
                         // There is a pending ReadAsync().
@@ -852,6 +859,11 @@ private static unsafe uint HandleEventRecv(State state, ref StreamEvent evt)
                         state.ReadState = ReadState.None;
 
                         readLength = CopyMsQuicBuffersToUserBuffer(new ReadOnlySpan<QuicBuffer>(receiveEvent.Buffers, (int)receiveEvent.BufferCount), state.ReceiveUserBuffer.Span);
+                        if (receiveEvent.Flags.HasFlag(QUIC_RECEIVE_FLAGS.FIN) && readLength == receiveEvent.BufferCount)
+                        {
+                            state.ReadState = ReadState.ReadsCompleted;
+                            state.FinReceived = true;
+                        }
                         state.ReceiveUserBuffer = null;
                         break;
                     default:

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/QuicStreamProvider.cs

@@ -13,6 +13,8 @@ internal abstract class QuicStreamProvider : IDisposable, IAsyncDisposable
 
         internal abstract bool CanRead { get; }
 
+        internal abstract bool ReadsCompleted { get; }
+
         internal abstract int Read(Span<byte> buffer);
 
         internal abstract ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default);

src/libraries/System.Net.Quic/src/System/Net/Quic/QuicStream.cs

@@ -71,6 +71,8 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati
 
         public override bool CanRead => _provider.CanRead;
 
+        public bool ReadsCompleted => _provider.ReadsCompleted;
+
         public override int Read(Span<byte> buffer) => _provider.Read(buffer);
 
         public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default) => _provider.ReadAsync(buffer, cancellationToken);

src/libraries/System.Net.Quic/tests/FunctionalTests/QuicStreamTests.cs

@@ -27,10 +27,12 @@ await RunClientServer(
                 serverFunction: async connection =>
                 {
                     await using QuicStream stream = await connection.AcceptStreamAsync();
+                    Assert.False(stream.ReadsCompleted);
 
                     byte[] buffer = new byte[s_data.Length];
                     int bytesRead = await ReadAll(stream, buffer);
 
+                    Assert.True(stream.ReadsCompleted);
                     Assert.Equal(s_data.Length, bytesRead);
                     Assert.Equal(s_data, buffer);
 
@@ -40,12 +42,14 @@ await RunClientServer(
                 clientFunction: async connection =>
                 {
                     await using QuicStream stream = connection.OpenBidirectionalStream();
+                    Assert.False(stream.ReadsCompleted);
 
                     await stream.WriteAsync(s_data, endStream: true);
 
                     byte[] buffer = new byte[s_data.Length];
                     int bytesRead = await ReadAll(stream, buffer);
 
+                    Assert.True(stream.ReadsCompleted);
                     Assert.Equal(s_data.Length, bytesRead);
                     Assert.Equal(s_data, buffer);
 
@@ -503,6 +507,30 @@ await RunBidirectionalClientServer(
                 });
         }
 
+        [Fact]
+        public async Task Read_ReadsCompleted_ReportedBeforeReturning0()
+        {
+            await RunBidirectionalClientServer(
+                async clientStream =>
+                {
+                    await clientStream.WriteAsync(new byte[1], endStream: true);
+                },
+                async serverStream =>
+                {
+                    Assert.False(serverStream.ReadsCompleted);
+                    var received = await serverStream.ReadAsync(new byte[1]);
+                    Assert.Equal(1, received);
+                    Assert.True(serverStream.ReadsCompleted);
+
+                    var task = serverStream.ReadAsync(new byte[1]);
+                    Assert.True(task.IsCompleted);
+
+                    received = await task;
+                    Assert.Equal(0, received);
+                    Assert.True(serverStream.ReadsCompleted);
+                });
+        }
+
         [Fact]
         public async Task ReadOutstanding_ReadAborted_Throws()
         {