conn pool: fix bugs leading to incorrect conns created (#39446)

Backport #39310 and #39349

If a connection starts draining while it has negative unused capacity
(which happens if a SETTINGS frame reduces allowed concurrency to below
the current number of requests), that connections unused capacity will
be included in total pool capacity even though it is unusable because it
is draining. This can result in not enough connections being established
for current pending requests.

This is most problematic for long-lived requests (such as streaming gRPC
requests or long-poll requests) because a connection could be in the
draining state for a long time.

Maybe fixes: #39238

Fixed an issue that could lead to too many connections when using
:ref:`AutoHttpConfig

<envoy_v3_api_msg_extensions.upstreams.http.v3.HttpProtocolOptions.AutoHttpConfig>`
if the
established connection is ``http/2`` and Envoy predicted it would have
lower concurrent capacity.

---------

Signed-off-by: Greg Greenway <[email protected]>
Signed-off-by: Dario Cillerai <[email protected]>

Author: ggreenway

changelogs/current.yaml

@@ -8,14 +8,19 @@ minor_behavior_changes:
 
 bug_fixes:
 # *Changes expected to improve the state of the world and are unlikely to have negative effects*
-- area: eds
+- area: conn_pool
   change: |
-    Fixed crash when creating an EDS cluster with invalid configuration.
-- area: url_template
+    Fixed an issue that could lead to too many connections when using
+    :ref:`AutoHttpConfig <envoy_v3_api_msg_extensions.upstreams.http.v3.HttpProtocolOptions.AutoHttpConfig>` if the
+    established connection is ``http/2`` and Envoy predicted it would have lower concurrent capacity.
+- area: conn_pool
   change: |
-    Included the asterisk ``*`` in the match pattern when using the * or ** operators in the URL template.
-    This behavioral change can be temporarily reverted by setting runtime guard
-    ``envoy.reloadable_features.uri_template_match_on_asterisk`` to ``false``.
+    Fixed an issue that could lead to insufficient connections for current pending requests. If a connection starts draining while it
+    has negative unused capacity (which happens if an HTTP/2 ``SETTINGS`` frame reduces allowed concurrency to below the current number
+    of requests), that connection's unused capacity will be included in total pool capacity even though it is unusable because it is
+    draining. This can result in not enough connections being established for current pending requests. This is most problematic for
+    long-lived requests (such as streaming gRPC requests or long-poll requests) because a connection could be in the draining state
+    for a long time.
 
 removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`

source/common/conn_pool/conn_pool_base.cc

@@ -261,7 +261,7 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
   // We don't update the capacity for HTTP/3 as the stream count should only
   // increase when a MAX_STREAMS frame is received.
   if (trackStreamCapacity()) {
-    // If the effective client capacity was limited by concurrency, increase connecting capacity.
+    // If the effective client capacity was limited by concurrency, increase connected capacity.
     bool limited_by_concurrency =
         client.remaining_streams_ > client.concurrent_stream_limit_ - client.numActiveStreams() - 1;
     // The capacity calculated by concurrency could be negative if a SETTINGS frame lowered the
@@ -878,12 +878,20 @@ void ActiveClient::onConnectionDurationTimeout() {
 }
 
 void ActiveClient::drain() {
-  if (currentUnusedCapacity() <= 0) {
-    return;
-  }
-  parent_.decrConnectingAndConnectedStreamCapacity(currentUnusedCapacity(), *this);
+  const int64_t unused = currentUnusedCapacity();
 
+  // Remove draining client's capacity from the pool.
+  //
+  // The code that adds capacity back to the pool in `onStreamClosed` will only add it back if
+  // it sees the connection as currently limited by concurrent capacity, not total lifetime streams.
+  // Setting this to zero ensures that the `limited_by_concurrency` check does not detect this
+  // connection as limited for that reason, because it is now being marked as having zero remaining
+  // lifetime requests.
   remaining_streams_ = 0;
+
+  if (unused > 0) {
+    parent_.decrConnectingAndConnectedStreamCapacity(unused, *this);
+  }
 }
 
 } // namespace ConnectionPool

source/common/http/mixed_conn_pool.cc

@@ -86,8 +86,8 @@ void HttpConnPoolImplMixed::onConnected(Envoy::ConnectionPool::ActiveClient& cli
   // it to reflect any difference between the TCP stream limits and HTTP/2
   // stream limits.
   if (new_client->effectiveConcurrentStreamLimit() > old_effective_limit) {
-    state_.incrConnectingAndConnectedStreamCapacity(new_client->effectiveConcurrentStreamLimit() -
-                                                    old_effective_limit);
+    incrConnectingAndConnectedStreamCapacity(
+        new_client->effectiveConcurrentStreamLimit() - old_effective_limit, client);
   }
   new_client->setState(ActiveClient::State::Connecting);
   LinkedList::moveIntoList(std::move(new_client), owningList(new_client->state()));

test/common/http/http2/conn_pool_test.cc

@@ -1805,6 +1805,97 @@ TEST_F(Http2ConnPoolImplTest, TestUnusedCapacity) {
   CHECK_STATE(0 /*active*/, 0 /*pending*/, 0 /*capacity*/);
 }
 
+TEST_F(Http2ConnPoolImplTest, TestNegativeUnusedCapacity) {
+  cluster_->http2_options_.mutable_max_concurrent_streams()->set_value(4);
+
+  expectClientsCreate(2);
+  ActiveTestRequest r1(*this, 0, false);
+  CHECK_STATE(0 /*active*/, 1 /*pending*/, 4 /*capacity*/);
+  expectClientConnect(0, r1);
+  CHECK_STATE(1 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  ActiveTestRequest r2(*this, 0, true);
+  ActiveTestRequest r3(*this, 0, true);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, 1 /*capacity*/);
+
+  // Settings frame results in negative unused capacity.
+  NiceMock<MockReceivedSettings> settings;
+  settings.max_concurrent_streams_ = 1;
+  test_clients_[0].codec_client_->onSettings(settings);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, -2 /*capacity*/);
+
+  completeRequest(r1);
+  CHECK_STATE(2 /*active*/, 0 /*pending*/, -1 /*capacity*/);
+
+  // With negative capacity, verify that a new request creates a new connection.
+  ActiveTestRequest r4(*this, 1, false);
+  CHECK_STATE(2 /*active*/, 1 /*pending*/, 3 /*capacity*/);
+  expectClientConnect(1, r4);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, 2 /*capacity*/);
+
+  completeRequest(r2);
+  CHECK_STATE(2 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  completeRequest(r3);
+  CHECK_STATE(1 /*active*/, 0 /*pending*/, 4 /*capacity*/);
+
+  completeRequest(r4);
+  CHECK_STATE(0 /*active*/, 0 /*pending*/, 5 /*capacity*/);
+
+  // Clean up with an outstanding stream.
+  pool_->drainConnections(Envoy::ConnectionPool::DrainBehavior::DrainExistingConnections);
+  closeAllClients();
+  CHECK_STATE(0 /*active*/, 0 /*pending*/, 0 /*capacity*/);
+}
+
+TEST_F(Http2ConnPoolImplTest, TestDrainingNegativeUnusedCapacity) {
+  cluster_->http2_options_.mutable_max_concurrent_streams()->set_value(4);
+
+  expectClientsCreate(2);
+  ActiveTestRequest r1(*this, 0, false);
+  CHECK_STATE(0 /*active*/, 1 /*pending*/, 4 /*capacity*/);
+  expectClientConnect(0, r1);
+  CHECK_STATE(1 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  ActiveTestRequest r2(*this, 0, true);
+  ActiveTestRequest r3(*this, 0, true);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, 1 /*capacity*/);
+
+  // Settings frame results in negative unused capacity.
+  NiceMock<MockReceivedSettings> settings;
+  settings.max_concurrent_streams_ = 2;
+  test_clients_[0].codec_client_->onSettings(settings);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, -1 /*capacity*/);
+
+  // Clean up with an outstanding stream.
+  pool_->drainConnections(Envoy::ConnectionPool::DrainBehavior::DrainExistingConnections);
+
+  completeRequest(r1);
+  CHECK_STATE(2 /*active*/, 0 /*pending*/, 0 /*capacity*/);
+
+  // With negative capacity, verify that a new request creates a new connection.
+  ActiveTestRequest r4(*this, 1, false);
+  CHECK_STATE(2 /*active*/, 1 /*pending*/, 4 /*capacity*/);
+  expectClientConnect(1, r4);
+  CHECK_STATE(3 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  // Completing this request does NOT add more capacity because the connection is
+  // draining, and that connection's capacity is no longer negative.
+  completeRequest(r2);
+  CHECK_STATE(2 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  // After r3 completes, the draining connection should close.
+  EXPECT_CALL(*this, onClientDestroy());
+  completeRequest(r3);
+  dispatcher_.clearDeferredDeleteList();
+  CHECK_STATE(1 /*active*/, 0 /*pending*/, 3 /*capacity*/);
+
+  completeRequest(r4);
+  CHECK_STATE(0 /*active*/, 0 /*pending*/, 4 /*capacity*/);
+
+  closeClient(1);
+}
+
 TEST_F(Http2ConnPoolImplTest, TestStateWithMultiplexing) {
   cluster_->http2_options_.mutable_max_concurrent_streams()->set_value(2);
   cluster_->max_requests_per_connection_ = 4;

test/common/http/mixed_conn_pool_test.cc

@@ -114,6 +114,18 @@ TEST_F(MixedConnPoolImplTest, HandshakeWithCachedLimit) {
   testAlpnHandshake({});
 }
 
+// Test that increasing the limit upon connect, versus what was in the cache before connection,
+// works correctly.
+TEST_F(MixedConnPoolImplTest, HandshakeWithCachedLimitAndEffectiveIncrease) {
+  expected_capacity_ = 1;
+
+  // This simulates a previous connection being http 1.
+  EXPECT_CALL(mock_cache_, getConcurrentStreams(_)).WillOnce(Return(1));
+
+  // This makes the new connection http 2, which has more than 1 stream available.
+  testAlpnHandshake(Protocol::Http2);
+}
+
 TEST_F(MixedConnPoolImplTest, HandshakeWithCachedLimitCapped) {
   EXPECT_CALL(mock_cache_, getConcurrentStreams(_))
       .WillOnce(Return(std::numeric_limits<uint32_t>::max()));