From ea3a257c1d62056acdb882c7389ad6bffcea6bcb Mon Sep 17 00:00:00 2001 From: "Masih H. Derkani" Date: Thu, 27 Aug 2026 18:30:20 +0100 Subject: [PATCH] Bound frozen RPC router batch allocations Limit JSON-RPC batches to 1,000 calls by default and reject oversized batches during streaming decode with one bounded error response. Add a configurable 30-second HTTP write timeout to release slow-client handlers. --- cmd/frozen-rpc-router/README.md | 4 ++ cmd/frozen-rpc-router/config.go | 12 +++++ cmd/frozen-rpc-router/config_test.go | 38 ++++++++++++++++ cmd/frozen-rpc-router/main.go | 3 +- cmd/frozen-rpc-router/router.go | 66 ++++++++++++++++++++++------ cmd/frozen-rpc-router/router_test.go | 36 ++++++++++++--- 6 files changed, 139 insertions(+), 20 deletions(-) diff --git a/cmd/frozen-rpc-router/README.md b/cmd/frozen-rpc-router/README.md index eebb8b1e60..31d2fee62c 100644 --- a/cmd/frozen-rpc-router/README.md +++ b/cmd/frozen-rpc-router/README.md @@ -30,3 +30,7 @@ the live node. Single-backend HTTP responses include `Sei-RPC-Route: frozen:` or `Sei-RPC-Route: live`. A batch split across backends returns `Sei-RPC-Route: mixed`. + +JSON-RPC batches are limited to 1,000 calls by default. Use +`--batch-request-limit` to change the limit. HTTP response writes time out after +30 seconds by default and can be configured with `--write-timeout`. diff --git a/cmd/frozen-rpc-router/config.go b/cmd/frozen-rpc-router/config.go index 03ba0c1a5f..aa29b0a0fd 100644 --- a/cmd/frozen-rpc-router/config.go +++ b/cmd/frozen-rpc-router/config.go @@ -15,6 +15,8 @@ const ( defaultListenAddress = "127.0.0.1:8545" defaultMaxRequestBodySize = int64(5 << 20) defaultMaxBlockReferenceDepth = 16 + defaultBatchRequestLimit = 1000 + defaultWriteTimeout = 30 * time.Second defaultShutdownTimeout = 10 * time.Second ) @@ -24,6 +26,8 @@ type config struct { frozenNodes frozenNodeFlags maxRequestBodySize int64 maxBlockReferenceDepth int + batchRequestLimit int + writeTimeout time.Duration shutdownTimeout time.Duration } @@ -52,6 +56,8 @@ func parseConfig(args []string, output io.Writer) (config, error) { flags.Var(&cfg.frozenNodes, "frozen-node", "freeze-height=ip:port pair; repeat once per frozen node") flags.Int64Var(&cfg.maxRequestBodySize, "max-request-body-bytes", defaultMaxRequestBodySize, "maximum JSON-RPC request body size") flags.IntVar(&cfg.maxBlockReferenceDepth, "max-block-reference-depth", defaultMaxBlockReferenceDepth, "maximum nested block reference depth") + flags.IntVar(&cfg.batchRequestLimit, "batch-request-limit", defaultBatchRequestLimit, "maximum number of calls in a JSON-RPC batch") + flags.DurationVar(&cfg.writeTimeout, "write-timeout", defaultWriteTimeout, "maximum duration for writing an HTTP response") flags.DurationVar(&cfg.shutdownTimeout, "shutdown-timeout", defaultShutdownTimeout, "graceful shutdown timeout") if err := flags.Parse(args); err != nil { return config{}, err @@ -68,6 +74,12 @@ func parseConfig(args []string, output io.Writer) (config, error) { if cfg.maxBlockReferenceDepth <= 0 { return config{}, errors.New("--max-block-reference-depth must be positive") } + if cfg.batchRequestLimit <= 0 { + return config{}, errors.New("--batch-request-limit must be positive") + } + if cfg.writeTimeout <= 0 { + return config{}, errors.New("--write-timeout must be positive") + } if cfg.shutdownTimeout <= 0 { return config{}, errors.New("--shutdown-timeout must be positive") } diff --git a/cmd/frozen-rpc-router/config_test.go b/cmd/frozen-rpc-router/config_test.go index f7151208ec..a2f3d21385 100644 --- a/cmd/frozen-rpc-router/config_test.go +++ b/cmd/frozen-rpc-router/config_test.go @@ -3,6 +3,7 @@ package main import ( "io" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -14,11 +15,15 @@ func TestParseConfig(t *testing.T) { "--frozen-node", "200=localhost:8547", "--frozen-node", "100=localhost:8546", "--max-block-reference-depth", "32", + "--batch-request-limit", "50", + "--write-timeout", "45s", }, io.Discard) require.NoError(t, err) require.Equal(t, "0.0.0.0:9000", cfg.listenAddress) require.Equal(t, "localhost:8545", cfg.liveNode) require.Equal(t, 32, cfg.maxBlockReferenceDepth) + require.Equal(t, 50, cfg.batchRequestLimit) + require.Equal(t, 45*time.Second, cfg.writeTimeout) nodes, err := parseFrozenNodes(cfg.frozenNodes) require.NoError(t, err) @@ -38,6 +43,39 @@ func TestParseConfigRejectsNonPositiveBlockReferenceDepth(t *testing.T) { require.EqualError(t, err, "--max-block-reference-depth must be positive") } +func TestParseConfigUsesResourceLimitDefaults(t *testing.T) { + cfg, err := parseConfig([]string{"--live-node", "localhost:8545"}, io.Discard) + require.NoError(t, err) + require.Equal(t, defaultBatchRequestLimit, cfg.batchRequestLimit) + require.Equal(t, 30*time.Second, cfg.writeTimeout) +} + +func TestParseConfigRejectsNonPositiveResourceLimits(t *testing.T) { + testCases := []struct { + name string + args []string + wantErr string + }{ + { + name: "batch request limit", + args: []string{"--batch-request-limit", "0"}, + wantErr: "--batch-request-limit must be positive", + }, + { + name: "write timeout", + args: []string{"--write-timeout", "0s"}, + wantErr: "--write-timeout must be positive", + }, + } + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + args := append([]string{"--live-node", "localhost:8545"}, testCase.args...) + _, err := parseConfig(args, io.Discard) + require.EqualError(t, err, testCase.wantErr) + }) + } +} + func TestParseFrozenNodesRejectsInvalidPairs(t *testing.T) { for _, value := range []string{"100", "=localhost:8545", "0=localhost:8545", "abc=localhost:8545", "9223372036854775808=localhost:8545", "100="} { t.Run(value, func(t *testing.T) { diff --git a/cmd/frozen-rpc-router/main.go b/cmd/frozen-rpc-router/main.go index a18ab64868..51b57cbbc1 100644 --- a/cmd/frozen-rpc-router/main.go +++ b/cmd/frozen-rpc-router/main.go @@ -32,7 +32,7 @@ func run() error { if err != nil { return err } - router, err := newRouter(cfg.liveNode, frozenNodes, nil, cfg.maxRequestBodySize, cfg.maxBlockReferenceDepth) + router, err := newRouter(cfg.liveNode, frozenNodes, nil, cfg.maxRequestBodySize, cfg.maxBlockReferenceDepth, cfg.batchRequestLimit) if err != nil { return err } @@ -41,6 +41,7 @@ func run() error { Addr: cfg.listenAddress, Handler: router, ReadHeaderTimeout: 10 * time.Second, + WriteTimeout: cfg.writeTimeout, IdleTimeout: 2 * time.Minute, } diff --git a/cmd/frozen-rpc-router/router.go b/cmd/frozen-rpc-router/router.go index 2cc6f58ca2..dbe1899cee 100644 --- a/cmd/frozen-rpc-router/router.go +++ b/cmd/frozen-rpc-router/router.go @@ -48,12 +48,15 @@ var blockParameterIndexes = map[string]int{ "eth_getUncleCountByBlockNumber": 0, } +var errBatchTooLarge = errors.New("batch too large") + type router struct { live *upstream frozen []*upstream client *http.Client maxRequestBodySize int64 maxBlockReferenceDepth int + batchRequestLimit int liveProxy *httputil.ReverseProxy } @@ -95,7 +98,7 @@ type batchGroup struct { err error } -func newRouter(liveAddress string, frozenConfigs []frozenNodeConfig, client *http.Client, maxRequestBodySize int64, maxBlockReferenceDepth int) (*router, error) { +func newRouter(liveAddress string, frozenConfigs []frozenNodeConfig, client *http.Client, maxRequestBodySize int64, maxBlockReferenceDepth, batchRequestLimit int) (*router, error) { liveURL, err := parseEndpoint(liveAddress) if err != nil { return nil, fmt.Errorf("invalid live node: %w", err) @@ -109,6 +112,9 @@ func newRouter(liveAddress string, frozenConfigs []frozenNodeConfig, client *htt if maxBlockReferenceDepth <= 0 { return nil, errors.New("maximum block reference depth must be positive") } + if batchRequestLimit <= 0 { + return nil, errors.New("batch request limit must be positive") + } live := &upstream{endpoint: liveURL} frozen := make([]*upstream, 0, len(frozenConfigs)) @@ -139,6 +145,7 @@ func newRouter(liveAddress string, frozenConfigs []frozenNodeConfig, client *htt client: client, maxRequestBodySize: maxRequestBodySize, maxBlockReferenceDepth: maxBlockReferenceDepth, + batchRequestLimit: batchRequestLimit, liveProxy: liveProxy, }, nil } @@ -219,25 +226,20 @@ func (r *router) serveSingle(w http.ResponseWriter, request *http.Request, body } func (r *router) serveBatch(w http.ResponseWriter, request *http.Request, body []byte) { - var rawCalls []json.RawMessage - if err := json.Unmarshal(body, &rawCalls); err != nil { + calls, err := decodeBatchCalls(body, r.batchRequestLimit) + if errors.Is(err, errBatchTooLarge) { + writeRPCError(w, nil, rpcError{Code: jsonRPCInvalidRequest, Message: "batch too large"}) + return + } + if err != nil { writeRPCError(w, nil, rpcError{Code: jsonRPCParseError, Message: "parse error"}) return } - if len(rawCalls) == 0 { + if len(calls) == 0 { writeRPCError(w, nil, rpcError{Code: jsonRPCInvalidRequest, Message: "invalid request"}) return } - calls := make([]rpcCall, 0, len(rawCalls)) - for _, raw := range rawCalls { - call, err := decodeCall(raw) - if err != nil { - call = rpcCall{raw: raw} - } - calls = append(calls, call) - } - if target, ok := r.singleBatchTarget(calls); ok { if err := r.proxy(w, request, target, body); err != nil { writeBatchUpstreamErrors(w, calls) @@ -259,6 +261,44 @@ func (r *router) serveBatch(w http.ResponseWriter, request *http.Request, body [ writeBatchResponses(w, responses) } +func decodeBatchCalls(body []byte, limit int) ([]rpcCall, error) { + decoder := json.NewDecoder(bytes.NewReader(body)) + opening, err := decoder.Token() + if err != nil { + return nil, err + } + if opening != json.Delim('[') { + return nil, errors.New("batch must be an array") + } + + var calls []rpcCall + for decoder.More() { + if len(calls) == limit { + return nil, errBatchTooLarge + } + var raw json.RawMessage + if err := decoder.Decode(&raw); err != nil { + return nil, err + } + call, err := decodeCall(raw) + if err != nil { + call = rpcCall{raw: raw} + } + calls = append(calls, call) + } + if _, err := decoder.Token(); err != nil { + return nil, err + } + var trailing json.RawMessage + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return nil, errors.New("batch has trailing JSON") + } + return nil, err + } + return calls, nil +} + func decodeCall(raw json.RawMessage) (rpcCall, error) { call := rpcCall{raw: raw} var fields map[string]json.RawMessage diff --git a/cmd/frozen-rpc-router/router_test.go b/cmd/frozen-rpc-router/router_test.go index dd592d3a2c..dd077d36fc 100644 --- a/cmd/frozen-rpc-router/router_test.go +++ b/cmd/frozen-rpc-router/router_test.go @@ -99,7 +99,7 @@ func TestRouteRanges(t *testing.T) { func TestRouterForwardsSingleRequest(t *testing.T) { live := newRPCBackend(t, "live") frozen := newRPCBackend(t, "frozen") - r, err := newRouter(live.server.URL, []frozenNodeConfig{{freezeHeight: 100, address: frozen.server.URL}}, live.server.Client(), defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth) + r, err := newRouter(live.server.URL, []frozenNodeConfig{{freezeHeight: 100, address: frozen.server.URL}}, live.server.Client(), defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth, defaultBatchRequestLimit) require.NoError(t, err) recorder := httptest.NewRecorder() @@ -121,7 +121,7 @@ func TestRouterSplitsMixedBatch(t *testing.T) { r, err := newRouter(live.server.URL, []frozenNodeConfig{ {freezeHeight: 200, address: frozen200.server.URL}, {freezeHeight: 100, address: frozen100.server.URL}, - }, live.server.Client(), defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth) + }, live.server.Client(), defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth, defaultBatchRequestLimit) require.NoError(t, err) body := `[ @@ -173,8 +173,29 @@ func TestRouterOmitsErrorForUnsupportedNotification(t *testing.T) { require.Empty(t, recorder.Body.String()) } +func TestRouterRejectsOversizedBatchWithSingleError(t *testing.T) { + r, err := newRouter("live:8545", nil, nil, defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth, defaultBatchRequestLimit) + require.NoError(t, err) + recorder := httptest.NewRecorder() + body := "[" + strings.Repeat("{},", defaultBatchRequestLimit) + "{}]" + request := httptest.NewRequest(http.MethodPost, "http://router/", strings.NewReader(body)) + r.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) + require.JSONEq(t, `{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"batch too large"}}`, recorder.Body.String()) +} + +func TestDecodeBatchCallsAcceptsLimit(t *testing.T) { + calls, err := decodeBatchCalls([]byte(`[{}, {}]`), 2) + require.NoError(t, err) + require.Len(t, calls, 2) + + _, err = decodeBatchCalls([]byte(`[{}, {}, {}]`), 2) + require.ErrorIs(t, err, errBatchTooLarge) +} + func TestRouterRejectsOversizedRequest(t *testing.T) { - r, err := newRouter("live:8545", nil, nil, 8, defaultMaxBlockReferenceDepth) + r, err := newRouter("live:8545", nil, nil, 8, defaultMaxBlockReferenceDepth, defaultBatchRequestLimit) require.NoError(t, err) recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodPost, "http://router/", bytes.NewReader([]byte("123456789"))) @@ -190,11 +211,14 @@ func TestNewRouterSortsAndValidatesFrozenNodes(t *testing.T) { _, err := newRouter("live:8545", []frozenNodeConfig{ {freezeHeight: 100, address: "one:8545"}, {freezeHeight: 100, address: "two:8545"}, - }, nil, defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth) + }, nil, defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth, defaultBatchRequestLimit) require.EqualError(t, err, "duplicate freeze height 100") - _, err = newRouter("live:8545", nil, nil, defaultMaxRequestBodySize, 0) + _, err = newRouter("live:8545", nil, nil, defaultMaxRequestBodySize, 0, defaultBatchRequestLimit) require.EqualError(t, err, "maximum block reference depth must be positive") + + _, err = newRouter("live:8545", nil, nil, defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth, 0) + require.EqualError(t, err, "batch request limit must be positive") } func newTestRouter(t *testing.T) *router { @@ -202,7 +226,7 @@ func newTestRouter(t *testing.T) *router { r, err := newRouter("live:8545", []frozenNodeConfig{ {freezeHeight: 200, address: "frozen-200:8545"}, {freezeHeight: 100, address: "frozen-100:8545"}, - }, nil, defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth) + }, nil, defaultMaxRequestBodySize, defaultMaxBlockReferenceDepth, defaultBatchRequestLimit) require.NoError(t, err) return r }