[3.4] cherry-pick protect execution TIP under RPC load (#19905) (#20297)

lupin012 · claude · web-flow · commit bebac8eed62a · 2026-04-03T09:13:21.000+07:00
Co-authored-by: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/.github/workflows/qa-rpc-performance-tests.yml b/.github/workflows/qa-rpc-performance-tests.yml
@@ -121,7 +121,7 @@ jobs:
         if: matrix.client == 'erigon' || needs.setup.outputs.run_geth == 'true'
         run: |
           rm -rf ${{runner.workspace}}/rpc-tests
-          git -c advice.detachedHead=false clone --depth 1 --branch v1.115.0 https://github.com/erigontech/rpc-tests ${{runner.workspace}}/rpc-tests
+          git -c advice.detachedHead=false clone --depth 1 --branch v1.124.0 https://github.com/erigontech/rpc-tests ${{runner.workspace}}/rpc-tests
           cd ${{runner.workspace}}/rpc-tests
 
       - name: Clean Erigon Build Directory
diff --git a/.github/workflows/qa-tip-tracking-with-load.yml b/.github/workflows/qa-tip-tracking-with-load.yml
@@ -82,7 +82,7 @@ jobs:
       if: matrix.client == 'erigon' || (matrix.client == 'geth' && github.event.inputs.run_geth == 'true')
       run: |
         rm -rf ${{runner.workspace}}/rpc-tests
-        git -c advice.detachedHead=false clone --depth 1 --branch v1.78.0 https://github.com/erigontech/rpc-tests ${{runner.workspace}}/rpc-tests
+        git -c advice.detachedHead=false clone --depth 1 --branch v1.124.0 https://github.com/erigontech/rpc-tests ${{runner.workspace}}/rpc-tests
         cd ${{runner.workspace}}/rpc-tests
 
     - name: Clean Erigon Build Directory
diff --git a/cmd/rpcdaemon/cli/config.go b/cmd/rpcdaemon/cli/config.go
@@ -129,6 +129,7 @@ func RootCommand() (*cobra.Command, *httpcfg.HttpCfg) {
 	rootCmd.PersistentFlags().BoolVar(&cfg.RpcStreamingDisable, utils.RpcStreamingDisableFlag.Name, false, utils.RpcStreamingDisableFlag.Usage)
 	rootCmd.PersistentFlags().BoolVar(&cfg.DebugSingleRequest, utils.HTTPDebugSingleFlag.Name, false, utils.HTTPDebugSingleFlag.Usage)
 	rootCmd.PersistentFlags().IntVar(&cfg.DBReadConcurrency, utils.DBReadConcurrencyFlag.Name, utils.DBReadConcurrencyFlag.Value, utils.DBReadConcurrencyFlag.Usage)
+	rootCmd.PersistentFlags().IntVar(&cfg.RpcMaxConcurrentRequests, utils.RpcMaxConcurrentRequestsFlag.Name, utils.RpcMaxConcurrentRequestsFlag.Value, utils.RpcMaxConcurrentRequestsFlag.Usage)
 	rootCmd.PersistentFlags().BoolVar(&cfg.TraceCompatibility, "trace.compat", false, "Bug for bug compatibility with OE for trace_ routines")
 	rootCmd.PersistentFlags().BoolVar(&cfg.GethCompatibility, "rpc.gethcompat", false, "Enables Geth-compatible storage iteration order for debug_storageRangeAt (sorted by keccak256 hash). Disabled by default for performance.")
 	rootCmd.PersistentFlags().BoolVar(&cfg.TestingEnabled, "rpc.testing", false, "Enables the testing_ RPC namespace (testing_buildBlockV1). WARNING: do not enable on production networks.")
@@ -744,7 +745,17 @@ func startRegularRpcServer(ctx context.Context, cfg *httpcfg.HttpCfg, rpcAPI []r
 		logger.Info("Socket Endpoint opened", "url", socketUrl)
 	}
 
-	httpHandler := node.NewHTTPHandlerStack(srv, cfg.HttpCORSDomain, cfg.HttpVirtualHost, cfg.HttpCompression)
+	// RPC admission limit: -1 = unlimited, 0 = use db.read.concurrency, >0 = explicit limit.
+	var rpcConcurrencyLimit int64
+	switch {
+	case cfg.RpcMaxConcurrentRequests == -1:
+		rpcConcurrencyLimit = 0 // disabled
+	case cfg.RpcMaxConcurrentRequests > 0:
+		rpcConcurrencyLimit = int64(cfg.RpcMaxConcurrentRequests)
+	default:
+		rpcConcurrencyLimit = int64(cfg.DBReadConcurrency)
+	}
+	httpHandler := node.NewHTTPHandlerStack(srv, cfg.HttpCORSDomain, cfg.HttpVirtualHost, cfg.HttpCompression, rpcConcurrencyLimit, true)
 	var wsHandler http.Handler
 	if cfg.WebsocketEnabled {
 		wsHandler = srv.WebsocketHandler([]string{"*"}, nil, cfg.WebsocketCompression, logger)
@@ -955,7 +966,9 @@ func createEngineListener(cfg *httpcfg.HttpCfg, engineApi []rpc.API, logger log.
 
 	wsHandler := engineSrv.WebsocketHandler([]string{"*"}, jwtSecret, cfg.WebsocketCompression, logger)
 
-	engineHttpHandler := node.NewHTTPHandlerStack(engineSrv, nil /* authCors */, cfg.AuthRpcVirtualHost, cfg.HttpCompression)
+	// Engine API (auth) is the CL↔EL protocol — not user RPC. Do not tag with TxPriorityRPC
+	// so execution-engine DB operations use blocking Acquire instead of fail-fast TryAcquire.
+	engineHttpHandler := node.NewHTTPHandlerStack(engineSrv, nil /* authCors */, cfg.AuthRpcVirtualHost, cfg.HttpCompression, 0, false)
 
 	graphQLHandler := graphql.CreateHandler(engineApi)
 
diff --git a/cmd/rpcdaemon/cli/httpcfg/http_cfg.go b/cmd/rpcdaemon/cli/httpcfg/http_cfg.go
@@ -71,6 +71,7 @@ type HttpCfg struct {
 	RpcStreamingDisable               bool
 	RpcFiltersConfig                  rpchelper.FiltersConfig
 	DBReadConcurrency                 int
+	RpcMaxConcurrentRequests          int  // HTTP admission control limit; -1 = unlimited
 	TraceCompatibility                bool // Bug for bug compatibility for trace_ routines with OpenEthereum
 	GethCompatibility                 bool // Geth-compatible storage iteration order for debug_storageRangeAt
 	TxPoolApiAddr                     string
diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
@@ -389,6 +389,11 @@ var (
 		Usage: "Does limit amount of parallel db reads. Default: equal to GOMAXPROCS (or number of CPU)",
 		Value: min(max(10, runtime.GOMAXPROCS(-1)*64), 9_000),
 	}
+	RpcMaxConcurrentRequestsFlag = cli.IntFlag{
+		Name:  "rpc.max.concurrency",
+		Usage: "Maximum number of concurrent HTTP RPC requests (HTTP admission control). 0 = use db.read.concurrency, -1 = unlimited (no admission control)",
+		Value: 0,
+	}
 	RpcAccessListFlag = cli.StringFlag{
 		Name:  "rpc.accessList",
 		Usage: "Specify granular (method-by-method) API allowlist",
diff --git a/db/kv/kv_interface.go b/db/kv/kv_interface.go
@@ -696,6 +696,23 @@ var (
 	//DbGcSelfPnlMergeCalls  = metrics.NewCounter(`db_gc_pnl{phase="slef_merge_calls"}`)                //nolint
 )
 
+// ErrServerOverloaded is returned by BeginRo when the DB semaphore is full and the caller is an RPC handler.
+var ErrServerOverloaded = errors.New("server overloaded, retry later")
+
+type nonBlockingAcquireKey struct{}
+
+// WithNonBlockingAcquire tags ctx to request fail-fast semaphore acquisition in BeginRo.
+// When set, BeginRo uses TryAcquire and returns ErrServerOverloaded immediately if the
+// read-tx semaphore is full, instead of blocking until a slot is available.
+func WithNonBlockingAcquire(ctx context.Context) context.Context {
+	return context.WithValue(ctx, nonBlockingAcquireKey{}, struct{}{})
+}
+
+// IsNonBlockingAcquire reports whether ctx was tagged by WithNonBlockingAcquire.
+func IsNonBlockingAcquire(ctx context.Context) bool {
+	return ctx.Value(nonBlockingAcquireKey{}) != nil
+}
+
 type Closer interface {
 	Close()
 }
diff --git a/db/kv/mdbx/kv_mdbx.go b/db/kv/mdbx/kv_mdbx.go
@@ -46,8 +46,11 @@ import (
 	"github.com/erigontech/erigon/db/kv/dbcfg"
 	"github.com/erigontech/erigon/db/kv/order"
 	"github.com/erigontech/erigon/db/kv/stream"
+	"github.com/erigontech/erigon/diagnostics/metrics"
 )
 
+var dbRoTxOverloaded = metrics.GetOrCreateCounter(`db_rotx_overloaded_total`)
+
 func init() {
 	mdbx.MapFullErrorMessage += " You can try remove the database files (e.g., by running rm -rf /path/to/db)"
 }
@@ -582,8 +585,13 @@ func (db *MdbxKV) BeginRo(ctx context.Context) (txn kv.Tx, err error) {
 		return nil, errors.New("db closed")
 	}
 
-	// will return nil err if context is cancelled (may appear to acquire the semaphore)
-	if semErr := db.roTxsLimiter.Acquire(ctx, 1); semErr != nil {
+	if kv.IsNonBlockingAcquire(ctx) {
+		if !db.roTxsLimiter.TryAcquire(1) {
+			db.trackTxEnd()
+			dbRoTxOverloaded.Inc()
+			return nil, kv.ErrServerOverloaded
+		}
+	} else if semErr := db.roTxsLimiter.Acquire(ctx, 1); semErr != nil {
 		db.trackTxEnd()
 		return nil, fmt.Errorf("mdbx.MdbxKV.BeginRo: roTxsLimiter error %w", semErr)
 	}
diff --git a/db/kv/remotedbserver/remotedbserver.go b/db/kv/remotedbserver/remotedbserver.go
@@ -135,7 +135,7 @@ func (s *KvServer) begin(ctx context.Context) (id uint64, err error) {
 	}
 	s.txsMapLock.Lock()
 	defer s.txsMapLock.Unlock()
-	tx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic
+	tx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic // tx is stored in s.txs and rolled back by rollback(); defer would close it prematurely
 	if errBegin != nil {
 		return 0, errBegin
 	}
@@ -157,7 +157,7 @@ func (s *KvServer) renew(ctx context.Context, id uint64) (err error) {
 		defer tx.Unlock()
 		tx.Rollback()
 	}
-	newTx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic
+	newTx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic // tx is stored in s.txs and rolled back by rollback(); defer would close it prematurely
 	if errBegin != nil {
 		return fmt.Errorf("kvserver: %w", errBegin)
 	}
diff --git a/node/cli/default_flags.go b/node/cli/default_flags.go
@@ -83,6 +83,7 @@ var DefaultFlags = []cli.Flag{
 	&utils.RpcBatchConcurrencyFlag,
 	&utils.RpcStreamingDisableFlag,
 	&utils.DBReadConcurrencyFlag,
+	&utils.RpcMaxConcurrentRequestsFlag,
 	&utils.RpcAccessListFlag,
 	&utils.RpcTraceCompatFlag,
 	&utils.RpcGethCompatFlag,
diff --git a/node/cli/flags.go b/node/cli/flags.go
@@ -447,6 +447,7 @@ func setEmbeddedRpcDaemon(ctx *cli.Context, cfg *nodecfg.Config, logger log.Logg
 		RpcBatchConcurrency:       ctx.Uint(utils.RpcBatchConcurrencyFlag.Name),
 		RpcStreamingDisable:       ctx.Bool(utils.RpcStreamingDisableFlag.Name),
 		DBReadConcurrency:         ctx.Int(utils.DBReadConcurrencyFlag.Name),
+		RpcMaxConcurrentRequests:  ctx.Int(utils.RpcMaxConcurrentRequestsFlag.Name),
 		RpcAllowListFilePath:      ctx.String(utils.RpcAccessListFlag.Name),
 		RpcFiltersConfig: rpchelper.FiltersConfig{
 			RpcSubscriptionFiltersMaxLogs:      ctx.Int(RpcSubscriptionFiltersMaxLogsFlag.Name),
diff --git a/node/rpcstack.go b/node/rpcstack.go
@@ -35,6 +35,8 @@ import (
 	"github.com/rs/cors"
 
 	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/db/kv"
+	"github.com/erigontech/erigon/diagnostics/metrics"
 	"github.com/erigontech/erigon/rpc"
 	"github.com/erigontech/erigon/rpc/rpccfg"
 )
@@ -46,6 +48,10 @@ type httpConfig struct {
 	Vhosts             []string
 	Compression        bool
 	prefix             string // path prefix on which to mount http handler
+	// RpcConcurrencyLimit is the maximum number of concurrent HTTP RPC requests.
+	// Requests beyond this limit receive an immediate 503 before touching any middleware.
+	// 0 means unlimited (admission control disabled).
+	RpcConcurrencyLimit int64
 }
 
 // wsConfig is the JSON-RPC/Websocket configuration
@@ -193,6 +199,11 @@ func (h *httpServer) start() error {
 
 func (h *httpServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// check if ws request and serve if ws enabled
+	// Note: WebSocket connections bypass rpcAdmissionHandler intentionally.
+	// HTTP admission control limits inflight requests per connection, but WebSocket
+	// is a persistent long-lived connection where the relevant limit is the number
+	// of concurrent connections, not inflight requests. A dedicated WebSocket
+	// connection limiter will be addressed in a separate PR.
 	ws := h.wsHandler.Load().(*rpcHandler)
 	if ws != nil && isWebsocket(r) {
 		if checkPath(r, h.wsConfig.prefix) {
@@ -280,7 +291,7 @@ func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig, allowList rpc.
 	}
 	h.httpConfig = config
 	h.httpHandler.Store(&rpcHandler{
-		Handler: NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts, config.Compression),
+		Handler: NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts, config.Compression, config.RpcConcurrencyLimit, true),
 		server:  srv,
 	})
 	return nil
@@ -345,17 +356,60 @@ func isWebsocket(r *http.Request) bool {
 		strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade")
 }
 
-// NewHTTPHandlerStack returns wrapped http-related handlers
-func NewHTTPHandlerStack(srv http.Handler, cors []string, vhosts []string, compression bool) http.Handler {
-	// Wrap the CORS-handler within a host-handler
+// NewHTTPHandlerStack returns wrapped http-related handlers.
+// When tagAsRPC is true and rpcConcurrencyLimit > 0, enforces admission control
+// (503 if inflight > limit) to prevent goroutine pile-up under load.
+func NewHTTPHandlerStack(srv http.Handler, cors []string, vhosts []string, compression bool, rpcConcurrencyLimit int64, tagAsRPC bool) http.Handler {
 	handler := newCorsHandler(srv, cors)
 	handler = newVHostHandler(vhosts, handler)
 	if compression {
 		handler = newGzipHandler(handler)
 	}
+	if tagAsRPC {
+		handler = newRPCAdmissionHandler(rpcConcurrencyLimit, handler)
+	}
 	return handler
 }
 
+// rpcAdmissionHandler limits the number of concurrent HTTP RPC requests.
+// Requests that exceed the limit receive an immediate HTTP 503 without going
+// through CORS, gzip, or JSON decoding.
+type rpcAdmissionHandler struct {
+	inflight atomic.Int64
+	limit    int64
+	next     http.Handler
+}
+
+var rpcAdmissionRejected = metrics.GetOrCreateCounter(`rpc_admission_rejected_total`)
+
+func newRPCAdmissionHandler(limit int64, next http.Handler) http.Handler {
+	return &rpcAdmissionHandler{limit: limit, next: next}
+}
+
+func (h *rpcAdmissionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if h.limit > 0 {
+		if h.inflight.Add(1) > h.limit {
+			h.inflight.Add(-1)
+			rpcAdmissionRejected.Inc()
+			w.Header().Set("Retry-After", "1")
+			// TODO: the 503 response here is plain text, not a JSON-RPC envelope.
+			// Similarly, when BeginRo returns ErrServerOverloaded (inner gate), the
+			// response is HTTP 200 with JSON-RPC code -32000 instead of -32005.
+			// Both paths should return HTTP 503 + JSON-RPC {"error":{"code":-32005,...}}.
+			// This requires buffering the response in rpc/http.go and will be
+			// addressed in a separate PR.
+			http.Error(w, "server overloaded, retry later", http.StatusServiceUnavailable)
+			return
+		}
+		defer h.inflight.Add(-1)
+	}
+	ctx := r.Context()
+	if h.limit > 0 {
+		ctx = kv.WithNonBlockingAcquire(ctx)
+	}
+	h.next.ServeHTTP(w, r.WithContext(ctx))
+}
+
 func newCorsHandler(srv http.Handler, allowedOrigins []string) http.Handler {
 	// disable CORS support if user has not specified a custom CORS configuration
 	if len(allowedOrigins) == 0 {
diff --git a/node/rpcstack_test.go b/node/rpcstack_test.go
@@ -24,9 +24,11 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/gorilla/websocket"
@@ -245,6 +247,11 @@ func Test_checkPath(t *testing.T) {
 func createAndStartServer(t *testing.T, conf *httpConfig, ws bool, wsConf *wsConfig) *httpServer {
 	t.Helper()
 
+	// Ensure RpcConcurrencyLimit is always set so admission control wiring is exercised.
+	// A high value avoids interfering with existing tests while still activating the path.
+	if conf.RpcConcurrencyLimit == 0 {
+		conf.RpcConcurrencyLimit = 1000
+	}
 	srv := newHTTPServer(testlog.Logger(t, log.LvlError), rpccfg.DefaultHTTPTimeouts)
 	require.NoError(t, srv.enableRPC(nil, *conf, nil))
 	if ws {
@@ -375,3 +382,63 @@ func TestHTTP2H2C(t *testing.T) {
 	require.NoError(t, err)
 	assert.Contains(t, string(result), "jsonrpc", "expected JSON-RPC response")
 }
+
+// TestRPCAdmissionHandler verifies that rpcAdmissionHandler correctly limits
+// concurrent requests and returns HTTP 503 when the limit is exceeded.
+func TestRPCAdmissionHandler(t *testing.T) {
+	okHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+
+	t.Run("disabled when limit is zero", func(t *testing.T) {
+		h := newRPCAdmissionHandler(0, okHandler)
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/", nil))
+		assert.Equal(t, http.StatusOK, rec.Code)
+	})
+
+	t.Run("allows requests under the limit", func(t *testing.T) {
+		h := newRPCAdmissionHandler(5, okHandler)
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/", nil))
+		assert.Equal(t, http.StatusOK, rec.Code)
+	})
+
+	t.Run("returns 503 when limit is exceeded", func(t *testing.T) {
+		// Use a gate channel to hold inflight requests open long enough to
+		// trigger the limit.
+		gate := make(chan struct{})
+		blockingHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			<-gate
+			w.WriteHeader(http.StatusOK)
+		})
+
+		const limit = 2
+		h := newRPCAdmissionHandler(limit, blockingHandler)
+
+		var wg sync.WaitGroup
+		for i := 0; i < limit; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/", nil))
+			}()
+		}
+
+		// Give goroutines time to enter the handler and increment the counter.
+		// We busy-wait on the inflight counter rather than sleeping.
+		admission := h.(*rpcAdmissionHandler)
+		for admission.inflight.Load() < limit {
+			// spin
+		}
+
+		// Now the limit is reached — next request must be rejected.
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/", nil))
+		assert.Equal(t, http.StatusServiceUnavailable, rec.Code)
+
+		// Release the held requests.
+		close(gate)
+		wg.Wait()
+	})
+}

Original file line number	Diff line number	Diff line change
`@@ -135,7 +135,7 @@ func (s *KvServer) begin(ctx context.Context) (id uint64, err error) {`
`135`	`135`	`}`
`136`	`136`	`s.txsMapLock.Lock()`
`137`	`137`	`defer s.txsMapLock.Unlock()`
`138`		`- tx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic`
	`138`	`+ tx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic // tx is stored in s.txs and rolled back by rollback(); defer would close it prematurely`
`139`	`139`	`if errBegin != nil {`
`140`	`140`	`return 0, errBegin`
`141`	`141`	`}`
`@@ -157,7 +157,7 @@ func (s *KvServer) renew(ctx context.Context, id uint64) (err error) {`
`157`	`157`	`defer tx.Unlock()`
`158`	`158`	`tx.Rollback()`
`159`	`159`	`}`
`160`		`- newTx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic`
	`160`	`+ newTx, errBegin := s.kv.BeginTemporalRo(ctx) //nolint:gocritic // tx is stored in s.txs and rolled back by rollback(); defer would close it prematurely`
`161`	`161`	`if errBegin != nil {`
`162`	`162`	`return fmt.Errorf("kvserver: %w", errBegin)`
`163`	`163`	`}`