fix(s3): drain in-flight requests on SIGTERM in standalone weed s3 (#11603)

Standalone weed s3 only shut its servers down when shutdownCtx was set,
which only weed mini does. On SIGTERM the interrupt hooks ran and the
process exited with requests still in flight, so clients saw connection
resets during rolling restarts.

Register an interrupt hook that drains every S3 HTTP(S) listener
(including the local and unix socket ones, and the Iceberg and Lance
servers) for up to 15s alongside a bounded gRPC GracefulStop, then closes
the S3 API server, reusing the filer's shutdown helper. Serve exits now
join that shutdown so the process does not exit mid-drain, and the
secondary listeners tolerate http.ErrServerClosed instead of exiting
fatally.

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/command/filer.go

@@ -505,21 +505,7 @@ func (fo *FilerOptions) startFiler() {
 	if gracefulTimeout <= 0 {
 		gracefulTimeout = 15 * time.Second
 	}
-	stopGrpcServer := func() {
-		glog.V(0).Infof("Gracefully stopping gRPC server")
-		stopped := make(chan struct{})
-		go func() {
-			grpcS.GracefulStop()
-			close(stopped)
-		}()
-		select {
-		case <-stopped:
-			glog.V(0).Infof("gRPC server stopped gracefully")
-		case <-time.After(gracefulTimeout):
-			glog.V(0).Infof("gRPC server graceful stop timed out after %s, forcing stop", gracefulTimeout)
-			grpcS.Stop()
-		}
-	}
+	stopGrpcServer := func() { gracefulStopGrpc(grpcS, gracefulTimeout) }
 
 	var socketServer *http.Server
 	if runtime.GOOS != "windows" {
@@ -640,26 +626,9 @@ func (fo *FilerOptions) startFiler() {
 // The filer leaves the lock ring first: peers and S3 gateways route keys to it
 // until the ring changes, and would hit refused connections once gRPC stops.
 func newFilerShutdown(leaveLockRing, stopGrpc, shutdownFiler func(), httpServers ...*http.Server) func() {
+	drain := newGracefulShutdown(stopGrpc, shutdownFiler, httpServers...)
 	return sync.OnceFunc(func() {
 		leaveLockRing()
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
-		defer cancel()
-		var drained sync.WaitGroup
-		drained.Add(1)
-		go func() {
-			defer drained.Done()
-			stopGrpc()
-		}()
-		for _, server := range httpServers {
-			drained.Add(1)
-			go func() {
-				defer drained.Done()
-				if err := server.Shutdown(shutdownCtx); err != nil {
-					glog.Warningf("filer HTTP shutdown: %v", err)
-				}
-			}()
-		}
-		drained.Wait()
-		shutdownFiler()
+		drain()
 	})
 }

weed/command/s3.go

@@ -83,10 +83,9 @@ type S3Options struct {
 	readerCacheSizeMB         *int64
 
 	allowUntrustedRemoteEndpoints *bool
-	// shutdownCtx, when non-nil, tells startS3Server/startIcebergServer to
-	// gracefully shut down their HTTP/gRPC servers once the ctx is cancelled.
-	// Used by weed mini to orchestrate an ordered shutdown; nil for standalone
-	// weed s3.
+	// shutdownCtx, when non-nil, tells startS3Server to gracefully shut down
+	// its HTTP/gRPC servers once the ctx is cancelled, in addition to on
+	// interrupt. Used by weed mini to orchestrate an ordered shutdown.
 	shutdownCtx context.Context
 }
 
@@ -395,16 +394,17 @@ func (s3opt *S3Options) startS3Server() bool {
 	if s3ApiServer_err != nil {
 		glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)
 	}
-	defer s3ApiServer.Shutdown()
+
+	var httpServers []*http.Server
 
 	// Start Iceberg REST Catalog server if enabled
 	if *s3opt.portIceberg > 0 {
-		go s3opt.startIcebergServer(s3ApiServer)
+		httpServers = append(httpServers, s3opt.startIcebergServer(s3ApiServer))
 	}
 
 	// Start Lance Namespace server if enabled
 	if s3opt.portLance != nil && *s3opt.portLance > 0 {
-		go s3opt.startLanceServer(s3ApiServer)
+		httpServers = append(httpServers, s3opt.startLanceServer(s3ApiServer))
 	}
 
 	if runtime.GOOS != "windows" {
@@ -415,13 +415,14 @@ func (s3opt *S3Options) startS3Server() bool {
 		if err := os.Remove(localSocket); err != nil && !os.IsNotExist(err) {
 			glog.Fatalf("Failed to remove %s, error: %s", localSocket, err.Error())
 		}
+		s3SocketListener, err := net.Listen("unix", localSocket)
+		if err != nil {
+			glog.Fatalf("Failed to listen on %s: %v", localSocket, err)
+		}
+		socketServer := newHttpServer(router, nil)
+		httpServers = append(httpServers, socketServer)
 		go func() {
-			// start on local unix socket
-			s3SocketListener, err := net.Listen("unix", localSocket)
-			if err != nil {
-				glog.Fatalf("Failed to listen on %s: %v", localSocket, err)
-			}
-			if err := newHttpServer(router, nil).Serve(s3SocketListener); err != nil && err != http.ErrServerClosed {
+			if err := socketServer.Serve(s3SocketListener); err != nil && err != http.ErrServerClosed {
 				glog.Fatalf("Failed to start S3 http server: %v", err)
 			}
 		}()
@@ -456,6 +457,7 @@ func (s3opt *S3Options) startS3Server() bool {
 	}
 	go grpcS.Serve(grpcL)
 	pb.ServeGrpcOnLocalSocket(grpcS, grpcPort)
+	stopGrpcServer := func() { gracefulStopGrpc(grpcS, 15*time.Second) }
 
 	if *s3opt.tlsPrivateKey != "" {
 		// Check for port conflict when both HTTP and HTTPS are enabled on the same port
@@ -496,23 +498,22 @@ func (s3opt *S3Options) startS3Server() bool {
 		if *s3opt.portHttps == 0 {
 			glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", version.Version(), *s3opt.port)
 			if s3ApiLocalListener != nil {
+				localServer := newHttpServer(router, tlsConfig)
+				httpServers = append(httpServers, localServer)
 				go func() {
-					if err = newHttpServer(router, tlsConfig).ServeTLS(s3ApiLocalListener, "", ""); err != nil {
+					if err := localServer.ServeTLS(s3ApiLocalListener, "", ""); err != nil && err != http.ErrServerClosed {
 						glog.Fatalf("S3 API Server Fail to serve: %v", err)
 					}
 				}()
 			}
 			httpS := newHttpServer(router, tlsConfig)
-			if s3opt.shutdownCtx != nil {
-				go func() {
-					<-s3opt.shutdownCtx.Done()
-					httpS.Shutdown(context.Background())
-					grpcS.Stop()
-				}()
-			}
+			httpServers = append(httpServers, httpS)
+			shutdown := s3opt.newShutdown(stopGrpcServer, s3ApiServer, httpServers)
 			if err = httpS.ServeTLS(s3ApiListener, "", ""); err != nil && err != http.ErrServerClosed {
 				glog.Fatalf("S3 API Server Fail to serve: %v", err)
 			}
+			// Serve returns when listeners close, before active requests finish.
+			shutdown()
 		} else {
 			glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", version.Version(), *s3opt.portHttps)
 			s3ApiListenerHttps, s3ApiLocalListenerHttps, err := util.NewIpAndLocalListeners(
@@ -521,14 +522,18 @@ func (s3opt *S3Options) startS3Server() bool {
 				glog.Fatalf("S3 API HTTPS listener on %s:%d error: %v", *s3opt.bindIp, *s3opt.portHttps, err)
 			}
 			if s3ApiLocalListenerHttps != nil {
+				localHttpsServer := newHttpServer(router, tlsConfig)
+				httpServers = append(httpServers, localHttpsServer)
 				go func() {
-					if err = newHttpServer(router, tlsConfig).ServeTLS(s3ApiLocalListenerHttps, "", ""); err != nil {
+					if err := localHttpsServer.ServeTLS(s3ApiLocalListenerHttps, "", ""); err != nil && err != http.ErrServerClosed {
 						glog.Fatalf("S3 API Server Fail to serve: %v", err)
 					}
 				}()
 			}
+			httpsServer := newHttpServer(router, tlsConfig)
+			httpServers = append(httpServers, httpsServer)
 			go func() {
-				if err = newHttpServer(router, tlsConfig).ServeTLS(s3ApiListenerHttps, "", ""); err != nil {
+				if err := httpsServer.ServeTLS(s3ApiListenerHttps, "", ""); err != nil && err != http.ErrServerClosed {
 					glog.Fatalf("S3 API Server Fail to serve: %v", err)
 				}
 			}()
@@ -537,31 +542,42 @@ func (s3opt *S3Options) startS3Server() bool {
 	if *s3opt.tlsPrivateKey == "" || *s3opt.portHttps > 0 {
 		glog.V(0).Infof("Start Seaweed S3 API Server %s at http port %d", version.Version(), *s3opt.port)
 		if s3ApiLocalListener != nil {
+			localServer := newHttpServer(router, nil)
+			httpServers = append(httpServers, localServer)
 			go func() {
-				if err = newHttpServer(router, nil).Serve(s3ApiLocalListener); err != nil {
+				if err := localServer.Serve(s3ApiLocalListener); err != nil && err != http.ErrServerClosed {
 					glog.Fatalf("S3 API Server Fail to serve: %v", err)
 				}
 			}()
 		}
 		httpS := newHttpServer(router, nil)
-		if s3opt.shutdownCtx != nil {
-			go func() {
-				<-s3opt.shutdownCtx.Done()
-				httpS.Shutdown(context.Background())
-				grpcS.Stop()
-			}()
-		}
+		httpServers = append(httpServers, httpS)
+		shutdown := s3opt.newShutdown(stopGrpcServer, s3ApiServer, httpServers)
 		if err = httpS.Serve(s3ApiListener); err != nil && err != http.ErrServerClosed {
 			glog.Fatalf("S3 API Server Fail to serve: %v", err)
 		}
+		// Serve returns when listeners close, before active requests finish.
+		shutdown()
 	}
 
 	return true
 
 }
 
+func (s3opt *S3Options) newShutdown(stopGrpc func(), s3ApiServer *s3api.S3ApiServer, httpServers []*http.Server) func() {
+	shutdown := newGracefulShutdown(stopGrpc, s3ApiServer.Shutdown, httpServers...)
+	grace.OnInterrupt(shutdown)
+	if s3opt.shutdownCtx != nil {
+		go func() {
+			<-s3opt.shutdownCtx.Done()
+			shutdown()
+		}()
+	}
+	return shutdown
+}
+
 // startIcebergServer starts the Iceberg REST Catalog server on a separate port.
-func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) {
+func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) *http.Server {
 	icebergRouter := mux.NewRouter().SkipClean(true)
 	// warehouse/parent query values may legally contain ';', which Go's
 	// url.ParseQuery would otherwise drop
@@ -591,12 +607,6 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) {
 	glog.V(0).Infof("Start Iceberg REST Catalog Server at http://%s", listenAddress)
 
 	httpS := newHttpServer(icebergRouter, nil)
-	if s3opt.shutdownCtx != nil {
-		go func() {
-			<-s3opt.shutdownCtx.Done()
-			httpS.Shutdown(context.Background())
-		}()
-	}
 	// Serve on localhost as well if we're bound to a different interface
 	if icebergLocalListener != nil {
 		go func() {
@@ -605,15 +615,18 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) {
 			}
 		}()
 	}
-	if err = httpS.Serve(icebergListener); err != nil && err != http.ErrServerClosed {
-		glog.Fatalf("Iceberg REST Catalog Server Fail to serve: %v", err)
-	}
+	go func() {
+		if err := httpS.Serve(icebergListener); err != nil && err != http.ErrServerClosed {
+			glog.Fatalf("Iceberg REST Catalog Server Fail to serve: %v", err)
+		}
+	}()
+	return httpS
 }
 
 // startLanceServer starts the Lance Namespace server on a separate port. It
 // shares the Iceberg catalog's credential role: one deployment vends table
 // credentials one way, whichever catalog the client speaks to.
-func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) {
+func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) *http.Server {
 	lanceRouter := mux.NewRouter().SkipClean(true)
 	lanceRouter.Use(util_http.EscapeSemicolonsInQuery)
 
@@ -636,12 +649,6 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) {
 	glog.V(0).Infof("Start Lance Namespace Server at http://%s", listenAddress)
 
 	httpS := newHttpServer(lanceRouter, nil)
-	if s3opt.shutdownCtx != nil {
-		go func() {
-			<-s3opt.shutdownCtx.Done()
-			httpS.Shutdown(context.Background())
-		}()
-	}
 	if lanceLocalListener != nil {
 		go func() {
 			if err := httpS.Serve(lanceLocalListener); err != nil && err != http.ErrServerClosed {
@@ -649,9 +656,12 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer) {
 			}
 		}()
 	}
-	if err = httpS.Serve(lanceListener); err != nil && err != http.ErrServerClosed {
-		glog.Fatalf("Lance Namespace Server Fail to serve: %v", err)
-	}
+	go func() {
+		if err := httpS.Serve(lanceListener); err != nil && err != http.ErrServerClosed {
+			glog.Fatalf("Lance Namespace Server Fail to serve: %v", err)
+		}
+	}()
+	return httpS
 }
 
 // deriveLanceStorageEndpoint picks the endpoint the Lance namespace puts in

weed/command/s3_shutdown_test.go

@@ -0,0 +1,81 @@
+package command
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api"
+)
+
+func TestS3ShutdownDrainsActiveRequestBeforeServeReturns(t *testing.T) {
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	finish := sync.OnceFunc(func() { close(release) })
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		close(entered)
+		<-release
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	t.Cleanup(server.Close)
+	t.Cleanup(finish)
+	httpClosing := make(chan struct{})
+	server.Config.RegisterOnShutdown(func() { close(httpClosing) })
+
+	requestDone := make(chan struct{})
+	go func() {
+		defer close(requestDone)
+		response, err := server.Client().Get(server.URL)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		response.Body.Close()
+		if response.StatusCode != http.StatusNoContent {
+			t.Errorf("active request returned %s", response.Status)
+		}
+	}()
+	select {
+	case <-entered:
+	case <-requestDone:
+		t.Fatal("request ended before reaching the handler")
+	case <-time.After(5 * time.Second):
+		t.Fatal("request did not reach the handler")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	grpcStopped := make(chan struct{})
+	s3opt := &S3Options{shutdownCtx: ctx}
+	shutdown := s3opt.newShutdown(func() { close(grpcStopped) }, &s3api.S3ApiServer{}, []*http.Server{server.Config})
+
+	cancel()
+	select {
+	case <-httpClosing:
+	case <-time.After(5 * time.Second):
+		t.Fatal("cancelling shutdownCtx did not start the HTTP drain")
+	}
+	<-grpcStopped
+
+	// Model the main Serve path joining shutdown once its listener closes.
+	serveReturned := make(chan struct{})
+	go func() { shutdown(); close(serveReturned) }()
+	select {
+	case <-serveReturned:
+		t.Fatal("Serve exit path returned while a request was still active")
+	case <-requestDone:
+		t.Fatal("active request ended before it was released")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	finish()
+	<-requestDone
+	select {
+	case <-serveReturned:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Serve exit path did not return after the request drained")
+	}
+}

weed/command/shutdown.go

@@ -0,0 +1,53 @@
+package command
+
+import (
+	"context"
+	"net/http"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+func gracefulStopGrpc(grpcS *grpc.Server, timeout time.Duration) {
+	glog.V(0).Infof("Gracefully stopping gRPC server")
+	stopped := make(chan struct{})
+	go func() {
+		grpcS.GracefulStop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+		glog.V(0).Infof("gRPC server stopped gracefully")
+	case <-time.After(timeout):
+		glog.V(0).Infof("gRPC server graceful stop timed out after %s, forcing stop", timeout)
+		grpcS.Stop()
+	}
+}
+
+// newGracefulShutdown joins shutdown callers while gRPC and HTTP drain concurrently.
+func newGracefulShutdown(stopGrpc, closeServer func(), httpServers ...*http.Server) func() {
+	return sync.OnceFunc(func() {
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+		defer cancel()
+		var drained sync.WaitGroup
+		drained.Add(1)
+		go func() {
+			defer drained.Done()
+			stopGrpc()
+		}()
+		for _, server := range httpServers {
+			drained.Add(1)
+			go func() {
+				defer drained.Done()
+				if err := server.Shutdown(shutdownCtx); err != nil {
+					glog.Warningf("HTTP shutdown: %v", err)
+				}
+			}()
+		}
+		drained.Wait()
+		closeServer()
+	})
+}

weed/command/shutdown_test.go

@@ -0,0 +1,96 @@
+package command
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
+)
+
+type blockingIamCache struct {
+	s3_pb.UnimplementedSeaweedS3IamCacheServer
+	entered chan struct{}
+	release chan struct{}
+}
+
+func (b *blockingIamCache) PutIdentity(ctx context.Context, _ *iam_pb.PutIdentityRequest) (*iam_pb.PutIdentityResponse, error) {
+	close(b.entered)
+	// Stop only cancels handler contexts; it cannot interrupt a handler that ignores them.
+	select {
+	case <-b.release:
+		return &iam_pb.PutIdentityResponse{}, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func startBlockingGrpc(t *testing.T) (grpcS *grpc.Server, release func(), rpcErr chan error) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	handler := &blockingIamCache{entered: make(chan struct{}), release: make(chan struct{})}
+	grpcS = grpc.NewServer()
+	s3_pb.RegisterSeaweedS3IamCacheServer(grpcS, handler)
+	go grpcS.Serve(listener)
+	t.Cleanup(grpcS.Stop)
+	release = sync.OnceFunc(func() { close(handler.release) })
+	t.Cleanup(release)
+
+	conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	rpcErr = make(chan error, 1)
+	go func() {
+		_, err := s3_pb.NewSeaweedS3IamCacheClient(conn).PutIdentity(context.Background(), &iam_pb.PutIdentityRequest{})
+		rpcErr <- err
+	}()
+	<-handler.entered
+	return grpcS, release, rpcErr
+}
+
+func TestGracefulStopGrpcWaitsForInFlightRPC(t *testing.T) {
+	grpcS, release, rpcErr := startBlockingGrpc(t)
+
+	stopped := make(chan struct{})
+	go func() { gracefulStopGrpc(grpcS, 5*time.Second); close(stopped) }()
+	select {
+	case <-stopped:
+		t.Fatal("gRPC stopped while an RPC was in flight")
+	case <-time.After(100 * time.Millisecond):
+	}
+	release()
+	if err := <-rpcErr; err != nil {
+		t.Errorf("in-flight RPC failed: %v", err)
+	}
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("gRPC graceful stop did not return after the RPC completed")
+	}
+}
+
+func TestGracefulStopGrpcForcesStopAfterTimeout(t *testing.T) {
+	grpcS, _, rpcErr := startBlockingGrpc(t)
+
+	stopped := make(chan struct{})
+	go func() { gracefulStopGrpc(grpcS, 50*time.Millisecond); close(stopped) }()
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("gRPC graceful stop did not honour its timeout")
+	}
+	if err := <-rpcErr; err == nil {
+		t.Error("in-flight RPC succeeded although the server was force-stopped")
+	}
+}