From b3c9c5e85de32f1d0fe89952f5b2413550bcecbc Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Mon, 27 Jul 2026 16:49:54 +0200 Subject: [PATCH 1/9] WIP --- evmrpc/config/config.go | 34 +++--- evmrpc/metrics.go | 19 +++- evmrpc/request_limiter.go | 11 +- evmrpc/request_limiter_test.go | 1 + evmrpc/rpcstack.go | 10 +- evmrpc/server.go | 4 +- evmrpc/ws_admission_test.go | 200 +++++++++++++++++++++++++++++++++ 7 files changed, 252 insertions(+), 27 deletions(-) create mode 100644 evmrpc/ws_admission_test.go diff --git a/evmrpc/config/config.go b/evmrpc/config/config.go index e645bc57f9..41ecb25f30 100644 --- a/evmrpc/config/config.go +++ b/evmrpc/config/config.go @@ -250,17 +250,18 @@ type Config struct { // batched JSON-RPC call (HTTP and WebSocket). Set to 0 to disable the limit. BatchResponseMaxSize int `mapstructure:"batch_response_max_size"` - // MaxRequestBodyBytes is the maximum size, in bytes, of a single HTTP - // JSON-RPC request body. Requests larger than this are rejected (HTTP 413) - // before the body is buffered or JSON-decoded. 0 uses the go-ethereum - // default (5 MiB). + // MaxRequestBodyBytes is the maximum size, in bytes, of a single HTTP (:8545) + // or WebSocket (:8546) JSON-RPC request/frame. HTTP requests larger than this + // are rejected (HTTP 413) before the body is buffered or JSON-decoded. + // WebSocket frames exceeding this limit are rejected by the read loop. + // 0 uses the go-ethereum default (5 MiB). MaxRequestBodyBytes int64 `mapstructure:"max_request_body_bytes"` - // MaxConcurrentRequestBytes bounds the total size, in bytes, of HTTP - // JSON-RPC request bodies admitted for processing concurrently, weighted by - // each request's Content-Length. Requests that would exceed the budget are - // rejected fast (HTTP 429) before decode, capping peak memory under load. - // Set to 0 to disable the limit. + // MaxConcurrentRequestBytes bounds the total size, in bytes, of HTTP and + // WebSocket JSON-RPC request bodies admitted for processing concurrently + // (independent budgets per plane). HTTP uses Content-Length weighting and + // rejects over-budget requests fast (HTTP 429). WebSocket blocks until + // budget frees or times out. Set to 0 to disable the limit on either plane. MaxConcurrentRequestBytes int64 `mapstructure:"max_concurrent_request_bytes"` // MaxOpenConnections caps the number of simultaneously accepted connections @@ -916,15 +917,16 @@ batch_request_limit = {{ .EVM.BatchRequestLimit }} # batched JSON-RPC call (HTTP and WebSocket). Set to 0 to disable the limit. batch_response_max_size = {{ .EVM.BatchResponseMaxSize }} -# max_request_body_bytes is the maximum size, in bytes, of a single HTTP -# JSON-RPC request body. Larger requests are rejected (HTTP 413) before the body -# is buffered or JSON-decoded. Set to 0 to use the default (5 MiB). +# max_request_body_bytes is the maximum size, in bytes, of a single HTTP (:8545) +# or WebSocket (:8546) JSON-RPC request/frame. HTTP larger requests are rejected +# (HTTP 413) before decode; WS frames exceeding this limit are rejected by the +# read loop. Set to 0 to use the default (5 MiB). max_request_body_bytes = {{ .EVM.MaxRequestBodyBytes }} -# max_concurrent_request_bytes bounds the total size, in bytes, of HTTP JSON-RPC -# request bodies admitted for processing concurrently (weighted by each request's -# Content-Length). Requests that would exceed the budget are rejected fast -# (HTTP 429) before decode, capping peak memory under load. Set to 0 to disable. +# max_concurrent_request_bytes bounds total request bytes admitted concurrently +# on HTTP (:8545) and WebSocket (:8546) (independent budgets per plane). HTTP +# rejects over-budget requests fast (HTTP 429); WS blocks until budget frees or +# times out. Set to 0 to disable on either plane. max_concurrent_request_bytes = {{ .EVM.MaxConcurrentRequestBytes }} # max_open_connections caps the number of simultaneously accepted connections on diff --git a/evmrpc/metrics.go b/evmrpc/metrics.go index ead49300e2..1e6fc7cf98 100644 --- a/evmrpc/metrics.go +++ b/evmrpc/metrics.go @@ -18,6 +18,9 @@ const ( errorClassKey = "error_class" jsonrpcCodeKey = "jsonrpc_code" rejectReasonKey = "reason" + planeKey = "plane" + planeHTTP = "http" + planeWS = "ws" // reject reason values for requestRejectedCount. rejectReasonOversize = "oversize" // body exceeded max_request_body_bytes rejectReasonBusy = "busy" // max_concurrent_request_bytes budget exhausted @@ -76,7 +79,7 @@ var ( )), requestRejectedCount: must(rpcTelemetryMeter.Int64Counter( "evmrpc_requests_rejected_total", - metric.WithDescription("Number of HTTP JSON-RPC requests rejected by pre-decode admission control (labeled by reason)"), + metric.WithDescription("Number of JSON-RPC requests rejected by admission control (labeled by plane and reason)"), metric.WithUnit("{count}"), )), } @@ -166,13 +169,19 @@ func recordHistoricalDebugTraceAttempt(ctx context.Context, endpoint, connection ) } -// recordRequestRejected counts an HTTP JSON-RPC request dropped by pre-decode -// admission control. reason is one of rejectReasonOversize / rejectReasonBusy. -// No endpoint dimension is recorded: the rejection happens before the JSON-RPC -// method is decoded, so it is not yet known. func recordRequestRejected(ctx context.Context, reason string) { metrics.requestRejectedCount.Add(ctx, 1, metric.WithAttributes( + attribute.String(planeKey, planeHTTP), + attribute.String(rejectReasonKey, reason), + ), + ) +} + +func recordWSAdmissionRejected(ctx context.Context, reason string) { + metrics.requestRejectedCount.Add(ctx, 1, + metric.WithAttributes( + attribute.String(planeKey, planeWS), attribute.String(rejectReasonKey, reason), ), ) diff --git a/evmrpc/request_limiter.go b/evmrpc/request_limiter.go index b138df93ff..a2a597351f 100644 --- a/evmrpc/request_limiter.go +++ b/evmrpc/request_limiter.go @@ -11,6 +11,13 @@ import ( // max_request_body_bytes is left at 0 ("use the default"). const defaultMaxRequestBodyBytes int64 = 5 * 1024 * 1024 +func effectiveMaxRequestBodyBytes(max int64) int64 { + if max <= 0 { + return defaultMaxRequestBodyBytes + } + return max +} + // requestSizeLimiter is an HTTP middleware that bounds peak decode-time memory by // admitting JSON-RPC requests *before* the body is buffered or decoded. It enforces: // @@ -34,9 +41,7 @@ type requestSizeLimiter struct { // global budget. If a positive budget is smaller than maxBody it is raised to maxBody // so that a single maximum-size request can always be admitted. func newRequestSizeLimiter(inner http.Handler, maxBody, maxConcurrentBytes int64) http.Handler { - if maxBody <= 0 { - maxBody = defaultMaxRequestBodyBytes - } + maxBody = effectiveMaxRequestBodyBytes(maxBody) l := &requestSizeLimiter{inner: inner, maxBody: maxBody} if maxConcurrentBytes > 0 { if maxConcurrentBytes < maxBody { diff --git a/evmrpc/request_limiter_test.go b/evmrpc/request_limiter_test.go index 9f5af56248..ee8c110615 100644 --- a/evmrpc/request_limiter_test.go +++ b/evmrpc/request_limiter_test.go @@ -58,6 +58,7 @@ func TestRequestSizeLimiter(t *testing.T) { }) t.Run("zero maxBody uses default cap", func(t *testing.T) { + require.Equal(t, defaultMaxRequestBodyBytes, effectiveMaxRequestBodyBytes(0)) l, ok := newRequestSizeLimiter(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), 0, 0).(*requestSizeLimiter) require.True(t, ok) require.Equal(t, defaultMaxRequestBodyBytes, l.maxBody) diff --git a/evmrpc/rpcstack.go b/evmrpc/rpcstack.go index 1fd9e6bead..2b80b085f4 100644 --- a/evmrpc/rpcstack.go +++ b/evmrpc/rpcstack.go @@ -381,7 +381,15 @@ func (h *HTTPServer) EnableWS(apis []rpc.API, config WsConfig) error { // Create RPC server and handler. srv := rpc.NewServer() srv.SetBatchLimits(config.batchItemLimit, config.batchResponseSizeLimit) - srv.SetReadLimits(config.readLimit) + readLimit := effectiveMaxRequestBodyBytes(config.readLimit) + if readLimit > math.MaxInt { + readLimit = math.MaxInt + } + srv.SetReadLimits(readLimit) + srv.SetWSConcurrentRequestBytes(config.maxConcurrentRequestBytes) + srv.SetWSAdmissionEventHook(func(reason string) { + recordWSAdmissionRejected(context.Background(), reason) + }) logger.Info("Registering apis for evm websocket") if err := RegisterApis(apis, config.Modules, srv); err != nil { return err diff --git a/evmrpc/server.go b/evmrpc/server.go index 58ad5d9055..243494b4d3 100644 --- a/evmrpc/server.go +++ b/evmrpc/server.go @@ -24,7 +24,6 @@ var ConnectionTypeWS ConnectionType = "websocket" var ConnectionTypeHTTP ConnectionType = "http" const LocalAddress = "0.0.0.0" -const DefaultWebsocketMaxMessageSize = 10 * 1024 * 1024 type EVMServer interface { Start() error @@ -339,7 +338,8 @@ func NewEVMWebSocketServer( } wsConfig := WsConfig{Origins: strings.Split(config.WSOrigins, ",")} - wsConfig.readLimit = DefaultWebsocketMaxMessageSize + wsConfig.readLimit = config.MaxRequestBodyBytes + wsConfig.maxConcurrentRequestBytes = config.MaxConcurrentRequestBytes wsConfig.batchItemLimit = config.BatchRequestLimit wsConfig.batchResponseSizeLimit = config.BatchResponseMaxSize if err := httpServer.EnableWS(apis, wsConfig); err != nil { diff --git a/evmrpc/ws_admission_test.go b/evmrpc/ws_admission_test.go new file mode 100644 index 0000000000..fede0deaf5 --- /dev/null +++ b/evmrpc/ws_admission_test.go @@ -0,0 +1,200 @@ +package evmrpc + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/rpc" + "github.com/gorilla/websocket" + "github.com/stretchr/testify/require" +) + +type wsAdmissionTestService struct{} + +func (wsAdmissionTestService) Sleep(_ context.Context, duration time.Duration) { + time.Sleep(duration) +} + +func TestEffectiveMaxRequestBodyBytes(t *testing.T) { + require.Equal(t, defaultMaxRequestBodyBytes, effectiveMaxRequestBodyBytes(0)) + require.Equal(t, defaultMaxRequestBodyBytes, effectiveMaxRequestBodyBytes(-1)) + require.Equal(t, int64(1024), effectiveMaxRequestBodyBytes(1024)) +} + +func TestEnableWSConcurrentRequestBytes(t *testing.T) { + const ( + sleepDuration = 200 * time.Millisecond + pad = 48 + ) + + makeMsg := func(id int, method string) string { + return fmt.Sprintf( + `{"jsonrpc":"2.0","id":%d,"method":"%s","params":[%d],"_pad":"%s"}`, + id, method, sleepDuration.Nanoseconds(), strings.Repeat("x", pad), + ) + } + payload := makeMsg(1, "test_sleep") + frameSize := int64(len(payload)) + + srv := startWSTestServer(t, WsConfig{ + Origins: []string{"*"}, + RPCEndpointConfig: RPCEndpointConfig{ + readLimit: frameSize, + maxConcurrentRequestBytes: frameSize, + }, + }) + + conn := dialWSTestServer(t, srv) + defer conn.Close() + + writeReq := func(id int) { + msg := makeMsg(id, "test_sleep") + require.Equal(t, len(payload), len(msg)) + require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(msg))) + } + + writeReq(1) + writeReq(2) + + var firstResp, secondResp rpcResponse + readJSON(t, conn, &firstResp) + readJSON(t, conn, &secondResp) + + require.Equal(t, json.Number("1"), firstResp.ID) + require.Equal(t, json.Number("2"), secondResp.ID) +} + +func TestEnableWSReadLimitDefault(t *testing.T) { + srv := NewHTTPServer(rpc.DefaultHTTPTimeouts) + wsConf := WsConfig{ + Origins: []string{"*"}, + RPCEndpointConfig: RPCEndpointConfig{ + readLimit: 0, + maxConcurrentRequestBytes: 0, + }, + } + require.NoError(t, srv.EnableWS([]rpc.API{ + {Namespace: "test", Service: wsAdmissionTestService{}}, + }, wsConf)) + require.Equal(t, defaultMaxRequestBodyBytes, effectiveMaxRequestBodyBytes(wsConf.readLimit)) +} + +func TestWSAdmissionHookBudgetWaitTimeout(t *testing.T) { + const ( + sleepDuration = 200 * time.Millisecond + pad = 48 + waitTimeout = 50 * time.Millisecond + ) + + srv := rpc.NewServer() + require.NoError(t, srv.RegisterName("test", wsAdmissionTestService{})) + + var ( + mu sync.Mutex + reasons []string + ) + srv.SetWSAdmissionEventHook(func(reason string) { + recordWSAdmissionRejected(context.Background(), reason) + mu.Lock() + reasons = append(reasons, reason) + mu.Unlock() + }) + srv.SetWSAdmissionTimeout(waitTimeout) + + makeMsg := func(id int) string { + return fmt.Sprintf( + `{"jsonrpc":"2.0","id":%d,"method":"test_sleep","params":[%d],"_pad":"%s"}`, + id, sleepDuration.Nanoseconds(), strings.Repeat("x", pad), + ) + } + payload := makeMsg(1) + frameSize := int64(len(payload)) + srv.SetReadLimits(frameSize) + srv.SetWSConcurrentRequestBytes(frameSize) + + p1, p2 := net.Pipe() + serveDone := make(chan struct{}) + go func() { + srv.ServeCodec(rpc.NewCodec(p1), 0) + close(serveDone) + }() + t.Cleanup(func() { + p2.Close() + p1.Close() + select { + case <-serveDone: + case <-time.After(2 * time.Second): + t.Error("ServeCodec did not exit within 2s") + } + srv.Stop() + }) + + _, err := io.WriteString(p2, payload) + require.NoError(t, err) + + deadline := time.Now().Add(waitTimeout + 300*time.Millisecond) + for { + mu.Lock() + got := append([]string(nil), reasons...) + mu.Unlock() + if len(got) > 0 { + require.Equal(t, rpc.WSAdmissionReasonBudgetWaitTimeout, got[0]) + return + } + if time.Now().After(deadline) { + t.Fatal("expected admission hook to fire on budget wait timeout") + } + time.Sleep(10 * time.Millisecond) + } +} + +type rpcResponse struct { + ID json.Number `json:"id"` + Result json.RawMessage `json:"result"` + Error *rpcError `json:"error"` + JSONRPC string `json:"jsonrpc"` +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +func startWSTestServer(t *testing.T, wsConf WsConfig) *HTTPServer { + t.Helper() + + srv := NewHTTPServer(rpc.DefaultHTTPTimeouts) + require.NoError(t, srv.EnableWS([]rpc.API{ + {Namespace: "test", Service: wsAdmissionTestService{}}, + }, wsConf)) + require.NoError(t, srv.SetListenAddr("127.0.0.1", 0)) + require.NoError(t, srv.Start()) + t.Cleanup(func() { srv.Stop() }) + return srv +} + +func dialWSTestServer(t *testing.T, srv *HTTPServer) *websocket.Conn { + t.Helper() + + wsURL := "ws://" + srv.ListenAddr() + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + require.NoError(t, err) + return conn +} + +func readJSON(t *testing.T, conn *websocket.Conn, dest any) { + t.Helper() + + _ = conn.SetReadDeadline(time.Now().Add(time.Second)) + _, data, err := conn.ReadMessage() + require.NoError(t, err) + require.NoError(t, json.Unmarshal(data, dest)) + _ = conn.SetReadDeadline(time.Time{}) +} From b55b32fa3a2bdb2c817d1558d8d0b29b724e58f6 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 11:37:19 +0200 Subject: [PATCH 2/9] ws_admission_test cleanup --- evmrpc/ws_admission_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/evmrpc/ws_admission_test.go b/evmrpc/ws_admission_test.go index fede0deaf5..2d756f9938 100644 --- a/evmrpc/ws_admission_test.go +++ b/evmrpc/ws_admission_test.go @@ -56,7 +56,6 @@ func TestEnableWSConcurrentRequestBytes(t *testing.T) { writeReq := func(id int) { msg := makeMsg(id, "test_sleep") - require.Equal(t, len(payload), len(msg)) require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(msg))) } @@ -101,7 +100,7 @@ func TestWSAdmissionHookBudgetWaitTimeout(t *testing.T) { reasons []string ) srv.SetWSAdmissionEventHook(func(reason string) { - recordWSAdmissionRejected(context.Background(), reason) + recordWSAdmissionRejected(t.Context(), reason) mu.Lock() reasons = append(reasons, reason) mu.Unlock() @@ -189,11 +188,12 @@ func dialWSTestServer(t *testing.T, srv *HTTPServer) *websocket.Conn { return conn } -func readJSON(t *testing.T, conn *websocket.Conn, dest any) { +func readJSON(t *testing.T, conn *websocket.Conn, dest *rpcResponse) { t.Helper() _ = conn.SetReadDeadline(time.Now().Add(time.Second)) - _, data, err := conn.ReadMessage() + msgType, data, err := conn.ReadMessage() + require.IsType(t, websocket.TextMessage, msgType) require.NoError(t, err) require.NoError(t, json.Unmarshal(data, dest)) _ = conn.SetReadDeadline(time.Time{}) From 6be73060a26be45d1bcbef25e0c0b8a4a0c58cad Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 12:29:24 +0200 Subject: [PATCH 3/9] Added comment for SetWSConcurrentRequestBytes, added matching test --- evmrpc/rpcstack.go | 3 +++ evmrpc/ws_admission_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/evmrpc/rpcstack.go b/evmrpc/rpcstack.go index 2b80b085f4..3d0203277b 100644 --- a/evmrpc/rpcstack.go +++ b/evmrpc/rpcstack.go @@ -386,6 +386,9 @@ func (h *HTTPServer) EnableWS(apis []rpc.API, config WsConfig) error { readLimit = math.MaxInt } srv.SetReadLimits(readLimit) + // maxConcurrentRequestBytes is passed through raw; rpc.Server.recomputeWSConcurrentBudget + // raises it to readLimit when smaller, matching + // newRequestSizeLimiter’s max(budget, maxBody) rule on the HTTP plane. srv.SetWSConcurrentRequestBytes(config.maxConcurrentRequestBytes) srv.SetWSAdmissionEventHook(func(reason string) { recordWSAdmissionRejected(context.Background(), reason) diff --git a/evmrpc/ws_admission_test.go b/evmrpc/ws_admission_test.go index 2d756f9938..5d5d49e982 100644 --- a/evmrpc/ws_admission_test.go +++ b/evmrpc/ws_admission_test.go @@ -70,6 +70,37 @@ func TestEnableWSConcurrentRequestBytes(t *testing.T) { require.Equal(t, json.Number("2"), secondResp.ID) } +func TestEnableWSConcurrentBudgetBelowReadLimitAdmitsMaxFrame(t *testing.T) { + const pad = 48 + + makeMsg := func(id int) string { + return fmt.Sprintf( + `{"jsonrpc":"2.0","id":%d,"method":"test_sleep","params":[0],"_pad":"%s"}`, + id, strings.Repeat("x", pad), + ) + } + payload := makeMsg(1) + frameSize := int64(len(payload)) + + srv := startWSTestServer(t, WsConfig{ + Origins: []string{"*"}, + RPCEndpointConfig: RPCEndpointConfig{ + readLimit: frameSize, + maxConcurrentRequestBytes: frameSize / 2, + }, + }) + + conn := dialWSTestServer(t, srv) + defer conn.Close() + + require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(payload))) + + var resp rpcResponse + readJSON(t, conn, &resp) + require.Equal(t, json.Number("1"), resp.ID) + require.Nil(t, resp.Error) +} + func TestEnableWSReadLimitDefault(t *testing.T) { srv := NewHTTPServer(rpc.DefaultHTTPTimeouts) wsConf := WsConfig{ From a5f1b6118d6a3b4b3532aa14e451112e71b1022b Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 12:32:34 +0200 Subject: [PATCH 4/9] Updated ws_admission_test for require.Equal --- evmrpc/ws_admission_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/evmrpc/ws_admission_test.go b/evmrpc/ws_admission_test.go index 5d5d49e982..247fb51411 100644 --- a/evmrpc/ws_admission_test.go +++ b/evmrpc/ws_admission_test.go @@ -224,7 +224,7 @@ func readJSON(t *testing.T, conn *websocket.Conn, dest *rpcResponse) { _ = conn.SetReadDeadline(time.Now().Add(time.Second)) msgType, data, err := conn.ReadMessage() - require.IsType(t, websocket.TextMessage, msgType) + require.Equal(t, websocket.TextMessage, msgType) require.NoError(t, err) require.NoError(t, json.Unmarshal(data, dest)) _ = conn.SetReadDeadline(time.Time{}) From 3a568d541ec280e5cbe1792f3fb49fde43c0c198 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 12:43:15 +0200 Subject: [PATCH 5/9] Reused effectiveMaxRequestBodyBytes method --- evmrpc/sei_legacy_http.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/evmrpc/sei_legacy_http.go b/evmrpc/sei_legacy_http.go index 8dfdea6ecb..d4f7331e6e 100644 --- a/evmrpc/sei_legacy_http.go +++ b/evmrpc/sei_legacy_http.go @@ -29,10 +29,7 @@ func wrapSeiLegacyHTTP(inner http.Handler, allowlist map[string]struct{}, maxBod if allowlist == nil { return inner } - if maxBody <= 0 { - maxBody = defaultMaxRequestBodyBytes - } - return &seiLegacyHTTPGate{inner: inner, allowlist: allowlist, maxBody: maxBody} + return &seiLegacyHTTPGate{inner: inner, allowlist: allowlist, maxBody: effectiveMaxRequestBodyBytes(maxBody)} } type seiLegacyHTTPGate struct { From 1f9d1f1ccb46fd9e7347f25892a1ffd1889c73b0 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 12:43:37 +0200 Subject: [PATCH 6/9] Added comment for context.Background --- evmrpc/rpcstack.go | 1 + 1 file changed, 1 insertion(+) diff --git a/evmrpc/rpcstack.go b/evmrpc/rpcstack.go index 3d0203277b..d9e4f9b3a4 100644 --- a/evmrpc/rpcstack.go +++ b/evmrpc/rpcstack.go @@ -391,6 +391,7 @@ func (h *HTTPServer) EnableWS(apis []rpc.API, config WsConfig) error { // newRequestSizeLimiter’s max(budget, maxBody) rule on the HTTP plane. srv.SetWSConcurrentRequestBytes(config.maxConcurrentRequestBytes) srv.SetWSAdmissionEventHook(func(reason string) { + // Hook carries no request context, and the fork's own connCtx is context.Background() too. recordWSAdmissionRejected(context.Background(), reason) }) logger.Info("Registering apis for evm websocket") From 63410e87065254d4eeeff71d3282cb6f844d8f2d Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 13:42:14 +0200 Subject: [PATCH 7/9] Addressing pr feedback --- evmrpc/ws_admission_test.go | 24 ++++++++++-------------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/evmrpc/ws_admission_test.go b/evmrpc/ws_admission_test.go index 247fb51411..ea9e6482c4 100644 --- a/evmrpc/ws_admission_test.go +++ b/evmrpc/ws_admission_test.go @@ -169,20 +169,16 @@ func TestWSAdmissionHookBudgetWaitTimeout(t *testing.T) { _, err := io.WriteString(p2, payload) require.NoError(t, err) - deadline := time.Now().Add(waitTimeout + 300*time.Millisecond) - for { + require.Eventually(t, func() bool { mu.Lock() - got := append([]string(nil), reasons...) - mu.Unlock() - if len(got) > 0 { - require.Equal(t, rpc.WSAdmissionReasonBudgetWaitTimeout, got[0]) - return - } - if time.Now().After(deadline) { - t.Fatal("expected admission hook to fire on budget wait timeout") - } - time.Sleep(10 * time.Millisecond) - } + defer mu.Unlock() + return len(reasons) > 0 + }, 2*time.Second, 10*time.Millisecond) + + mu.Lock() + got := append([]string(nil), reasons...) + mu.Unlock() + require.Equal(t, rpc.WSAdmissionReasonBudgetWaitTimeout, got[0]) } type rpcResponse struct { @@ -224,8 +220,8 @@ func readJSON(t *testing.T, conn *websocket.Conn, dest *rpcResponse) { _ = conn.SetReadDeadline(time.Now().Add(time.Second)) msgType, data, err := conn.ReadMessage() - require.Equal(t, websocket.TextMessage, msgType) require.NoError(t, err) require.NoError(t, json.Unmarshal(data, dest)) + require.Equal(t, websocket.TextMessage, msgType) _ = conn.SetReadDeadline(time.Time{}) } From 1e025cf4c17a9e79620cf965af7f8c54ec92408d Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 16:53:28 +0200 Subject: [PATCH 8/9] Wired wsAdmissionTimeout from config --- evmrpc/config/config.go | 22 ++++++++++++++++++-- evmrpc/config/config_test.go | 8 ++++++++ evmrpc/rpcstack.go | 8 +++++--- evmrpc/server.go | 1 + evmrpc/ws_admission_test.go | 40 ++++++++++++++++++++++++++++++++++++ 5 files changed, 74 insertions(+), 5 deletions(-) diff --git a/evmrpc/config/config.go b/evmrpc/config/config.go index 41ecb25f30..f37ac9b838 100644 --- a/evmrpc/config/config.go +++ b/evmrpc/config/config.go @@ -261,9 +261,15 @@ type Config struct { // WebSocket JSON-RPC request bodies admitted for processing concurrently // (independent budgets per plane). HTTP uses Content-Length weighting and // rejects over-budget requests fast (HTTP 429). WebSocket blocks until - // budget frees or times out. Set to 0 to disable the limit on either plane. + // budget frees or WSAdmissionTimeout elapses. Set to 0 to disable the limit + // on either plane. MaxConcurrentRequestBytes int64 `mapstructure:"max_concurrent_request_bytes"` + // WSAdmissionTimeout bounds how long a WebSocket connection waits for + // concurrent-byte budget to free before the next frame is read or committed. + // Zero or negative values use the go-ethereum default (30s). + WSAdmissionTimeout time.Duration `mapstructure:"ws_admission_timeout"` + // MaxOpenConnections caps the number of simultaneously accepted connections // on the EVM HTTP and WebSocket listeners. The limit is applied per listener // (HTTP and WS each get their own budget). Excess connections block in the @@ -327,6 +333,7 @@ var DefaultConfig = Config{ BatchResponseMaxSize: 25 * 1000 * 1000, // 25MB MaxRequestBodyBytes: 5 * 1024 * 1024, // 5 MiB (matches go-ethereum rpc default body limit) MaxConcurrentRequestBytes: 128 * 1024 * 1024, // 128 MiB of request bodies admitted concurrently + WSAdmissionTimeout: 30 * time.Second, // matches go-ethereum rpc defaultWSAdmissionTimeout MaxOpenConnections: 2000, } @@ -382,6 +389,7 @@ const ( flagBatchResponseMaxSize = "evm.batch_response_max_size" flagMaxRequestBodyBytes = "evm.max_request_body_bytes" flagMaxConcurrentRequestBytes = "evm.max_concurrent_request_bytes" + flagWSAdmissionTimeout = "evm.ws_admission_timeout" flagMaxOpenConnections = "evm.max_open_connections" ) @@ -652,6 +660,11 @@ func ReadConfig(opts servertypes.AppOptions) (Config, error) { return cfg, err } } + if v := opts.Get(flagWSAdmissionTimeout); v != nil { + if cfg.WSAdmissionTimeout, err = cast.ToDurationE(v); err != nil { + return cfg, err + } + } if v := opts.Get(flagMaxOpenConnections); v != nil { if cfg.MaxOpenConnections, err = cast.ToIntE(v); err != nil { return cfg, err @@ -926,9 +939,14 @@ max_request_body_bytes = {{ .EVM.MaxRequestBodyBytes }} # max_concurrent_request_bytes bounds total request bytes admitted concurrently # on HTTP (:8545) and WebSocket (:8546) (independent budgets per plane). HTTP # rejects over-budget requests fast (HTTP 429); WS blocks until budget frees or -# times out. Set to 0 to disable on either plane. +# ws_admission_timeout elapses. Set to 0 to disable on either plane. max_concurrent_request_bytes = {{ .EVM.MaxConcurrentRequestBytes }} +# ws_admission_timeout bounds how long a WebSocket connection waits for +# concurrent-byte budget before the next frame is read or committed. Zero or +# negative values use the go-ethereum default (30s). +ws_admission_timeout = "{{ .EVM.WSAdmissionTimeout }}" + # max_open_connections caps the number of simultaneously accepted connections on # the EVM HTTP and WebSocket listeners. Set to 0 to disable the limit. max_open_connections = {{ .EVM.MaxOpenConnections }} diff --git a/evmrpc/config/config_test.go b/evmrpc/config/config_test.go index baa061f58f..039ce91701 100644 --- a/evmrpc/config/config_test.go +++ b/evmrpc/config/config_test.go @@ -48,6 +48,7 @@ type opts struct { batchResponseMaxSize interface{} maxRequestBodyBytes interface{} maxConcurrentRequestBytes interface{} + wsAdmissionTimeout interface{} maxOpenConnections interface{} maxTraceStructLogBytes interface{} traceAllowedTracers interface{} @@ -189,6 +190,9 @@ func (o *opts) Get(k string) interface{} { if k == "evm.max_concurrent_request_bytes" { return o.maxConcurrentRequestBytes } + if k == "evm.ws_admission_timeout" { + return o.wsAdmissionTimeout + } if k == "evm.max_open_connections" { return o.maxOpenConnections } @@ -252,6 +256,7 @@ func getDefaultOpts() opts { 25 * 1000 * 1000, int64(5 * 1024 * 1024), int64(128 * 1024 * 1024), + 30 * time.Second, 2000, uint64(256 * 1024 * 1024), []string{"callTracer", "prestateTracer"}, @@ -519,15 +524,18 @@ func TestReadConfigRequestSizeLimits(t *testing.T) { require.NoError(t, err) require.Equal(t, config.DefaultConfig.MaxRequestBodyBytes, cfg.MaxRequestBodyBytes) require.Equal(t, config.DefaultConfig.MaxConcurrentRequestBytes, cfg.MaxConcurrentRequestBytes) + require.Equal(t, config.DefaultConfig.WSAdmissionTimeout, cfg.WSAdmissionTimeout) // Custom values (including 0 to use default / disable) flow through. o := getDefaultOpts() o.maxRequestBodyBytes = int64(1024) o.maxConcurrentRequestBytes = int64(0) + o.wsAdmissionTimeout = 10 * time.Second cfg, err = config.ReadConfig(&o) require.NoError(t, err) require.Equal(t, int64(1024), cfg.MaxRequestBodyBytes) require.Equal(t, int64(0), cfg.MaxConcurrentRequestBytes) + require.Equal(t, 10*time.Second, cfg.WSAdmissionTimeout) } func TestReadConfigMaxOpenConnections(t *testing.T) { diff --git a/evmrpc/rpcstack.go b/evmrpc/rpcstack.go index d9e4f9b3a4..870e08e42c 100644 --- a/evmrpc/rpcstack.go +++ b/evmrpc/rpcstack.go @@ -53,9 +53,10 @@ type HTTPConfig struct { // WsConfig is the JSON-RPC/Websocket configuration type WsConfig struct { - Origins []string - Modules []string - prefix string // path prefix on which to mount ws handler + Origins []string + Modules []string + prefix string // path prefix on which to mount ws handler + wsAdmissionTimeout time.Duration RPCEndpointConfig } @@ -390,6 +391,7 @@ func (h *HTTPServer) EnableWS(apis []rpc.API, config WsConfig) error { // raises it to readLimit when smaller, matching // newRequestSizeLimiter’s max(budget, maxBody) rule on the HTTP plane. srv.SetWSConcurrentRequestBytes(config.maxConcurrentRequestBytes) + srv.SetWSAdmissionTimeout(config.wsAdmissionTimeout) srv.SetWSAdmissionEventHook(func(reason string) { // Hook carries no request context, and the fork's own connCtx is context.Background() too. recordWSAdmissionRejected(context.Background(), reason) diff --git a/evmrpc/server.go b/evmrpc/server.go index 243494b4d3..1cd9c8fdfa 100644 --- a/evmrpc/server.go +++ b/evmrpc/server.go @@ -340,6 +340,7 @@ func NewEVMWebSocketServer( wsConfig := WsConfig{Origins: strings.Split(config.WSOrigins, ",")} wsConfig.readLimit = config.MaxRequestBodyBytes wsConfig.maxConcurrentRequestBytes = config.MaxConcurrentRequestBytes + wsConfig.wsAdmissionTimeout = config.WSAdmissionTimeout wsConfig.batchItemLimit = config.BatchRequestLimit wsConfig.batchResponseSizeLimit = config.BatchResponseMaxSize if err := httpServer.EnableWS(apis, wsConfig); err != nil { diff --git a/evmrpc/ws_admission_test.go b/evmrpc/ws_admission_test.go index ea9e6482c4..f4be05dad0 100644 --- a/evmrpc/ws_admission_test.go +++ b/evmrpc/ws_admission_test.go @@ -116,6 +116,46 @@ func TestEnableWSReadLimitDefault(t *testing.T) { require.Equal(t, defaultMaxRequestBodyBytes, effectiveMaxRequestBodyBytes(wsConf.readLimit)) } +func TestEnableWSAdmissionTimeout(t *testing.T) { + const ( + sleepDuration = 200 * time.Millisecond + pad = 48 + waitTimeout = 50 * time.Millisecond + ) + + makeMsg := func(id int) string { + return fmt.Sprintf( + `{"jsonrpc":"2.0","id":%d,"method":"test_sleep","params":[%d],"_pad":"%s"}`, + id, sleepDuration.Nanoseconds(), strings.Repeat("x", pad), + ) + } + payload := makeMsg(1) + frameSize := int64(len(payload)) + + srv := startWSTestServer(t, WsConfig{ + Origins: []string{"*"}, + wsAdmissionTimeout: waitTimeout, + RPCEndpointConfig: RPCEndpointConfig{ + readLimit: frameSize, + maxConcurrentRequestBytes: frameSize, + }, + }) + + conn := dialWSTestServer(t, srv) + defer conn.Close() + + require.NoError(t, conn.WriteMessage(websocket.TextMessage, []byte(payload))) + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + _, _, err := conn.ReadMessage() + if err != nil { + return + } + } + t.Fatal("expected websocket connection to close after admission timeout") +} + func TestWSAdmissionHookBudgetWaitTimeout(t *testing.T) { const ( sleepDuration = 200 * time.Millisecond From 6f5a17020731aef3ff0600a960d45d6459d4381c Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Tue, 28 Jul 2026 17:33:42 +0200 Subject: [PATCH 9/9] Renamed plane to protocol --- evmrpc/config/config.go | 8 ++++---- evmrpc/metrics.go | 12 ++++++------ evmrpc/rpcstack.go | 2 +- ratelimiter/registry.go | 2 +- 4 files changed, 12 insertions(+), 12 deletions(-) diff --git a/evmrpc/config/config.go b/evmrpc/config/config.go index f37ac9b838..eb1f7d065a 100644 --- a/evmrpc/config/config.go +++ b/evmrpc/config/config.go @@ -259,10 +259,10 @@ type Config struct { // MaxConcurrentRequestBytes bounds the total size, in bytes, of HTTP and // WebSocket JSON-RPC request bodies admitted for processing concurrently - // (independent budgets per plane). HTTP uses Content-Length weighting and + // (independent budgets per protocol). HTTP uses Content-Length weighting and // rejects over-budget requests fast (HTTP 429). WebSocket blocks until // budget frees or WSAdmissionTimeout elapses. Set to 0 to disable the limit - // on either plane. + // on either protocol. MaxConcurrentRequestBytes int64 `mapstructure:"max_concurrent_request_bytes"` // WSAdmissionTimeout bounds how long a WebSocket connection waits for @@ -937,9 +937,9 @@ batch_response_max_size = {{ .EVM.BatchResponseMaxSize }} max_request_body_bytes = {{ .EVM.MaxRequestBodyBytes }} # max_concurrent_request_bytes bounds total request bytes admitted concurrently -# on HTTP (:8545) and WebSocket (:8546) (independent budgets per plane). HTTP +# on HTTP (:8545) and WebSocket (:8546) (independent budgets per protocol). HTTP # rejects over-budget requests fast (HTTP 429); WS blocks until budget frees or -# ws_admission_timeout elapses. Set to 0 to disable on either plane. +# ws_admission_timeout elapses. Set to 0 to disable on either protocol. max_concurrent_request_bytes = {{ .EVM.MaxConcurrentRequestBytes }} # ws_admission_timeout bounds how long a WebSocket connection waits for diff --git a/evmrpc/metrics.go b/evmrpc/metrics.go index 1e6fc7cf98..e12a3e7ff6 100644 --- a/evmrpc/metrics.go +++ b/evmrpc/metrics.go @@ -18,9 +18,9 @@ const ( errorClassKey = "error_class" jsonrpcCodeKey = "jsonrpc_code" rejectReasonKey = "reason" - planeKey = "plane" - planeHTTP = "http" - planeWS = "ws" + protocolKey = "protocol" + protocolHTTP = "http" + protocolWS = "ws" // reject reason values for requestRejectedCount. rejectReasonOversize = "oversize" // body exceeded max_request_body_bytes rejectReasonBusy = "busy" // max_concurrent_request_bytes budget exhausted @@ -79,7 +79,7 @@ var ( )), requestRejectedCount: must(rpcTelemetryMeter.Int64Counter( "evmrpc_requests_rejected_total", - metric.WithDescription("Number of JSON-RPC requests rejected by admission control (labeled by plane and reason)"), + metric.WithDescription("Number of JSON-RPC requests rejected by admission control (labeled by protocol and reason)"), metric.WithUnit("{count}"), )), } @@ -172,7 +172,7 @@ func recordHistoricalDebugTraceAttempt(ctx context.Context, endpoint, connection func recordRequestRejected(ctx context.Context, reason string) { metrics.requestRejectedCount.Add(ctx, 1, metric.WithAttributes( - attribute.String(planeKey, planeHTTP), + attribute.String(protocolKey, protocolHTTP), attribute.String(rejectReasonKey, reason), ), ) @@ -181,7 +181,7 @@ func recordRequestRejected(ctx context.Context, reason string) { func recordWSAdmissionRejected(ctx context.Context, reason string) { metrics.requestRejectedCount.Add(ctx, 1, metric.WithAttributes( - attribute.String(planeKey, planeWS), + attribute.String(protocolKey, protocolWS), attribute.String(rejectReasonKey, reason), ), ) diff --git a/evmrpc/rpcstack.go b/evmrpc/rpcstack.go index 870e08e42c..c6d3dd8f76 100644 --- a/evmrpc/rpcstack.go +++ b/evmrpc/rpcstack.go @@ -389,7 +389,7 @@ func (h *HTTPServer) EnableWS(apis []rpc.API, config WsConfig) error { srv.SetReadLimits(readLimit) // maxConcurrentRequestBytes is passed through raw; rpc.Server.recomputeWSConcurrentBudget // raises it to readLimit when smaller, matching - // newRequestSizeLimiter’s max(budget, maxBody) rule on the HTTP plane. + // newRequestSizeLimiter’s max(budget, maxBody) rule on the HTTP protocol. srv.SetWSConcurrentRequestBytes(config.maxConcurrentRequestBytes) srv.SetWSAdmissionTimeout(config.wsAdmissionTimeout) srv.SetWSAdmissionEventHook(func(reason string) { diff --git a/ratelimiter/registry.go b/ratelimiter/registry.go index cddf8d20fe..74e2e2f6f9 100644 --- a/ratelimiter/registry.go +++ b/ratelimiter/registry.go @@ -86,7 +86,7 @@ func New(cfg Config) (*Registry, error) { }, nil } -// Allow reports whether the request from ip should be allowed for the given plane. +// Allow reports whether the request from ip should be allowed for the given protocol. // Rejections increment rpc_rate_limit_rejected_total{plane}. func (r *Registry) Allow(ctx context.Context, ip, plane string) bool { if r.cfg.RPS <= 0 || r.cfg.Burst <= 0 {