http2: fix a bug where memory will leak if stream is reset before send headers (#41260)

To close #40253.

Risk Level: mid. core http2 code change.
Testing: unit.
Docs Changes: n/a.
Release Notes: added.

---------

Signed-off-by: WangBaiping <[email protected]>
Signed-off-by: code <[email protected]>

Author: wbpcode

changelogs/current.yaml

@@ -24,6 +24,11 @@ minor_behavior_changes:
 
 bug_fixes:
 # *Changes expected to improve the state of the world and are unlikely to have negative effects*
+- area: http2
+  change: |
+    Fixed a bug where Envoy will leak memory if the HTTP2 stream is reset before the request headers are
+    sent. For example, if one upstream http filter send a local reply after the connection is established but
+    before the request headers are sent, the memory allocated for the stream will not be released.
 - area: lua
   change: |
     Fix a bug where Lua filters may result in Envoy crashes when setting response body to a

source/common/http/http2/codec_impl.cc

@@ -408,10 +408,8 @@ void ConnectionImpl::StreamImpl::processBufferedData() {
     ENVOY_CONN_LOG(debug, "invoking onStreamClose for stream: {} via processBufferedData",
                    parent_.connection_, stream_id_);
     // We only buffer the onStreamClose if we had no errors.
-    if (Status status = parent_.onStreamClose(this, 0); !status.ok()) {
-      ENVOY_CONN_LOG(debug, "error invoking onStreamClose: {}", parent_.connection_,
-                     status.message()); // LCOV_EXCL_LINE
-    }
+    Status status = parent_.onStreamClose(this, 0);
+    ASSERT(status.ok());
   }
 }
 
