quic: fixing the disconnect between quiche stream count and Envoy

envoy/http/codec.h

@@ -394,6 +394,13 @@ class ConnectionCallbacks {
    * @param ReceivedSettings the settings received from the peer.
    */
   virtual void onSettings(ReceivedSettings& settings) { UNREFERENCED_PARAMETER(settings); }
+
+  /**
+   * Fires when the MAX_STREAMS frame is received from the peer.
+   * This may occur multiple times across the lifetime of an HTTP/3 connection.
+   * @param num_streams the number of streams now allowed to be opened.
+   */
+  virtual void onMaxStreamsChanged(uint32_t num_streams) { UNREFERENCED_PARAMETER(num_streams); }
 };
 
 /**

source/common/conn_pool/conn_pool_base.cc

@@ -175,18 +175,23 @@ void ConnPoolImplBase::attachStreamToClient(Envoy::ConnectionPool::ActiveClient&
   }
   ENVOY_CONN_LOG(debug, "creating stream", client);
 
+  // Latch capacity before updating remaining streams.
+  uint64_t capacity = client.currentUnusedCapacity();
   client.remaining_streams_--;
   if (client.remaining_streams_ == 0) {
     ENVOY_CONN_LOG(debug, "maximum streams per connection, DRAINING", client);
     host_->cluster().stats().upstream_cx_max_requests_.inc();
     transitionActiveClientState(client, Envoy::ConnectionPool::ActiveClient::State::DRAINING);
-  } else if (client.numActiveStreams() + 1 >= client.concurrent_stream_limit_) {
+  } else if (capacity <= 1) {
     // As soon as the new stream is created, the client will be maxed out.
     transitionActiveClientState(client, Envoy::ConnectionPool::ActiveClient::State::BUSY);
   }
 
   // Decrement the capacity, as there's one less stream available for serving.
-  state_.decrConnectingAndConnectedStreamCapacity(1);
+  // For HTTP/3, the capacity is updated in newStreamEncoder.
+  if (trackStreamCapacity()) {
+    state_.decrConnectingAndConnectedStreamCapacity(1);
+  }
   // Track the new active stream.
   state_.incrActiveStreams(1);
   num_active_streams_++;
@@ -213,14 +218,17 @@ void ConnPoolImplBase::onStreamClosed(Envoy::ConnectionPool::ActiveClient& clien
   // If the effective client capacity was limited by concurrency, increase connecting capacity.
   // If the effective client capacity was limited by max total streams, this will not result in an
   // increment as no capacity is freed up.
-  if (client.remaining_streams_ > client.concurrent_stream_limit_ - client.numActiveStreams() - 1 ||
-      had_negative_capacity) {
+  // 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() && (client.remaining_streams_ > client.concurrent_stream_limit_ -
+                                                                client.numActiveStreams() - 1 ||
+                                had_negative_capacity)) {
     state_.incrConnectingAndConnectedStreamCapacity(1);
   }
   if (client.state() == ActiveClient::State::DRAINING && client.numActiveStreams() == 0) {
     // Close out the draining client if we no longer have active streams.
     client.close();
-  } else if (client.state() == ActiveClient::State::BUSY) {
+  } else if (client.state() == ActiveClient::State::BUSY && client.currentUnusedCapacity() != 0) {
     transitionActiveClientState(client, ActiveClient::State::READY);
     if (!delay_attaching_stream) {
       onUpstreamReady();

source/common/conn_pool/conn_pool_base.h

@@ -56,7 +56,7 @@ class ActiveClient : public LinkedObject<ActiveClient>,
   // Returns the application protocol, or absl::nullopt for TCP.
   virtual absl::optional<Http::Protocol> protocol() const PURE;
 
-  int64_t currentUnusedCapacity() const {
+  virtual int64_t currentUnusedCapacity() const {
     int64_t remaining_concurrent_streams =
         static_cast<int64_t>(concurrent_stream_limit_) - numActiveStreams();
 
@@ -102,6 +102,10 @@ class ActiveClient : public LinkedObject<ActiveClient>,
   virtual void drain();
 
   ConnPoolImplBase& parent_;
+  // The count of remaining streams allowed for this connection.
+  // This will start out as the total number of streams per connection if capped
+  // by configuration, or it will be set to std::numeric_limits<uint32_t>::max() to be
+  // (functionally) unlimited.
   uint32_t remaining_streams_;
   uint32_t concurrent_stream_limit_;
   Upstream::HostDescriptionConstSharedPtr real_host_description_;
@@ -148,6 +152,10 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   virtual ~ConnPoolImplBase();
 
   void deleteIsPendingImpl();
+  // By default, the connection pool will track connected and connecting stream
+  // capacity as streams are created and destroyed. QUIC does custom stream
+  // accounting so will override this to false.
+  virtual bool trackStreamCapacity() { return true; }
 
   // A helper function to get the specific context type from the base class context.
   template <class T> T& typedContext(AttachContext& context) {
@@ -234,6 +242,9 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   void decrClusterStreamCapacity(uint32_t delta) {
     state_.decrConnectingAndConnectedStreamCapacity(delta);
   }
+  void incrClusterStreamCapacity(uint32_t delta) {
+    state_.incrConnectingAndConnectedStreamCapacity(delta);
+  }
   void dumpState(std::ostream& os, int indent_level = 0) const {
     const char* spaces = spacesForLevel(indent_level);
     os << spaces << "ConnPoolImplBase " << this << DUMP_MEMBER(ready_clients_.size())

source/common/http/codec_client.h

@@ -158,6 +158,11 @@ class CodecClient : protected Logger::Loggable<Logger::Id::client>,
       codec_callbacks_->onSettings(settings);
     }
   }
+  void onMaxStreamsChanged(uint32_t num_streams) override {
+    if (codec_callbacks_) {
+      codec_callbacks_->onMaxStreamsChanged(num_streams);
+    }
+  }
 
   void onIdleTimeout() {
     host_->cluster().stats().upstream_cx_idle_timeout_.inc();

source/common/http/http3/conn_pool.cc

@@ -30,6 +30,13 @@ ActiveClient::ActiveClient(Envoy::Http::HttpConnPoolImplBase& parent,
                                   parent.host()->cluster().stats().upstream_cx_http3_total_, data) {
 }
 
+void ActiveClient::onMaxStreamsChanged(uint32_t num_streams) {
+  updateCapacity(num_streams);
+  if (state() == ActiveClient::State::BUSY && currentUnusedCapacity() != 0) {
+    parent_.transitionActiveClientState(*this, ActiveClient::State::READY);
+  }
+}
+
 void Http3ConnPoolImpl::setQuicConfigFromClusterConfig(const Upstream::ClusterInfo& cluster,
                                                        quic::QuicConfig& quic_config) {
   // TODO(alyssawilk) use and test other defaults.

source/common/http/http3/conn_pool.h

@@ -23,6 +23,56 @@ class ActiveClient : public MultiplexedActiveClientBase {
 public:
   ActiveClient(Envoy::Http::HttpConnPoolImplBase& parent,
                Upstream::Host::CreateConnectionData& data);
+
+  // Http::ConnectionCallbacks
+  void onMaxStreamsChanged(uint32_t num_streams) override;
+
+  RequestEncoder& newStreamEncoder(ResponseDecoder& response_decoder) override {
+    ASSERT(quiche_capacity_ != 0);
+    // Each time a quic stream is allocated the quic capacity needs to get
+    // decremented. See comments by quiche_capacity_.
+    updateCapacity(quiche_capacity_ - 1);
+    return MultiplexedActiveClientBase::newStreamEncoder(response_decoder);
+  }
+
+  // Overload the default capacity calculations to return the quic capacity
+  // (modified by any stream limits in Envoy config)
+  int64_t currentUnusedCapacity() const override {
+    return std::min<int64_t>(quiche_capacity_, effectiveConcurrentStreamLimit());
+  }
+
+  void updateCapacity(uint64_t new_quiche_capacity) {
+    // Each time we update the capacity make sure to reflect the update in the
+    // connection pool.
+    //
+    // Due to interplay between the max number of concurrent streams Envoy will
+    // allow and the max number of streams per connection this is not as simple
+    // as just updating based on the delta between quiche_capacity_ and
+    // new_quiche_capacity, so we use the delta between the actual calculated
+    // capacity before and after the update.
+    uint64_t old_capacity = currentUnusedCapacity();
+    quiche_capacity_ = new_quiche_capacity;
+    uint64_t new_capacity = currentUnusedCapacity();
+
+    if (new_capacity < old_capacity) {
+      parent_.decrClusterStreamCapacity(old_capacity - new_capacity);
+    } else if (old_capacity < new_capacity) {
+      parent_.incrClusterStreamCapacity(new_capacity - old_capacity);
+    }
+  }
+
+  // Unlike HTTP/2 and HTTP/1, rather than having a cap on the number of active
+  // streams, QUIC has a fixed number of streams available which is updated via
+  // the MAX_STREAMS frame.
+  //
+  // As such each time we create a new stream for QUIC, the capacity goes down
+  // by one, but unlike the other two codecs it is _not_ restored on stream
+  // closure.
+  //
+  // We track the QUIC capacity here, and overload currentUnusedCapacity so the
+  // connection pool can accurately keep track of when it is safe to create new
+  // streams.
+  uint64_t quiche_capacity_ = 100;
 };
 
 // Http3 subclass of FixedHttpConnPoolImpl which exists to store quic data.
@@ -45,6 +95,9 @@ class Http3ConnPoolImpl : public FixedHttpConnPoolImpl {
                                              quic::QuicConfig& quic_config);
 
   Quic::PersistentQuicInfoImpl& quicInfo() { return *quic_info_; }
+  // For HTTP/3 the base connection pool does not track stream capacity, rather
+  // the HTTP3 active client does.
+  bool trackStreamCapacity() override { return false; }
 
 private:
   // Store quic helpers which can be shared between connections and must live

source/common/quic/envoy_quic_client_session.cc

@@ -4,6 +4,21 @@
 
 #include "quic_filter_manager_connection_impl.h"
 
+namespace quic {
+namespace test {
+
+// TODO(alyssawilk) add the necessary accessors to quiche and remove this.
+class QuicSessionPeer {
+public:
+  static quic::QuicStreamIdManager&
+  getStreamIdManager(Envoy::Quic::EnvoyQuicClientSession* session) {
+    return session->ietf_streamid_manager_.bidirectional_stream_id_manager_;
+  }
+};
+
+} // namespace test
+} // namespace quic
+
 namespace Envoy {
 namespace Quic {
 
@@ -87,6 +102,15 @@ void EnvoyQuicClientSession::SetDefaultEncryptionLevel(quic::EncryptionLevel lev
   }
 }
 
+void EnvoyQuicClientSession::OnCanCreateNewOutgoingStream(bool unidirectional) {
+  if (http_connection_callbacks_ && !unidirectional) {
+    quic::QuicStreamIdManager& manager = quic::test::QuicSessionPeer::getStreamIdManager(this);
+    ASSERT(manager.outgoing_max_streams() >= manager.outgoing_stream_count());
+    uint32_t streams_available = manager.outgoing_max_streams() - manager.outgoing_stream_count();
+    http_connection_callbacks_->onMaxStreamsChanged(streams_available);
+  }
+}
+
 std::unique_ptr<quic::QuicSpdyClientStream> EnvoyQuicClientSession::CreateClientStream() {
   ASSERT(codec_stats_.has_value() && http3_options_.has_value());
   return std::make_unique<EnvoyQuicClientStream>(GetNextOutgoingBidirectionalStreamId(), this,

source/common/quic/envoy_quic_client_session.h

@@ -11,6 +11,8 @@
 namespace Envoy {
 namespace Quic {
 
+class EnvoyQuicClientSession;
+
 // Act as a Network::ClientConnection to ClientCodec.
 // TODO(danzh) This class doesn't need to inherit Network::FilterManager
 // interface but need all other Network::Connection implementation in
@@ -71,6 +73,13 @@ class EnvoyQuicClientSession : public QuicFilterManagerConnectionImpl,
   // QuicFilterManagerConnectionImpl
   void setHttp3Options(const envoy::config::core::v3::Http3ProtocolOptions& http3_options) override;
 
+  bool QuicheShouldCreateOutgoingBidirectionalStream() {
+    return quic::QuicSpdyClientSession::ShouldCreateOutgoingBidirectionalStream();
+  }
+
+  // Notify any registered connection pool when new streams are available.
+  void OnCanCreateNewOutgoingStream(bool) override;
+
   using quic::QuicSpdyClientSession::PerformActionOnActiveStreams;
 
 protected:

source/common/quic/envoy_quic_client_stream.cc

@@ -286,7 +286,9 @@ void EnvoyQuicClientStream::ResetWithError(quic::QuicResetStreamError error) {
   stats_.tx_reset_.inc();
   // Upper layers expect calling resetStream() to immediately raise reset callbacks.
   runResetCallbacks(quicRstErrorToEnvoyLocalResetReason(error.internal_code()));
-  quic::QuicSpdyClientStream::ResetWithError(error);
+  if (session()->connection()->connected()) {
+    quic::QuicSpdyClientStream::ResetWithError(error);
+  }
 }
 
 void EnvoyQuicClientStream::OnConnectionClosed(quic::QuicErrorCode error,

test/integration/multiplexed_upstream_integration_test.cc

@@ -205,7 +205,6 @@ TEST_P(Http2UpstreamIntegrationTest, LargeSimultaneousRequestWithBufferLimits) {
 void Http2UpstreamIntegrationTest::manySimultaneousRequests(uint32_t request_bytes,
                                                             uint32_t max_response_bytes,
                                                             uint32_t num_requests) {
-  autonomous_allow_incomplete_streams_ = true;
   TestRandomGenerator rand;
   std::vector<Http::RequestEncoder*> encoders;
   std::vector<IntegrationStreamDecoderPtr> responses;
@@ -236,11 +235,8 @@ void Http2UpstreamIntegrationTest::manySimultaneousRequests(uint32_t request_byt
     ASSERT_TRUE(responses[i]->waitForEndStream());
     if (i % 2 != 0) {
       EXPECT_TRUE(responses[i]->complete());
-      // TODO(18160) remove this if and always check for 200 and body length.
-      if (num_requests <= 100 || upstreamProtocol() != Http::CodecType::HTTP3) {
-        EXPECT_EQ("200", responses[i]->headers().getStatusValue());
-        EXPECT_EQ(response_bytes[i], responses[i]->body().length());
-      }
+      EXPECT_EQ("200", responses[i]->headers().getStatusValue());
+      EXPECT_EQ(response_bytes[i], responses[i]->body().length());
     } else {
       // Upstream stream reset.
       EXPECT_EQ("503", responses[i]->headers().getStatusValue());
@@ -255,13 +251,9 @@ TEST_P(Http2UpstreamIntegrationTest, ManySimultaneousRequest) {
   manySimultaneousRequests(1024, 1024, 100);
 }
 
-#ifdef NDEBUG
-// TODO(alyssawilk) this causes crashes in debug mode for QUIC due to a race
-// condition between Envoy's stream accounting and QUICE's. Debug and fix.
 TEST_P(Http2UpstreamIntegrationTest, TooManySimultaneousRequests) {
   manySimultaneousRequests(1024, 1024, 200);
 }
-#endif
 
 TEST_P(Http2UpstreamIntegrationTest, ManyLargeSimultaneousRequestWithBufferLimits) {
   config_helper_.setBufferLimits(1024, 1024); // Set buffer limits upstream and downstream.