Fix re-entrancy in IncomingStream::ReadFromPipeAndEnqueue()

The method blink::IncomingStream::ReadFromPipeAndEnqueue() can be called
re-entrantly when

1. A read() is pending
2. OnHandleReady() calls ReadFromPipeAndEnqueue()
3. Enqueue() results in a call to pull() (to re-fill the queue)
4. pull() calls ReadFromPipeAndEnqueue()

Since there cannot be two calls to BeginReadData() in progress, this
causes a fatal error.

Add a flag |in_two_phase_read_| which causes ReadFromPipeAndEnqueue() to
return early. Add another flag |read_pending_| which indicates that
ReadFromPipeAndEnqueue() has returned early and that it should be called
again when the current read completes.

Also add a unit test for this condition.

BUG=1064434

Change-Id: I7b367e466a0c5c2e1abccc0a688d2098e17143f9
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2124331
Commit-Queue: Adam Rice <ricea@chromium.org>
Reviewed-by: Yutaka Hirano <yhirano@chromium.org>
Cr-Commit-Position: refs/heads/master@{#754436}

third_party/blink/renderer/modules/webtransport/incoming_stream.cc

@@ -228,18 +228,36 @@ void IncomingStream::ProcessClose() {
 }
 
 void IncomingStream::ReadFromPipeAndEnqueue() {
-  DVLOG(1) << "IncomingStream::ReadFromPipeAndEnqueue() this=" << this;
+  DVLOG(1) << "IncomingStream::ReadFromPipeAndEnqueue() this=" << this
+           << " in_two_phase_read_=" << in_two_phase_read_
+           << " read_pending_=" << read_pending_;
+
+  // Protect against re-entrancy.
+  if (in_two_phase_read_) {
+    read_pending_ = true;
+    return;
+  }
+  DCHECK(!read_pending_);
 
   const void* buffer = nullptr;
   uint32_t buffer_num_bytes = 0;
   auto result = data_pipe_->BeginReadData(&buffer, &buffer_num_bytes,
                                           MOJO_BEGIN_READ_DATA_FLAG_NONE);
-
   switch (result) {
-    case MOJO_RESULT_OK:
+    case MOJO_RESULT_OK: {
+      in_two_phase_read_ = true;
+      // EnqueueBytes() may re-enter this method via pull().
       EnqueueBytes(buffer, buffer_num_bytes);
       data_pipe_->EndReadData(buffer_num_bytes);
+      in_two_phase_read_ = false;
+      if (read_pending_) {
+        read_pending_ = false;
+        // pull() will not be called when another pull() is in progress, so the
+        // maximum recursion depth is 1.
+        ReadFromPipeAndEnqueue();
+      }
       break;
+    }
 
     case MOJO_RESULT_SHOULD_WAIT:
       read_watcher_.ArmOrNotify();
@@ -250,7 +268,7 @@ void IncomingStream::ReadFromPipeAndEnqueue() {
       return;
 
     default:
-      DLOG(FATAL) << "FATAL ERROR";
+      NOTREACHED() << "Unexpected result: " << result;
       return;
   }
 }

third_party/blink/renderer/modules/webtransport/incoming_stream.h

@@ -153,6 +153,14 @@ class MODULES_EXPORT IncomingStream
   // True when |data_pipe_| has been detected to be closed. The close is not
   // processed until |fin_received_| is also set.
   bool is_pipe_closed_ = false;
+
+  // Indicates if we are currently performing a two-phase read from the pipe and
+  // so can't start another read.
+  bool in_two_phase_read_ = false;
+
+  // Indicates if we need to perform another read after the current one
+  // completes.
+  bool read_pending_ = false;
 };
 
 }  // namespace blink

third_party/blink/renderer/modules/webtransport/incoming_stream_test.cc

@@ -72,8 +72,8 @@ class IncomingStreamTest : public ::testing::Test {
   void ClosePipe() { data_pipe_producer_.reset(); }
 
   // Copies the contents of a v8::Value containing a Uint8Array to a Vector.
-  Vector<uint8_t> ToVector(const V8TestingScope& scope,
-                           v8::Local<v8::Value> v8value) {
+  static Vector<uint8_t> ToVector(const V8TestingScope& scope,
+                                  v8::Local<v8::Value> v8value) {
     Vector<uint8_t> ret;
 
     DOMUint8Array* value =
@@ -94,19 +94,23 @@ class IncomingStreamTest : public ::testing::Test {
 
   // Performs a single read from |reader|, converting the output to the
   // Iterator type. Assumes that the readable stream is not errored.
-  Iterator Read(const V8TestingScope& scope,
-                ReadableStreamDefaultReader* reader) {
+  static Iterator Read(const V8TestingScope& scope,
+                       ReadableStreamDefaultReader* reader) {
     auto* script_state = scope.GetScriptState();
     ScriptPromise read_promise =
         reader->read(script_state, ASSERT_NO_EXCEPTION);
     ScriptPromiseTester tester(script_state, read_promise);
     tester.WaitUntilSettled();
     EXPECT_TRUE(tester.IsFulfilled());
-    v8::Local<v8::Value> result = tester.Value().V8Value();
+    return IteratorFromReadResult(scope, tester.Value().V8Value());
+  }
+
+  static Iterator IteratorFromReadResult(const V8TestingScope& scope,
+                                         v8::Local<v8::Value> result) {
     CHECK(result->IsObject());
     Iterator ret;
     v8::Local<v8::Value> v8value;
-    if (!V8UnpackIteratorResult(script_state, result.As<v8::Object>(),
+    if (!V8UnpackIteratorResult(scope.GetScriptState(), result.As<v8::Object>(),
                                 &ret.done)
              .ToLocal(&v8value)) {
       ADD_FAILURE() << "Couldn't unpack iterator";
@@ -378,6 +382,29 @@ TEST_F(IncomingStreamTest, GarbageCollectionRemoteClose) {
   EXPECT_FALSE(incoming_stream);
 }
 
+TEST_F(IncomingStreamTest, WriteToPipeWithPendingRead) {
+  V8TestingScope scope;
+
+  auto* incoming_stream = CreateIncomingStream(scope);
+  auto* script_state = scope.GetScriptState();
+  auto* reader =
+      incoming_stream->readable()->getReader(script_state, ASSERT_NO_EXCEPTION);
+  ScriptPromise read_promise = reader->read(script_state, ASSERT_NO_EXCEPTION);
+  ScriptPromiseTester tester(script_state, read_promise);
+
+  test::RunPendingTasks();
+
+  WriteToPipe({'A'});
+
+  tester.WaitUntilSettled();
+  EXPECT_TRUE(tester.IsFulfilled());
+
+  Iterator result = IteratorFromReadResult(scope, tester.Value().V8Value());
+  EXPECT_FALSE(result.done);
+  EXPECT_THAT(result.value, ElementsAre('A'));
+  EXPECT_CALL(*mock_close_proxy_, ForgetStream());
+}
+
 }  // namespace
 
 }  // namespace blink