diff --git a/node/node.go b/node/node.go index ec7382e725..9f8778f62e 100644 --- a/node/node.go +++ b/node/node.go @@ -28,6 +28,7 @@ import ( "slices" "strings" "sync" + "time" "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" @@ -519,6 +520,7 @@ func (n *Node) stopRPC() { n.ws.stop() n.httpAuth.stop() n.wsAuth.stop() + time.Sleep(rpc.StopPendingRequestTimeout) n.ipc.stop() n.stopInProc() } diff --git a/rpc/http.go b/rpc/http.go index f4b99429ef..91cd9deead 100644 --- a/rpc/http.go +++ b/rpc/http.go @@ -309,7 +309,11 @@ func (t *httpServerConn) SetWriteDeadline(time.Time) error { return nil } func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Permit dumb empty requests for remote health-checks (AWS) if r.Method == http.MethodGet && r.ContentLength == 0 && r.URL.RawQuery == "" { - w.WriteHeader(http.StatusOK) + if s.ready { + w.WriteHeader(http.StatusOK) + } else { + w.WriteHeader(http.StatusServiceUnavailable) + } return } if code, err := s.validateRequest(r); err != nil { diff --git a/rpc/server.go b/rpc/server.go index 42b59f8f6f..2be3f54ee4 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -23,6 +23,7 @@ import ( "net" "sync" "sync/atomic" + "time" "github.com/ethereum/go-ethereum/log" ) @@ -36,6 +37,10 @@ const EngineApi = "engine" type CodecOption int const ( + // give pending requests StopPendingRequestTimeout the time to finish when the server is stopped + // if readiness probe period is 5 seconds, this is enough time for health check to be triggered + StopPendingRequestTimeout = 6 * time.Second + // OptionMethodInvocation is an indication that the codec supports RPC method calls OptionMethodInvocation CodecOption = 1 << iota @@ -51,6 +56,7 @@ type Server struct { mutex sync.Mutex codecs map[ServerCodec]struct{} run atomic.Bool + ready bool batchItemLimit int batchResponseLimit int httpBodyLimit int @@ -68,6 +74,7 @@ func NewServer() *Server { // as the services and methods it offers. rpcService := &RPCService{server} server.RegisterName(MetadataApi, rpcService) + server.ready = true return server } @@ -184,15 +191,19 @@ func messageForReadError(err error) string { // requests to finish, then closes all codecs which will cancel pending requests and // subscriptions. func (s *Server) Stop() { - s.mutex.Lock() - defer s.mutex.Unlock() + log.Debug("RPC server draining pending requests") + s.ready = false + time.AfterFunc(StopPendingRequestTimeout, func() { + s.mutex.Lock() + defer s.mutex.Unlock() - if s.run.CompareAndSwap(true, false) { - log.Debug("RPC server shutting down") - for codec := range s.codecs { - codec.close() + if s.run.CompareAndSwap(true, false) { + log.Debug("RPC server shutting down") + for codec := range s.codecs { + codec.close() + } } - } + }) } // RPCService gives meta information about the server.