Merged PR 40029: Lock around Http3Stream data frame processing

Follows the same pattern as Http2Stream:
https://github.com/dotnet/aspnetcore/blob/main/src/Servers/Kestrel/Core/src/Internal/Http2/Http2Stream.cs#L463 and https://github.com/dotnet/aspnetcore/blob/38fe7dd594bde10cc9c9c3c710947af14d3117b3/src/Servers/Kestrel/Core/src/Internal/Http2/Http2Stream.cs#L490

Author: BrennanConroy

src/Servers/Kestrel/Core/src/Internal/Http3/Http3Stream.cs

@@ -63,6 +63,7 @@ internal abstract partial class Http3Stream : HttpProtocol, IHttp3Stream, IHttpS
 
     public bool EndStreamReceived => (_completionState & StreamCompletionFlags.EndStreamReceived) == StreamCompletionFlags.EndStreamReceived;
     public bool IsAborted => (_completionState & StreamCompletionFlags.Aborted) == StreamCompletionFlags.Aborted;
+    private bool IsAbortedRead => (_completionState & StreamCompletionFlags.AbortedRead) == StreamCompletionFlags.AbortedRead;
     public bool IsCompleted => (_completionState & StreamCompletionFlags.Completed) == StreamCompletionFlags.Completed;
 
     public Pipe RequestBodyPipe { get; private set; } = default!;
@@ -892,12 +893,20 @@ private Task ProcessDataFrameAsync(in ReadOnlySequence<byte> payload)
             InputRemaining -= payload.Length;
         }
 
-        foreach (var segment in payload)
+        lock (_completionLock)
         {
-            RequestBodyPipe.Writer.Write(segment.Span);
-        }
+            if (IsAborted || IsAbortedRead)
+            {
+                return Task.CompletedTask;
+            }
 
-        return RequestBodyPipe.Writer.FlushAsync().GetAsTask();
+            foreach (var segment in payload)
+            {
+                RequestBodyPipe.Writer.Write(segment.Span);
+            }
+
+            return RequestBodyPipe.Writer.FlushAsync().GetAsTask();
+        }
     }
 
     protected override void OnReset()