CVE-2022-23606: Uncontrolled Recursion (CWE-674)
value: Envoy is an open source edge and service proxy, designed for cloud-native applications. When a cluster is deleted via Cluster Discovery Service (CDS) all idle connections established to endpoints in that cluster are disconnected. A recursion was introduced in the procedure of disconnecting idle connections that can lead to stack exhaustion and abnormal process termination when a cluster has a large number of idle connections. This infinite recursion causes Envoy to crash. Users are advised to upgrade.
CWE description: The product does not properly control the amount of recursion that takes place,  consuming excessive resources, such as allocated memory or the program stack.
CVSS v3 base score: 6.5
Published: 2022-02-22T23:15Z
Repository: https://github.com/envoyproxy/envoy
Language: C++
Files changed: docs/root/version_history/current.rst, source/common/conn_pool/BUILD, source/common/conn_pool/conn_pool_base.cc, source/common/conn_pool/conn_pool_base.h, test/config/utility.cc, test/config/utility.h, test/integration/BUILD, test/integration/cds_integration_test.cc
Commit date: 2022-02-22 17:58:22 +0000
Commit message: CVE-2022-23606

Avoid closing other connections to prevent deep recursion when a large number of idle connections are closed at the start of a pool drain, when a connection is closed.

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

Diff: diff --git a/docs/root/version_history/current.rst b/docs/root/version_history/current.rst
index 4bc41e24b30..f420dcf3598 100644
--- a/docs/root/version_history/current.rst
+++ b/docs/root/version_history/current.rst
@@ -26,10 +26,11 @@ Bug Fixes
 
 * access_log: fix memory leak when reopening an access log fails. Access logs will now try to be reopened on each subsequent flush attempt after a failure.
 * data plane: fixing error handling where writing to a socket failed while under the stack of processing. This should genreally affect HTTP/3. This behavioral change can be reverted by setting ``envoy.reloadable_features.allow_upstream_inline_write`` to false.
 * eds: fix the eds cluster update by allowing update on the locality of the cluster endpoints. This behavioral change can be temporarily reverted by setting runtime guard ``envoy.reloadable_features.support_locality_update_on_eds_cluster_endpoints`` to false.
 * tls: fix a bug while matching a certificate SAN with an exact value in ``match_typed_subject_alt_names`` of a listener where wildcard ``*`` character is not the only character of the dns label. Example, ``baz*.example.net`` and ``*baz.example.net`` and ``b*z.example.net`` will match ``baz1.example.net`` and ``foobaz.example.net`` and ``buzz.example.net``, respectively.