@@ -810,6 +808,11 @@ void ConnectionImpl::StreamImpl::resetStream(StreamResetReason reason) {
 void ConnectionImpl::StreamImpl::resetStreamWorker(StreamResetReason reason) {
   if (stream_id_ == -1) {
     // Handle the case where client streams are reset before headers are created.
+    // For example, if we send local reply after the stream is created but before
+    // headers are sent, we will end up here.
+    ENVOY_CONN_LOG(trace, "Stream {} reset before headers sent.", parent_.connection_, stream_id_);
+    Status status = parent_.onStreamClose(this, 0);
+    ASSERT(status.ok());
     return;
   }
   if (codec_callbacks_) {
@@ -1883,6 +1886,7 @@ void ConnectionImpl::Http2Visitor::OnRstStream(Http2StreamId stream_id, Http2Err
 bool ConnectionImpl::Http2Visitor::OnCloseStream(Http2StreamId stream_id,
                                                  Http2ErrorCode error_code) {
   Status status = connection_->onStreamClose(stream_id, static_cast<uint32_t>(error_code));
+  ASSERT(status.ok());
   if (stream_close_listener_) {
     ENVOY_CONN_LOG(trace, "Http2Visitor invoking stream close listener for stream {}",
                    connection_->connection_, stream_id);

source/common/http/http2/codec_impl.h

@@ -800,7 +800,7 @@ class ConnectionImpl : public virtual Connection,
   bool raised_goaway_ : 1;
   Event::SchedulableCallbackPtr protocol_constraint_violation_callback_;
   Random::RandomGenerator& random_;
-  MonotonicTime last_received_data_time_{};
+  MonotonicTime last_received_data_time_;
   Event::TimerPtr keepalive_send_timer_;
   Event::TimerPtr keepalive_timeout_timer_;
   std::chrono::milliseconds keepalive_interval_;

test/common/http/http2/codec_impl_test.cc

@@ -252,6 +252,7 @@ class Http2CodecImplTestFixture {
     driveToCompletion();
 
     EXPECT_CALL(server_callbacks_, newStream(_, _))
+        .Times(AnyNumber())
         .WillRepeatedly(Invoke([&](ResponseEncoder& encoder, bool) -> RequestDecoder& {
           response_encoder_ = &encoder;
           encoder.getStream().addCallbacks(server_stream_callbacks_);
@@ -1009,6 +1010,14 @@ TEST_P(Http2CodecImplTest, RefusedStreamReset) {
   driveToCompletion();
 }
 
+TEST_P(Http2CodecImplTest, ResetBeforeHeadersSent) {
+  initialize();
+
+  EXPECT_EQ(1, TestUtility::findGauge(client_stats_store_, "http2.streams_active")->value());
+  request_encoder_->getStream().resetStream(StreamResetReason::LocalReset);
+  EXPECT_EQ(0, TestUtility::findGauge(client_stats_store_, "http2.streams_active")->value());
+}
+
 TEST_P(Http2CodecImplTest, InvalidHeadersFrameMissing) {
   initialize();
 #ifdef ENVOY_ENABLE_UHV

test/integration/BUILD

@@ -1023,6 +1023,7 @@ envoy_cc_test(
         "//source/common/http:header_map_lib",
         "//source/extensions/filters/http/buffer:config",
         "//test/integration/filters:encoder_decoder_buffer_filter_lib",
+        "//test/integration/filters:local_reply_during_decoding_filter_lib",
         "//test/integration/filters:random_pause_filter_lib",
         "//test/test_common:utility_lib",
         "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",

test/integration/filters/local_reply_during_decoding_filter.cc

@@ -11,11 +11,42 @@
 
 namespace Envoy {
 
-class LocalReplyDuringDecode : public Http::PassThroughFilter {
+class LocalReplyDuringDecode : public Http::PassThroughFilter, public Http::UpstreamCallbacks {
 public:
   constexpr static char name[] = "local-reply-during-decode";
 
-  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& request_headers, bool) override {
+  void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
+    decoder_callbacks_ = &callbacks;
+    if (auto cb = decoder_callbacks_->upstreamCallbacks(); cb) {
+      cb->addUpstreamCallbacks(*this);
+    }
+  }
+
+  void onUpstreamConnectionEstablished() override {
+    if (latched_end_stream_.has_value()) {
+      const bool end_stream = *latched_end_stream_;
+      latched_end_stream_.reset();
+      Http::FilterHeadersStatus status = decodeHeaders(*request_headers_, end_stream);
+      if (status == Http::FilterHeadersStatus::Continue) {
+        decoder_callbacks_->continueDecoding();
+      }
+    }
+  }
+
+  Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& request_headers,
+                                          bool end_stream) override {
+    // If this filter is being used as an upstream filter and the upstream connection is not yet
+    // established, check for the "wait-upstream-connection" header to determine whether to wait
+    // for the connection to be established before continuing decoding.
+    if (auto cb = decoder_callbacks_->upstreamCallbacks(); cb && !cb->upstream()) {
+      auto result = request_headers.get(Http::LowerCaseString("wait-upstream-connection"));
+      if (!result.empty() && result[0]->value() == "true") {
+        request_headers_ = &request_headers;
+        latched_end_stream_ = end_stream;
+        return Http::FilterHeadersStatus::StopAllIterationAndBuffer;
+      }
+    }
+
     auto result = request_headers.get(Http::LowerCaseString("skip-local-reply"));
     if (!result.empty() && result[0]->value() == "true") {
       header_local_reply_skipped_ = true;
@@ -47,6 +78,8 @@ class LocalReplyDuringDecode : public Http::PassThroughFilter {
   }
 
 private:
+  Http::RequestHeaderMap* request_headers_{};
+  absl::optional<bool> latched_end_stream_;
   bool header_local_reply_skipped_ = false;
   bool local_reply_during_data_ = false;
 };

test/integration/multiplexed_upstream_integration_test.cc

@@ -652,6 +652,37 @@ TEST_P(MultiplexedUpstreamIntegrationTest, MultipleRequestsLowStreamLimit) {
   cleanupUpstreamAndDownstream();
 }
 
+TEST_P(MultiplexedUpstreamIntegrationTest, UpstreamFilterSendLocalReply) {
+  if (upstreamProtocol() != Http::CodecType::HTTP2) {
+    return;
+  }
+  autonomous_upstream_ = true;
+  envoy::config::core::v3::Http2ProtocolOptions config;
+  config.mutable_max_concurrent_streams()->set_value(20000);
+  mergeOptions(config);
+  config_helper_.prependFilter(fmt::format(R"EOF(
+  name: local-reply-during-decode
+)EOF"),
+                               false);
+
+  initialize();
+  codec_client_ = makeHttpConnection(lookupPort("http"));
+
+  // Start sending the request, but ensure no end stream will be sent, so the
+  // stream will stay in use.
+  auto response = codec_client_->makeHeaderOnlyRequest(
+      Http::TestRequestHeaderMapImpl{{":method", "POST"},
+                                     {":path", "/test/long/url"},
+                                     {":scheme", "http"},
+                                     {":authority", "sni.lyft.com"},
+                                     {"wait-upstream-connection", "true"}});
+  // Wait until the response is sent to ensure the SETTINGS frame has been read
+  // by Envoy.
+  ASSERT_TRUE(response->waitForEndStream());
+
+  EXPECT_EQ(0, test_server_->gauge("cluster.cluster_0.http2.streams_active")->value());
+}
+
 // Regression test for https://github.com/envoyproxy/envoy/issues/13933
 TEST_P(MultiplexedUpstreamIntegrationTest, UpstreamGoaway) {
   initialize();