Merge 8b2b0afe7c into 7c107c2691

cmd/geth/main.go

@@ -188,6 +188,7 @@ var (
 		utils.AllowUnprotectedTxs,
 		utils.BatchRequestLimit,
 		utils.BatchResponseMaxSize,
+		utils.ShutdownDelay,
 	}
 
 	metricsFlags = []cli.Flag{

cmd/utils/flags.go

@@ -791,6 +791,12 @@ var (
 		Value:    node.DefaultConfig.BatchResponseMaxSize,
 		Category: flags.APICategory,
 	}
+	ShutdownDelay = &cli.DurationFlag{
+		Name:     "shutdown-delay",
+		Usage:    "Sets a delay between receiving SIGTERM and shutting down the node",
+		Value:    node.DefaultConfig.ShutdownDelay,
+		Category: flags.APICategory,
+	}
 
 	// Network Settings
 	MaxPeersFlag = &cli.IntFlag{
@@ -1235,6 +1241,10 @@ func setHTTP(ctx *cli.Context, cfg *node.Config) {
 	if ctx.IsSet(BatchResponseMaxSize.Name) {
 		cfg.BatchResponseMaxSize = ctx.Int(BatchResponseMaxSize.Name)
 	}
+
+	if ctx.IsSet(ShutdownDelay.Name) {
+		cfg.ShutdownDelay = ctx.Duration(ShutdownDelay.Name)
+	}
 }
 
 // setGraphQL creates the GraphQL listener interface string from the set

node/config.go

@@ -24,6 +24,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -211,6 +212,9 @@ type Config struct {
 	EnablePersonal bool `toml:"-"`
 
 	DBEngine string `toml:",omitempty"`
+
+	// ShutdownDelay is the delay between receiving SIGTERM and shutting down the node.
+	ShutdownDelay time.Duration `toml:",omitempty"`
 }
 
 // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into

node/defaults.go

@@ -21,6 +21,7 @@ import (
 	"os/user"
 	"path/filepath"
 	"runtime"
+	"time"
 
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/nat"
@@ -73,6 +74,7 @@ var DefaultConfig = Config{
 		NAT:        nat.Any(),
 	},
 	DBEngine: "", // Use whatever exists, will default to Pebble if non-existent and supported
+	ShutdownDelay: 0 * time.Second,
 }
 
 // DefaultDataDir is the default data directory to use for the databases and other

node/node.go

@@ -150,10 +150,10 @@ func New(conf *Config) (*Node, error) {
 	}
 
 	// Configure RPC servers.
-	node.http = newHTTPServer(node.log, conf.HTTPTimeouts)
-	node.httpAuth = newHTTPServer(node.log, conf.HTTPTimeouts)
-	node.ws = newHTTPServer(node.log, rpc.DefaultHTTPTimeouts)
-	node.wsAuth = newHTTPServer(node.log, rpc.DefaultHTTPTimeouts)
+	node.http = newHTTPServer(node.log, conf.HTTPTimeouts, conf.ShutdownDelay)
+	node.httpAuth = newHTTPServer(node.log, conf.HTTPTimeouts, conf.ShutdownDelay)
+	node.ws = newHTTPServer(node.log, rpc.DefaultHTTPTimeouts, conf.ShutdownDelay)
+	node.wsAuth = newHTTPServer(node.log, rpc.DefaultHTTPTimeouts, conf.ShutdownDelay)
 	node.ipc = newIPCServer(node.log, conf.IPCEndpoint())
 
 	return node, nil
@@ -519,6 +519,12 @@ func (n *Node) stopRPC() {
 	n.ws.stop()
 	n.httpAuth.stop()
 	n.wsAuth.stop()
+
+	n.http.wait()
+	n.ws.wait()
+	n.httpAuth.wait()
+	n.wsAuth.wait()
+
 	n.ipc.stop()
 	n.stopInProc()
 }

node/rpcstack.go

@@ -74,9 +74,11 @@ type httpServer struct {
 	mu            sync.Mutex
 	server        *http.Server
 	listener      net.Listener // non-nil when server is running
+	ready         bool
+	shutdownWG    sync.WaitGroup // WG to wait for shutdown
+	shutdownDelay time.Duration
 
 	// HTTP RPC handler things.
-
 	httpConfig  httpConfig
 	httpHandler atomic.Pointer[rpcHandler]
 
@@ -96,8 +98,8 @@ const (
 	shutdownTimeout = 5 * time.Second
 )
 
-func newHTTPServer(log log.Logger, timeouts rpc.HTTPTimeouts) *httpServer {
-	h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string)}
+func newHTTPServer(log log.Logger, timeouts rpc.HTTPTimeouts, shutdownDelay time.Duration) *httpServer {
+	h := &httpServer{log: log, timeouts: timeouts, handlerNames: make(map[string]string), shutdownDelay: shutdownDelay}
 	return h
 }
 
@@ -165,6 +167,7 @@ func (h *httpServer) start() error {
 		}
 		h.log.Info("WebSocket enabled", "url", url)
 	}
+	h.ready = true
 	// if server is websocket only, return after logging
 	if !h.rpcAllowed() {
 		return nil
@@ -195,6 +198,22 @@ func (h *httpServer) start() error {
 }
 
 func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// server health probe endpoints
+	if r.Method == http.MethodGet {
+		// readiness probe fails during shutdown
+		if r.URL.Path == "/readyz" {
+			if h.ready {
+				w.WriteHeader(http.StatusNoContent)
+			} else {
+				w.WriteHeader(http.StatusServiceUnavailable)
+			}
+			return
+			// liveness probe always succeeds
+		} else if r.URL.Path == "/livez" {
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+	}
 	// check if ws request and serve if ws enabled
 	ws := h.wsHandler.Load()
 	if ws != nil && isWebsocket(r) {
@@ -255,8 +274,20 @@ func validatePrefix(what, path string) error {
 // stop shuts down the HTTP server.
 func (h *httpServer) stop() {
 	h.mu.Lock()
+	// unit test executes stop multiple times, so we cannot increment the WG in the start method
+	h.shutdownWG = sync.WaitGroup{}
+	h.shutdownWG.Add(1)
+	h.ready = false
+	time.AfterFunc(h.shutdownDelay, func() {
 		defer h.mu.Unlock()
 		h.doStop()
+		h.shutdownWG.Done()
+	})
+}
+
+// wait waits for the server to shutdown.
+func (h *httpServer) wait() {
+	h.shutdownWG.Wait()
 }
 
 func (h *httpServer) doStop() {

node/rpcstack_test.go

@@ -241,7 +241,7 @@ func createAndStartServer(t *testing.T, conf *httpConfig, ws bool, wsConf *wsCon
 	if timeouts == nil {
 		timeouts = &rpc.DefaultHTTPTimeouts
 	}
-	srv := newHTTPServer(testlog.Logger(t, log.LvlDebug), *timeouts)
+	srv := newHTTPServer(testlog.Logger(t, log.LvlDebug), *timeouts, 0*time.Second)
 	assert.NoError(t, srv.enableRPC(apis(), *conf))
 	if ws {
 		assert.NoError(t, srv.enableWS(nil, *wsConf))