fix(ws): pin near-tip getBlock to EvmLeaderUpstream

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.

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