Update oghttp2 to close streams above a received GOAWAY's last_stream_id.

diannahu · copybara-github · commit 9af10d5cc822 · 2022-04-07T15:37:37.000-07:00
This CL causes oghttp2 to respond to the receipt of a GOAWAY by:
  - Closing active streams in its stream map above the GOAWAY's last_stream_id.
  - Avoiding sending frames with stream ID values above the last_stream_id.
These two changes handle cases where request HEADERS (or other frames for a
stream) have already been enqueued at the time a GOAWAY is received, while also
preventing subsequent enqueues for these active streams (by closing them).

This CL does not handle the cases of new or pending streams, i.e., does not
prevent subsequent requests or the start of requests that have been queued as
pending. These cases will be handled in follow-up cl/439616811.

This change increases oghttp2 parity with nghttp2.

PiperOrigin-RevId: 440217657
diff --git a/quiche/http2/adapter/nghttp2_adapter_test.cc b/quiche/http2/adapter/nghttp2_adapter_test.cc
@@ -1585,6 +1585,10 @@ TEST(NgHttp2AdapterTest, ClientReceivesGoAway) {
               EqualsFrames({SpdyFrameType::HEADERS, SpdyFrameType::HEADERS}));
   visitor.Clear();
 
+  // Submit a pending WINDOW_UPDATE for a stream that will be closed due to
+  // GOAWAY. The WINDOW_UPDATE should not be sent.
+  adapter->SubmitWindowUpdate(3, 42);
+
   const std::string stream_frames =
       TestFrameSequence()
           .ServerPreface()
@@ -1611,11 +1615,90 @@ TEST(NgHttp2AdapterTest, ClientReceivesGoAway) {
   const int64_t stream_result = adapter->ProcessBytes(stream_frames);
   EXPECT_EQ(stream_frames.size(), stream_result);
 
-  // Intriguingly, despite want_write() being true, the sent data is empty; the
-  // receipt of the GOAWAY prevents the SETTINGS ack from being written.
+  EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x1));
+  EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x1, 0));
+
+  // SETTINGS ack (but only after the enqueue of the seemingly unrelated
+  // WINDOW_UPDATE). The WINDOW_UPDATE is not written.
   EXPECT_TRUE(adapter->want_write());
   result = adapter->Send();
   EXPECT_EQ(0, result);
