feat: add GRPC_MAX_CONCURRENT_STREAMS to cap concurrent gRPC streams

README.md

@@ -923,6 +923,7 @@ 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_MAX_CONCURRENT_STREAMS`: caps the maximum number of concurrent gRPC streams the server allows **per HTTP/2 connection**. Defaults to `0`, which leaves the grpc-go server default (unlimited, `math.MaxUint32`) in place. This is a per-connection limit, so the aggregate number of concurrent streams a pod can process is roughly `(number of active connections) x GRPC_MAX_CONCURRENT_STREAMS`. Setting this gives operators a lever to bound the number of handler goroutines spawned during a downstream stall (e.g. a slow Redis), preventing unbounded goroutine growth and OOMs.
 
 ## Health-check
 

src/server/server_impl.go

@@ -231,6 +231,15 @@ func NewServer(s settings.Settings, name string, statsManager stats.Manager, loc
 	return newServer(s, name, statsManager, localCache, opts...)
 }
 
+// maxConcurrentStreamsOptions returns the ServerOption that caps concurrent gRPC
+// streams per HTTP/2 connection, or no options when the cap is disabled (0).
+func maxConcurrentStreamsOptions(maxStreams uint32) []grpc.ServerOption {
+	if maxStreams > 0 {
+		return []grpc.ServerOption{grpc.MaxConcurrentStreams(maxStreams)}
+	}
+	return nil
+}
+
 func newServer(s settings.Settings, name string, statsManager stats.Manager, localCache *freecache.Cache, opts ...settings.Option) *server {
 	for _, opt := range opts {
 		opt(&s)
@@ -258,6 +267,7 @@ func newServer(s settings.Settings, name string, statsManager stats.Manager, loc
 		),
 		grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()),
 	}
+	grpcOptions = append(grpcOptions, maxConcurrentStreamsOptions(s.GrpcMaxConcurrentStreams)...)
 	if s.GrpcServerUseTLS {
 		grpcServerTlsConfig := s.GrpcServerTlsConfig
 		ret.grpcCertProvider = provider.NewCertProvider(s, ret.store, s.GrpcServerTlsCert, s.GrpcServerTlsKey)

src/server/server_impl_internal_test.go

@@ -0,0 +1,17 @@
+package server
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMaxConcurrentStreamsOptions_Disabled(t *testing.T) {
+	opts := maxConcurrentStreamsOptions(0)
+	assert.Empty(t, opts)
+}
+
+func TestMaxConcurrentStreamsOptions_Enabled(t *testing.T) {
+	opts := maxConcurrentStreamsOptions(100)
+	assert.Len(t, opts, 1)
+}

src/settings/settings.go

@@ -33,6 +33,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 caps the maximum number of concurrent gRPC streams the server allows PER HTTP/2 connection.
+	// 0 disables the cap, leaving the grpc-go default (unlimited) in place.
+	GrpcMaxConcurrentStreams uint32 `envconfig:"GRPC_MAX_CONCURRENT_STREAMS" default:"0" description:"Maximum number of concurrent gRPC streams allowed per HTTP/2 connection. 0 disables the cap (grpc-go default, unlimited)."`
 	// 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

@@ -128,6 +128,24 @@ func TestRedisClusterPipelineParallelism_Auto(t *testing.T) {
 	assert.Equal(t, 0, settings.RedisClusterPipelineParallelism)
 }
 
+// Tests for GrpcMaxConcurrentStreams
+func TestGrpcMaxConcurrentStreams_Default(t *testing.T) {
+	os.Unsetenv("GRPC_MAX_CONCURRENT_STREAMS")
+
+	settings := NewSettings()
+
+	assert.Equal(t, uint32(0), settings.GrpcMaxConcurrentStreams)
+}
+
+func TestGrpcMaxConcurrentStreams_Configured(t *testing.T) {
+	os.Setenv("GRPC_MAX_CONCURRENT_STREAMS", "100")
+	defer os.Unsetenv("GRPC_MAX_CONCURRENT_STREAMS")
+
+	settings := NewSettings()
+
+	assert.Equal(t, uint32(100), settings.GrpcMaxConcurrentStreams)
+}
+
 // Test both pools can be configured independently
 func TestRedisPoolOnEmptyBehavior_IndependentConfiguration(t *testing.T) {
 	os.Setenv("REDIS_POOL_ON_EMPTY_BEHAVIOR", "ERROR")