Enable max connection age for API server.

Author: cody-littley

disperser/apiserver/server_v2.go

@@ -8,8 +8,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
-
 	"github.com/Layr-Labs/eigenda/api"
 	pbcommon "github.com/Layr-Labs/eigenda/api/grpc/common"
 	pbv1 "github.com/Layr-Labs/eigenda/api/grpc/disperser"
@@ -23,7 +21,9 @@ import (
 	"github.com/Layr-Labs/eigenda/encoding"
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	gethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/prometheus/client_golang/prometheus"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/reflection"
 )
 
@@ -136,6 +136,12 @@ func (s *DispersalServerV2) Start(ctx context.Context) error {
 	}
 
 	// Serve grpc requests
+	keepAliveConfig := grpc.KeepaliveParams(keepalive.ServerParameters{
+		MaxConnectionIdle:     s.serverConfig.MaxIdleConnectionAge,
+		MaxConnectionAge:      s.serverConfig.MaxConnectionAge,
+		MaxConnectionAgeGrace: s.serverConfig.MaxConnectionAgeGrace,
+	})
+
 	addr := fmt.Sprintf("%s:%s", disperser.Localhost, s.serverConfig.GrpcPort)
 	listener, err := net.Listen("tcp", addr)
 	if err != nil {
@@ -146,7 +152,7 @@ func (s *DispersalServerV2) Start(ctx context.Context) error {
 	gs := grpc.NewServer(
 		grpc.ChainUnaryInterceptor(
 			s.metrics.grpcMetrics.UnaryServerInterceptor(),
-		), opt)
+		), opt, keepAliveConfig)
 	reflection.Register(gs)
 	pb.RegisterDisperserServer(gs, s)
 

disperser/cmd/apiserver/flags/flags.go

@@ -46,11 +46,33 @@ var (
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "GRPC_STREAM_TIMEOUT"),
 		Value:    time.Second * 10,
 	}
+	MaxConnectionAgeFlag = cli.DurationFlag{
+		Name: common.PrefixFlag(FlagPrefix, "max-connection-age"),
+		Usage: "Maximum age of a gRPC connection before it is closed. " +
+			"If zero, then the server will not close connections based on age.",
+		Required: false,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "MAX_CONNECTION_AGE_SECONDS"),
+		Value:    5 * time.Minute,
+	}
+	MaxConnectionAgeGraceFlag = cli.DurationFlag{
+		Name:     common.PrefixFlag(FlagPrefix, "max-connection-age-grace"),
+		Usage:    "Grace period after MaxConnectionAge before the connection is forcibly closed.",
+		Required: false,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "MAX_CONNECTION_AGE_GRACE_SECONDS"),
+		Value:    30 * time.Second,
+	}
+	MaxIdleConnectionAgeFlag = cli.DurationFlag{
+		Name:     common.PrefixFlag(FlagPrefix, "max-idle-connection-age"),
+		Usage:    "Maximum time a connection can be idle before it is closed.",
+		Required: false,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "MAX_IDLE_CONNECTION_AGE_SECONDS"),
+		Value:    time.Minute,
+	}
 	BlsOperatorStateRetrieverFlag = cli.StringFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "bls-operator-state-retriever"),
 		Usage:    "[Deprecated: use EigenDADirectory instead] Address of the BLS operator state Retriever",
 		Required: false,
-		EnvVar:   common.PrefixEnvVar(envVarPrefix, "BLS_OPERATOR_STATE_RETRIVER"),
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "BLS_OPERATOR_STATE_RETRIVER"), // sigh
 	}
 	EigenDAServiceManagerFlag = cli.StringFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "eigenda-service-manager"),
@@ -272,6 +294,9 @@ var optionalFlags = []cli.Flag{
 	EnablePaymentMeterer,
 	BucketStoreSize,
 	GrpcTimeoutFlag,
+	MaxConnectionAgeFlag,
+	MaxConnectionAgeGraceFlag,
+	MaxIdleConnectionAgeFlag,
 	MaxBlobSize,
 	ReservationsTableName,
 	OnDemandTableName,

disperser/cmd/apiserver/lib/config.go

@@ -98,10 +98,13 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 		DisperserVersion: DisperserVersion(version),
 		AwsClientConfig:  aws.ReadClientConfig(ctx, flags.FlagPrefix),
 		ServerConfig: disperser.ServerConfig{
-			GrpcPort:      ctx.GlobalString(flags.GrpcPortFlag.Name),
-			GrpcTimeout:   ctx.GlobalDuration(flags.GrpcTimeoutFlag.Name),
-			PprofHttpPort: ctx.GlobalString(flags.PprofHttpPort.Name),
-			EnablePprof:   ctx.GlobalBool(flags.EnablePprof.Name),
+			GrpcPort:              ctx.GlobalString(flags.GrpcPortFlag.Name),
+			GrpcTimeout:           ctx.GlobalDuration(flags.GrpcTimeoutFlag.Name),
+			MaxConnectionAge:      ctx.GlobalDuration(flags.MaxConnectionAgeFlag.Name),
+			MaxConnectionAgeGrace: ctx.GlobalDuration(flags.MaxConnectionAgeGraceFlag.Name),
+			MaxIdleConnectionAge:  ctx.GlobalDuration(flags.MaxIdleConnectionAgeFlag.Name),
+			PprofHttpPort:         ctx.GlobalString(flags.PprofHttpPort.Name),
+			EnablePprof:           ctx.GlobalBool(flags.EnablePprof.Name),
 		},
 		BlobstoreConfig: blobstore.Config{
 			BucketName: ctx.GlobalString(flags.S3BucketNameFlag.Name),

disperser/server_config.go

@@ -7,9 +7,23 @@ const (
 )
 
 type ServerConfig struct {
-	GrpcPort    string
+	GrpcPort string
+
+	// This timeout is used control the maximum age of a DisperseBlobAuthenticated() RPC call
+	// (via a context with a timeout).
 	GrpcTimeout time.Duration
 
+	// The maximum permissible age of a GRPC connection before it is closed. If zero, then the server will not close
+	// connections based on age.
+	MaxConnectionAge time.Duration
+
+	// When the server closes a connection due to MaxConnectionAgeSeconds, it will wait for this grace period before
+	// forcibly closing the connection. This allows in-flight requests to complete.
+	MaxConnectionAgeGrace time.Duration
+
+	// MaxIdleConnectionAge is the maximum time a connection can be idle before it is closed.
+	MaxIdleConnectionAge time.Duration
+
 	PprofHttpPort string
 	EnablePprof   bool
 }