diff --git a/erpc/http_server_ws_tip_leader_test.go b/erpc/http_server_ws_tip_leader_test.go index 5dda808b3..fdc5f9341 100644 --- a/erpc/http_server_ws_tip_leader_test.go +++ b/erpc/http_server_ws_tip_leader_test.go @@ -145,6 +145,141 @@ func TestHttpServer_GetBlockByNumberLatest_RefetchPinsEvmLeaderUpstream(t *testi "EnforceHighestBlock must pin the tip re-fetch to EvmLeaderUpstream") } +// Direct eth_getBlockByNumber(tip) must pin to EvmLeaderUpstream on first +// forward (same idea as EnforceHighestBlock tip re-fetch), not hit a lagging +// sibling that is preferred by selection order. +func TestHttpServer_GetBlockByNumber_NearTipPinsEvmLeaderUpstream(t *testing.T) { + util.ResetGock() + defer util.ResetGock() + util.SetupMocksForEvmStatePoller() + // rpc1 tip-null Persist mock is intentionally unused when the pin works. + defer util.AssertNoPendingMocks(t, 1) + + const tip = int64(0x33338889) + tipHex := "0x33338889" + var leaderHits atomic.Int64 + var laggingHits atomic.Int64 + + gock.New("http://rpc1.localhost"). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + if r.URL.Host != "rpc1.localhost" { + return false + } + body := util.SafeReadBody(r) + if strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, tipHex) { + laggingHits.Add(1) + return true + } + return false + }). + Reply(200). + JSON([]byte(`{"result":null}`)) + + gock.New("http://rpc2.localhost"). + Post(""). + Filter(func(r *http.Request) bool { + if r.URL.Host != "rpc2.localhost" { + return false + } + body := util.SafeReadBody(r) + if strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, tipHex) { + leaderHits.Add(1) + return true + } + return false + }). + Reply(200). + JSON([]byte(`{"result":{"number":"0x33338889","hash":"0xcccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","parentHash":"0xdddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd","timestamp":"0x6702a8f1"}}`)) + + cfg := &common.Config{ + Server: &common.ServerConfig{ + MaxTimeout: common.Duration(100 * time.Second).Ptr(), + }, + Projects: []*common.ProjectConfig{ + { + Id: "test_project", + Networks: []*common.NetworkConfig{ + { + Architecture: "evm", + Evm: &common.EvmNetworkConfig{ + ChainId: 123, + }, + Failsafe: []*common.FailsafeConfig{ + { + Retry: &common.RetryPolicyConfig{MaxAttempts: 2}, + }, + }, + }, + }, + Upstreams: []*common.UpstreamConfig{ + { + Id: "rpc1", + Endpoint: "http://rpc1.localhost", + Type: common.UpstreamTypeEvm, + Evm: &common.EvmUpstreamConfig{ + ChainId: 123, + StatePollerInterval: common.Duration(10 * time.Second), + }, + }, + { + Id: "rpc2", + Endpoint: "http://rpc2.localhost", + Type: common.UpstreamTypeEvm, + Evm: &common.EvmUpstreamConfig{ + ChainId: 123, + StatePollerInterval: common.Duration(10 * time.Second), + }, + }, + }, + }, + }, + } + + sendRequest, _, _, shutdown, erpcInstance := createServerTestFixtures(cfg, t) + defer shutdown() + + prj, err := erpcInstance.GetProject("test_project") + require.NoError(t, err) + policy.OverrideAllForTest(prj.policyEngine) + policy.OverrideOrderForTest(prj.policyEngine, "evm:123", "rpc1", "rpc2") + + time.Sleep(500 * time.Millisecond) + + nw, err := prj.GetNetwork(context.Background(), "evm:123") + require.NoError(t, err) + + var leader *upstream.Upstream + for _, u := range nw.upstreamsRegistry.GetNetworkUpstreams(context.Background(), "evm:123") { + if u.Id() == "rpc2" { + leader = u + break + } + } + require.NotNil(t, leader) + leader.EvmStatePoller().SuggestLatestBlock(tip) + require.Equal(t, "rpc2", nw.EvmLeaderUpstream(context.Background()).Id()) + + statusCode, _, body := sendRequest(`{ + "jsonrpc": "2.0", + "id": 1, + "method": "eth_getBlockByNumber", + "params": ["`+tipHex+`", false] + }`, nil, nil) + + require.Equal(t, http.StatusOK, statusCode, "body=%s", body) + var respObject map[string]interface{} + require.NoError(t, sonic.UnmarshalString(body, &respObject)) + result, ok := respObject["result"].(map[string]interface{}) + require.True(t, ok, "got: %s", body) + assert.Equal(t, tipHex, result["number"]) + assert.GreaterOrEqual(t, leaderHits.Load(), int64(1), + "near-tip getBlock must pin to EvmLeaderUpstream") + assert.Equal(t, int64(0), laggingHits.Load(), + "lagging sibling must not receive the pinned near-tip getBlock") +} + // When TipHW is ahead of every upstream's concrete block response, // EnforceHighestBlock must NOT fail-open to the stale "latest" — that is // the MultiNode FOOS trigger once WS has already delivered the higher head. diff --git a/erpc/networks.go b/erpc/networks.go index 6ee3e0821..6a6374176 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -725,9 +725,15 @@ func (n *Network) Forward(ctx context.Context, req *common.NormalizedRequest) (* // upstream demonstrably having the block. partitionUpstreamsByLatestBlock // is stable so it composes with the tier and score orderings layered // on top. + // + // Near-tip eth_getBlockByNumber also pins UseUpstream to EvmLeaderUpstream + // (typically the WS ingress that SuggestLatestBlock advanced) so the + // first attempt hits the node that already has the head — same idea as + // EnforceHighestBlock's tip re-fetch pin, but for direct client tip reads. if n.Architecture() == common.ArchitectureEvm { if bn := requestBlockNumber(ctx, req); bn > 0 { upsList = partitionUpstreamsByLatestBlock(upsList, bn) + n.pinNearTipGetBlockToLeader(ctx, req, method, bn) } } @@ -2232,6 +2238,40 @@ func partitionUpstreamsByLatestBlock(ups []common.Upstream, bn int64) []common.U return out } +// pinNearTipGetBlockToLeader sets UseUpstream to EvmLeaderUpstream when the +// request is eth_getBlockByNumber for the leader tip or tip+1 (sibling import +// race window). Skips if the client/config already set UseUpstream. +func (n *Network) pinNearTipGetBlockToLeader(ctx context.Context, req *common.NormalizedRequest, method string, bn int64) { + if n == nil || req == nil || method != "eth_getBlockByNumber" || bn <= 0 { + return + } + leader := n.EvmLeaderUpstream(ctx) + if leader == nil { + return + } + eu, ok := leader.(common.EvmUpstream) + if !ok { + return + } + sp := eu.EvmStatePoller() + if sp == nil || sp.IsObjectNull() { + return + } + l := sp.LatestBlock() + if l <= 0 || bn < l || bn > l+1 { + return + } + dr := req.Directives() + if dr == nil { + dr = &common.RequestDirectives{} + req.SetDirectives(dr) + } + if dr.UseUpstream != "" { + return + } + dr.UseUpstream = leader.Id() +} + // requestBlockNumber resolves the specific block number a request targets, // or 0 when the request has no block reference. Mirrors the extraction // path in checkUpstreamBlockAvailability so routing and gating see the