feat(proxyd): consensus-gated /readyz and idle graceful shutdown

proxyd/backend_test.go

@@ -140,6 +140,8 @@ func TestClientDisconnectionFlow499(t *testing.T) {
 		nil,                                    // limExemptKeys
 		TxValidationMiddlewareConfig{},         // txValidationConfig
 		10*time.Second,                         // gracefulShutdownDuration
+		false,                                  // gracefulShutdownIdle
+		10*time.Second,                         // gracefulShutdownIdleDuration
 	)
 	require.NoError(t, err)
 

proxyd/config.go

@@ -37,6 +37,14 @@ type ServerConfig struct {
 	// GracefulShutdownSeconds is the duration to wait during drain before shutting down.
 	// Defaults to 0 (no drain delay). Set to a positive value to enable graceful drain.
 	GracefulShutdownSeconds int `toml:"graceful_shutdown_seconds"`
+
+	// GracefulShutdownIdle, when true, replaces the drain-then-sleep shutdown
+	// behaviour with an idle-based one: on shutdown signal /readyz starts
+	// returning 503 (so k8s removes the pod from the Service), but the RPC
+	// and WS endpoints keep serving. The process exits only after no
+	// non-healthcheck request has been received for GracefulShutdownIdleSeconds.
+	GracefulShutdownIdle        bool `toml:"graceful_shutdown_idle"`
+	GracefulShutdownIdleSeconds int  `toml:"graceful_shutdown_idle_seconds"`
 }
 
 type CacheConfig struct {

proxyd/proxyd.go

@@ -509,6 +509,8 @@ func Start(config *Config) (*Server, func(), error) {
 		apiKeys,
 		config.TxValidationMiddlewareConfig,
 		time.Duration(config.Server.GracefulShutdownSeconds)*time.Second,
+		config.Server.GracefulShutdownIdle,
+		gracefulShutdownIdleDuration(config.Server.GracefulShutdownIdleSeconds),
 	)
 	if err != nil {
 		return nil, nil, fmt.Errorf("error creating server: %w", err)
@@ -689,6 +691,13 @@ func Start(config *Config) (*Server, func(), error) {
 	return srv, shutdownFunc, nil
 }
 
+func gracefulShutdownIdleDuration(seconds int) time.Duration {
+	if seconds <= 0 {
+		return 10 * time.Second
+	}
+	return time.Duration(seconds) * time.Second
+}
+
 func validateReceiptsTarget(val string) (string, error) {
 	if val == "" {
 		val = ReceiptsTargetDebugGetRawReceipts

proxyd/server.go

@@ -92,14 +92,18 @@ type Server struct {
 	publicAccess            bool
 	enableTxHashLogging     bool
 
-	enableTxValidation       bool
-	txValidationFn           TxValidationFunc
-	txValidationEndpoint     string
-	txValidationMethods      TxValidationMethodSet
-	txValidationClient       *TxValidationClient
-	txValidationFailOpen     bool
-	isDraining               atomic.Bool
-	gracefulShutdownDuration time.Duration
+	enableTxValidation           bool
+	txValidationFn               TxValidationFunc
+	txValidationEndpoint         string
+	txValidationMethods          TxValidationMethodSet
+	txValidationClient           *TxValidationClient
+	txValidationFailOpen         bool
+	isDraining                   atomic.Bool
+	gracefulShutdownDuration     time.Duration
+	gracefulShutdownIdle         bool
+	gracefulShutdownIdleDuration time.Duration
+	isShuttingDown               atomic.Bool
+	lastRequestNanos             atomic.Int64
 }
 
 type limiterFunc func(method string) bool
@@ -131,6 +135,8 @@ func NewServer(
 	limExemptKeys []string,
 	txValidationConfig TxValidationMiddlewareConfig,
 	gracefulShutdownDuration time.Duration,
+	gracefulShutdownIdle bool,
+	gracefulShutdownIdleDuration time.Duration,
 ) (*Server, error) {
 	if cache == nil {
 		cache = &NoopRPCCache{}
@@ -221,7 +227,7 @@ func NewServer(
 		txValidationFailOpen = *txValidationConfig.FailOpen
 	}
 
-	return &Server{
+	srv := &Server{
 		BackendGroups:        backendGroups,
 		wsBackendGroup:       wsBackendGroup,
 		wsMethodWhitelist:    wsMethodWhitelist,
@@ -239,33 +245,38 @@ func NewServer(
 		upgrader: &websocket.Upgrader{
 			HandshakeTimeout: defaultWSHandshakeTimeout,
 		},
-		mainLim:                  mainLim,
-		overrideLims:             overrideLims,
-		globallyLimitedMethods:   globalMethodLims,
-		senderLim:                senderLim,
-		interopSenderLim:         interopSenderLim,
-		allowedChainIds:          senderRateLimitConfig.AllowedChainIds,
-		limExemptOrigins:         limExemptOrigins,
-		limExemptUserAgents:      limExemptUserAgents,
-		limExemptKeys:            limExemptKeys,
-		rateLimitHeader:          rateLimitHeader,
-		interopValidatingConfig:  interopValidatingConfig,
-		interopStrategy:          interopStrategy,
-		enableTxHashLogging:      enableTxHashLogging,
-		enableTxValidation:       txValidationConfig.Enabled,
-		txValidationFn:           txValidationClient.Validate,
-		txValidationEndpoint:     txValidationConfig.Endpoint,
-		txValidationMethods:      txValidationMethods,
-		txValidationClient:       txValidationClient,
-		txValidationFailOpen:     txValidationFailOpen,
-		gracefulShutdownDuration: gracefulShutdownDuration,
-	}, nil
+		mainLim:                      mainLim,
+		overrideLims:                 overrideLims,
+		globallyLimitedMethods:       globalMethodLims,
+		senderLim:                    senderLim,
+		interopSenderLim:             interopSenderLim,
+		allowedChainIds:              senderRateLimitConfig.AllowedChainIds,
+		limExemptOrigins:             limExemptOrigins,
+		limExemptUserAgents:          limExemptUserAgents,
+		limExemptKeys:                limExemptKeys,
+		rateLimitHeader:              rateLimitHeader,
+		interopValidatingConfig:      interopValidatingConfig,
+		interopStrategy:              interopStrategy,
+		enableTxHashLogging:          enableTxHashLogging,
+		enableTxValidation:           txValidationConfig.Enabled,
+		txValidationFn:               txValidationClient.Validate,
+		txValidationEndpoint:         txValidationConfig.Endpoint,
+		txValidationMethods:          txValidationMethods,
+		txValidationClient:           txValidationClient,
+		txValidationFailOpen:         txValidationFailOpen,
+		gracefulShutdownDuration:     gracefulShutdownDuration,
+		gracefulShutdownIdle:         gracefulShutdownIdle,
+		gracefulShutdownIdleDuration: gracefulShutdownIdleDuration,
+	}
+	srv.lastRequestNanos.Store(time.Now().UnixNano())
+	return srv, nil
 }
 
 func (s *Server) RPCListenAndServe(host string, port int) error {
 	s.srvMu.Lock()
 	hdlr := mux.NewRouter()
 	hdlr.HandleFunc("/healthz", s.HandleHealthz).Methods("GET")
+	hdlr.HandleFunc("/readyz", s.HandleReadyz).Methods("GET")
 	hdlr.HandleFunc("/", s.HandleRPC).Methods("POST")
 	hdlr.HandleFunc("/{authorization}", s.HandleRPC).Methods("POST")
 	c := cors.New(cors.Options{
@@ -300,6 +311,21 @@ func (s *Server) WSListenAndServe(host string, port int) error {
 }
 
 func (s *Server) Drain() {
+	if s.gracefulShutdownIdle {
+		s.isShuttingDown.Store(true)
+		log.Info("graceful shutdown: waiting for idle window",
+			"idle", s.gracefulShutdownIdleDuration)
+		ticker := time.NewTicker(time.Second)
+		defer ticker.Stop()
+		for range ticker.C {
+			last := time.Unix(0, s.lastRequestNanos.Load())
+			if time.Since(last) >= s.gracefulShutdownIdleDuration {
+				log.Info("graceful shutdown: idle window elapsed, shutting down")
+				return
+			}
+		}
+		return
+	}
 	s.isDraining.Store(true)
 	time.Sleep(s.gracefulShutdownDuration)
 }
@@ -326,7 +352,26 @@ func (s *Server) HandleHealthz(w http.ResponseWriter, r *http.Request) {
 	_, _ = w.Write([]byte("OK"))
 }
 
+func (s *Server) HandleReadyz(w http.ResponseWriter, r *http.Request) {
+	if s.isDraining.Load() || s.isShuttingDown.Load() {
+		http.Error(w, "Server is draining", http.StatusServiceUnavailable)
+		return
+	}
+	for name, bg := range s.BackendGroups {
+		if bg.Consensus == nil {
+			continue
+		}
+		if len(bg.Consensus.GetConsensusGroup()) == 0 || bg.Consensus.GetLatestBlockNumber() == 0 {
+			http.Error(w, fmt.Sprintf("consensus not ready: %s", name), http.StatusServiceUnavailable)
+			return
+		}
+	}
+	_, _ = w.Write([]byte("OK"))
+}
+
 func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) {
+	s.lastRequestNanos.Store(time.Now().UnixNano())
+
 	ctx := s.populateContext(w, r)
 	if ctx == nil {
 		return
@@ -782,6 +827,8 @@ func (s *Server) handleBatchRPC(ctx context.Context, reqs []json.RawMessage, isL
 }
 
 func (s *Server) HandleWS(w http.ResponseWriter, r *http.Request) {
+	s.lastRequestNanos.Store(time.Now().UnixNano())
+
 	ctx := s.populateContext(w, r)
 	if ctx == nil {
 		return

proxyd/server_test.go

@@ -3,9 +3,11 @@ package proxyd
 import (
 	"context"
 	"encoding/json"
+	"net/http/httptest"
 	"os"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -96,3 +98,81 @@ func TestIsValidAPIKey(t *testing.T) {
 		})
 	}
 }
+
+func TestHandleReadyz(t *testing.T) {
+	newConsensusGroup := func(members int, latest hexutil.Uint64) *BackendGroup {
+		bg := &BackendGroup{}
+		cg := make([]*Backend, members)
+		for i := range cg {
+			cg[i] = &Backend{}
+		}
+		tracker := NewInMemoryConsensusTracker()
+		tracker.SetState(ConsensusTrackerState{Latest: latest})
+		bg.Consensus = &ConsensusPoller{
+			backendGroup:   bg,
+			consensusGroup: cg,
+			tracker:        tracker,
+		}
+		return bg
+	}
+
+	tests := []struct {
+		name     string
+		draining bool
+		groups   map[string]*BackendGroup
+		want     int
+	}{
+		{
+			name:     "draining returns 503",
+			draining: true,
+			groups:   map[string]*BackendGroup{"main": newConsensusGroup(2, 100)},
+			want:     503,
+		},
+		{
+			name:   "no backend groups returns 200",
+			groups: map[string]*BackendGroup{},
+			want:   200,
+		},
+		{
+			name:   "group without consensus is skipped",
+			groups: map[string]*BackendGroup{"main": {}},
+			want:   200,
+		},
+		{
+			name:   "empty consensus group returns 503",
+			groups: map[string]*BackendGroup{"main": newConsensusGroup(0, 100)},
+			want:   503,
+		},
+		{
+			name:   "consensus group with latest=0 returns 503",
+			groups: map[string]*BackendGroup{"main": newConsensusGroup(2, 0)},
+			want:   503,
+		},
+		{
+			name:   "ready consensus group returns 200",
+			groups: map[string]*BackendGroup{"main": newConsensusGroup(2, 100)},
+			want:   200,
+		},
+		{
+			name: "any unready group fails the gate",
+			groups: map[string]*BackendGroup{
+				"main":  newConsensusGroup(2, 100),
+				"other": newConsensusGroup(0, 100),
+			},
+			want: 503,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &Server{BackendGroups: tt.groups}
+			s.isDraining.Store(tt.draining)
+
+			req := httptest.NewRequest("GET", "/readyz", nil)
+			rec := httptest.NewRecorder()
+			s.HandleReadyz(rec, req)
+
+			require.Equal(t, tt.want, rec.Code)
+		})
+	}
+}