Abort async QPACK header decompression when stream is closed.

This is to make sure QuicSpdyStream::OnHeadersDecoded() is not called by
QpackDecodedHeadersAccumulator after the stream is closed.  This is important,
because OnHeadersDecoded() calls OnDataAvailable(), which can cause issues in
higher layers.

Protected by FLAGS_quic_reloadable_flag_quic_abort_qpack_on_stream_close.

PiperOrigin-RevId: 332106028
Change-Id: I0d2e2069c63931cf8682d1f9f79d34644561d266

quic/core/http/quic_spdy_stream.cc

diff --git a/quic/core/http/quic_spdy_stream.cc b/quic/core/http/quic_spdy_stream.cc
index 67c24e4..1c5f964 100644
--- a/quic/core/http/quic_spdy_stream.cc
+++ b/quic/core/http/quic_spdy_stream.cc
@@ -801,6 +801,11 @@
 void QuicSpdyStream::OnClose() {
   QuicStream::OnClose();
 
+  if (GetQuicReloadableFlag(quic_abort_qpack_on_stream_close)) {
+    QUIC_RELOADABLE_FLAG_COUNT(quic_abort_qpack_on_stream_close);
+    qpack_decoded_headers_accumulator_.reset();
+  }
+
   if (visitor_) {
     Visitor* visitor = visitor_;
     // Calling Visitor::OnClose() may result the destruction of the visitor,

quic/core/http/quic_spdy_stream_test.cc

diff --git a/quic/core/http/quic_spdy_stream_test.cc b/quic/core/http/quic_spdy_stream_test.cc
index cc0e3ec..85e6467 100644
--- a/quic/core/http/quic_spdy_stream_test.cc
+++ b/quic/core/http/quic_spdy_stream_test.cc
@@ -2416,6 +2416,64 @@
   session_->qpack_decoder()->OnInsertWithoutNameReference("trailing", "foobar");
 }
 
+// Regression test for b/132603592: QPACK decoding unblocked after stream is
+// closed.
+TEST_P(QuicSpdyStreamTest, HeaderDecodingUnblockedAfterStreamClosed) {
+  if (!UsesHttp3()) {
+    return;
+  }
+
+  Initialize(kShouldProcessData);
+  testing::InSequence s;
+  session_->qpack_decoder()->OnSetDynamicTableCapacity(1024);
+  StrictMock<MockHttp3DebugVisitor> debug_visitor;
+  session_->set_debug_visitor(&debug_visitor);
+
+  // HEADERS frame referencing first dynamic table entry.
+  std::string encoded_headers = quiche::QuicheTextUtils::HexDecode("020080");
+  std::string headers = HeadersFrame(encoded_headers);
+  EXPECT_CALL(debug_visitor,
+              OnHeadersFrameReceived(stream_->id(), encoded_headers.length()));
+  stream_->OnStreamFrame(QuicStreamFrame(stream_->id(), false, 0, headers));
+
+  // Decoding is blocked because dynamic table entry has not been received yet.
+  EXPECT_FALSE(stream_->headers_decompressed());
+
+  // Decoder stream type and stream cancellation instruction.
+  auto decoder_send_stream =
+      QuicSpdySessionPeer::GetQpackDecoderSendStream(session_.get());
+  EXPECT_CALL(*session_,
+              WritevData(decoder_send_stream->id(), /* write_length = */ 1,
+                         /* offset = */ 0, _, _, _));
+  EXPECT_CALL(*session_,
+              WritevData(decoder_send_stream->id(), /* write_length = */ 1,
+                         /* offset = */ 1, _, _, _));
+
+  // Reset stream.
+  EXPECT_CALL(*session_,
+              SendRstStream(stream_->id(), QUIC_STREAM_CANCELLED, _, _));
+  stream_->Reset(QUIC_STREAM_CANCELLED);
+
+  if (!GetQuicReloadableFlag(quic_abort_qpack_on_stream_close)) {
+    // Header acknowledgement.
+    EXPECT_CALL(*session_,
+                WritevData(decoder_send_stream->id(), /* write_length = */ 1,
+                           /* offset = */ 2, _, _, _));
+    EXPECT_CALL(debug_visitor, OnHeadersDecoded(stream_->id(), _));
+  }
+
+  // Deliver dynamic table entry to decoder.
+  session_->qpack_decoder()->OnInsertWithoutNameReference("foo", "bar");
+
+  if (GetQuicReloadableFlag(quic_abort_qpack_on_stream_close)) {
+    EXPECT_FALSE(stream_->headers_decompressed());
+  } else {
+    // Verify headers.
+    EXPECT_TRUE(stream_->headers_decompressed());
+    EXPECT_THAT(stream_->header_list(), ElementsAre(Pair("foo", "bar")));
+  }
+}
+
 class QuicSpdyStreamIncrementalConsumptionTest : public QuicSpdyStreamTest {
  protected:
   QuicSpdyStreamIncrementalConsumptionTest() : offset_(0), consumed_bytes_(0) {}