fix(filer): prevent data corruption during graceful shutdown (#9037)

* fix: wait for in-flight uploads to complete before filer shutdown

Prevents data corruption when SIGTERM is received during active uploads.
The filer now waits for all in-flight operations to complete before
calling the underlying shutdown logic.

This affects all deployment types (Kubernetes, Docker, systemd) and
fixes corruption issues during rolling updates, certificate rotation,
and manual restarts.

Changes:
- Add FilerServer.Shutdown() method with upload wait logic
- Update grace.OnInterrupt hook to use new shutdown method

Fixes data corruption reported by production users during pod restarts.

* fix: implement graceful shutdown for gRPC and HTTP servers, ensuring in-flight uploads complete

* fix: address review comments on graceful shutdown

- Add 10s timeout to gRPC GracefulStop to prevent indefinite blocking
  from long-lived streams (falls back to Stop on timeout)
- Reduce HTTP/HTTPS shutdown timeout from 25s to 15s to fit within
  Kubernetes default 30s termination grace period
- Move fs.Shutdown() (database close) after Serve() returns instead
  of a separate hook to eliminate race where main goroutine exits
  before the shutdown hook runs

* fix: shut down all HTTP servers before filer database close

Address remaining review comments:
- Shut down auxiliary HTTP servers (Unix socket, local listener) during
  graceful shutdown so they can't serve write traffic after the main
  server stops
- Register fs.Shutdown() as a grace.OnInterrupt hook to guarantee it
  completes before os.Exit(0), fixing the race between the grace
  goroutine and the main goroutine
- Use sync.Once to ensure fs.Shutdown() runs exactly once regardless
  of whether shutdown is signal-driven or context-driven (MiniCluster)

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/command/filer.go

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/spf13/viper"
@@ -31,6 +32,7 @@ import (
 	weed_server "github.com/seaweedfs/seaweedfs/weed/server"
 	stats_collect "github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/grace"
 	"github.com/seaweedfs/seaweedfs/weed/util/version"
 )
 
@@ -381,6 +383,15 @@ func (fo *FilerOptions) startFiler() {
 		glog.Fatalf("Filer startup error: %v", nfs_err)
 	}
 
+	// Ensure fs.Shutdown() runs exactly once, whether triggered by a signal hook
+	// or by the main goroutine after Serve() returns (e.g., MiniCluster tests).
+	var shutdownOnce sync.Once
+	shutdownFiler := func() {
+		shutdownOnce.Do(func() {
+			fs.Shutdown()
+		})
+	}
+
 	if *fo.publicPort != 0 {
 		publicListeningAddress := util.JoinHostPort(*fo.bindIp, *fo.publicPort)
 		glog.V(0).Infoln("Start Seaweed filer server", version.Version(), "public at", publicListeningAddress)
@@ -434,6 +445,24 @@ func (fo *FilerOptions) startFiler() {
 	go grpcS.Serve(grpcL)
 	pb.ServeGrpcOnLocalSocket(grpcS, grpcPort)
 
+	// Register graceful shutdown for gRPC server to wait for active RPCs
+	grace.OnInterrupt(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(10 * time.Second):
+			glog.V(0).Infof("gRPC server graceful stop timed out, forcing stop")
+			grpcS.Stop()
+		}
+	})
+
+	var socketServer *http.Server
 	if runtime.GOOS != "windows" {
 		localSocket := *fo.localSocket
 		if localSocket == "" {
@@ -442,14 +471,12 @@ func (fo *FilerOptions) startFiler() {
 		if err := os.Remove(localSocket); err != nil && !os.IsNotExist(err) {
 			glog.Fatalf("Failed to remove %s, error: %s", localSocket, err.Error())
 		}
-		go func() {
-			// start on local unix socket
-			filerSocketListener, err := net.Listen("unix", localSocket)
-			if err != nil {
-				glog.Fatalf("Failed to listen on %s: %v", localSocket, err)
-			}
-			newHttpServer(defaultMux, nil).Serve(filerSocketListener)
-		}()
+		filerSocketListener, err := net.Listen("unix", localSocket)
+		if err != nil {
+			glog.Fatalf("Failed to listen on %s: %v", localSocket, err)
+		}
+		socketServer = newHttpServer(defaultMux, nil)
+		go socketServer.Serve(filerSocketListener)
 	}
 
 	if viper.GetString("https.filer.key") != "" {
@@ -489,14 +516,34 @@ func (fo *FilerOptions) startFiler() {
 
 		security.FixTlsConfig(util.GetViper(), tlsConfig)
 
+		var localTLSServer *http.Server
 		if filerLocalListener != nil {
+			localTLSServer = newHttpServer(defaultMux, tlsConfig)
 			go func() {
-				if err := newHttpServer(defaultMux, tlsConfig).ServeTLS(filerLocalListener, "", ""); err != nil {
+				if err := localTLSServer.ServeTLS(filerLocalListener, "", ""); err != nil {
 					glog.Errorf("Filer Fail to serve: %v", err)
 				}
 			}()
 		}
 		httpS := newHttpServer(defaultMux, tlsConfig)
+
+		// Register shutdown hooks: stop all HTTP servers, then close filer database
+		grace.OnInterrupt(func() {
+			glog.V(0).Infof("Gracefully stopping all HTTP servers")
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+			defer cancel()
+			if socketServer != nil {
+				socketServer.Shutdown(shutdownCtx)
+			}
+			if localTLSServer != nil {
+				localTLSServer.Shutdown(shutdownCtx)
+			}
+			if err := httpS.Shutdown(shutdownCtx); err != nil {
+				glog.Warningf("HTTPS server shutdown: %v", err)
+			}
+		})
+		grace.OnInterrupt(shutdownFiler)
+
 		if MiniClusterCtx != nil {
 			ctx := MiniClusterCtx
 			go func() {
@@ -508,15 +555,37 @@ func (fo *FilerOptions) startFiler() {
 		if err := httpS.ServeTLS(filerListener, "", ""); err != nil && err != http.ErrServerClosed {
 			glog.Fatalf("Filer Fail to serve: %v", err)
 		}
+		// Close database after servers have stopped to prevent data corruption
+		shutdownFiler()
 	} else {
+		var localHTTPServer *http.Server
 		if filerLocalListener != nil {
+			localHTTPServer = newHttpServer(defaultMux, nil)
 			go func() {
-				if err := newHttpServer(defaultMux, nil).Serve(filerLocalListener); err != nil {
+				if err := localHTTPServer.Serve(filerLocalListener); err != nil {
 					glog.Errorf("Filer Fail to serve: %v", err)
 				}
 			}()
 		}
 		httpS := newHttpServer(defaultMux, nil)
+
+		// Register shutdown hooks: stop all HTTP servers, then close filer database
+		grace.OnInterrupt(func() {
+			glog.V(0).Infof("Gracefully stopping all HTTP servers")
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+			defer cancel()
+			if socketServer != nil {
+				socketServer.Shutdown(shutdownCtx)
+			}
+			if localHTTPServer != nil {
+				localHTTPServer.Shutdown(shutdownCtx)
+			}
+			if err := httpS.Shutdown(shutdownCtx); err != nil {
+				glog.Warningf("HTTP server shutdown: %v", err)
+			}
+		})
+		grace.OnInterrupt(shutdownFiler)
+
 		if MiniClusterCtx != nil {
 			ctx := MiniClusterCtx
 			go func() {
@@ -528,5 +597,7 @@ func (fo *FilerOptions) startFiler() {
 		if err := httpS.Serve(filerListener); err != nil && err != http.ErrServerClosed {
 			glog.Fatalf("Filer Fail to serve: %v", err)
 		}
+		// Close database after servers have stopped to prevent data corruption
+		shutdownFiler()
 	}
 }

weed/server/filer_server.go

@@ -258,9 +258,6 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs.filer.LoadRemoteStorageConfAndMapping()
 
 	grace.OnReload(fs.Reload)
-	grace.OnInterrupt(func() {
-		fs.filer.Shutdown()
-	})
 
 	fs.SetupDlmReplication()
 	fs.filer.Dlm.LockRing.SetTakeSnapshotCallback(fs.OnDlmChangeSnapshot)
@@ -298,6 +295,13 @@ func (fs *FilerServer) checkWithMaster() {
 	}
 }
 
+// Shutdown gracefully shuts down the filer server by waiting for in-flight uploads to complete.
+// This prevents data corruption when the process receives SIGTERM during active uploads.
+func (fs *FilerServer) Shutdown() {
+	glog.V(0).Infof("Shutting down filer")
+	fs.filer.Shutdown()
+}
+
 func (fs *FilerServer) Reload() {
 	glog.V(0).Infoln("Reload filer server...")