+* upstream: fix stack overflow when a cluster with large number of idle connections is removed.
 * xray: fix the AWS X-Ray tracer extension to not sample the trace if ``sampled=`` keyword is not present in the header ``x-amzn-trace-id``.
 
 Removed Config or Runtime
 -------------------------
 *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`
diff --git a/source/common/conn_pool/BUILD b/source/common/conn_pool/BUILD
index d526dfbf956..048aa131fbb 100644
--- a/source/common/conn_pool/BUILD
+++ b/source/common/conn_pool/BUILD
@@ -12,10 +12,11 @@ envoy_cc_library(
     name = "conn_pool_base_lib",
     srcs = ["conn_pool_base.cc"],
     hdrs = ["conn_pool_base.h"],
     deps = [
         "//envoy/stats:timespan_interface",
+        "//source/common/common:debug_recursion_checker_lib",
         "//source/common/common:linked_object",
         "//source/common/stats:timespan_lib",
         "//source/common/upstream:upstream_lib",
     ],
 )
diff --git a/source/common/conn_pool/conn_pool_base.cc b/source/common/conn_pool/conn_pool_base.cc
index 1493b757d2f..4d0cb8acfb6 100644
--- a/source/common/conn_pool/conn_pool_base.cc
+++ b/source/common/conn_pool/conn_pool_base.cc
@@ -1,8 +1,9 @@
 #include "source/common/conn_pool/conn_pool_base.h"
 
 #include "source/common/common/assert.h"
+#include "source/common/common/debug_recursion_checker.h"
 #include "source/common/network/transport_socket_options_impl.h"
 #include "source/common/runtime/runtime_features.h"
 #include "source/common/stats/timespan_impl.h"
 #include "source/common/upstream/upstream_impl.h"
 
@@ -347,10 +348,12 @@ void ConnPoolImplBase::transitionActiveClientState(ActiveClient& client,
 }
 
 void ConnPoolImplBase::addIdleCallbackImpl(Instance::IdleCb cb) { idle_callbacks_.push_back(cb); }
 
 void ConnPoolImplBase::closeIdleConnectionsForDrainingPool() {
+  Common::AutoDebugRecursionChecker assert_not_in(recursion_checker_);
+
   // Create a separate list of elements to close to avoid mutate-while-iterating problems.
   std::list<ActiveClient*> to_close;
 
   for (auto& client : ready_clients_) {
     if (client->numActiveStreams() == 0) {
@@ -401,24 +404,28 @@ void ConnPoolImplBase::drainConnectionsImpl(DrainBehavior drain_behavior) {
 bool ConnPoolImplBase::isIdleImpl() const {
   return pending_streams_.empty() && ready_clients_.empty() && busy_clients_.empty() &&
          connecting_clients_.empty();
 }
 
-void ConnPoolImplBase::checkForIdleAndCloseIdleConnsIfDraining() {
-  if (is_draining_for_deletion_) {
-    closeIdleConnectionsForDrainingPool();
-  }
-
+void ConnPoolImplBase::checkForIdleAndNotify() {
   if (isIdleImpl()) {
     ENVOY_LOG(debug, "invoking idle callbacks - is_draining_for_deletion_={}",
               is_draining_for_deletion_);
     for (const Instance::IdleCb& cb : idle_callbacks_) {
       cb();
     }
   }
 }
 
+void ConnPoolImplBase::checkForIdleAndCloseIdleConnsIfDraining() {
+  if (is_draining_for_deletion_) {
+    closeIdleConnectionsForDrainingPool();
+  }
+
+  checkForIdleAndNotify();
+}
+
 void ConnPoolImplBase::onConnectionEvent(ActiveClient& client, absl::string_view failure_reason,
                                          Network::ConnectionEvent event) {
   if (client.state() == ActiveClient::State::CONNECTING) {
     ASSERT(connecting_stream_capacity_ >= client.effectiveConcurrentStreamLimit());
     connecting_stream_capacity_ -= client.effectiveConcurrentStreamLimit();
@@ -485,11 +492,19 @@ void ConnPoolImplBase::onConnectionEvent(ActiveClient& client, absl::string_view
       client.connection_duration_timer_.reset();
     }
 
     dispatcher_.deferredDelete(client.removeFromList(owningList(client.state())));
 
-    checkForIdleAndCloseIdleConnsIfDraining();
+    // Check if the pool transitioned to idle state after removing closed client
+    // from one of the client tracking lists.
+    // There is no need to check if other connections are idle in a draining pool
+    // because the pool will close all idle connection when it is starting to
+    // drain.
+    // Trying to close other connections here can lead to deep recursion when
+    // a large number idle connections are closed at the start of pool drain.
+    // See CdsIntegrationTest.CdsClusterDownWithLotsOfIdleConnections for an example.
+    checkForIdleAndNotify();
 
     client.setState(ActiveClient::State::CLOSED);
 
     // If we have pending streams and we just lost a connection we should make a new one.
     if (!pending_streams_.empty()) {
diff --git a/source/common/conn_pool/conn_pool_base.h b/source/common/conn_pool/conn_pool_base.h
index b127fc234dc..d475ecf0988 100644
--- a/source/common/conn_pool/conn_pool_base.h
+++ b/source/common/conn_pool/conn_pool_base.h
@@ -4,10 +4,11 @@
 #include "envoy/event/dispatcher.h"
 #include "envoy/network/connection.h"
 #include "envoy/stats/timespan.h"
 #include "envoy/upstream/cluster_manager.h"
 
+#include "source/common/common/debug_recursion_checker.h"
 #include "source/common/common/dump_state_utils.h"
 #include "source/common/common/linked_object.h"
 
 #include "absl/strings/string_view.h"
 
@@ -220,10 +221,13 @@ public:
   void transitionActiveClientState(ActiveClient& client, ActiveClient::State new_state);
 
   void onConnectionEvent(ActiveClient& client, absl::string_view failure_reason,
                          Network::ConnectionEvent event);
 
+  // Check if the pool has gone idle and invoke idle notification callbacks.
+  void checkForIdleAndNotify();
+
   // See if the pool has gone idle. If we're draining, this will also close idle connections.
   void checkForIdleAndCloseIdleConnsIfDraining();
 
   void scheduleOnUpstreamReady();
   ConnectionPool::Cancellable* newStreamImpl(AttachContext& context, bool can_send_early_data);
@@ -369,9 +373,10 @@ private:
 
   // True iff this object is in the deferred delete list.
   bool deferred_deleting_{false};
 
   Event::SchedulableCallbackPtr upstream_ready_cb_;
+  Common::DebugRecursionChecker recursion_checker_;
 };
 
 } // namespace ConnectionPool
 } // namespace Envoy
diff --git a/test/config/utility.cc b/test/config/utility.cc
index a8248fd5109..d2d9bacc285 100644
--- a/test/config/utility.cc
+++ b/test/config/utility.cc
@@ -440,10 +440,38 @@ envoy::config::cluster::v3::Cluster ConfigHelper::buildStaticCluster(const std::
             http2_protocol_options: {{}}
     )EOF",
                   name, name, address, port, lb_policy));
 }
 
+envoy::config::cluster::v3::Cluster ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits(
+    const std::string& name, int port, const std::string& address, const std::string& lb_policy) {
+  return TestUtility::parseYaml<envoy::config::cluster::v3::Cluster>(
+      fmt::format(R"EOF(
+      name: {}
+      connect_timeout: 50s
+      type: STATIC
+      circuit_breakers:
+        thresholds:
+        - priority: DEFAULT
+          max_connections: 10000
+          max_pending_requests: 10000
+          max_requests: 10000
+          max_retries: 10000
+      load_assignment:
+        cluster_name: {}
+        endpoints:
+        - lb_endpoints:
+          - endpoint:
+              address:
+                socket_address:
+                  address: {}
+                  port_value: {}
+      lb_policy: {}
+    )EOF",
+                  name, name, address, port, lb_policy));
+}
+
 envoy::config::cluster::v3::Cluster ConfigHelper::buildCluster(const std::string& name,
                                                                const std::string& lb_policy) {
   API_NO_BOOST(envoy::config::cluster::v3::Cluster) cluster;
   TestUtility::loadFromYaml(fmt::format(R"EOF(
       name: {}
diff --git a/test/config/utility.h b/test/config/utility.h
index bd6830f57f0..be4d0f1d042 100644
--- a/test/config/utility.h
+++ b/test/config/utility.h
@@ -151,10 +151,15 @@ public:
   // Builds a standard Cluster config fragment, with a single endpoint (at address:port).
   static envoy::config::cluster::v3::Cluster
   buildStaticCluster(const std::string& name, int port, const std::string& address,
                      const std::string& lb_policy = "ROUND_ROBIN");
 
+  static envoy::config::cluster::v3::Cluster
+  buildH1ClusterWithHighCircuitBreakersLimits(const std::string& name, int port,
+                                              const std::string& address,
+                                              const std::string& lb_policy = "ROUND_ROBIN");
+
   // ADS configurations
   static envoy::config::cluster::v3::Cluster
   buildCluster(const std::string& name, const std::string& lb_policy = "ROUND_ROBIN");
 
   static envoy::config::cluster::v3::Cluster
diff --git a/test/integration/BUILD b/test/integration/BUILD
index c9f17d22f0b..990e17b2d03 100644
--- a/test/integration/BUILD
+++ b/test/integration/BUILD
@@ -108,14 +108,16 @@ envoy_proto_library(
     srcs = [":h2_capture_fuzz.proto"],
 )
 
 envoy_cc_test(
     name = "cds_integration_test",
+    size = "large",
     srcs = ["cds_integration_test.cc"],
     data = [
         "//test/config/integration/certs",
     ],
+    shard_count = 4,
     deps = [
         ":http_integration_lib",
         "//source/common/config:protobuf_link_hacks",
         "//source/common/protobuf:utility_lib",
         "//test/common/grpc:grpc_client_integration_lib",
diff --git a/test/integration/cds_integration_test.cc b/test/integration/cds_integration_test.cc
index 53f82125617..58c737f57af 100644
--- a/test/integration/cds_integration_test.cc
+++ b/test/integration/cds_integration_test.cc
@@ -35,11 +35,12 @@ public:
       : HttpIntegrationTest(Http::CodecType::HTTP2, ipVersion(),
                             ConfigHelper::discoveredClustersBootstrap(
                                 sotwOrDelta() == Grpc::SotwOrDelta::Sotw ||
                                         sotwOrDelta() == Grpc::SotwOrDelta::UnifiedSotw
                                     ? "GRPC"
-                                    : "DELTA_GRPC")) {
+                                    : "DELTA_GRPC")),
+        cluster_creator_(&ConfigHelper::buildStaticCluster) {
     if (sotwOrDelta() == Grpc::SotwOrDelta::UnifiedSotw ||
         sotwOrDelta() == Grpc::SotwOrDelta::UnifiedDelta) {
       config_helper_.addRuntimeOverride("envoy.reloadable_features.unified_mux", "true");
     }
     use_lds_ = false;
@@ -77,18 +78,18 @@ public:
     HttpIntegrationTest::initialize();
 
     // Create the regular (i.e. not an xDS server) upstreams. We create them manually here after
     // initialize() because finalize() expects all fake_upstreams_ to correspond to a static
     // cluster in the bootstrap config - which we don't want since we're testing dynamic CDS!
-    addFakeUpstream(Http::CodecType::HTTP2);
-    addFakeUpstream(Http::CodecType::HTTP2);
-    cluster1_ = ConfigHelper::buildStaticCluster(
+    addFakeUpstream(upstream_codec_type_);
+    addFakeUpstream(upstream_codec_type_);
+    cluster1_ = cluster_creator_(
         ClusterName1, fake_upstreams_[UpstreamIndex1]->localAddress()->ip()->port(),
-        Network::Test::getLoopbackAddressString(ipVersion()));
-    cluster2_ = ConfigHelper::buildStaticCluster(
+        Network::Test::getLoopbackAddressString(ipVersion()), "ROUND_ROBIN");
+    cluster2_ = cluster_creator_(
         ClusterName2, fake_upstreams_[UpstreamIndex2]->localAddress()->ip()->port(),
-        Network::Test::getLoopbackAddressString(ipVersion()));
+        Network::Test::getLoopbackAddressString(ipVersion()), "ROUND_ROBIN");
 
     // Let Envoy establish its connection to the CDS server.
     acceptXdsConnection();
 
     // Do the initial compareDiscoveryRequest / sendDiscoveryResponse for cluster_1.
@@ -131,10 +132,14 @@ public:
 
   envoy::config::cluster::v3::Cluster cluster1_;
   envoy::config::cluster::v3::Cluster cluster2_;
   // True if we decided not to run the test after all.
   bool test_skipped_{true};
+  Http::CodecType upstream_codec_type_{Http::CodecType::HTTP2};
+  std::function<envoy::config::cluster::v3::Cluster(const std::string&, int, const std::string&,
+                                                    const std::string&)>
+      cluster_creator_;
 };
 
 INSTANTIATE_TEST_SUITE_P(IpVersionsClientTypeDelta, CdsIntegrationTest,
                          DELTA_SOTW_GRPC_CLIENT_INTEGRATION_PARAMS);
 
@@ -341,7 +346,162 @@ TEST_P(CdsIntegrationTest, VersionsRememberedAfterReconnect) {
   testRouterHeaderOnlyRequestAndResponse(nullptr, UpstreamIndex2, "/cluster2");
   cleanupUpstreamAndDownstream();
   ASSERT_TRUE(codec_client_->waitForDisconnect());
 }
 
+// This test verifies that Envoy can delete a cluster with a lot of idle connections.
+// The original problem was recursive closure of idle connections that can run out
+// of stack when there are a lot of idle connections.
+TEST_P(CdsIntegrationTest, CdsClusterDownWithLotsOfIdleConnections) {
+  constexpr int num_requests = 2000;
+  // Make upstream H/1 so it creates connection for each request
+  upstream_codec_type_ = Http::CodecType::HTTP1;
+  // Relax default circuit breaker limits and timeouts so Envoy can accumulate a lot of idle
+  // connections
+  cluster_creator_ = &ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits;
+  config_helper_.addConfigModifier(
+      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+              hcm) -> void {
+        hcm.mutable_route_config()
+            ->mutable_virtual_hosts(0)
+            ->mutable_routes(0)
+            ->mutable_route()
+            ->mutable_timeout()
+            ->set_seconds(600);
+        hcm.mutable_route_config()
+            ->mutable_virtual_hosts(0)
+            ->mutable_routes(0)
+            ->mutable_route()
+            ->mutable_idle_timeout()
+            ->set_seconds(600);
+      });
+  initialize();
+  std::vector<IntegrationStreamDecoderPtr> responses;
+  std::vector<FakeHttpConnectionPtr> upstream_connections;
+  std::vector<FakeStreamPtr> upstream_requests;
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+  // The first loop establishes a lot of open connections with active requests to upstream
+  for (int i = 0; i < num_requests; ++i) {
+    Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"},
+                                                   {":path", "/cluster1"},
+                                                   {":scheme", "http"},
+                                                   {":authority", "host"},
+                                                   {"x-lyft-user-id", absl::StrCat(i)}};
+
+    auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
+    responses.push_back(std::move(response));
+
+    FakeHttpConnectionPtr fake_upstream_connection;
+    waitForNextUpstreamConnection({UpstreamIndex1}, TestUtility::DefaultTimeout,
+                                  fake_upstream_connection);
+    // Wait for the next stream on the upstream connection.
+    FakeStreamPtr upstream_request;
+    AssertionResult result =
+        fake_upstream_connection->waitForNewStream(*dispatcher_, upstream_request);
+    RELEASE_ASSERT(result, result.message());
+    // Wait for the stream to be completely received.
+    result = upstream_request->waitForEndStream(*dispatcher_);
+    RELEASE_ASSERT(result, result.message());
+    upstream_connections.push_back(std::move(fake_upstream_connection));
+    upstream_requests.push_back(std::move(upstream_request));
+  }
+
+  // This loop completes all requests making the all upstream connections idle
+  for (int i = 0; i < num_requests; ++i) {
+    // Send response headers, and end_stream if there is no response body.
+    upstream_requests[i]->encodeHeaders(default_response_headers_, true);
+    // Wait for the response to be read by the codec client.
+    RELEASE_ASSERT(responses[i]->waitForEndStream(), "unexpected timeout");
+    ASSERT_TRUE(responses[i]->complete());
+    EXPECT_EQ("200", responses[i]->headers().getStatusValue());
+  }
+
+  test_server_->waitForCounterGe("cluster_manager.cluster_added", 1);
+
+  // Tell Envoy that cluster_1 is gone. Envoy will try to close all idle connections
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "55", {}, {}, {}));
+  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster, {}, {},
+                                                             {ClusterName1}, "42");
+  // We can continue the test once we're sure that Envoy's ClusterManager has made use of
+  // the DiscoveryResponse that says cluster_1 is gone.
+  test_server_->waitForCounterGe("cluster_manager.cluster_removed", 1);
+
+  // If we made it this far then everything is ok.
+  for (int i = 0; i < num_requests; ++i) {
+    AssertionResult result = upstream_connections[i]->close();
+    RELEASE_ASSERT(result, result.message());
+    result = upstream_connections[i]->waitForDisconnect();
+    RELEASE_ASSERT(result, result.message());
+  }
+  upstream_connections.clear();
+  cleanupUpstreamAndDownstream();
+  ASSERT_TRUE(codec_client_->waitForDisconnect());
+}
+
+// This test verifies that Envoy can delete a cluster with a lot of connections in the connecting
+// state and associated pending requests. The recursion guard in the
+// ConnPoolImplBase::closeIdleConnectionsForDrainingPool() would fire if it was called recursively.
+//
+// Test is currently disabled as there is presently no reliable way of making upstream connections
+// hang in connecting state.
+TEST_P(CdsIntegrationTest, DISABLED_CdsClusterDownWithLotsOfConnectingConnections) {
+  // Use low number of pending connections to prevent bumping into the default
+  // limit of 128, since the upstream will be prevented below from
+  // accepting connections.
+  constexpr int num_requests = 64;
+  // Make upstream H/1 so it creates connection for each request
+  upstream_codec_type_ = Http::CodecType::HTTP1;
+  cluster_creator_ = &ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits;
+  config_helper_.addConfigModifier(
+      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
+              hcm) -> void {
+        hcm.mutable_route_config()
+            ->mutable_virtual_hosts(0)
+            ->mutable_routes(0)
+            ->mutable_route()
+            ->mutable_timeout()
+            ->set_seconds(600);
+        hcm.mutable_route_config()
+            ->mutable_virtual_hosts(0)
+            ->mutable_routes(0)
+            ->mutable_route()
+            ->mutable_idle_timeout()
+            ->set_seconds(600);
+      });
+  initialize();
+  test_server_->waitForCounterGe("cluster_manager.cluster_added", 1);
+  std::vector<IntegrationStreamDecoderPtr> responses;
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+  // Stop upstream at UpstreamIndex1 dispatcher, to prevent it from accepting TCP connections.
+  // This will cause Envoy's connections to that upstream hang in the connecting state.
+  fake_upstreams_[UpstreamIndex1]->dispatcher()->exit();
+  for (int i = 0; i < num_requests; ++i) {
+    Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"},
+                                                   {":path", "/cluster1"},
+                                                   {":scheme", "http"},
+                                                   {":authority", "host"},
+                                                   {"x-lyft-user-id", absl::StrCat(i)}};
+
+    auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
+    responses.push_back(std::move(response));
+  }
+
+  // Wait for Envoy to try to establish all expected connections
+  test_server_->waitForCounterEq("cluster.cluster_1.upstream_cx_total", num_requests);
+
+  // Tell Envoy that cluster_1 is gone. Envoy will try to close all pending connections
+  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "55", {}, {}, {}));
+  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster, {}, {},
+                                                             {ClusterName1}, "42");
+  // We can continue the test once we're sure that Envoy's ClusterManager has made use of
+  // the DiscoveryResponse that says cluster_1 is gone.
+  test_server_->waitForCounterGe("cluster_manager.cluster_removed", 1);
+
+  cleanupUpstreamAndDownstream();
+  ASSERT_TRUE(codec_client_->waitForDisconnect());
+  // If we got here it means that the recursion guard in the
+  // ConnPoolImplBase::closeIdleConnectionsForDrainingPool() did not fire, which is what this test
+  // validates.
+}
+
 } // namespace
 } // namespace Envoy

Vulnerable code: void ConnPoolImplBase::checkForIdleAndCloseIdleConnsIfDraining() {
  if (is_draining_for_deletion_) {
    closeIdleConnectionsForDrainingPool();
  }

    checkForIdleAndCloseIdleConnsIfDraining();
                                    : "DELTA_GRPC")) {
    addFakeUpstream(Http::CodecType::HTTP2);
    addFakeUpstream(Http::CodecType::HTTP2);
    cluster1_ = ConfigHelper::buildStaticCluster(
        Network::Test::getLoopbackAddressString(ipVersion()));
    cluster2_ = ConfigHelper::buildStaticCluster(
        Network::Test::getLoopbackAddressString(ipVersion()));
Fixed code: * upstream: fix stack overflow when a cluster with large number of idle connections is removed.
        "//source/common/common:debug_recursion_checker_lib",
#include "source/common/common/debug_recursion_checker.h"
  Common::AutoDebugRecursionChecker assert_not_in(recursion_checker_);

void ConnPoolImplBase::checkForIdleAndNotify() {
void ConnPoolImplBase::checkForIdleAndCloseIdleConnsIfDraining() {
  if (is_draining_for_deletion_) {
    closeIdleConnectionsForDrainingPool();
  }

  checkForIdleAndNotify();
}

    // Check if the pool transitioned to idle state after removing closed client
    // from one of the client tracking lists.
    // There is no need to check if other connections are idle in a draining pool
    // because the pool will close all idle connection when it is starting to
    // drain.
    // Trying to close other connections here can lead to deep recursion when
    // a large number idle connections are closed at the start of pool drain.
    // See CdsIntegrationTest.CdsClusterDownWithLotsOfIdleConnections for an example.
    checkForIdleAndNotify();
#include "source/common/common/debug_recursion_checker.h"
  // Check if the pool has gone idle and invoke idle notification callbacks.
  void checkForIdleAndNotify();

  Common::DebugRecursionChecker recursion_checker_;
envoy::config::cluster::v3::Cluster ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits(
    const std::string& name, int port, const std::string& address, const std::string& lb_policy) {
  return TestUtility::parseYaml<envoy::config::cluster::v3::Cluster>(
      fmt::format(R"EOF(
      name: {}
      connect_timeout: 50s
      type: STATIC
      circuit_breakers:
        thresholds:
        - priority: DEFAULT
          max_connections: 10000
          max_pending_requests: 10000
          max_requests: 10000
          max_retries: 10000
      load_assignment:
        cluster_name: {}
        endpoints:
        - lb_endpoints:
          - endpoint:
              address:
                socket_address:
                  address: {}
                  port_value: {}
      lb_policy: {}
    )EOF",
                  name, name, address, port, lb_policy));
}

  static envoy::config::cluster::v3::Cluster
  buildH1ClusterWithHighCircuitBreakersLimits(const std::string& name, int port,
                                              const std::string& address,
                                              const std::string& lb_policy = "ROUND_ROBIN");

    size = "large",
    shard_count = 4,
                                    : "DELTA_GRPC")),
        cluster_creator_(&ConfigHelper::buildStaticCluster) {
    addFakeUpstream(upstream_codec_type_);
    addFakeUpstream(upstream_codec_type_);
    cluster1_ = cluster_creator_(
        Network::Test::getLoopbackAddressString(ipVersion()), "ROUND_ROBIN");
    cluster2_ = cluster_creator_(
        Network::Test::getLoopbackAddressString(ipVersion()), "ROUND_ROBIN");
  Http::CodecType upstream_codec_type_{Http::CodecType::HTTP2};
  std::function<envoy::config::cluster::v3::Cluster(const std::string&, int, const std::string&,
                                                    const std::string&)>
      cluster_creator_;
// This test verifies that Envoy can delete a cluster with a lot of idle connections.
// The original problem was recursive closure of idle connections that can run out
// of stack when there are a lot of idle connections.
TEST_P(CdsIntegrationTest, CdsClusterDownWithLotsOfIdleConnections) {
  constexpr int num_requests = 2000;
  // Make upstream H/1 so it creates connection for each request
  upstream_codec_type_ = Http::CodecType::HTTP1;
  // Relax default circuit breaker limits and timeouts so Envoy can accumulate a lot of idle
  // connections
  cluster_creator_ = &ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits;
  config_helper_.addConfigModifier(
      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
              hcm) -> void {
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_timeout()
            ->set_seconds(600);
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_idle_timeout()
            ->set_seconds(600);
      });
  initialize();
  std::vector<IntegrationStreamDecoderPtr> responses;
  std::vector<FakeHttpConnectionPtr> upstream_connections;
  std::vector<FakeStreamPtr> upstream_requests;
  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
  // The first loop establishes a lot of open connections with active requests to upstream
  for (int i = 0; i < num_requests; ++i) {
    Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"},
                                                   {":path", "/cluster1"},
                                                   {":scheme", "http"},
                                                   {":authority", "host"},
                                                   {"x-lyft-user-id", absl::StrCat(i)}};

    auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
    responses.push_back(std::move(response));

    FakeHttpConnectionPtr fake_upstream_connection;
    waitForNextUpstreamConnection({UpstreamIndex1}, TestUtility::DefaultTimeout,
                                  fake_upstream_connection);
    // Wait for the next stream on the upstream connection.
    FakeStreamPtr upstream_request;
    AssertionResult result =
        fake_upstream_connection->waitForNewStream(*dispatcher_, upstream_request);
    RELEASE_ASSERT(result, result.message());
    // Wait for the stream to be completely received.
    result = upstream_request->waitForEndStream(*dispatcher_);
    RELEASE_ASSERT(result, result.message());
    upstream_connections.push_back(std::move(fake_upstream_connection));
    upstream_requests.push_back(std::move(upstream_request));
  }

  // This loop completes all requests making the all upstream connections idle
  for (int i = 0; i < num_requests; ++i) {
    // Send response headers, and end_stream if there is no response body.
    upstream_requests[i]->encodeHeaders(default_response_headers_, true);
    // Wait for the response to be read by the codec client.
    RELEASE_ASSERT(responses[i]->waitForEndStream(), "unexpected timeout");
    ASSERT_TRUE(responses[i]->complete());
    EXPECT_EQ("200", responses[i]->headers().getStatusValue());
  }

  test_server_->waitForCounterGe("cluster_manager.cluster_added", 1);

  // Tell Envoy that cluster_1 is gone. Envoy will try to close all idle connections
  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "55", {}, {}, {}));
  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster, {}, {},
                                                             {ClusterName1}, "42");
  // We can continue the test once we're sure that Envoy's ClusterManager has made use of
  // the DiscoveryResponse that says cluster_1 is gone.
  test_server_->waitForCounterGe("cluster_manager.cluster_removed", 1);

  // If we made it this far then everything is ok.
  for (int i = 0; i < num_requests; ++i) {
    AssertionResult result = upstream_connections[i]->close();
    RELEASE_ASSERT(result, result.message());
    result = upstream_connections[i]->waitForDisconnect();
    RELEASE_ASSERT(result, result.message());
  }
  upstream_connections.clear();
  cleanupUpstreamAndDownstream();
  ASSERT_TRUE(codec_client_->waitForDisconnect());
}

// This test verifies that Envoy can delete a cluster with a lot of connections in the connecting
// state and associated pending requests. The recursion guard in the
// ConnPoolImplBase::closeIdleConnectionsForDrainingPool() would fire if it was called recursively.
//
// Test is currently disabled as there is presently no reliable way of making upstream connections
// hang in connecting state.
TEST_P(CdsIntegrationTest, DISABLED_CdsClusterDownWithLotsOfConnectingConnections) {
  // Use low number of pending connections to prevent bumping into the default
  // limit of 128, since the upstream will be prevented below from
  // accepting connections.
  constexpr int num_requests = 64;
  // Make upstream H/1 so it creates connection for each request
  upstream_codec_type_ = Http::CodecType::HTTP1;
  cluster_creator_ = &ConfigHelper::buildH1ClusterWithHighCircuitBreakersLimits;
  config_helper_.addConfigModifier(
      [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
              hcm) -> void {
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_timeout()
            ->set_seconds(600);
        hcm.mutable_route_config()
            ->mutable_virtual_hosts(0)
            ->mutable_routes(0)
            ->mutable_route()
            ->mutable_idle_timeout()
            ->set_seconds(600);
      });
  initialize();
  test_server_->waitForCounterGe("cluster_manager.cluster_added", 1);
  std::vector<IntegrationStreamDecoderPtr> responses;
  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
  // Stop upstream at UpstreamIndex1 dispatcher, to prevent it from accepting TCP connections.
  // This will cause Envoy's connections to that upstream hang in the connecting state.
  fake_upstreams_[UpstreamIndex1]->dispatcher()->exit();
  for (int i = 0; i < num_requests; ++i) {
    Http::TestRequestHeaderMapImpl request_headers{{":method", "GET"},
                                                   {":path", "/cluster1"},
                                                   {":scheme", "http"},
                                                   {":authority", "host"},
                                                   {"x-lyft-user-id", absl::StrCat(i)}};

    auto response = codec_client_->makeHeaderOnlyRequest(request_headers);
    responses.push_back(std::move(response));
  }

  // Wait for Envoy to try to establish all expected connections
  test_server_->waitForCounterEq("cluster.cluster_1.upstream_cx_total", num_requests);

  // Tell Envoy that cluster_1 is gone. Envoy will try to close all pending connections
  EXPECT_TRUE(compareDiscoveryRequest(Config::TypeUrl::get().Cluster, "55", {}, {}, {}));
  sendDiscoveryResponse<envoy::config::cluster::v3::Cluster>(Config::TypeUrl::get().Cluster, {}, {},
                                                             {ClusterName1}, "42");
  // We can continue the test once we're sure that Envoy's ClusterManager has made use of
  // the DiscoveryResponse that says cluster_1 is gone.
  test_server_->waitForCounterGe("cluster_manager.cluster_removed", 1);

  cleanupUpstreamAndDownstream();
  ASSERT_TRUE(codec_client_->waitForDisconnect());
  // If we got here it means that the recursion guard in the
  // ConnPoolImplBase::closeIdleConnectionsForDrainingPool() did not fire, which is what this test
  // validates.
}