server: configurable gRPC max concurrent streams

README.md

@@ -40,6 +40,7 @@
     - [xDS Management Server Based Configuration Loading](#xds-management-server-based-configuration-loading)
   - [Log Format](#log-format)
   - [GRPC Keepalive](#grpc-keepalive)
+  - [GRPC Concurrency](#grpc-concurrency)
   - [Health-check](#health-check)
     - [Health-check configurations](#health-check-configurations)
   - [GRPC server](#grpc-server)
@@ -932,6 +933,10 @@ The behavior can be fixed by configuring the following env variables for the rat
 - `GRPC_MAX_CONNECTION_AGE`: a duration for the maximum amount of time a connection may exist before it will be closed by sending a GoAway. A random jitter of +/-10% will be added to MaxConnectionAge to spread out connection storms.
 - `GRPC_MAX_CONNECTION_AGE_GRACE`: an additive period after MaxConnectionAge after which the connection will be forcibly closed.
 
+## GRPC Concurrency
+
+- `GRPC_MAX_CONCURRENT_STREAMS`: maximum number of concurrent calls on each gRPC connection. Calls beyond it wait on the client side (HTTP/2 flow control) instead of piling up in the rate limit service while it is slow, e.g. because Redis is. Size it above the normal number of concurrent calls per connection (calls per second per connection × latency). `0` keeps the gRPC default (unbounded). Default: `0`
+
 ## Health-check
 
 Health check status is determined internally by individual components.
@@ -1360,7 +1365,7 @@ The deployment type can be specified with the `REDIS_TYPE` / `REDIS_PERSECOND_TY
 
 ### Connection Timeout
 
-Controls the maximum duration for Redis connection establishment, read operations, and write operations.
+Controls the maximum duration for Redis connection establishment.
 
 1. `REDIS_TIMEOUT`: sets the timeout for Redis connection and I/O operations. Default: `10s`
 1. `REDIS_PERSECOND_TIMEOUT`: sets the timeout for per-second Redis connection and I/O operations. Default: `10s`

src/server/server_impl.go

@@ -255,6 +255,9 @@ func newServer(s settings.Settings, name string, statsManager stats.Manager, loc
 		grpc.UnaryInterceptor(s.GrpcUnaryInterceptor),
 		grpc.StatsHandler(otelgrpc.NewServerHandler()),
 	}
+	if s.GrpcMaxConcurrentStreams > 0 {
+		grpcOptions = append(grpcOptions, grpc.MaxConcurrentStreams(s.GrpcMaxConcurrentStreams))
+	}
 	if s.GrpcServerUseTLS {
 		grpcServerTlsConfig := s.GrpcServerTlsConfig
 		ret.grpcCertProvider = provider.NewCertProvider(s, ret.store, s.GrpcServerTlsCert, s.GrpcServerTlsKey)

src/settings/settings.go

@@ -34,6 +34,9 @@ type Settings struct {
 	GrpcMaxConnectionAge time.Duration `envconfig:"GRPC_MAX_CONNECTION_AGE" default:"24h" description:"Duration a connection may exist before it will be closed by sending a GoAway."`
 	// GrpcMaxConnectionAgeGrace is an additive period after MaxConnectionAge after which the connection will be forcibly closed.
 	GrpcMaxConnectionAgeGrace time.Duration `envconfig:"GRPC_MAX_CONNECTION_AGE_GRACE" default:"1h" description:"Period after MaxConnectionAge after which the connection will be forcibly closed."`
+	// GrpcMaxConcurrentStreams bounds the number of concurrent calls (streams)
+	// on each gRPC connection. 0 keeps the gRPC default (unbounded).
+	GrpcMaxConcurrentStreams uint32 `envconfig:"GRPC_MAX_CONCURRENT_STREAMS" default:"0"`
 	// GrpcServerUseTLS enables gprc connections to server over TLS
 	GrpcServerUseTLS bool `envconfig:"GRPC_SERVER_USE_TLS" default:"false"`
 	// Allow to set the server certificate and key for TLS connections.

src/settings/settings_test.go

@@ -144,6 +144,13 @@ func TestRedisPoolOnEmptyBehavior_IndependentConfiguration(t *testing.T) {
 	assert.Equal(t, "CREATE", settings.RedisPerSecondPoolOnEmptyBehavior)
 }
 
+func TestGrpcMaxConcurrentStreams(t *testing.T) {
+	assert.Equal(t, uint32(0), NewSettings().GrpcMaxConcurrentStreams)
+
+	t.Setenv("GRPC_MAX_CONCURRENT_STREAMS", "128")
+	assert.Equal(t, uint32(128), NewSettings().GrpcMaxConcurrentStreams)
+}
+
 // Tests for WeekRateLimitResetDay
 func TestWeekRateLimitResetDay_Default(t *testing.T) {
 	os.Unsetenv("WEEK_RATE_LIMIT_RESET_DAY")