diff --git a/.github/workflows/integration-test-matrix.json b/.github/workflows/integration-test-matrix.json index b3c32d8d2f..bc44a7ad94 100644 --- a/.github/workflows/integration-test-matrix.json +++ b/.github/workflows/integration-test-matrix.json @@ -237,6 +237,13 @@ "./integration_test/evm_module/scripts/evm_rpc_tests.sh" ] }, + { + "name": "Frozen RPC Router", + "cluster": "frozen-rpc-router", + "scripts": [ + "go test -tags frozen_rpc_integration -v -count=1 -timeout 5m ./integration_test/frozen_rpc_router/..." + ] + }, { "name": "EVM RPC Parity (geth reference)", "env": "GIGA_STORAGE=true", diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index 29a377fe17..26d4bdc656 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -132,7 +132,7 @@ jobs: - uses: actions/setup-go@v6 with: go-version: '1.25.6' - - name: Build seid binary in localnode container + - name: Build seid and frozen RPC router binaries in localnode container env: DOCKER_PLATFORM: linux/amd64 run: make build-seid-in-localnode-ci @@ -142,7 +142,7 @@ jobs: docker tag sei-chain/localnode "${GHCR_LOCALNODE}:${{ github.run_id }}" push_with_retry "${GHCR_LOCALNODE}:${{ github.run_id }}" - name: Package CI artifacts - run: tar -czf integration-build.tar.gz build/seid + run: tar -czf integration-build.tar.gz build/seid build/frozen-rpc-router - name: Upload integration CI artifacts uses: actions/upload-artifact@v4 with: @@ -206,7 +206,7 @@ jobs: - name: Load prebuilt seid and pull Docker images run: | tar -xzf integration-build.tar.gz - # The seid binary is now extracted to build/seid; drop the + # The binaries are now extracted under build/; drop the # tarball so its copy does not sit on disk for the rest of the job. rm -f integration-build.tar.gz source .github/scripts/docker-registry-retry.sh @@ -218,8 +218,13 @@ jobs: # retagged as sei-chain/*; keeps a single tag per image. docker image rm "${GHCR_LOCALNODE}:${{ github.run_id }}" "${GHCR_RPCNODE}:${{ github.run_id }}" || true - name: Start 4 node docker cluster + if: ${{ matrix.test.cluster != 'frozen-rpc-router' }} run: DOCKER_DETACH=true INVARIANT_CHECK_INTERVAL=10 ${{matrix.test.env}} make docker-cluster-start-ci + - name: Start frozen RPC router docker cluster + if: ${{ matrix.test.cluster == 'frozen-rpc-router' }} + run: DOCKER_DETACH=true INVARIANT_CHECK_INTERVAL=10 make docker-frozen-rpc-cluster-start-ci + - name: Wait for docker cluster to start run: | until [ $(cat build/generated/launch.complete |wc -l) = 4 ] @@ -229,9 +234,11 @@ jobs: sleep 10 - name: Start rpc node + if: ${{ matrix.test.cluster != 'frozen-rpc-router' }} run: ${{matrix.test.env}} make run-rpc-node-integration-ci & - name: Verify Sei Chain is running + if: ${{ matrix.test.cluster != 'frozen-rpc-router' }} run: go test -tags yaml_integration -v -timeout 5m ./integration_test/runner/... -run TestStartup - name: ${{ matrix.test.name }} @@ -259,6 +266,8 @@ jobs: tail -200 "build/generated/logs/seid-${NODE_ID}.log" || true fi done + echo "==================== sei-frozen-rpc-router ====================" + docker logs --tail 200 sei-frozen-rpc-router || true - name: Prepare log artifact name if: ${{ always() }} @@ -356,7 +365,7 @@ jobs: - name: Load prebuilt seid and pull Docker images run: | tar -xzf integration-build.tar.gz - # The seid binary is now extracted to build/seid; drop the + # The binaries are now extracted under build/; drop the # tarball so its copy does not sit on disk for the rest of the job. rm -f integration-build.tar.gz source .github/scripts/docker-registry-retry.sh diff --git a/Makefile b/Makefile index 16b333e95c..9ded00c1bf 100644 --- a/Makefile +++ b/Makefile @@ -200,6 +200,11 @@ build: go build $(BUILD_FLAGS) -o ./build/seid ./cmd/seid .PHONY: build +build-frozen-rpc-router: + mkdir -p ./build + go build -o ./build/frozen-rpc-router ./cmd/frozen-rpc-router +.PHONY: build-frozen-rpc-router + build-verbose: mkdir -p ./build go build -x -v $(BUILD_FLAGS) -o ./build/seid ./cmd/seid @@ -268,7 +273,7 @@ build-seid-in-localnode: build-docker-node -w /sei-protocol/sei-chain \ -e LEDGER_ENABLED=false \ sei-chain/localnode \ - bash -c 'export PATH=/usr/local/go/bin:$$PATH && make clean && make build-linux && mkdir -p build/generated && echo DONE > build/generated/build.complete' + bash -c 'export PATH=/usr/local/go/bin:$$PATH && make clean && make build-linux && make build-frozen-rpc-router && mkdir -p build/generated && echo DONE > build/generated/build.complete' .PHONY: build-seid-in-localnode # CI variant: assumes localnode image already built by Buildx in prepare-cluster (skips docker build). @@ -283,10 +288,10 @@ build-seid-in-localnode-ci: ensure-integration-ci-images -w /sei-protocol/sei-chain \ -e LEDGER_ENABLED=false \ sei-chain/localnode \ - bash -c 'export PATH=/usr/local/go/bin:$$PATH && make clean && make build-linux && mkdir -p build/generated && echo DONE > build/generated/build.complete' + bash -c 'export PATH=/usr/local/go/bin:$$PATH && make clean && make build-linux && make build-frozen-rpc-router && mkdir -p build/generated && echo DONE > build/generated/build.complete' .PHONY: build-seid-in-localnode-ci -# Images + seid binary for integration-test CI (see .github/workflows/integration-test.yml). +# Images plus seid and frozen-rpc-router binaries for integration-test CI. # build-seid-in-localnode already depends on build-docker-node, so omit it here to avoid building localnode twice. build-integration-ci-artifacts: build-rpc-node build-seid-in-localnode .PHONY: build-integration-ci-artifacts @@ -398,6 +403,8 @@ CLUSTER_ENV_VARS = DOCKER_PLATFORM=$(DOCKER_PLATFORM) USERID=$(shell id -u) GROU GIGA_MIGRATE_FROM_MEMIAVL=$(GIGA_MIGRATE_FROM_MEMIAVL) \ GIGA_FLATKV_ONLY=$(GIGA_FLATKV_ONLY) +FROZEN_RPC_COMPOSE_FILES = -f docker-compose.yml -f docker-compose.frozen-rpc-router.yml + # Run a 4-node docker containers docker-cluster-start: docker-cluster-stop build-docker-node @rm -rf $(PROJECT_HOME)/build/generated @@ -445,6 +452,48 @@ docker-cluster-stop: @cd docker && DOCKER_PLATFORM=$(DOCKER_PLATFORM) USERID=$(shell id -u) GROUPID=$(shell id -g) GOCACHE=$(shell go env GOCACHE) docker compose down .PHONY: localnet-stop +# Run the localnet with two non-validating archival nodes frozen before heights +# 10 and 20, two live validators, and frozen-rpc-router published on port 8553. +docker-frozen-rpc-cluster-start: docker-frozen-rpc-cluster-stop build-docker-node + @rm -rf $(PROJECT_HOME)/build/generated + @mkdir -p $(shell go env GOMODCACHE) + @mkdir -p $(shell go env GOCACHE) + @cd docker && \ + if [ "$${DOCKER_DETACH:-}" = "true" ]; then \ + DETACH_FLAG="-d"; \ + else \ + DETACH_FLAG=""; \ + fi; \ + $(CLUSTER_ENV_VARS) docker compose $(FROZEN_RPC_COMPOSE_FILES) up $$DETACH_FLAG +.PHONY: docker-frozen-rpc-cluster-start + +# CI variant: use the prebuilt image and binaries from integration-build.tar.gz. +docker-frozen-rpc-cluster-start-ci: docker-frozen-rpc-cluster-stop ensure-integration-ci-images + @rm -rf $(PROJECT_HOME)/build/generated + @test -f $(PROJECT_HOME)/build/seid || (echo "build/seid missing; download integration-build.tar.gz from prepare-cluster" && exit 1) + @test -f $(PROJECT_HOME)/build/frozen-rpc-router || (echo "build/frozen-rpc-router missing; download integration-build.tar.gz from prepare-cluster" && exit 1) + @mkdir -p $(shell go env GOMODCACHE) + @mkdir -p $(shell go env GOCACHE) + @cd docker && \ + if [ "$${DOCKER_DETACH:-}" = "true" ]; then \ + DETACH_FLAG="-d"; \ + else \ + DETACH_FLAG=""; \ + fi; \ + $(CLUSTER_ENV_VARS) SKIP_BUILD=true docker compose $(FROZEN_RPC_COMPOSE_FILES) up $$DETACH_FLAG +.PHONY: docker-frozen-rpc-cluster-start-ci + +docker-frozen-rpc-cluster-stop: + @cd docker && DOCKER_PLATFORM=$(DOCKER_PLATFORM) USERID=$(shell id -u) GROUPID=$(shell id -g) GOCACHE=$(shell go env GOCACHE) docker compose $(FROZEN_RPC_COMPOSE_FILES) down --remove-orphans +.PHONY: docker-frozen-rpc-cluster-stop + +frozen-rpc-router-integration-test: + @set -e; \ + trap '$(MAKE) docker-frozen-rpc-cluster-stop' EXIT; \ + DOCKER_DETACH=true $(MAKE) docker-frozen-rpc-cluster-start; \ + go test -tags frozen_rpc_integration -v -count=1 -timeout 5m ./integration_test/frozen_rpc_router/... +.PHONY: frozen-rpc-router-integration-test + # Start 4-node cluster with Prometheus and Grafana monitoring docker-cluster-start-monitoring: docker-cluster-stop-monitoring build-docker-node @rm -rf $(PROJECT_HOME)/build/generated diff --git a/cmd/frozen-rpc-router/README.md b/cmd/frozen-rpc-router/README.md new file mode 100644 index 0000000000..eebb8b1e60 --- /dev/null +++ b/cmd/frozen-rpc-router/README.md @@ -0,0 +1,32 @@ +# Frozen RPC router + +`frozen-rpc-router` exposes an HTTP EVM JSON-RPC endpoint backed by a live node +and any number of nodes running with `freeze-height`. + +A freeze height is an exclusive boundary. A node started with +`freeze-height = 100` serves blocks through height 99, so the router sends +height 99 to that node and height 100 to the next configured interval (or the +live node). + +```sh +go run ./cmd/frozen-rpc-router \ + --listen-address 0.0.0.0:8545 \ + --live-node localhost:9545 \ + --frozen-node 1000000=localhost:9546 \ + --frozen-node 2000000=10.0.0.12:8545 +``` + +Repeat `--frozen-node` for every `freeze-height=ip:port` pair. HTTP and HTTPS +URLs are also accepted. Frozen nodes may be listed in any order, but freeze +heights must be unique. + +Methods with explicit numeric block parameters are routed to the matching +interval. `eth_getLogs` and `eth_feeHistory` are routed only when their entire +explicit range belongs to one interval; ranges crossing an interval boundary +return JSON-RPC error `-32000`. Latest-style block tags, methods without block +numbers, stateful filter methods, subscriptions, and WebSocket connections use +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`. diff --git a/cmd/frozen-rpc-router/config.go b/cmd/frozen-rpc-router/config.go new file mode 100644 index 0000000000..d2c56f6abe --- /dev/null +++ b/cmd/frozen-rpc-router/config.go @@ -0,0 +1,88 @@ +package main + +import ( + "errors" + "flag" + "fmt" + "io" + "math" + "strconv" + "strings" + "time" +) + +const ( + defaultListenAddress = "127.0.0.1:8545" + defaultMaxRequestBodySize = int64(5 << 20) + defaultShutdownTimeout = 10 * time.Second +) + +type config struct { + listenAddress string + liveNode string + frozenNodes frozenNodeFlags + maxRequestBodySize int64 + shutdownTimeout time.Duration +} + +type frozenNodeConfig struct { + freezeHeight uint64 + address string +} + +type frozenNodeFlags []string + +func (f *frozenNodeFlags) String() string { + return strings.Join(*f, ",") +} + +func (f *frozenNodeFlags) Set(value string) error { + *f = append(*f, value) + return nil +} + +func parseConfig(args []string, output io.Writer) (config, error) { + cfg := config{} + flags := flag.NewFlagSet("frozen-rpc-router", flag.ContinueOnError) + flags.SetOutput(output) + flags.StringVar(&cfg.listenAddress, "listen-address", defaultListenAddress, "address on which the router listens") + flags.StringVar(&cfg.liveNode, "live-node", "", "HTTP RPC address of the live node (required)") + 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.DurationVar(&cfg.shutdownTimeout, "shutdown-timeout", defaultShutdownTimeout, "graceful shutdown timeout") + if err := flags.Parse(args); err != nil { + return config{}, err + } + if flags.NArg() != 0 { + return config{}, fmt.Errorf("unexpected positional arguments: %s", strings.Join(flags.Args(), " ")) + } + if strings.TrimSpace(cfg.liveNode) == "" { + return config{}, errors.New("--live-node is required") + } + if cfg.maxRequestBodySize <= 0 { + return config{}, errors.New("--max-request-body-bytes must be positive") + } + if cfg.shutdownTimeout <= 0 { + return config{}, errors.New("--shutdown-timeout must be positive") + } + return cfg, nil +} + +func parseFrozenNodes(values frozenNodeFlags) ([]frozenNodeConfig, error) { + nodes := make([]frozenNodeConfig, 0, len(values)) + for _, value := range values { + heightText, address, ok := strings.Cut(value, "=") + if !ok || strings.TrimSpace(heightText) == "" || strings.TrimSpace(address) == "" { + return nil, fmt.Errorf("invalid frozen node %q: expected freeze-height=ip:port", value) + } + freezeHeight, err := strconv.ParseUint(strings.TrimSpace(heightText), 10, 64) + if err != nil || freezeHeight == 0 || freezeHeight > math.MaxInt64 { + return nil, fmt.Errorf("invalid freeze height %q", heightText) + } + nodes = append(nodes, frozenNodeConfig{ + freezeHeight: freezeHeight, + address: strings.TrimSpace(address), + }) + } + return nodes, nil +} diff --git a/cmd/frozen-rpc-router/config_test.go b/cmd/frozen-rpc-router/config_test.go new file mode 100644 index 0000000000..f15cf6c741 --- /dev/null +++ b/cmd/frozen-rpc-router/config_test.go @@ -0,0 +1,41 @@ +package main + +import ( + "io" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestParseConfig(t *testing.T) { + cfg, err := parseConfig([]string{ + "--listen-address", "0.0.0.0:9000", + "--live-node", "localhost:8545", + "--frozen-node", "200=localhost:8547", + "--frozen-node", "100=localhost:8546", + }, io.Discard) + require.NoError(t, err) + require.Equal(t, "0.0.0.0:9000", cfg.listenAddress) + require.Equal(t, "localhost:8545", cfg.liveNode) + + nodes, err := parseFrozenNodes(cfg.frozenNodes) + require.NoError(t, err) + require.Equal(t, []frozenNodeConfig{ + {freezeHeight: 200, address: "localhost:8547"}, + {freezeHeight: 100, address: "localhost:8546"}, + }, nodes) +} + +func TestParseConfigRejectsMissingLiveNode(t *testing.T) { + _, err := parseConfig(nil, io.Discard) + require.EqualError(t, err, "--live-node is required") +} + +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) { + _, err := parseFrozenNodes(frozenNodeFlags{value}) + require.Error(t, err) + }) + } +} diff --git a/cmd/frozen-rpc-router/main.go b/cmd/frozen-rpc-router/main.go new file mode 100644 index 0000000000..eb4f70072d --- /dev/null +++ b/cmd/frozen-rpc-router/main.go @@ -0,0 +1,70 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" +) + +func main() { + if err := run(); err != nil { + if errors.Is(err, flag.ErrHelp) { + return + } + log.Printf("frozen RPC router failed: %v", err) + os.Exit(1) + } +} + +func run() error { + cfg, err := parseConfig(os.Args[1:], os.Stderr) + if err != nil { + return err + } + frozenNodes, err := parseFrozenNodes(cfg.frozenNodes) + if err != nil { + return err + } + router, err := newRouter(cfg.liveNode, frozenNodes, nil, cfg.maxRequestBodySize) + if err != nil { + return err + } + + server := &http.Server{ + Addr: cfg.listenAddress, + Handler: router, + ReadHeaderTimeout: 10 * time.Second, + IdleTimeout: 2 * time.Minute, + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + serveErr := make(chan error, 1) + go func() { + log.Printf("frozen RPC router listening on %s", cfg.listenAddress) + serveErr <- server.ListenAndServe() + }() + + select { + case err := <-serveErr: + if errors.Is(err, http.ErrServerClosed) { + return nil + } + return err + case <-ctx.Done(): + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), cfg.shutdownTimeout) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + return fmt.Errorf("shut down HTTP server: %w", err) + } + return nil +} diff --git a/cmd/frozen-rpc-router/router.go b/cmd/frozen-rpc-router/router.go new file mode 100644 index 0000000000..226c3df69c --- /dev/null +++ b/cmd/frozen-rpc-router/router.go @@ -0,0 +1,687 @@ +package main + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "net/http" + "net/http/httputil" + "net/url" + "sort" + "strconv" + "strings" + "sync" +) + +const ( + jsonRPCParseError = -32700 + jsonRPCInvalidRequest = -32600 + jsonRPCUnsupportedError = -32000 + jsonRPCUpstreamError = -32001 + rpcRouteHeader = "Sei-RPC-Route" +) + +var blockParameterIndexes = map[string]int{ + "debug_getRawBlock": 0, + "debug_getRawHeader": 0, + "debug_getRawReceipts": 0, + "debug_traceBlockByNumber": 0, + "debug_traceCall": 1, + "eth_call": 1, + "eth_createAccessList": 1, + "eth_estimateGas": 1, + "eth_estimateGasAfterCalls": 2, + "eth_getBalance": 1, + "eth_getBlockByNumber": 0, + "eth_getBlockReceipts": 0, + "eth_getBlockTransactionCountByNumber": 0, + "eth_getCode": 1, + "eth_getProof": 2, + "eth_getRawTransactionByBlockNumberAndIndex": 0, + "eth_getStorageAt": 2, + "eth_getTransactionByBlockNumberAndIndex": 0, + "eth_getTransactionCount": 1, + "eth_getUncleByBlockNumberAndIndex": 0, + "eth_getUncleCountByBlockNumber": 0, +} + +type router struct { + live *upstream + frozen []*upstream + client *http.Client + maxRequestBodySize int64 + liveProxy *httputil.ReverseProxy +} + +type upstream struct { + freezeHeight uint64 + endpoint *url.URL +} + +type rpcCall struct { + raw json.RawMessage + method string + params json.RawMessage + id json.RawMessage + hasID bool + isValid bool +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +type rpcErrorResponse struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Error rpcError `json:"error"` +} + +type blockReference struct { + height uint64 + known bool + live bool +} + +type batchGroup struct { + upstream *upstream + calls []rpcCall + responses []json.RawMessage + err error +} + +func newRouter(liveAddress string, frozenConfigs []frozenNodeConfig, client *http.Client, maxRequestBodySize int64) (*router, error) { + liveURL, err := parseEndpoint(liveAddress) + if err != nil { + return nil, fmt.Errorf("invalid live node: %w", err) + } + if client == nil { + client = &http.Client{} + } + if maxRequestBodySize <= 0 { + return nil, errors.New("maximum request body size must be positive") + } + + live := &upstream{endpoint: liveURL} + frozen := make([]*upstream, 0, len(frozenConfigs)) + seenHeights := make(map[uint64]struct{}, len(frozenConfigs)) + for _, cfg := range frozenConfigs { + if cfg.freezeHeight == 0 { + return nil, errors.New("freeze height must be positive") + } + if _, exists := seenHeights[cfg.freezeHeight]; exists { + return nil, fmt.Errorf("duplicate freeze height %d", cfg.freezeHeight) + } + seenHeights[cfg.freezeHeight] = struct{}{} + endpoint, err := parseEndpoint(cfg.address) + if err != nil { + return nil, fmt.Errorf("invalid frozen node at height %d: %w", cfg.freezeHeight, err) + } + frozen = append(frozen, &upstream{freezeHeight: cfg.freezeHeight, endpoint: endpoint}) + } + sort.Slice(frozen, func(i, j int) bool { + return frozen[i].freezeHeight < frozen[j].freezeHeight + }) + + liveProxy := httputil.NewSingleHostReverseProxy(liveURL) + liveProxy.Transport = client.Transport + return &router{ + live: live, + frozen: frozen, + client: client, + maxRequestBodySize: maxRequestBodySize, + liveProxy: liveProxy, + }, nil +} + +func parseEndpoint(address string) (*url.URL, error) { + address = strings.TrimSpace(address) + if !strings.Contains(address, "://") { + address = "http://" + address + } + endpoint, err := url.Parse(address) + if err != nil { + return nil, err + } + if endpoint.Scheme != "http" && endpoint.Scheme != "https" { + return nil, fmt.Errorf("scheme must be http or https") + } + if endpoint.Host == "" { + return nil, errors.New("address must include a host") + } + if endpoint.Fragment != "" { + return nil, errors.New("address must not include a fragment") + } + return endpoint, nil +} + +func (r *router) ServeHTTP(w http.ResponseWriter, request *http.Request) { + if request.Method != http.MethodPost { + r.liveProxy.ServeHTTP(w, request) + return + } + + request.Body = http.MaxBytesReader(w, request.Body, r.maxRequestBodySize) + body, err := io.ReadAll(request.Body) + if err != nil { + var maxBytesErr *http.MaxBytesError + if errors.As(err, &maxBytesErr) { + http.Error(w, "request body too large", http.StatusRequestEntityTooLarge) + return + } + http.Error(w, "failed to read request body", http.StatusBadRequest) + return + } + + trimmed := bytes.TrimSpace(body) + if len(trimmed) > 0 && trimmed[0] == '[' { + r.serveBatch(w, request, body) + return + } + r.serveSingle(w, request, body) +} + +func (r *router) serveSingle(w http.ResponseWriter, request *http.Request, body []byte) { + call, err := decodeCall(body) + if err != nil { + if json.Valid(body) { + writeRPCError(w, nil, rpcError{Code: jsonRPCInvalidRequest, Message: "invalid request"}) + } else { + writeRPCError(w, nil, rpcError{Code: jsonRPCParseError, Message: "parse error"}) + } + return + } + if !call.isValid { + writeRPCError(w, nil, rpcError{Code: jsonRPCInvalidRequest, Message: "invalid request"}) + return + } + target, routingErr := r.route(call) + if routingErr != nil { + if call.hasID { + writeRPCError(w, call.id, *routingErr) + } + return + } + if err := r.proxy(w, request, target, body); err != nil { + if call.hasID { + writeRPCError(w, call.id, rpcError{Code: jsonRPCUpstreamError, Message: "upstream request failed"}) + } + } +} + +func (r *router) serveBatch(w http.ResponseWriter, request *http.Request, body []byte) { + var rawCalls []json.RawMessage + if err := json.Unmarshal(body, &rawCalls); err != nil { + writeRPCError(w, nil, rpcError{Code: jsonRPCParseError, Message: "parse error"}) + return + } + if len(rawCalls) == 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) + } + return + } + + groups, localResponses := r.groupBatch(calls) + r.fetchBatchGroups(request, groups) + responses := append([]json.RawMessage(nil), localResponses...) + for _, group := range groups { + if group.err != nil { + responses = append(responses, upstreamErrorResponses(group.calls)...) + continue + } + responses = append(responses, group.responses...) + } + w.Header().Set(rpcRouteHeader, "mixed") + writeBatchResponses(w, responses) +} + +func decodeCall(raw json.RawMessage) (rpcCall, error) { + call := rpcCall{raw: raw} + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + return call, err + } + if fields == nil { + return call, nil + } + methodRaw, ok := fields["method"] + if !ok || json.Unmarshal(methodRaw, &call.method) != nil || call.method == "" { + return call, nil + } + call.params = fields["params"] + call.id, call.hasID = fields["id"] + call.isValid = true + return call, nil +} + +func (r *router) singleBatchTarget(calls []rpcCall) (*upstream, bool) { + var target *upstream + for _, call := range calls { + if !call.isValid { + return nil, false + } + callTarget, routingErr := r.route(call) + if routingErr != nil { + return nil, false + } + if target == nil { + target = callTarget + continue + } + if target != callTarget { + return nil, false + } + } + return target, target != nil +} + +func (r *router) groupBatch(calls []rpcCall) ([]*batchGroup, []json.RawMessage) { + groupsByTarget := make(map[*upstream]*batchGroup) + groups := make([]*batchGroup, 0) + localResponses := make([]json.RawMessage, 0) + for _, call := range calls { + if !call.isValid { + localResponses = append(localResponses, marshalRPCError(nil, rpcError{Code: jsonRPCInvalidRequest, Message: "invalid request"})) + continue + } + target, routingErr := r.route(call) + if routingErr != nil { + if call.hasID { + localResponses = append(localResponses, marshalRPCError(call.id, *routingErr)) + } + continue + } + group := groupsByTarget[target] + if group == nil { + group = &batchGroup{upstream: target} + groupsByTarget[target] = group + groups = append(groups, group) + } + group.calls = append(group.calls, call) + } + return groups, localResponses +} + +func (r *router) fetchBatchGroups(request *http.Request, groups []*batchGroup) { + var wg sync.WaitGroup + for _, group := range groups { + wg.Add(1) + go func(group *batchGroup) { + defer wg.Done() + payload := make([]json.RawMessage, 0, len(group.calls)) + for _, call := range group.calls { + payload = append(payload, call.raw) + } + body, err := json.Marshal(payload) + if err != nil { + group.err = err + return + } + responseBody, err := r.callUpstream(request, group.upstream, body) + if err != nil { + group.err = err + return + } + group.responses, group.err = decodeBatchResponses(responseBody) + }(group) + } + wg.Wait() +} + +func decodeBatchResponses(body []byte) ([]json.RawMessage, error) { + if len(bytes.TrimSpace(body)) == 0 { + return nil, nil + } + var responses []json.RawMessage + if err := json.Unmarshal(body, &responses); err == nil { + return responses, nil + } + var response json.RawMessage + if err := json.Unmarshal(body, &response); err != nil { + return nil, errors.New("upstream returned invalid JSON") + } + return []json.RawMessage{response}, nil +} + +func (r *router) route(call rpcCall) (*upstream, *rpcError) { + switch call.method { + case "eth_getLogs": + return r.routeGetLogs(call.params) + case "eth_feeHistory": + return r.routeFeeHistory(call.params) + default: + parameterIndex, ok := blockParameterIndexes[call.method] + if !ok { + return r.live, nil + } + parameter, ok := positionalParameter(call.params, parameterIndex) + if !ok { + return r.live, nil + } + return r.upstreamForReference(parseBlockReference(parameter)), nil + } +} + +func (r *router) routeGetLogs(params json.RawMessage) (*upstream, *rpcError) { + filterRaw, ok := positionalParameter(params, 0) + if !ok { + return r.live, nil + } + var filter map[string]json.RawMessage + if json.Unmarshal(filterRaw, &filter) != nil || filter == nil { + return r.live, nil + } + if blockHash, exists := filter["blockHash"]; exists && !bytes.Equal(bytes.TrimSpace(blockHash), []byte("null")) { + return r.live, nil + } + + fromRaw, hasFrom := filter["fromBlock"] + toRaw, hasTo := filter["toBlock"] + if !hasFrom && hasTo { + fromRaw = toRaw + hasFrom = true + } + from := blockReference{live: true} + to := blockReference{live: true} + if hasFrom { + from = parseBlockReference(fromRaw) + } + if hasTo { + to = parseBlockReference(toRaw) + } + return r.routeRange(from, to) +} + +func (r *router) routeFeeHistory(params json.RawMessage) (*upstream, *rpcError) { + countRaw, hasCount := positionalParameter(params, 0) + lastRaw, hasLast := positionalParameter(params, 1) + if !hasCount || !hasLast { + return r.live, nil + } + count, ok := parseQuantity(countRaw) + if !ok { + return r.live, nil + } + last := parseBlockReference(lastRaw) + if !last.known || last.live { + return r.live, nil + } + firstHeight := last.height + if count > 1 { + if count-1 > last.height { + firstHeight = 0 + } else { + firstHeight = last.height - count + 1 + } + } + return r.routeRange(blockReference{height: firstHeight, known: true}, last) +} + +func (r *router) routeRange(from, to blockReference) (*upstream, *rpcError) { + if (!from.known && !from.live) || (!to.known && !to.live) { + return r.live, nil + } + if from.known && to.known && from.height > to.height { + return r.live, nil + } + fromTarget := r.upstreamForReference(from) + toTarget := r.upstreamForReference(to) + if fromTarget != toTarget { + return nil, &rpcError{ + Code: jsonRPCUnsupportedError, + Message: "block ranges spanning multiple frozen-node intervals are not supported", + } + } + return fromTarget, nil +} + +func (r *router) upstreamForReference(reference blockReference) *upstream { + if !reference.known || reference.live { + return r.live + } + for _, frozen := range r.frozen { + if reference.height < frozen.freezeHeight { + return frozen + } + } + return r.live +} + +func parseBlockReference(raw json.RawMessage) blockReference { + trimmed := bytes.TrimSpace(raw) + if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) { + return blockReference{} + } + if trimmed[0] == '{' { + var object map[string]json.RawMessage + if json.Unmarshal(trimmed, &object) != nil { + return blockReference{} + } + blockNumber, ok := object["blockNumber"] + if !ok { + return blockReference{} + } + return parseBlockReference(blockNumber) + } + + var value string + if json.Unmarshal(trimmed, &value) != nil { + return blockReference{} + } + switch value { + case "earliest": + return blockReference{height: 0, known: true} + case "latest", "pending", "safe", "finalized": + return blockReference{live: true} + } + height, err := strconv.ParseUint(strings.TrimPrefix(value, "0x"), 16, 64) + if err != nil || !strings.HasPrefix(value, "0x") || height > math.MaxInt64 { + return blockReference{} + } + return blockReference{height: height, known: true} +} + +func parseQuantity(raw json.RawMessage) (uint64, bool) { + trimmed := bytes.TrimSpace(raw) + var value string + if json.Unmarshal(trimmed, &value) == nil { + if !strings.HasPrefix(value, "0x") { + return 0, false + } + quantity, err := strconv.ParseUint(strings.TrimPrefix(value, "0x"), 16, 64) + return quantity, err == nil + } + var quantity uint64 + if json.Unmarshal(trimmed, &quantity) != nil { + return 0, false + } + return quantity, true +} + +func positionalParameter(params json.RawMessage, index int) (json.RawMessage, bool) { + if len(params) == 0 { + return nil, false + } + var values []json.RawMessage + if json.Unmarshal(params, &values) != nil || index < 0 || index >= len(values) { + return nil, false + } + return values[index], true +} + +func (r *router) proxy(w http.ResponseWriter, request *http.Request, target *upstream, body []byte) error { + upstreamRequest, err := r.newUpstreamRequest(request, target, body) + if err != nil { + return err + } + response, err := r.client.Do(upstreamRequest) + if err != nil { + return err + } + + copyResponseHeaders(w.Header(), response.Header) + w.Header().Set(rpcRouteHeader, target.routeName()) + w.WriteHeader(response.StatusCode) + _, _ = io.Copy(w, response.Body) + _ = response.Body.Close() + return nil +} + +func (u *upstream) routeName() string { + if u.freezeHeight == 0 { + return "live" + } + return fmt.Sprintf("frozen:%d", u.freezeHeight) +} + +func (r *router) callUpstream(request *http.Request, target *upstream, body []byte) ([]byte, error) { + upstreamRequest, err := r.newUpstreamRequest(request, target, body) + if err != nil { + return nil, err + } + response, err := r.client.Do(upstreamRequest) + if err != nil { + return nil, err + } + responseBody, err := io.ReadAll(response.Body) + _ = response.Body.Close() + if err != nil { + return nil, err + } + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf("upstream returned HTTP %d", response.StatusCode) + } + return responseBody, nil +} + +func (r *router) newUpstreamRequest(request *http.Request, target *upstream, body []byte) (*http.Request, error) { + endpoint := joinedURL(target.endpoint, request.URL) + upstreamRequest, err := http.NewRequestWithContext(request.Context(), request.Method, endpoint.String(), bytes.NewReader(body)) + if err != nil { + return nil, err + } + upstreamRequest.Header = request.Header.Clone() + removeHopByHopHeaders(upstreamRequest.Header) + upstreamRequest.Header.Del("Accept-Encoding") + if host := clientIP(request.RemoteAddr); host != "" { + prior := upstreamRequest.Header.Get("X-Forwarded-For") + if prior != "" { + host = prior + ", " + host + } + upstreamRequest.Header.Set("X-Forwarded-For", host) + } + return upstreamRequest, nil +} + +func joinedURL(base, incoming *url.URL) *url.URL { + target := *base + if incoming.Path != "" && incoming.Path != "/" { + target.Path = strings.TrimRight(base.Path, "/") + "/" + strings.TrimLeft(incoming.Path, "/") + target.RawPath = "" + } + if target.RawQuery == "" { + target.RawQuery = incoming.RawQuery + } else if incoming.RawQuery != "" { + target.RawQuery += "&" + incoming.RawQuery + } + return &target +} + +func clientIP(remoteAddress string) string { + if index := strings.LastIndex(remoteAddress, ":"); index >= 0 { + return strings.Trim(remoteAddress[:index], "[]") + } + return remoteAddress +} + +func removeHopByHopHeaders(header http.Header) { + for _, name := range strings.Split(header.Get("Connection"), ",") { + header.Del(strings.TrimSpace(name)) + } + for _, name := range []string{ + "Connection", + "Keep-Alive", + "Proxy-Authenticate", + "Proxy-Authorization", + "Te", + "Trailer", + "Transfer-Encoding", + "Upgrade", + } { + header.Del(name) + } +} + +func copyResponseHeaders(destination, source http.Header) { + cloned := source.Clone() + removeHopByHopHeaders(cloned) + for name, values := range cloned { + for _, value := range values { + destination.Add(name, value) + } + } +} + +func writeRPCError(w http.ResponseWriter, id json.RawMessage, rpcErr rpcError) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(marshalRPCError(id, rpcErr)) +} + +func marshalRPCError(id json.RawMessage, rpcErr rpcError) json.RawMessage { + if len(id) == 0 { + id = json.RawMessage("null") + } + response, _ := json.Marshal(rpcErrorResponse{ + JSONRPC: "2.0", + ID: id, + Error: rpcErr, + }) + return response +} + +func writeBatchUpstreamErrors(w http.ResponseWriter, calls []rpcCall) { + writeBatchResponses(w, upstreamErrorResponses(calls)) +} + +func upstreamErrorResponses(calls []rpcCall) []json.RawMessage { + responses := make([]json.RawMessage, 0, len(calls)) + for _, call := range calls { + if call.hasID { + responses = append(responses, marshalRPCError(call.id, rpcError{ + Code: jsonRPCUpstreamError, + Message: "upstream request failed", + })) + } + } + return responses +} + +func writeBatchResponses(w http.ResponseWriter, responses []json.RawMessage) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + if len(responses) == 0 { + return + } + response, _ := json.Marshal(responses) + _, _ = w.Write(response) +} diff --git a/cmd/frozen-rpc-router/router_test.go b/cmd/frozen-rpc-router/router_test.go new file mode 100644 index 0000000000..9c5960cf14 --- /dev/null +++ b/cmd/frozen-rpc-router/router_test.go @@ -0,0 +1,244 @@ +package main + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRouteBlockParameters(t *testing.T) { + r := newTestRouter(t) + testCases := []struct { + name string + method string + params string + wantHost string + }{ + {name: "unclassified method", method: "eth_blockNumber", params: "[]", wantHost: "live:8545"}, + {name: "first interval genesis", method: "eth_getBlockByNumber", params: `["0x0",false]`, wantHost: "frozen-100:8545"}, + {name: "first interval upper edge", method: "eth_getBlockByNumber", params: `["0x63",false]`, wantHost: "frozen-100:8545"}, + {name: "second interval lower edge", method: "eth_getBlockByNumber", params: `["0x64",false]`, wantHost: "frozen-200:8545"}, + {name: "second interval upper edge", method: "debug_traceBlockByNumber", params: `["0xc7",{}]`, wantHost: "frozen-200:8545"}, + {name: "live interval lower edge", method: "eth_getBlockReceipts", params: `["0xc8"]`, wantHost: "live:8545"}, + {name: "latest tag", method: "eth_getBalance", params: `["0x0000000000000000000000000000000000000000","latest"]`, wantHost: "live:8545"}, + {name: "earliest tag", method: "eth_getCode", params: `["0x0000000000000000000000000000000000000000","earliest"]`, wantHost: "frozen-100:8545"}, + {name: "EIP-1898 number", method: "eth_call", params: `[{}, {"blockNumber":"0x64"}]`, wantHost: "frozen-200:8545"}, + {name: "EIP-1898 hash", method: "eth_call", params: `[{}, {"blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000"}]`, wantHost: "live:8545"}, + {name: "method-specific position", method: "eth_getStorageAt", params: `["0x0000000000000000000000000000000000000000","0x0","0x63"]`, wantHost: "frozen-100:8545"}, + {name: "optional block omitted", method: "eth_estimateGas", params: `[{}]`, wantHost: "live:8545"}, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + call := mustDecodeCall(t, testCase.method, testCase.params) + target, routingErr := r.route(call) + require.Nil(t, routingErr) + require.Equal(t, testCase.wantHost, target.endpoint.Host) + }) + } +} + +func TestRouteRanges(t *testing.T) { + r := newTestRouter(t) + testCases := []struct { + name string + method string + params string + wantHost string + wantError bool + }{ + {name: "get logs defaults to latest", method: "eth_getLogs", params: `[{}]`, wantHost: "live:8545"}, + {name: "get logs only to block", method: "eth_getLogs", params: `[{"toBlock":"0x63"}]`, wantHost: "frozen-100:8545"}, + {name: "get logs within second interval", method: "eth_getLogs", params: `[{"fromBlock":"0x64","toBlock":"0xc7"}]`, wantHost: "frozen-200:8545"}, + {name: "get logs within live interval", method: "eth_getLogs", params: `[{"fromBlock":"0xc8","toBlock":"0xfa"}]`, wantHost: "live:8545"}, + {name: "get logs by hash", method: "eth_getLogs", params: `[{"blockHash":"0x0000000000000000000000000000000000000000000000000000000000000000"}]`, wantHost: "live:8545"}, + {name: "get logs crosses frozen intervals", method: "eth_getLogs", params: `[{"fromBlock":"0x63","toBlock":"0x64"}]`, wantError: true}, + {name: "get logs crosses into live", method: "eth_getLogs", params: `[{"fromBlock":"0xc7","toBlock":"0xc8"}]`, wantError: true}, + {name: "fee history within interval", method: "eth_feeHistory", params: `["0x2","0x65",[]]`, wantHost: "frozen-200:8545"}, + {name: "fee history crosses intervals", method: "eth_feeHistory", params: `["0x2","0x64",[]]`, wantError: true}, + {name: "fee history crosses into live", method: "eth_feeHistory", params: `["0x2","0xc8",[]]`, wantError: true}, + {name: "fee history at latest", method: "eth_feeHistory", params: `["0x2","latest",[]]`, wantHost: "live:8545"}, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + call := mustDecodeCall(t, testCase.method, testCase.params) + target, routingErr := r.route(call) + if testCase.wantError { + require.Nil(t, target) + require.NotNil(t, routingErr) + require.Equal(t, jsonRPCUnsupportedError, routingErr.Code) + return + } + require.Nil(t, routingErr) + require.Equal(t, testCase.wantHost, target.endpoint.Host) + }) + } +} + +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) + require.NoError(t, err) + + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "http://router/", strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"eth_getBlockByNumber","params":["0x63",false]}`)) + r.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Equal(t, "frozen", recorder.Header().Get("X-Upstream")) + require.Equal(t, "frozen:100", recorder.Header().Get(rpcRouteHeader)) + require.JSONEq(t, `{"jsonrpc":"2.0","id":1,"result":"frozen"}`, recorder.Body.String()) + require.EqualValues(t, 0, live.hits.Load()) + require.EqualValues(t, 1, frozen.hits.Load()) +} + +func TestRouterSplitsMixedBatch(t *testing.T) { + live := newRPCBackend(t, "live") + frozen100 := newRPCBackend(t, "frozen-100") + frozen200 := newRPCBackend(t, "frozen-200") + r, err := newRouter(live.server.URL, []frozenNodeConfig{ + {freezeHeight: 200, address: frozen200.server.URL}, + {freezeHeight: 100, address: frozen100.server.URL}, + }, live.server.Client(), defaultMaxRequestBodySize) + require.NoError(t, err) + + body := `[ + {"jsonrpc":"2.0","id":1,"method":"eth_getBlockByNumber","params":["0x63",false]}, + {"jsonrpc":"2.0","id":2,"method":"eth_getBlockByNumber","params":["0x64",false]}, + {"jsonrpc":"2.0","id":3,"method":"net_version","params":[]}, + {"jsonrpc":"2.0","id":4,"method":"eth_getLogs","params":[{"fromBlock":"0x63","toBlock":"0x64"}]}, + {"jsonrpc":"2.0","method":"eth_blockNumber","params":[]} + ]` + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "http://router/", strings.NewReader(body)) + r.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Equal(t, "mixed", recorder.Header().Get(rpcRouteHeader)) + var responses []struct { + ID json.RawMessage `json:"id"` + Result string `json:"result"` + Error *rpcError `json:"error"` + } + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &responses)) + require.Len(t, responses, 4) + byID := make(map[string]struct { + result string + err *rpcError + }) + for _, response := range responses { + byID[string(response.ID)] = struct { + result string + err *rpcError + }{result: response.Result, err: response.Error} + } + require.Equal(t, "frozen-100", byID["1"].result) + require.Equal(t, "frozen-200", byID["2"].result) + require.Equal(t, "live", byID["3"].result) + require.Equal(t, jsonRPCUnsupportedError, byID["4"].err.Code) + require.EqualValues(t, 1, frozen100.hits.Load()) + require.EqualValues(t, 1, frozen200.hits.Load()) + require.EqualValues(t, 1, live.hits.Load()) +} + +func TestRouterOmitsErrorForUnsupportedNotification(t *testing.T) { + r := newTestRouter(t) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "http://router/", strings.NewReader(`{"jsonrpc":"2.0","method":"eth_getLogs","params":[{"fromBlock":"0x63","toBlock":"0x64"}]}`)) + r.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Empty(t, recorder.Body.String()) +} + +func TestRouterRejectsOversizedRequest(t *testing.T) { + r, err := newRouter("live:8545", nil, nil, 8) + require.NoError(t, err) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "http://router/", bytes.NewReader([]byte("123456789"))) + r.ServeHTTP(recorder, request) + require.Equal(t, http.StatusRequestEntityTooLarge, recorder.Code) +} + +func TestNewRouterSortsAndValidatesFrozenNodes(t *testing.T) { + r := newTestRouter(t) + require.Equal(t, uint64(100), r.frozen[0].freezeHeight) + require.Equal(t, uint64(200), r.frozen[1].freezeHeight) + + _, err := newRouter("live:8545", []frozenNodeConfig{ + {freezeHeight: 100, address: "one:8545"}, + {freezeHeight: 100, address: "two:8545"}, + }, nil, defaultMaxRequestBodySize) + require.EqualError(t, err, "duplicate freeze height 100") +} + +func newTestRouter(t *testing.T) *router { + t.Helper() + r, err := newRouter("live:8545", []frozenNodeConfig{ + {freezeHeight: 200, address: "frozen-200:8545"}, + {freezeHeight: 100, address: "frozen-100:8545"}, + }, nil, defaultMaxRequestBodySize) + require.NoError(t, err) + return r +} + +func mustDecodeCall(t *testing.T, method, params string) rpcCall { + t.Helper() + raw := json.RawMessage(`{"jsonrpc":"2.0","id":1,"method":` + strconv.Quote(method) + `,"params":` + params + `}`) + call, err := decodeCall(raw) + require.NoError(t, err) + require.True(t, call.isValid) + return call +} + +type rpcBackend struct { + server *httptest.Server + hits atomic.Int64 +} + +func newRPCBackend(t *testing.T, name string) *rpcBackend { + t.Helper() + backend := &rpcBackend{} + backend.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + backend.hits.Add(1) + body, err := io.ReadAll(request.Body) + require.NoError(t, err) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-Upstream", name) + trimmed := bytes.TrimSpace(body) + if len(trimmed) > 0 && trimmed[0] == '[' { + var calls []map[string]json.RawMessage + require.NoError(t, json.Unmarshal(trimmed, &calls)) + responses := make([]map[string]any, 0, len(calls)) + for _, call := range calls { + id, hasID := call["id"] + if !hasID { + continue + } + responses = append(responses, map[string]any{"jsonrpc": "2.0", "id": id, "result": name}) + } + if len(responses) != 0 { + require.NoError(t, json.NewEncoder(w).Encode(responses)) + } + return + } + var call map[string]json.RawMessage + require.NoError(t, json.Unmarshal(trimmed, &call)) + id, hasID := call["id"] + if hasID { + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": id, "result": name})) + } + })) + t.Cleanup(backend.server.Close) + return backend +} diff --git a/docker/README.md b/docker/README.md index 7e4ed68c5b..cadf841528 100644 --- a/docker/README.md +++ b/docker/README.md @@ -20,6 +20,26 @@ https://docs.docker.com/compose/install/other/ Detailed instruction: see the `Makefile` in the root of [the repo](https://github.com/sei-protocol/sei-chain/blob/main/Makefile) +### Frozen RPC router cluster + +The frozen-router topology runs two live validators, archival full nodes frozen +before heights 10 and 20, and `frozen-rpc-router` on host port 8553: + +```sh +DOCKER_DETACH=true make docker-frozen-rpc-cluster-start +go test -tags frozen_rpc_integration -v -count=1 -timeout 5m ./integration_test/frozen_rpc_router/... +make docker-frozen-rpc-cluster-stop +``` + +Run the complete start-test-stop flow with: + +```sh +make frozen-rpc-router-integration-test +``` + +The direct EVM RPC ports are 8547 for the height-10 frozen node, 8549 for the +height-20 frozen node, and 8545 for the live node. + **To start a single local node (Not Recommended)** ```sh diff --git a/docker/docker-compose.frozen-rpc-router.yml b/docker/docker-compose.frozen-rpc-router.yml new file mode 100644 index 0000000000..f32d7e1aea --- /dev/null +++ b/docker/docker-compose.frozen-rpc-router.yml @@ -0,0 +1,61 @@ +# Docker Compose override for the frozen RPC router integration topology. +# Nodes 1 and 2 are non-validating archival nodes, so freezing them does not +# remove voting power from the two-validator live chain (nodes 0 and 3). + +services: + node0: + environment: + - GIGA_EXECUTOR=false + - GIGA_OCC=false + + node1: + environment: + - VALIDATOR=false + - FREEZE_HEIGHT=10 + - GIGA_EXECUTOR=false + - GIGA_OCC=false + + node2: + environment: + - VALIDATOR=false + - FREEZE_HEIGHT=20 + - GIGA_EXECUTOR=false + - GIGA_OCC=false + + node3: + environment: + - GIGA_EXECUTOR=false + - GIGA_OCC=false + + frozen-rpc-router: + platform: ${DOCKER_PLATFORM:-linux/amd64} + container_name: sei-frozen-rpc-router + image: "sei-chain/localnode" + user: "${USERID}:${GROUPID}" + depends_on: + - node0 + - node1 + - node2 + environment: + - SKIP_BUILD + entrypoint: + - /bin/bash + - -c + command: + - | + if [ -z "$$SKIP_BUILD" ]; then + until [ -f build/generated/build.complete ]; do sleep 1; done + fi + until [ -x build/frozen-rpc-router ]; do sleep 1; done + exec build/frozen-rpc-router \ + --listen-address 0.0.0.0:8545 \ + --live-node http://node0:8545 \ + --frozen-node 10=http://node1:8545 \ + --frozen-node 20=http://node2:8545 + ports: + - "8553:8545" + volumes: + - "${PROJECT_HOME}:/sei-protocol/sei-chain:Z" + networks: + localnet: + ipv4_address: 192.168.10.14 diff --git a/docker/localnode/scripts/step0_build.sh b/docker/localnode/scripts/step0_build.sh index 3573ca9b0b..d29c44716d 100755 --- a/docker/localnode/scripts/step0_build.sh +++ b/docker/localnode/scripts/step0_build.sh @@ -18,5 +18,6 @@ else echo "Building with standard configuration..." make build-linux fi +make build-frozen-rpc-router mkdir -p build/generated echo "DONE" > build/generated/build.complete diff --git a/docker/localnode/scripts/step1_configure_init.sh b/docker/localnode/scripts/step1_configure_init.sh index 167ecc3cbd..fa2a7f1681 100755 --- a/docker/localnode/scripts/step1_configure_init.sh +++ b/docker/localnode/scripts/step1_configure_init.sh @@ -3,6 +3,7 @@ # Input parameters NODE_ID=${ID:-0} NUM_ACCOUNTS=${NUM_ACCOUNTS:-5} +VALIDATOR=${VALIDATOR:-true} echo "Configure and initialize environment" cp build/seid "$GOBIN"/ @@ -17,7 +18,8 @@ mkdir -p "$NODE_DIR" seid version # Uncomment the below line if there are any dependency issues # ldd build/seid -# Initialize validator node +# Initialize the node home. The topology selects validator or full-node mode +# after all nodes have contributed their shared genesis inputs. MONIKER="sei-node-$NODE_ID" seid init "$MONIKER" --chain-id sei >/dev/null 2>&1 @@ -54,9 +56,12 @@ echo "$GENESIS_ACCOUNT_ADDRESS" >> build/generated/genesis_accounts.txt # Add funds to genesis account seid add-genesis-account "$GENESIS_ACCOUNT_ADDRESS" 10000000usei,10000000uusdc,10000000uatom -# Create gentx -printf "12345678\n" | seid gentx "$ACCOUNT_NAME" 10000000usei --chain-id sei -cp ~/.sei/config/gentx/* build/generated/gentx/ +# Only validator nodes contribute a gentx to the genesis validator set. Full +# nodes still receive funded accounts and persistent-peer configuration. +if [ "$VALIDATOR" = "true" ]; then + printf "12345678\n" | seid gentx "$ACCOUNT_NAME" 10000000usei --chain-id sei + cp ~/.sei/config/gentx/* build/generated/gentx/ +fi # Creating some testing accounts echo "Creating $NUM_ACCOUNTS accounts" diff --git a/docker/localnode/scripts/step4_config_override.sh b/docker/localnode/scripts/step4_config_override.sh index b8a2ad7b96..f0afa2f97c 100755 --- a/docker/localnode/scripts/step4_config_override.sh +++ b/docker/localnode/scripts/step4_config_override.sh @@ -1,6 +1,7 @@ #!/usr/bin/env sh NODE_ID=${ID:-0} +VALIDATOR=${VALIDATOR:-true} # Defaults mirror the app's DefaultConfig (giga+OCC on): unset runs what an # unconfigured seid would; only an explicit false selects V2. GIGA_EXECUTOR=${GIGA_EXECUTOR:-true} @@ -31,6 +32,10 @@ cp build/generated/genesis.json ~/.sei/config/genesis.json cp "$APP_CONFIG_FILE" ~/.sei/config/app.toml cp "$TENDERMINT_CONFIG_FILE" ~/.sei/config/config.toml +if [ "$VALIDATOR" != "true" ]; then + sed -i 's/^mode = "validator"/mode = "full"/' ~/.sei/config/config.toml +fi + # Override up persistent peers NODE_IP=$(hostname -i | awk '{print $1}') PEERS=$(cat build/generated/persistent_peers.txt |grep -v "$NODE_IP" | paste -sd "," -) diff --git a/docker/localnode/scripts/step5_start_sei.sh b/docker/localnode/scripts/step5_start_sei.sh index 4474d768f5..78b4360b48 100755 --- a/docker/localnode/scripts/step5_start_sei.sh +++ b/docker/localnode/scripts/step5_start_sei.sh @@ -2,13 +2,14 @@ NODE_ID=${ID:-0} INVARIANT_CHECK_INTERVAL=${INVARIANT_CHECK_INTERVAL:-0} +FREEZE_HEIGHT=${FREEZE_HEIGHT:-0} LOG_DIR="build/generated/logs" mkdir -p $LOG_DIR -echo "Starting the seid process for node $NODE_ID with invariant check interval=$INVARIANT_CHECK_INTERVAL..." +echo "Starting the seid process for node $NODE_ID with invariant check interval=$INVARIANT_CHECK_INTERVAL and freeze height=$FREEZE_HEIGHT..." -seid start --chain-id sei --inv-check-period ${INVARIANT_CHECK_INTERVAL} > "$LOG_DIR/seid-$NODE_ID.log" 2>&1 & +seid start --chain-id sei --inv-check-period "${INVARIANT_CHECK_INTERVAL}" --freeze-height "${FREEZE_HEIGHT}" > "$LOG_DIR/seid-$NODE_ID.log" 2>&1 & SEID_PID=$! echo "Node $NODE_ID seid is started now" diff --git a/integration_test/frozen_rpc_router/router_test.go b/integration_test/frozen_rpc_router/router_test.go new file mode 100644 index 0000000000..c1734a1232 --- /dev/null +++ b/integration_test/frozen_rpc_router/router_test.go @@ -0,0 +1,170 @@ +//go:build frozen_rpc_integration + +// Package frozenrpcrouter verifies the Docker frozen-node routing topology. +package frozenrpcrouter + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strconv" + "strings" + "testing" + "time" +) + +const ( + defaultRouterURL = "http://127.0.0.1:8553" + defaultFrozen10URL = "http://127.0.0.1:8547" + defaultFrozen20URL = "http://127.0.0.1:8549" + defaultLiveURL = "http://127.0.0.1:8545" + routeHeader = "Sei-RPC-Route" +) + +type rpcResponse struct { + Result json.RawMessage `json:"result"` + Error *rpcError `json:"error"` +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +func TestFrozenRPCRouterReachesEveryNode(t *testing.T) { + client := &http.Client{Timeout: 10 * time.Second} + routerURL := envOrDefault("FROZEN_RPC_ROUTER_URL", defaultRouterURL) + frozen10URL := envOrDefault("FROZEN_RPC_NODE_10_URL", defaultFrozen10URL) + frozen20URL := envOrDefault("FROZEN_RPC_NODE_20_URL", defaultFrozen20URL) + liveURL := envOrDefault("FROZEN_RPC_LIVE_NODE_URL", defaultLiveURL) + + waitForHead(t, client, frozen10URL, func(height uint64) bool { return height == 9 }, "frozen node at height 10") + waitForHead(t, client, frozen20URL, func(height uint64) bool { return height == 19 }, "frozen node at height 20") + waitForHead(t, client, liveURL, func(height uint64) bool { return height > 20 }, "live node past height 20") + waitForHead(t, client, routerURL, func(height uint64) bool { return height > 20 }, "router live-node forwarding") + + for _, testCase := range []struct { + name string + height uint64 + wantRoute string + }{ + {name: "height 9 reaches first frozen node", height: 9, wantRoute: "frozen:10"}, + {name: "height 15 reaches second frozen node", height: 15, wantRoute: "frozen:20"}, + {name: "height 20 reaches live node", height: 20, wantRoute: "live"}, + } { + t.Run(testCase.name, func(t *testing.T) { + response, route, err := callRPC(t.Context(), client, routerURL, "eth_getBlockByNumber", []any{fmt.Sprintf("0x%x", testCase.height), false}) + if err != nil { + t.Fatal(err) + } + if response.Error != nil { + t.Fatalf("RPC returned error %d: %s", response.Error.Code, response.Error.Message) + } + if route != testCase.wantRoute { + t.Fatalf("route header = %q, want %q", route, testCase.wantRoute) + } + var block struct { + Number string `json:"number"` + } + if err := json.Unmarshal(response.Result, &block); err != nil { + t.Fatalf("decode block response: %v", err) + } + wantNumber := fmt.Sprintf("0x%x", testCase.height) + if block.Number != wantNumber { + t.Fatalf("block number = %q, want %q", block.Number, wantNumber) + } + }) + } + + t.Run("range crossing frozen nodes is rejected", func(t *testing.T) { + response, _, err := callRPC(t.Context(), client, routerURL, "eth_getLogs", []any{map[string]any{ + "fromBlock": "0x9", + "toBlock": "0xa", + }}) + if err != nil { + t.Fatal(err) + } + if response.Error == nil || response.Error.Code != -32000 { + t.Fatalf("RPC error = %+v, want code -32000", response.Error) + } + }) +} + +func waitForHead(t *testing.T, client *http.Client, endpoint string, ready func(uint64) bool, description string) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Minute) + defer cancel() + + var lastHeight uint64 + var lastErr error + for ctx.Err() == nil { + response, _, err := callRPC(ctx, client, endpoint, "eth_blockNumber", []any{}) + if err == nil && response.Error == nil { + var encodedHeight string + err = json.Unmarshal(response.Result, &encodedHeight) + if err == nil { + lastHeight, err = strconv.ParseUint(strings.TrimPrefix(encodedHeight, "0x"), 16, 64) + } + if err == nil && ready(lastHeight) { + return + } + } else if err == nil { + err = fmt.Errorf("RPC error %d: %s", response.Error.Code, response.Error.Message) + } + lastErr = err + timer := time.NewTimer(time.Second) + select { + case <-ctx.Done(): + timer.Stop() + case <-timer.C: + } + } + t.Fatalf("timed out waiting for %s at %s (last height %d, last error %v)", description, endpoint, lastHeight, lastErr) +} + +func callRPC(ctx context.Context, client *http.Client, endpoint, method string, params any) (rpcResponse, string, error) { + payload, err := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": 1, + "method": method, + "params": params, + }) + if err != nil { + return rpcResponse{}, "", err + } + request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return rpcResponse{}, "", err + } + request.Header.Set("Content-Type", "application/json") + response, err := client.Do(request) + if err != nil { + return rpcResponse{}, "", err + } + defer func() { + _ = response.Body.Close() + }() + body, err := io.ReadAll(response.Body) + if err != nil { + return rpcResponse{}, "", err + } + if response.StatusCode != http.StatusOK { + return rpcResponse{}, "", fmt.Errorf("HTTP status %d: %s", response.StatusCode, body) + } + var result rpcResponse + if err := json.Unmarshal(body, &result); err != nil { + return rpcResponse{}, "", fmt.Errorf("decode RPC response %q: %w", body, err) + } + return result, response.Header.Get(routeHeader), nil +} + +func envOrDefault(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +}