Merge pull request #1848 from versity/sis/fiber-rate-limiter

feat: adds fiber max connections and in-flight requests limiter

cmd/versitygw/gateway_test.go

@@ -32,6 +32,8 @@ func initEnv(dir string) {
 	rootUserSecret = "pass"
 	iamDir = dir
 	port = "127.0.0.1:7070"
+	maxConnections = 250000
+	maxRequests = 100000
 
 	// client
 	awsID = "user"

cmd/versitygw/main.go

@@ -43,6 +43,8 @@ var (
 	rootUserAccess                         string
 	rootUserSecret                         string
 	region                                 string
+	maxConnections, maxRequests            int
+	adminMaxConnections, adminMaxRequests  int
 	corsAllowOrigin                        string
 	admCertFile, admKeyFile                string
 	certFile, keyFile                      string
@@ -217,6 +219,22 @@ func initFlags() []cli.Flag {
 			Destination: &region,
 			Aliases:     []string{"r"},
 		},
+		&cli.IntFlag{
+			Name:        "max-connections",
+			Usage:       "maximum number of concurrent connections s3 api server may serve",
+			EnvVars:     []string{"VGW_MAX_CONNECTIONS"},
+			Value:       250000,
+			Destination: &maxConnections,
+			Aliases:     []string{"mc"},
+		},
+		&cli.IntFlag{
+			Name:        "max-requests",
+			Usage:       "maximum number of in-flight requests s3 api server may serve",
+			EnvVars:     []string{"VGW_MAX_REQUESTS"},
+			Value:       100000,
+			Destination: &maxRequests,
+			Aliases:     []string{"mr"},
+		},
 		&cli.StringFlag{
 			Name:        "cors-allow-origin",
 			Usage:       "default CORS Access-Control-Allow-Origin value (applied when no bucket CORS configuration exists, and for admin APIs)",
@@ -242,6 +260,22 @@ func initFlags() []cli.Flag {
 			Destination: &admPort,
 			Aliases:     []string{"ap"},
 		},
+		&cli.IntFlag{
+			Name:        "admin-max-connections",
+			Usage:       "maximum number of concurrent connections s3 admin server may handle",
+			EnvVars:     []string{"VGW_ADMIN_MAX_CONNECTIONS"},
+			Value:       250000,
+			Destination: &adminMaxConnections,
+			Aliases:     []string{"amc"},
+		},
+		&cli.IntFlag{
+			Name:        "admin-max-requests",
+			Usage:       "maximum number of in-flight requests s3 admin server may handle",
+			EnvVars:     []string{"VGW_ADMIN_MAX_REQUESTS"},
+			Value:       100000,
+			Destination: &adminMaxRequests,
+			Aliases:     []string{"amr"},
+		},
 		&cli.StringFlag{
 			Name:        "admin-cert",
 			Usage:       "TLS cert file for admin server",
@@ -673,6 +707,17 @@ func runGateway(ctx context.Context, be backend.Backend) error {
 		return fmt.Errorf("root user access and secret key must be provided")
 	}
 
+	if maxConnections < 1 {
+		log.Fatal("max-connections must be positive")
+	}
+	if maxRequests < 1 {
+		log.Fatal("max-requests must be positive")
+	}
+	if maxRequests > maxConnections {
+		log.Printf("WARNING: max-requests (%d) exceeds max-connections (%d) which could allow for gateway to panic before throttling requests",
+			maxRequests, maxConnections)
+	}
+
 	webuiAddr = strings.TrimSpace(webuiAddr)
 	if webuiAddr != "" && isAllDigits(webuiAddr) {
 		webuiAddr = ":" + webuiAddr
@@ -719,7 +764,9 @@ func runGateway(ctx context.Context, be backend.Backend) error {
 		}()
 	}
 
-	var opts []s3api.Option
+	opts := []s3api.Option{
+		s3api.WithConcurrencyLimiter(maxConnections, maxRequests),
+	}
 	if corsAllowOrigin != "" {
 		opts = append(opts, s3api.WithCORSAllowOrigin(corsAllowOrigin))
 	}
@@ -860,7 +907,20 @@ func runGateway(ctx context.Context, be backend.Backend) error {
 	var admSrv *s3api.S3AdminServer
 
 	if admPort != "" {
-		var opts []s3api.AdminOpt
+		if adminMaxConnections < 1 {
+			log.Fatal("admin-max-connections must be positive")
+		}
+		if adminMaxRequests < 1 {
+			log.Fatal("admin-max-requests must be positive")
+		}
+		if adminMaxRequests > adminMaxConnections {
+			log.Printf("WARNING: admin-max-requests (%d) exceeds admin-max-connections (%d) which could allow for gateway to panic before throttling requests",
+				adminMaxRequests, adminMaxConnections)
+		}
+
+		opts := []s3api.AdminOpt{
+			s3api.WithAdminConcurrencyLimiter(adminMaxConnections, adminMaxRequests),
+		}
 		if corsAllowOrigin != "" {
 			opts = append(opts, s3api.WithAdminCORSAllowOrigin(corsAllowOrigin))
 		}

s3api/admin-server.go

@@ -36,6 +36,8 @@ type S3AdminServer struct {
 	quiet           bool
 	debug           bool
 	corsAllowOrigin string
+	maxConnections  int
+	maxRequests     int
 }
 
 func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, port, region string, iam auth.IAMService, l s3log.AuditLogger, ctrl controllers.S3ApiController, opts ...AdminOpt) *S3AdminServer {
@@ -57,6 +59,7 @@ func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, port, r
 		Network:               fiber.NetworkTCP,
 		DisableStartupMessage: true,
 		ErrorHandler:          globalErrorHandler,
+		Concurrency:           server.maxConnections,
 	})
 
 	server.app = app
@@ -73,6 +76,10 @@ func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, port, r
 			Format: "${time} | adm | ${status} | ${latency} | ${ip} | ${method} | ${path} | ${error} | ${queryParams}\n",
 		}))
 	}
+
+	// initialize total requests cap limiter middleware
+	app.Use(middlewares.RateLimiter(server.maxRequests, nil, l))
+
 	app.Use(controllers.WrapMiddleware(middlewares.DecodeURL, l, nil))
 
 	// initialize the debug logger in debug mode
@@ -107,6 +114,15 @@ func WithAdminCORSAllowOrigin(origin string) AdminOpt {
 	return func(s *S3AdminServer) { s.corsAllowOrigin = origin }
 }
 
+// WithAdminConcurrencyLimiter sets the admin standalone server's maximum
+// connection limit and the hard limit for in-flight requests.
+func WithAdminConcurrencyLimiter(maxConnections, maxRequests int) AdminOpt {
+	return func(s *S3AdminServer) {
+		s.maxConnections = maxConnections
+		s.maxRequests = maxRequests
+	}
+}
+
 func (sa *S3AdminServer) Serve() (err error) {
 	if sa.CertStorage != nil {
 		ln, err := utils.NewTLSListener(sa.app.Config().Network, sa.port, sa.CertStorage.GetCertificate)

s3api/middlewares/rate-limiter.go

@@ -0,0 +1,50 @@
+// Copyright 2026 Versity Software
+// This file is licensed under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package middlewares
+
+import (
+	"github.com/gofiber/fiber/v2"
+	"github.com/versity/versitygw/metrics"
+	"github.com/versity/versitygw/s3err"
+	"github.com/versity/versitygw/s3log"
+	"golang.org/x/sync/semaphore"
+)
+
+// RateLimiter hard-limits the number of in-flight requests.
+// If the limit is reached, an immediate SlowDown error is returned
+func RateLimiter(limit int, mm metrics.Manager, logger s3log.AuditLogger) fiber.Handler {
+	sem := semaphore.NewWeighted(int64(limit))
+
+	return func(ctx *fiber.Ctx) error {
+		if !sem.TryAcquire(1) {
+			// limit reached
+			err := s3err.GetAPIError(s3err.ErrSlowDown)
+
+			if mm != nil {
+				mm.Send(ctx, err, metrics.ActionUndetected, 0, 0)
+			}
+			if logger != nil {
+				logger.Log(ctx, err, ctx.Body(), s3log.LogMeta{
+					Action: metrics.ActionUndetected,
+				})
+			}
+
+			ctx.Status(err.HTTPStatusCode)
+			return ctx.Send(s3err.GetAPIErrorResponse(err, "", "", ""))
+		}
+		defer sem.Release(1)
+		return ctx.Next()
+	}
+}

s3api/server.go

@@ -51,6 +51,8 @@ type S3ApiServer struct {
 	health          string
 	virtualDomain   string
 	corsAllowOrigin string
+	maxConnections  int
+	maxRequests     int
 }
 
 func New(
@@ -82,6 +84,7 @@ func New(
 		Network:               fiber.NetworkTCP,
 		DisableStartupMessage: true,
 		ErrorHandler:          globalErrorHandler,
+		Concurrency:           server.maxConnections,
 	})
 
 	server.app = app
@@ -106,6 +109,9 @@ func New(
 		})
 	}
 
+	// initialize total requests cap limiter middleware
+	app.Use(middlewares.RateLimiter(server.maxRequests, mm, l))
+
 	// initilaze the default value setter middleware
 	app.Use(middlewares.SetDefaultValues(root, region))
 
@@ -171,6 +177,15 @@ func WithCORSAllowOrigin(origin string) Option {
 	return func(s *S3ApiServer) { s.corsAllowOrigin = origin }
 }
 
+// WithConcurrencyLimiter sets the server's maximum connection limit
+// and the hard limit for in-flight requests.
+func WithConcurrencyLimiter(maxConnections, maxRequests int) Option {
+	return func(s *S3ApiServer) {
+		s.maxConnections = maxConnections
+		s.maxRequests = maxRequests
+	}
+}
+
 func (sa *S3ApiServer) Serve() (err error) {
 	if sa.CertStorage != nil {
 		ln, err := utils.NewTLSListener(sa.app.Config().Network, sa.port, sa.CertStorage.GetCertificate)

s3err/s3err.go

@@ -184,6 +184,7 @@ const (
 	ErrInvalidArgument
 	ErrMalformedTrailer
 	ErrInvalidChunkSize
+	ErrSlowDown
 
 	// Non-AWS errors
 	ErrExistingObjectIsDirectory
@@ -829,6 +830,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "Only the last chunk is allowed to have a size less than 8192 bytes",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrSlowDown: {
+		Code:           "SlowDown",
+		Description:    "Please reduce your request rate.",
+		HTTPStatusCode: http.StatusServiceUnavailable,
+	},
 
 	// non aws errors
 	ErrExistingObjectIsDirectory: {