Frame ordering

Author: ladeak

src/CHttpServer/CHttpServer/Http3/Http3Stream.cs

@@ -18,7 +18,9 @@ internal sealed partial class Http3Stream
     private enum StreamReadingStatus
     {
         ReadingFrameHeader,
-        ReadingPayload,
+        ReadingPayloadHeader,
+        ReadingPayloadData,
+        ReadingPayloadReserved,
     }
 
     private QuicStream? _quicStream;
@@ -96,7 +98,6 @@ public async void ProcessStream<TContext>(IHttpApplication<TContext> application
     public async Task ProcessStreamAsync<TContext>(IHttpApplication<TContext> application, CancellationToken token)
         where TContext : notnull
     {
-        ulong frameType = 0;
         long payloadRemainingLength = 0;
         Task? applicationProcessing = null;
         var streamReadingState = StreamReadingStatus.ReadingFrameHeader;
@@ -112,7 +113,7 @@ public async Task ProcessStreamAsync<TContext>(IHttpApplication<TContext> applic
 
                 while (!buffer.IsEmpty)
                 {
-                    long bufferConsumed = ProcessStreamAsync(application, ref frameType, ref payloadRemainingLength, ref applicationProcessing, ref streamReadingState, buffer, token);
+                    long bufferConsumed = ProcessStreamAsync(application, ref payloadRemainingLength, ref applicationProcessing, ref streamReadingState, buffer, token);
 
                     // Could not further process. Break the inner loop to read more data
                     if (bufferConsumed == 0)
@@ -130,6 +131,11 @@ public async Task ProcessStreamAsync<TContext>(IHttpApplication<TContext> applic
         {
             // Ignore
         }
+        catch (Http3ConnectionException connectionError)
+        {
+            _connection?.StreamError(connectionError.ErrorCode);
+            throw;
+        }
         catch (Exception ex)
         {
             Debug.WriteLine(ex);
@@ -141,7 +147,6 @@ public async Task ProcessStreamAsync<TContext>(IHttpApplication<TContext> applic
     }
 
     private long ProcessStreamAsync<TContext>(IHttpApplication<TContext> application,
-        ref ulong frameType,
         ref long payloadRemainingLength,
         ref Task? applicationProcessing,
         ref StreamReadingStatus streamReadingState,
@@ -153,7 +158,7 @@ private long ProcessStreamAsync<TContext>(IHttpApplication<TContext> application
         if (streamReadingState == StreamReadingStatus.ReadingFrameHeader)
         {
             // FrameType is a single byte.
-            if (!VariableLenghtIntegerDecoder.TryRead(buffer.FirstSpan, out frameType, out int bytesReadFrameType))
+            if (!VariableLenghtIntegerDecoder.TryRead(buffer.FirstSpan, out var frameType, out int bytesReadFrameType))
             {
                 // Not enough data to read payload length.
                 return 0;
@@ -165,49 +170,59 @@ private long ProcessStreamAsync<TContext>(IHttpApplication<TContext> application
                 return 0;
             }
             payloadRemainingLength = checked((long)payloadLength);
-            streamReadingState = StreamReadingStatus.ReadingPayload;
+            streamReadingState = NextRequestReadingState(applicationProcessing, frameType);
             buffer = buffer.Slice(bytesReadPayloadLength);
             bufferConsumed += bytesReadFrameType + bytesReadPayloadLength;
             return bufferConsumed;
         }
 
-        if (streamReadingState == StreamReadingStatus.ReadingPayload)
+        if (payloadRemainingLength < buffer.Length)
+            buffer = buffer.Slice(0, payloadRemainingLength);
+
+        if (streamReadingState == StreamReadingStatus.ReadingPayloadHeader)
         {
-            switch (frameType)
-            {
-                case 0x0: // DATA
-                    if (payloadRemainingLength < buffer.Length)
-                        buffer = buffer.Slice(0, payloadRemainingLength);
-                    bufferConsumed = ProcessDataFrame(buffer);
-                    payloadRemainingLength -= bufferConsumed;
-                    break;
-                case 0x1: // HEADERS
-                    if (payloadRemainingLength < buffer.Length)
-                        buffer = buffer.Slice(0, payloadRemainingLength);
-                    bufferConsumed = ProcessHeaderFrame(buffer);
-                    payloadRemainingLength -= bufferConsumed;
-                    if (payloadRemainingLength == 0)
-                        applicationProcessing = Task.Run(() => StartApplicationProcessing(application, token), token);
-                    break;
-                default:
-                    if ((frameType - 0x21) % 0x1f != 0)
-                    {
-                        _connection?.StreamError(ErrorCodes.H3FrameUnexpected);
-                        throw new Http3ConnectionException(ErrorCodes.H3FrameUnexpected);
-                    }
-
-                    // Read the complete reserved frame.
-                    bufferConsumed = payloadRemainingLength > buffer.Length ? buffer.Length : payloadRemainingLength;
-                    payloadRemainingLength -= bufferConsumed;
-                    break;
-            }
-            if (payloadRemainingLength == 0)
-                streamReadingState = StreamReadingStatus.ReadingFrameHeader;
+            bufferConsumed = ProcessHeaderFrame(buffer);
+            if (payloadRemainingLength == bufferConsumed)
+                applicationProcessing = Task.Run(() => StartApplicationProcessing(application, token), token);
         }
 
+        if (streamReadingState == StreamReadingStatus.ReadingPayloadData)
+            bufferConsumed = ProcessDataFrame(buffer);
+
+        if (streamReadingState == StreamReadingStatus.ReadingPayloadReserved)
+            bufferConsumed = buffer.Length; // Read the complete reserved frame.
+
+        payloadRemainingLength -= bufferConsumed;
+        if (payloadRemainingLength == 0)
+            streamReadingState = StreamReadingStatus.ReadingFrameHeader;
         return bufferConsumed;
     }
 
+    private StreamReadingStatus NextRequestReadingState(Task? applicationProcessing, ulong frameType)
+    {
+        StreamReadingStatus streamReadingState;
+        switch (frameType)
+        {
+            case 0x0: // DATA
+                if (applicationProcessing == null)
+                    throw new Http3ConnectionException(ErrorCodes.H3FrameUnexpected);
+                streamReadingState = StreamReadingStatus.ReadingPayloadData;
+                break;
+            case 0x1: // HEADERS
+                if (applicationProcessing != null)
+                    throw new Http3ConnectionException(ErrorCodes.H3FrameUnexpected);
+                streamReadingState = StreamReadingStatus.ReadingPayloadHeader;
+                break;
+            default:
+                if ((frameType - 0x21) % 0x1f != 0)
+                    throw new Http3ConnectionException(ErrorCodes.H3FrameUnexpected);
+                streamReadingState = StreamReadingStatus.ReadingPayloadReserved;
+                break;
+        }
+
+        return streamReadingState;
+    }
+
     private async Task CloseStreamAsync()
     {
         await _dataReader.CompleteAsync();
@@ -221,7 +236,7 @@ private async Task CloseStreamAsync()
     private long ProcessDataFrame(ReadOnlySequence<byte> buffer)
     {
         // 'Copy' to Body
-        return 0;
+        return buffer.Length;
 
     }
 
@@ -337,4 +352,4 @@ internal partial class Http3Stream : IHttpResponseTrailersFeature
 {
     private readonly Http3ResponseHeaderCollection _responseTrailers;
     public IHeaderDictionary Trailers { get => _responseTrailers; set => throw new PlatformNotSupportedException(); }
-}
+}

tests/CHttpServer.Tests/Http3/Http3StreamTests.cs

@@ -178,6 +178,25 @@ public async Task SingleWrite_HeaderFrame_ReservedFrame_DataFrame()
         await quicConnection.DisposeAsync();
     }
 
+    [Fact]
+    public async Task SingleWrite_InOrderFrames_Throw()
+    {
+        var quicConnection = await QuicConnectionFixture.SetupConnectionAsync(Port, TestContext.Current.CancellationToken);
+        var serverStreamTask = Task.Run(async () => await quicConnection.ServerConnection.AcceptInboundStreamAsync());
+        var sut = new Http3Stream([]);
+
+        var clientStream = await quicConnection.ClientConnection.OpenOutboundStreamAsync(QuicStreamType.Bidirectional, TestContext.Current.CancellationToken);
+        byte[] data = [.. GetData(30), .. await GetHeadersFrame()]; //DATA before HEADERS is in-order
+        await clientStream.WriteAsync(data, TestContext.Current.CancellationToken);
+        await clientStream.FlushAsync(TestContext.Current.CancellationToken);
+
+        sut.Initialize(null, await serverStreamTask);
+        var testApp = new TestBase.TestApplication(ctx => Task.CompletedTask);
+        var processing = await Assert.ThrowsAsync<Http3ConnectionException>(() => sut.ProcessStreamAsync(testApp, TestContext.Current.CancellationToken));
+        clientStream.Close();
+        await quicConnection.DisposeAsync();
+    }
+
     private static async Task<byte[]> GetHeadersFrame()
     {
         var encoder = new QPackDecoder();