Proactively disconnect connections flooded in pending flush timeout (#…

…13442)

Signed-off-by: Yan Avlasov <yavlasov@google.com>

source/common/http/http2/codec_impl.cc

@@ -466,6 +466,7 @@ void ConnectionImpl::StreamImpl::onPendingFlushTimer() {
   auto status = parent_.sendPendingFrames();
   // See comment in the `encodeHeadersBase()` method about this RELEASE_ASSERT.
   RELEASE_ASSERT(status.ok(), "sendPendingFrames() failure in non dispatching context");
+  parent_.checkProtocolConstraintViolation();
 }
 
 void ConnectionImpl::StreamImpl::encodeData(Buffer::Instance& data, bool end_stream) {

source/common/http/http2/codec_impl_legacy.cc

@@ -450,6 +450,7 @@ void ConnectionImpl::StreamImpl::onPendingFlushTimer() {
   // will be run because higher layers think the stream is already finished.
   resetStreamWorker(StreamResetReason::LocalReset);
   parent_.sendPendingFrames();
+  parent_.checkProtocolConstraintViolation();
 }
 
 void ConnectionImpl::StreamImpl::encodeData(Buffer::Instance& data, bool end_stream) {

test/common/http/http2/codec_impl_test.cc

@@ -1431,6 +1431,59 @@ TEST_P(Http2CodecImplFlowControlTest, WindowUpdateOnReadResumingFlood) {
   EXPECT_EQ(1, server_stats_store_.counter("http2.outbound_flood").value());
 }
 
+// Verify detection of outbound queue flooding by the RST_STREAM frame sent by the pending flush
+// timeout.
+TEST_P(Http2CodecImplFlowControlTest, RstStreamOnPendingFlushTimeoutFlood) {
+  // This test sets initial stream window to 65535 bytes.
+  initialize();
+
+  TestRequestHeaderMapImpl request_headers;
+  HttpTestUtility::addDefaultHeaders(request_headers);
+  EXPECT_CALL(request_decoder_, decodeHeaders_(_, false));
+  request_encoder_->encodeHeaders(request_headers, false);
+
+  int frame_count = 0;
+  Buffer::OwnedImpl buffer;
+  ON_CALL(server_connection_, write(_, _))
+      .WillByDefault(Invoke([&buffer, &frame_count](Buffer::Instance& frame, bool) {
+        ++frame_count;
+        buffer.move(frame);
+      }));
+
+  auto* violation_callback =
+      new NiceMock<Event::MockSchedulableCallback>(&server_connection_.dispatcher_);
+
+  TestResponseHeaderMapImpl response_headers{{":status", "200"}};
+  response_encoder_->encodeHeaders(response_headers, false);
+  // Account for the single HEADERS frame above and pre-fill outbound queue with 6 byte DATA frames
+  for (uint32_t i = 0; i < CommonUtility::OptionsLimits::DEFAULT_MAX_OUTBOUND_FRAMES - 2; ++i) {
+    Buffer::OwnedImpl data(std::string(6, '0'));
+    EXPECT_NO_THROW(response_encoder_->encodeData(data, false));
+  }
+
+  // client stream windows should have 5535 bytes left and the next frame should overflow it.
+  // nghttp2 sends 1 DATA frame for the remainder of the client window and it should make
+  // outbound frame queue 1 away from overflow.
+  auto flush_timer = new Event::MockTimer(&server_connection_.dispatcher_);
+  EXPECT_CALL(*flush_timer, enableTimer(std::chrono::milliseconds(30000), _));
+  Buffer::OwnedImpl large_body(std::string(6 * 1024, '1'));
+  response_encoder_->encodeData(large_body, true);
+
+  EXPECT_FALSE(violation_callback->enabled_);
+  EXPECT_CALL(server_stream_callbacks_, onResetStream(_, _));
+
+  // Pending flush timeout causes RST_STREAM to be sent and overflow the outbound frame queue.
+  flush_timer->invokeCallback();
+
+  EXPECT_TRUE(violation_callback->enabled_);
+  EXPECT_CALL(server_connection_, close(Envoy::Network::ConnectionCloseType::NoFlush));
+  violation_callback->invokeCallback();
+
+  EXPECT_EQ(1, server_stats_store_.counter("http2.tx_flush_timeout").value());
+  EXPECT_EQ(frame_count, CommonUtility::OptionsLimits::DEFAULT_MAX_OUTBOUND_FRAMES + 1);
+  EXPECT_EQ(1, server_stats_store_.counter("http2.outbound_flood").value());
+}
+
 TEST_P(Http2CodecImplTest, WatermarkUnderEndStream) {
   initialize();
   MockStreamCallbacks callbacks;

test/integration/http2_integration_test.cc

@@ -1698,7 +1698,8 @@ void Http2FloodMitigationTest::floodServer(absl::string_view host, absl::string_
             test_server_->counter("http.config_test.downstream_cx_delayed_close_timeout")->value());
 }
 
-void Http2FloodMitigationTest::prefillOutboundDownstreamQueue(uint32_t data_frame_count) {
+void Http2FloodMitigationTest::prefillOutboundDownstreamQueue(uint32_t data_frame_count,
+                                                              uint32_t data_frame_size) {
   // Set large buffer limits so the test is not affected by the flow control.
   config_helper_.setBufferLimits(1024 * 1024 * 1024, 1024 * 1024 * 1024);
   autonomous_upstream_ = true;
@@ -1715,6 +1716,7 @@ void Http2FloodMitigationTest::prefillOutboundDownstreamQueue(uint32_t data_fram
   const auto request = Http2Frame::makeRequest(
       Http2Frame::makeClientStreamId(0), "host", "/test/long/url",
       {Http2Frame::Header("response_data_blocks", absl::StrCat(data_frame_count)),
+       Http2Frame::Header("response_size_bytes", absl::StrCat(data_frame_size)),
        Http2Frame::Header("no_trailers", "0")});
   sendFrame(request);
 
@@ -2117,6 +2119,42 @@ TEST_P(Http2FloodMitigationTest, RST_STREAM) {
             test_server_->counter("http.config_test.downstream_cx_delayed_close_timeout")->value());
 }
 
+// Verify detection of flood by the RST_STREAM frame sent on pending flush timeout
+TEST_P(Http2FloodMitigationTest, RstStreamOverflowOnPendingFlushTimeout) {
+  config_helper_.addConfigModifier(
+      [](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+             hcm) {
+        hcm.mutable_stream_idle_timeout()->set_seconds(0);
+        constexpr uint64_t IdleTimeoutMs = 400;
+        hcm.mutable_stream_idle_timeout()->set_nanos(IdleTimeoutMs * 1000 * 1000);
+      });
+
+  // Pending flush timer is started when upstream response has completed but there is no window to
+  // send DATA downstream. The test downstream client does not update WINDOW and as such Envoy will
+  // use the default 65535 bytes. First, pre-fill outbound queue with 65 byte frames, which should
+  // consume 65 * 997 = 64805 bytes of downstream connection window.
+  prefillOutboundDownstreamQueue(AllFrameFloodLimit - 3, 65);
+
+  // At this point the outbound downstream frame queue should be 3 away from overflowing with 730
+  // byte window. Make response to be 1 DATA frame with 1024 payload. This should overflow the
+  // available downstream window and start pending flush timer. Envoy proxies 2 frames downstream,
+  // HEADERS and partial DATA frame, which makes the frame queue 1 away from overflow.
+  const auto request2 = Http2Frame::makeRequest(
+      Http2Frame::makeClientStreamId(1), "host", "/test/long/url",
+      {Http2Frame::Header("response_data_blocks", "1"),
+       Http2Frame::Header("response_size_bytes", "1024"), Http2Frame::Header("no_trailers", "0")});
+  sendFrame(request2);
+
+  // Pending flush timer sends RST_STREAM frame which should overflow outbound frame queue and
+  // disconnect the connection.
+  tcp_client_->waitForDisconnect();
+
+  // Verify that the flood check was triggered
+  EXPECT_EQ(1, test_server_->counter("http2.outbound_flood")->value());
+  // Verify that pending flush timeout was hit
+  EXPECT_EQ(1, test_server_->counter("http2.tx_flush_timeout")->value());
+}
+
 // Verify detection of frame flood when sending second GOAWAY frame on drain timeout
 TEST_P(Http2FloodMitigationTest, GoAwayOverflowOnDrainTimeout) {
   config_helper_.addConfigModifier(

test/integration/http2_integration_test.h

@@ -138,7 +138,7 @@ class Http2FloodMitigationTest : public SocketInterfaceSwap, public Http2FrameIn
 
   void setNetworkConnectionBufferSize();
   void beginSession() override;
-  void prefillOutboundDownstreamQueue(uint32_t data_frame_count);
+  void prefillOutboundDownstreamQueue(uint32_t data_frame_count, uint32_t data_frame_size = 10);
   void triggerListenerDrain();
 };
 } // namespace Envoy