+  EXPECT_THAT(visitor.data(), EqualsFrames({SpdyFrameType::SETTINGS}));
+}
+
+TEST(NgHttp2AdapterTest, ClientReceivesMultipleGoAways) {
+  DataSavingVisitor visitor;
+  auto adapter = NgHttp2Adapter::CreateClientAdapter(visitor);
+
+  testing::InSequence s;
+
+  const std::vector<Header> headers1 =
+      ToHeaders({{":method", "GET"},
+                 {":scheme", "http"},
+                 {":authority", "example.com"},
+                 {":path", "/this/is/request/one"}});
+
+  const int32_t stream_id1 = adapter->SubmitRequest(headers1, nullptr, nullptr);
+  ASSERT_GT(stream_id1, 0);
+
+  EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x5));
+  EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x5, 0));
+
+  int result = adapter->Send();
+  EXPECT_EQ(0, result);
+  absl::string_view data = visitor.data();
+  EXPECT_THAT(data, testing::StartsWith(spdy::kHttp2ConnectionHeaderPrefix));
+  data.remove_prefix(strlen(spdy::kHttp2ConnectionHeaderPrefix));
+  EXPECT_THAT(data, EqualsFrames({SpdyFrameType::HEADERS}));
+  visitor.Clear();
+
+  const std::string initial_frames =
+      TestFrameSequence()
+          .ServerPreface()
+          .GoAway(kMaxStreamId, Http2ErrorCode::INTERNAL_ERROR, "indigestion")
+          .Serialize();
+
+  EXPECT_CALL(visitor, OnFrameHeader(0, 0, SETTINGS, 0));
+  EXPECT_CALL(visitor, OnSettingsStart());
+  EXPECT_CALL(visitor, OnSettingsEnd());
+  EXPECT_CALL(visitor, OnFrameHeader(0, _, GOAWAY, 0));
+  EXPECT_CALL(visitor, OnGoAway(kMaxStreamId, Http2ErrorCode::INTERNAL_ERROR,
+                                "indigestion"));
+
+  const int64_t initial_result = adapter->ProcessBytes(initial_frames);
+  EXPECT_EQ(initial_frames.size(), static_cast<size_t>(initial_result));
+
+  // Submit a WINDOW_UPDATE for the open stream. Because the stream is below the
+  // GOAWAY's last_stream_id, it should be sent.
+  adapter->SubmitWindowUpdate(1, 42);
+
+  EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, 0, 0x1));
+  EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, 0, 0x1, 0));
+  EXPECT_CALL(visitor, OnBeforeFrameSent(WINDOW_UPDATE, 1, 4, 0x0));
+  EXPECT_CALL(visitor, OnFrameSent(WINDOW_UPDATE, 1, 4, 0x0, 0));
+
+  result = adapter->Send();
+  EXPECT_EQ(0, result);
+  EXPECT_THAT(visitor.data(), EqualsFrames({SpdyFrameType::SETTINGS,
+                                            SpdyFrameType::WINDOW_UPDATE}));
+  visitor.Clear();
+
+  const std::string final_frames =
+      TestFrameSequence()
+          .GoAway(0, Http2ErrorCode::INTERNAL_ERROR, "indigestion")
+          .Serialize();
+
+  EXPECT_CALL(visitor, OnFrameHeader(0, _, GOAWAY, 0));
+  EXPECT_CALL(visitor,
+              OnGoAway(0, Http2ErrorCode::INTERNAL_ERROR, "indigestion"));
+  EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::REFUSED_STREAM));
+
+  const int64_t final_result = adapter->ProcessBytes(final_frames);
+  EXPECT_EQ(final_frames.size(), static_cast<size_t>(final_result));
+
+  EXPECT_FALSE(adapter->want_write());
+  result = adapter->Send();
+  EXPECT_EQ(0, result);
   EXPECT_THAT(visitor.data(), testing::IsEmpty());
 }
 
diff --git a/quiche/http2/adapter/oghttp2_adapter_test.cc b/quiche/http2/adapter/oghttp2_adapter_test.cc
@@ -1955,6 +1955,10 @@ TEST(OgHttp2AdapterTest, ClientReceivesGoAway) {
                             SpdyFrameType::HEADERS}));
   visitor.Clear();
 
+  // Submit a pending WINDOW_UPDATE for a stream that will be closed due to
+  // GOAWAY. The WINDOW_UPDATE should not be sent.
+  adapter->SubmitWindowUpdate(3, 42);
+
   const std::string stream_frames =
       TestFrameSequence()
           .ServerPreface()
