WebTransport: Add state checks to BidrectionalStream

When the server closed a server-initiated bidirectional stream, Blink
would send a reset. Fix it by tracking state in BidirectionalStream so
that we don't send a reset after receiving a fin.

The state tracking is adequate to fix known bugs but is not yet
comprehensive. A followup CL will make the state machine robust.

BUG=1123772

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

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

@@ -39,11 +39,12 @@ void BidirectionalStream::Init() {
 
 void BidirectionalStream::OnIncomingStreamClosed(bool fin_received) {
   incoming_stream_->OnIncomingStreamClosed(fin_received);
-  // TODO(ricea): Review this behaviour when adding detail to the specification.
-  if (!sent_fin_) {
+  if (outgoing_stream_->GetState() == OutgoingStream::State::kSentFin) {
+    return;
+  }
+
   ScriptState::Scope scope(outgoing_stream_->GetScriptState());
   outgoing_stream_->Reset();
-  }
 }
 
 void BidirectionalStream::Reset() {
@@ -59,15 +60,15 @@ void BidirectionalStream::ContextDestroyed() {
 
 void BidirectionalStream::SendFin() {
   quic_transport_->SendFin(stream_id_);
-  sent_fin_ = true;
   // The IncomingStream will be closed on the network service side.
 }
 
 void BidirectionalStream::OnOutgoingStreamAbort() {
-  DCHECK(!sent_fin_);
   quic_transport_->AbortStream(stream_id_);
   quic_transport_->ForgetStream(stream_id_);
+  if (incoming_stream_->GetState() == IncomingStream::State::kOpen) {
     incoming_stream_->Reset();
+  }
 }
 
 void BidirectionalStream::Trace(Visitor* visitor) const {
@@ -81,6 +82,9 @@ void BidirectionalStream::Trace(Visitor* visitor) const {
 
 void BidirectionalStream::OnIncomingStreamAbort() {
   quic_transport_->ForgetStream(stream_id_);
+  if (outgoing_stream_->GetState() == OutgoingStream::State::kAborted) {
+    return;
+  }
   ScriptState::Scope scope(outgoing_stream_->GetScriptState());
   outgoing_stream_->Reset();
 }

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

@@ -77,7 +77,6 @@ class MODULES_EXPORT BidirectionalStream final : public ScriptWrappable,
   const Member<IncomingStream> incoming_stream_;
   const Member<QuicTransport> quic_transport_;
   const uint32_t stream_id_;
-  bool sent_fin_ = false;
 };
 
 }  // namespace blink

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

@@ -22,6 +22,7 @@
 #include "third_party/blink/renderer/bindings/core/v8/script_promise_tester.h"
 #include "third_party/blink/renderer/bindings/core/v8/script_value.h"
 #include "third_party/blink/renderer/bindings/core/v8/v8_binding_for_testing.h"
+#include "third_party/blink/renderer/bindings/core/v8/v8_iterator_result_value.h"
 #include "third_party/blink/renderer/bindings/core/v8/v8_uint8_array.h"
 #include "third_party/blink/renderer/bindings/modules/v8/v8_bidirectional_stream.h"
 #include "third_party/blink/renderer/bindings/modules/v8/v8_quic_transport_options.h"
@@ -380,6 +381,25 @@ TEST(BidirectionalStreamTest, IncomingStreamCleanClose) {
       kDefaultStreamId, true);
   scoped_quic_transport.Stub()->InputProducer().reset();
 
+  auto* script_state = scope.GetScriptState();
+  auto* reader = bidirectional_stream->readable()->getReader(
+      script_state, ASSERT_NO_EXCEPTION);
+
+  ScriptPromise read_promise = reader->read(script_state, ASSERT_NO_EXCEPTION);
+
+  ScriptPromiseTester read_tester(script_state, read_promise);
+  read_tester.WaitUntilSettled();
+  EXPECT_TRUE(read_tester.IsFulfilled());
+
+  v8::Local<v8::Value> result = read_tester.Value().V8Value();
+  DCHECK(result->IsObject());
+  v8::Local<v8::Value> v8value;
+  bool done = false;
+  EXPECT_TRUE(
+      V8UnpackIteratorResult(script_state, result.As<v8::Object>(), &done)
+          .ToLocal(&v8value));
+  EXPECT_TRUE(done);
+
   ScriptPromiseTester tester(scope.GetScriptState(),
                              bidirectional_stream->writingAborted());
   tester.WaitUntilSettled();

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

@@ -105,6 +105,9 @@ void IncomingStream::OnIncomingStreamClosed(bool fin_received) {
   DVLOG(1) << "IncomingStream::OnIncomingStreamClosed(" << fin_received
            << ") this=" << this;
 
+  DCHECK_NE(state_, State::kClosed);
+  state_ = State::kClosed;
+
   DCHECK(!fin_received_.has_value());
 
   fin_received_ = fin_received;
@@ -314,6 +317,8 @@ void IncomingStream::AbortAndReset() {
     reading_aborted_resolver_ = nullptr;
   }
 
+  state_ = State::kAborted;
+
   if (on_abort_) {
     // Cause QuicTransport to drop its reference to us.
     std::move(on_abort_).Run();

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

@@ -35,6 +35,12 @@ class MODULES_EXPORT IncomingStream final
   USING_PRE_FINALIZER(IncomingStream, Dispose);
 
  public:
+  enum class State {
+    kOpen,
+    kAborted,
+    kClosed,
+  };
+
   IncomingStream(ScriptState*,
                  base::OnceClosure on_abort,
                  mojo::ScopedDataPipeConsumerHandle);
@@ -68,6 +74,8 @@ class MODULES_EXPORT IncomingStream final
   // Does not execute JavaScript.
   void ContextDestroyed();
 
+  State GetState() const { return state_; }
+
   void Trace(Visitor*) const;
 
  private:
@@ -136,6 +144,8 @@ class MODULES_EXPORT IncomingStream final
   ScriptPromise reading_aborted_;
   Member<ScriptPromiseResolver> reading_aborted_resolver_;
 
+  State state_ = State::kOpen;
+
   // This is set when OnIncomingStreamClosed() is called.
   base::Optional<bool> fin_received_;
 

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

@@ -66,6 +66,7 @@ class OutgoingStream::UnderlyingSink final : public UnderlyingSinkBase {
     DCHECK(!outgoing_stream_->write_promise_resolver_);
 
     if (outgoing_stream_->client_) {
+      outgoing_stream_->state_ = State::kSentFin;
       outgoing_stream_->client_->SendFin();
       outgoing_stream_->client_ = nullptr;
     }
@@ -389,6 +390,8 @@ void OutgoingStream::AbortAndReset() {
   }
 
   if (client_) {
+    DCHECK_EQ(state_, State::kOpen);
+    state_ = State::kAborted;
     client_->OnOutgoingStreamAbort();
     client_ = nullptr;
   }

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

@@ -52,6 +52,12 @@ class MODULES_EXPORT OutgoingStream final
     virtual void OnOutgoingStreamAbort() = 0;
   };
 
+  enum class State {
+    kOpen,
+    kSentFin,
+    kAborted,
+  };
+
   OutgoingStream(ScriptState*, Client*, mojo::ScopedDataPipeProducerHandle);
   ~OutgoingStream();
 
@@ -76,6 +82,8 @@ class MODULES_EXPORT OutgoingStream final
   // scope to be entered.
   void Reset();
 
+  State GetState() const { return state_; }
+
   // Called from QuicTransport rather than using
   // ExecutionContextLifecycleObserver to ensure correct destruction order.
   // Does not execute JavaScript.
@@ -180,6 +188,8 @@ class MODULES_EXPORT OutgoingStream final
   // If an asynchronous write() on the underlying sink object is pending, this
   // will be non-null.
   Member<ScriptPromiseResolver> write_promise_resolver_;
+
+  State state_ = State::kOpen;
 };
 
 }  // namespace blink