[chttp2] Fix flaky max_concurrent_streams test (grpc#34774)

This test tests the strict http2 requirements on the protocol - but
chttp2 chooses to apply pushback earlier in some cases.

src/core/ext/transport/chttp2/transport/chttp2_transport.cc

@@ -638,6 +638,10 @@ static void read_channel_args(grpc_chttp2_transport* t,
       channel_args.GetInt(GRPC_ARG_HTTP2_PING_ON_RST_STREAM_PERCENT)
           .value_or(1),
       0, 100);
+
+  t->max_concurrent_streams_overload_protection =
+      channel_args.GetBool(GRPC_ARG_MAX_CONCURRENT_STREAMS_OVERLOAD_PROTECTION)
+          .value_or(grpc_core::IsOverloadProtectionEnabled());
 }
 
 static void init_keepalive_pings_if_enabled_locked(

src/core/ext/transport/chttp2/transport/internal.h

@@ -552,6 +552,9 @@ struct grpc_chttp2_transport final
   bool ack_pings = true;
   /// True if the keepalive system wants to see some data incoming
   bool keepalive_incoming_data_wanted = false;
+  /// True if we count stream allocation (instead of HTTP2 concurrency) for
+  /// MAX_CONCURRENT_STREAMS
+  bool max_concurrent_streams_overload_protection = false;
 
   // What percentage of rst_stream frames on the server should cause a ping
   // frame to be generated.
@@ -690,6 +693,12 @@ struct grpc_chttp2_stream {
 
 #define GRPC_ARG_PING_TIMEOUT_MS "grpc.http2.ping_timeout_ms"
 
+// EXPERIMENTAL: provide protection against overloading a server with too many
+// requests: wait for streams to be deallocated before they stop counting
+// against MAX_CONCURRENT_STREAMS
+#define GRPC_ARG_MAX_CONCURRENT_STREAMS_OVERLOAD_PROTECTION \
+  "grpc.http.overload_protection"
+
 /// Transport writing call flow:
 /// grpc_chttp2_initiate_write() is called anywhere that we know bytes need to
 /// go out on the wire.

src/core/ext/transport/chttp2/transport/parsing.cc

@@ -663,7 +663,7 @@ static grpc_error_handle init_header_frame_parser(grpc_chttp2_transport* t,
         return GRPC_ERROR_CREATE("Max stream count exceeded");
       }
     } else if (GPR_UNLIKELY(
-                   grpc_core::IsOverloadProtectionEnabled() &&
+                   t->max_concurrent_streams_overload_protection &&
                    t->streams_allocated.load(std::memory_order_relaxed) >
                        t->max_concurrent_streams_policy.AdvertiseValue())) {
       // We have more streams allocated than we'd like, so apply some pushback

test/core/end2end/tests/max_concurrent_streams.cc

@@ -23,6 +23,7 @@
 #include <grpc/impl/channel_arg_names.h>
 #include <grpc/status.h>
 
+#include "src/core/ext/transport/chttp2/transport/internal.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/gprpp/time.h"
 #include "test/core/end2end/end2end_tests.h"
@@ -58,7 +59,10 @@ void SimpleRequestBody(CoreEnd2endTest& test) {
 
 CORE_END2END_TEST(Http2SingleHopTest, MaxConcurrentStreams) {
   SKIP_IF_MINSTACK();
-  InitServer(ChannelArgs().Set(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1));
+  InitServer(
+      ChannelArgs()
+          .Set(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1)
+          .Set(GRPC_ARG_MAX_CONCURRENT_STREAMS_OVERLOAD_PROTECTION, false));
   InitClient(ChannelArgs());
   // perform a ping-pong to ensure that settings have had a chance to round
   // trip
@@ -152,7 +156,10 @@ CORE_END2END_TEST(Http2SingleHopTest, MaxConcurrentStreams) {
 
 CORE_END2END_TEST(Http2SingleHopTest, MaxConcurrentStreamsTimeoutOnFirst) {
   SKIP_IF_MINSTACK();
-  InitServer(ChannelArgs().Set(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1));
+  InitServer(
+      ChannelArgs()
+          .Set(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1)
+          .Set(GRPC_ARG_MAX_CONCURRENT_STREAMS_OVERLOAD_PROTECTION, false));
   InitClient(ChannelArgs());
   // perform a ping-pong to ensure that settings have had a chance to round
   // trip
@@ -197,7 +204,10 @@ CORE_END2END_TEST(Http2SingleHopTest, MaxConcurrentStreamsTimeoutOnFirst) {
 
 CORE_END2END_TEST(Http2SingleHopTest, MaxConcurrentStreamsTimeoutOnSecond) {
   SKIP_IF_MINSTACK();
-  InitServer(ChannelArgs().Set(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1));
+  InitServer(
+      ChannelArgs()
+          .Set(GRPC_ARG_MAX_CONCURRENT_STREAMS, 1)
+          .Set(GRPC_ARG_MAX_CONCURRENT_STREAMS_OVERLOAD_PROTECTION, false));
   InitClient(ChannelArgs());
   // perform a ping-pong to ensure that settings have had a chance to round
   // trip