@@ -1973,7 +1977,7 @@ TEST(OgHttp2AdapterTest, ClientReceivesGoAway) {
   EXPECT_CALL(visitor, OnFrameHeader(0, _, GOAWAY, 0));
   // Currently, oghttp2 does not pass the opaque data to the visitor.
   EXPECT_CALL(visitor, OnGoAway(1, Http2ErrorCode::INTERNAL_ERROR, ""));
-  // TODO(b/228093860): Close stream IDs above the GOAWAY's last_stream_id.
+  EXPECT_CALL(visitor, OnCloseStream(3, Http2ErrorCode::REFUSED_STREAM));
   EXPECT_CALL(visitor, OnFrameHeader(0, 4, WINDOW_UPDATE, 0));
   EXPECT_CALL(visitor, OnWindowUpdate(0, 42));
   EXPECT_CALL(visitor, OnFrameHeader(1, 4, WINDOW_UPDATE, 0));
@@ -1990,6 +1994,87 @@ TEST(OgHttp2AdapterTest, ClientReceivesGoAway) {
   EXPECT_THAT(visitor.data(), EqualsFrames({SpdyFrameType::SETTINGS}));
 }
 
+TEST(OgHttp2AdapterTest, ClientReceivesMultipleGoAways) {
+  DataSavingVisitor visitor;
+  OgHttp2Adapter::Options options{.perspective = Perspective::kClient};
+  auto adapter = OgHttp2Adapter::Create(visitor, options);
+
+  testing::InSequence s;
+
+  const std::vector<Header> headers1 =
+      ToHeaders({{":method", "GET"},
+                 {":scheme", "http"},
+                 {":authority", "example.com"},
+                 {":path", "/this/is/request/one"}});
+
+  const int32_t stream_id1 = adapter->SubmitRequest(headers1, nullptr, nullptr);
+  ASSERT_GT(stream_id1, 0);
+
+  EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, _, 0x0));
+  EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, _, 0x0, 0));
+  EXPECT_CALL(visitor, OnBeforeFrameSent(HEADERS, stream_id1, _, 0x5));
+  EXPECT_CALL(visitor, OnFrameSent(HEADERS, stream_id1, _, 0x5, 0));
+
+  int result = adapter->Send();
+  EXPECT_EQ(0, result);
+  absl::string_view data = visitor.data();
+  EXPECT_THAT(data, testing::StartsWith(spdy::kHttp2ConnectionHeaderPrefix));
+  data.remove_prefix(strlen(spdy::kHttp2ConnectionHeaderPrefix));
+  EXPECT_THAT(data,
+              EqualsFrames({SpdyFrameType::SETTINGS, SpdyFrameType::HEADERS}));
+  visitor.Clear();
+
+  const std::string initial_frames =
+      TestFrameSequence()
+          .ServerPreface()
+          .GoAway(kMaxStreamId, Http2ErrorCode::INTERNAL_ERROR, "indigestion")
+          .Serialize();
+
+  EXPECT_CALL(visitor, OnFrameHeader(0, 0, SETTINGS, 0));
+  EXPECT_CALL(visitor, OnSettingsStart());
+  EXPECT_CALL(visitor, OnSettingsEnd());
+  EXPECT_CALL(visitor, OnFrameHeader(0, _, GOAWAY, 0));
+  // Currently, oghttp2 does not pass the opaque data to the visitor.
+  EXPECT_CALL(visitor,
+              OnGoAway(kMaxStreamId, Http2ErrorCode::INTERNAL_ERROR, ""));
+
+  const int64_t initial_result = adapter->ProcessBytes(initial_frames);
+  EXPECT_EQ(initial_frames.size(), static_cast<size_t>(initial_result));
+
+  // Submit a WINDOW_UPDATE for the open stream. Because the stream is below the
+  // GOAWAY's last_stream_id, it should be sent.
+  adapter->SubmitWindowUpdate(1, 42);
+
+  EXPECT_CALL(visitor, OnBeforeFrameSent(SETTINGS, 0, 0, 0x1));
+  EXPECT_CALL(visitor, OnFrameSent(SETTINGS, 0, 0, 0x1, 0));
+  EXPECT_CALL(visitor, OnBeforeFrameSent(WINDOW_UPDATE, 1, 4, 0x0));
+  EXPECT_CALL(visitor, OnFrameSent(WINDOW_UPDATE, 1, 4, 0x0, 0));
+
+  result = adapter->Send();
+  EXPECT_EQ(0, result);
+  EXPECT_THAT(visitor.data(), EqualsFrames({SpdyFrameType::SETTINGS,
+                                            SpdyFrameType::WINDOW_UPDATE}));
+  visitor.Clear();
+
+  const std::string final_frames =
+      TestFrameSequence()
+          .GoAway(0, Http2ErrorCode::INTERNAL_ERROR, "indigestion")
+          .Serialize();
+
+  EXPECT_CALL(visitor, OnFrameHeader(0, _, GOAWAY, 0));
+  // Currently, oghttp2 does not pass the opaque data to the visitor.
+  EXPECT_CALL(visitor, OnGoAway(0, Http2ErrorCode::INTERNAL_ERROR, ""));
+  EXPECT_CALL(visitor, OnCloseStream(1, Http2ErrorCode::REFUSED_STREAM));
+
+  const int64_t final_result = adapter->ProcessBytes(final_frames);
+  EXPECT_EQ(final_frames.size(), static_cast<size_t>(final_result));
+
+  EXPECT_FALSE(adapter->want_write());
+  result = adapter->Send();
+  EXPECT_EQ(0, result);
+  EXPECT_THAT(visitor.data(), testing::IsEmpty());
+}
+
 TEST(OgHttp2AdapterTest, ClientFailsOnGoAway) {
   DataSavingVisitor visitor;
   OgHttp2Adapter::Options options{.perspective = Perspective::kClient};
diff --git a/quiche/http2/adapter/oghttp2_session.cc b/quiche/http2/adapter/oghttp2_session.cc
@@ -2,6 +2,7 @@
 
 #include <cstdint>
 #include <utility>
+#include <vector>
 
 #include "absl/memory/memory.h"
 #include "absl/strings/escaping.h"
@@ -678,6 +679,13 @@ OgHttp2Session::SendResult OgHttp2Session::SendQueuedFrames() {
       DecrementQueuedFrameCount(c.stream_id(), c.frame_type());
       frames_.pop_front();
       continue;
+    } else if (received_goaway_ &&
+               c.stream_id() >
+                   static_cast<uint32_t>(received_goaway_stream_id_)) {
+      // This frame will be ignored by the peer, so don't send it. The stream
+      // should already have been closed.
+      frames_.pop_front();
+      continue;
     }
     // Frames can't accurately report their own length; the actual serialized
     // length must be used instead.
@@ -1288,12 +1296,28 @@ void OgHttp2Session::OnPing(spdy::SpdyPingId unique_id, bool is_ack) {
 void OgHttp2Session::OnGoAway(spdy::SpdyStreamId last_accepted_stream_id,
                               spdy::SpdyErrorCode error_code) {
   received_goaway_ = true;
+  // TODO(diannahu): Validate that `last_accepted_stream_id` is non-increasing.
+  received_goaway_stream_id_ = last_accepted_stream_id;
   const bool result = visitor_.OnGoAway(last_accepted_stream_id,
                                         TranslateErrorCode(error_code), "");
   if (!result) {
     fatal_visitor_callback_failure_ = true;
     decoder_.StopProcessing();
   }
+
+  // Close the streams above `last_accepted_stream_id`.
+  if (last_accepted_stream_id == spdy::kMaxStreamId) {
+    return;
+  }
+  std::vector<Http2StreamId> streams_to_close;
+  for (const auto& [stream_id, stream_state] : stream_map_) {
+    if (static_cast<spdy::SpdyStreamId>(stream_id) > last_accepted_stream_id) {
+      streams_to_close.push_back(stream_id);
+    }
+  }
+  for (Http2StreamId stream_id : streams_to_close) {
+    CloseStream(stream_id, Http2ErrorCode::REFUSED_STREAM);
+  }
 }
 
 bool OgHttp2Session::OnGoAwayFrameData(const char* /*goaway_data*/, size_t
@@ -1710,7 +1734,6 @@ void OgHttp2Session::CloseStream(Http2StreamId stream_id,
   metadata_ready_.erase(stream_id);
   streams_reset_.erase(stream_id);
   queued_frames_.erase(stream_id);
-  stream_map_.erase(stream_id);
   if (write_scheduler_.StreamRegistered(stream_id)) {
     write_scheduler_.UnregisterStream(stream_id);
   }
diff --git a/quiche/http2/adapter/oghttp2_session.h b/quiche/http2/adapter/oghttp2_session.h
@@ -498,6 +498,7 @@ class QUICHE_EXPORT_PRIVATE OgHttp2Session
   Http2StreamId highest_received_stream_id_ = 0;
   Http2StreamId highest_processed_stream_id_ = 0;
   Http2StreamId metadata_stream_id_ = 0;
+  Http2StreamId received_goaway_stream_id_ = 0;
   size_t metadata_length_ = 0;
   int32_t connection_send_window_ = kInitialFlowControlWindowSize;
   // The initial flow control receive window size for any newly created streams.
