From a785c15e5e0d61e60b79c9ff3b994409a16d32c0 Mon Sep 17 00:00:00 2001 From: shpookas Date: Thu, 30 Jul 2026 14:20:59 +0200 Subject: [PATCH 1/7] fix(ws): skip tier:fallback escape on TipHW tip re-fetch Empty tip races on healthy primaries were fanning out to pay-per-call fallbacks (Infura etc.) via the emptyish escape hatch. Latest TipHW tip re-fetch now sets SkipFallbackEscape so TipHW still refuses stale latest without burning 3P quota; finalized re-fetch keeps escape for HA. Direct client concrete-block escapes are unchanged. --- architecture/evm/eth_getBlockByNumber.go | 15 ++- common/request.go | 8 ++ erpc/http_server_ws_tip_leader_test.go | 150 +++++++++++++++++++++++ erpc/networks.go | 12 +- internal/policy/testing.go | 28 +++-- 5 files changed, 196 insertions(+), 17 deletions(-) diff --git a/architecture/evm/eth_getBlockByNumber.go b/architecture/evm/eth_getBlockByNumber.go index 451a979fc..4bf0614f2 100644 --- a/architecture/evm/eth_getBlockByNumber.go +++ b/architecture/evm/eth_getBlockByNumber.go @@ -251,7 +251,9 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common // Do not use pickHighestBlock against the stale "latest" response — // that helper fail-opens to stale when the tip re-fetch misses, which // is exactly the MultiNode FOOS / EnforceRepeatableRead trigger. - nnr, ferr := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, useUpstream) + // SkipFallbackEscape: empty tip races must refuse-stale rather than + // escape to pay-per-call tier:fallback upstreams (Infura etc.). + nnr, ferr := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, useUpstream, true) if meetsTipFloor(ctx, nnr, highestBlockNumber) { if nr != nil { nr.Release() @@ -264,9 +266,8 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common // Pinned / excluded re-fetch missed the tip (sibling fullnode // lag, WS JSON-RPC miss, etc.). Retry with no UseUpstream pin - // so every upstream (including fallbacks via escape) can serve - // the concrete TipHW block. - nnr2, ferr2 := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, "") + // so remaining primaries can serve the concrete TipHW block. + nnr2, ferr2 := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, "", true) if meetsTipFloor(ctx, nnr2, highestBlockNumber) { if nr != nil { nr.Release() @@ -330,7 +331,9 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common if respBlockNumber > 0 { useUpstream = fmt.Sprintf("!%s", nr.UpstreamId()) } - nnr, err := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, useUpstream) + // Finalized re-fetch keeps fallback escape available for HA — + // SkipFallbackEscape is tip/latest-only (paid tip-race burn). + nnr, err := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, useUpstream, false) return pickHighestBlock(ctx, nnr, nr, err) default: return nr, re @@ -392,6 +395,7 @@ func forwardGetBlockByNumber( blockNumber int64, includeTx bool, useUpstream string, + skipFallbackEscape bool, ) (*common.NormalizedResponse, error) { request, err := BuildGetBlockByNumberRequest(blockNumber, includeTx) if err != nil { @@ -404,6 +408,7 @@ func forwardGetBlockByNumber( dr := original.Directives().Clone() dr.SkipCacheRead = "true" dr.UseUpstream = useUpstream + dr.SkipFallbackEscape = skipFallbackEscape newReq.SetDirectives(dr) newReq.SetNetwork(network) newReq.CopyHttpContextFrom(original) diff --git a/common/request.go b/common/request.go index f7dd6161c..4400d80c5 100644 --- a/common/request.go +++ b/common/request.go @@ -147,6 +147,12 @@ type RequestDirectives struct { // timeout still applies. Never set from HTTP headers. IsInternal bool `json:"-"` + // SkipFallbackEscape suppresses the per-request tier:fallback escape + // hatch for this request. Used by TipHW tip re-fetch so empty tip races + // on healthy primaries do not fan out to pay-per-call fallbacks + // (Infura etc.). Never set from HTTP headers. + SkipFallbackEscape bool `json:"-"` + // Instruct the normalization layer to avoid mutating JSON-RPC params for block tag interpolation. // When true, the system will still compute and cache block references (for finality/metrics), // but will NOT replace tags like "latest"/"finalized" with hex numbers in outbound requests. @@ -244,6 +250,8 @@ func (d *RequestDirectives) Clone() *RequestDirectives { SkipCacheRead: d.SkipCacheRead, UseUpstream: d.UseUpstream, ByPassMethodExclusion: d.ByPassMethodExclusion, + IsInternal: d.IsInternal, + SkipFallbackEscape: d.SkipFallbackEscape, SkipInterpolation: d.SkipInterpolation, SkipConsensus: d.SkipConsensus, EnforceHighestBlock: d.EnforceHighestBlock, diff --git a/erpc/http_server_ws_tip_leader_test.go b/erpc/http_server_ws_tip_leader_test.go index 5dda808b3..59a3c226e 100644 --- a/erpc/http_server_ws_tip_leader_test.go +++ b/erpc/http_server_ws_tip_leader_test.go @@ -11,9 +11,11 @@ import ( "github.com/bytedance/sonic" "github.com/erpc/erpc/common" "github.com/erpc/erpc/internal/policy" + "github.com/erpc/erpc/telemetry" "github.com/erpc/erpc/upstream" "github.com/erpc/erpc/util" "github.com/h2non/gock" + promUtil "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -269,3 +271,151 @@ func TestHttpServer_GetBlockByNumberLatest_RefusesStaleFailOpen(t *testing.T) { require.True(t, hasErr || statusCode >= 400, "expected error when tip re-fetch cannot reach TipHW, got status=%d body=%s", statusCode, body) } + +// TipHW tip re-fetch must refuse-stale when primaries miss the concrete tip, +// without escaping to tier:fallback pay-per-call upstreams. Goes through the +// HTTP → project doForward → HandleNetworkPostForward path (not bare +// Network.Forward, which skips TipHW enforcement). +func TestHttpServer_GetBlockByNumberLatest_TipRefetchSkipsFallbackEscape(t *testing.T) { + util.ResetGock() + defer util.ResetGock() + util.SetupMocksForEvmStatePoller() + // Persist tip mocks on primary + fallback remain pending by design. + defer util.AssertNoPendingMocks(t, 2) + + const tip = int64(0x11118889) + tipHex := "0x11118889" + staleHex := "0x11118888" + + var fallbackHits atomic.Int64 + // Primary tip re-fetch misses. + gock.New("http://rpc1.localhost"). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + return strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, tipHex) + }). + Reply(200). + JSON([]byte(`{"result":null}`)) + // Fallback would serve TipHW — must not be reached via escape on tip re-fetch. + gock.New("http://rpc2.localhost"). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + if strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, tipHex) { + fallbackHits.Add(1) + return true + } + return false + }). + Reply(200). + JSON([]byte(`{"result":{"number":"0x11118889","hash":"0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","parentHash":"0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","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, + Integrity: &common.EvmIntegrityConfig{ + EnforceHighestBlock: util.BoolPtr(true), + }, + }, + // Freeze policy ticks so a lazy method-slot eval cannot + // re-introduce the cordoned fallback mid-request. + SelectionPolicy: &common.SelectionPolicyConfig{ + EvalInterval: 0, + }, + Failover: &common.FailoverConfig{ + OnDefaultsExhausted: util.BoolPtr(true), + }, + 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, + Tags: []string{common.TagTierFallback}, + 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) + // Pin ordered list to primary only — mirrors preferTag cordoning + // fallbacks while a primary is healthy. Escape would be the only way + // to reach rpc2; SkipFallbackEscape must block that. + policy.OverrideOrderForTest(prj.policyEngine, "evm:123", "rpc1") + + time.Sleep(500 * time.Millisecond) + + nw, err := prj.GetNetwork(context.Background(), "evm:123") + require.NoError(t, err) + require.Equal(t, []string{"rpc1"}, nw.PolicyOrderedUpstreams("eth_getBlockByNumber"), + "fallback must stay cordoned so only escape could reach it") + nw.NoteObservedLatestBlock(context.Background(), tip) + require.Equal(t, tip, nw.EvmHighestLatestBlockNumber(context.Background())) + + escapeCounter := telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues( + "test_project", "evm:123", "eth_getBlockByNumber", + ) + escapeBefore := promUtil.ToFloat64(escapeCounter) + + statusCode, _, body := sendRequest(`{ + "jsonrpc": "2.0", + "id": 1, + "method": "eth_getBlockByNumber", + "params": ["latest", false] + }`, nil, nil) + + var respObject map[string]interface{} + require.NoError(t, sonic.UnmarshalString(body, &respObject)) + if result, ok := respObject["result"].(map[string]interface{}); ok { + require.NotEqual(t, staleHex, result["number"], + "must not fail-open to stale tip below TipHW; status=%d body=%s", statusCode, body) + require.NotEqual(t, tipHex, result["number"], + "must not serve TipHW from fallback escape; body=%s", body) + } + _, hasErr := respObject["error"] + require.True(t, hasErr || statusCode >= 400, + "expected refuse-stale error when tip re-fetch misses without fallback escape; status=%d body=%s", + statusCode, body) + + assert.Equal(t, escapeBefore, promUtil.ToFloat64(escapeCounter), + "TipHW tip re-fetch must not fire fallback escape") + // fallbackHits may still move from background pollers probing tip hex; + // escape counter + refuse-stale response are the request-path proofs. + _ = fallbackHits +} diff --git a/erpc/networks.go b/erpc/networks.go index 6ee3e0821..fa97e91b9 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -536,8 +536,10 @@ func (n *Network) EvmLowestFinalizedBlockNumber(ctx context.Context) int64 { // latest tip. Fallback-tier upstreams are ignored while any primary is up — // otherwise tip re-fetch pins UseUpstream to a cordoned fallback that is not // in the ordered primary list. TipHW may still advance from fallback WS -// (fan-out invariant); unconstrained tip re-fetch + emptyish escape reaches -// those fallbacks when primaries miss. +// (fan-out invariant). Tip re-fetch sets SkipFallbackEscape so empty tip +// races on healthy primaries refuse-stale instead of burning pay-per-call +// fallbacks; real HA still reaches fallbacks via selectionPolicy when +// primaries are down. func (n *Network) EvmLeaderUpstream(ctx context.Context) common.Upstream { var leader, fallbackLeader common.Upstream var leaderLastBlock, fallbackLastBlock int64 @@ -1085,8 +1087,14 @@ func (n *Network) Forward(ctx context.Context, req *common.NormalizedRequest) (* // couldn't serve; try a different one" — exactly the escape's job. // - Consensus requires strict per-upstream semantics; don't modify // the candidate set mid-execution. + // - SkipFallbackEscape (TipHW tip re-fetch) blocks escape so tip + // races on healthy primaries refuse-stale instead of fanning + // out to pay-per-call tier:fallback upstreams. + dirs := effectiveReq.Directives() + skipFallbackEscape := dirs != nil && dirs.SkipFallbackEscape bestRespEmptyish := bestResp != nil && bestResp.IsResultEmptyish() if (bestResp == nil || bestRespEmptyish) && + !skipFallbackEscape && !effectiveReq.HasEscalatedToFallbacks() && lastErr != nil && n.cfg.Failover != nil && n.cfg.Failover.Enabled() && diff --git a/internal/policy/testing.go b/internal/policy/testing.go index 47bd1a333..457c77610 100644 --- a/internal/policy/testing.go +++ b/internal/policy/testing.go @@ -19,18 +19,12 @@ import ( // the duration of the test should set `EvalInterval: 0` on the network's // SelectionPolicy. func OverrideOrderForTest(e *Engine, networkID string, ids ...string) { - e.mu.RLock() - slot, ok := e.slots[slotKey{networkID, "*", "*"}] + e.mu.Lock() + defer e.mu.Unlock() reg := e.networks[networkID] - e.mu.RUnlock() - if !ok || reg == nil { + if reg == nil { return } - // Stop the ticker for the lifetime of this override — tests that pin - // don't want background re-eval clobbering their cache mid-test, and - // thousands of test fixtures running at 1s tick each pushes the - // race-detector CI suite past its 20-minute budget. - slot.stop() ups := reg.upstreamsFn() index := make(map[string]common.Upstream, len(ups)) for _, u := range ups { @@ -49,7 +43,21 @@ func OverrideOrderForTest(e *Engine, networkID string, ids ...string) { ordered = append(ordered, u) } } - slot.cache.Store(&ordered) + // Pin every slot for this network (wildcard + method/finality-narrow). + // GetOrdered prefers a populated narrow slot over the wildcard, so + // overriding only ("*", "*") left tip-re-fetch tests selecting + // cordoned fallbacks from a stale eth_getBlockByNumber cache. + for k, slot := range e.slots { + if k.network != networkID { + continue + } + // Stop the ticker for the lifetime of this override — tests that pin + // don't want background re-eval clobbering their cache mid-test, and + // thousands of test fixtures running at 1s tick each pushes the + // race-detector CI suite past its 20-minute budget. + slot.stop() + slot.cache.Store(&ordered) + } } // OverrideAllForTest applies OverrideOrderForTest to EVERY network this From 738ca8657d406e4c3babca4ea4dade522d7d1ece Mon Sep 17 00:00:00 2001 From: shpookas Date: Thu, 30 Jul 2026 15:23:20 +0200 Subject: [PATCH 2/7] fix(ws): pin tip re-fetch to WS tip-source upstream Cross-node tip race: TipHW from reth-0 WS then getBlock on lagging reth-1 returns null. Record tip-source on SuggestLatestBlock, bump HTTP twin poller, and pin EnforceHighestBlock re-fetch to that source (fallback EvmLeaderUpstream). Keep SkipFallbackEscape so misses refuse-stale instead of burning Infura. Co-authored-by: Cursor --- architecture/evm/eth_getBlockByNumber.go | 39 +++-- erpc/http_server_ws_tip_leader_test.go | 112 +++++++++++++ erpc/networks.go | 63 ++++++++ erpc/networks_ws_tip_test.go | 195 +++++++++++++++++++++++ erpc/subscription_manager.go | 81 +++++++++- 5 files changed, 472 insertions(+), 18 deletions(-) diff --git a/architecture/evm/eth_getBlockByNumber.go b/architecture/evm/eth_getBlockByNumber.go index 4bf0614f2..6b4f268b0 100644 --- a/architecture/evm/eth_getBlockByNumber.go +++ b/architecture/evm/eth_getBlockByNumber.go @@ -19,6 +19,13 @@ type tipRefresher interface { EvmRefreshHighestLatestBlockNumber(ctx context.Context) int64 } +// tipSourceProvider is implemented by *erpc.Network. Optional so test +// doubles keep compiling. Returns the WS tip-source upstream id for a +// TipHW block when known on this pod. +type tipSourceProvider interface { + EvmTipSourceUpstreamId(blockNumber int64) string +} + func refreshHighestLatestBlockNumber(ctx context.Context, network common.Network) int64 { if r, ok := network.(tipRefresher); ok { return r.EvmRefreshHighestLatestBlockNumber(ctx) @@ -225,21 +232,25 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common ).Inc() } - // Prefer the upstream whose poller already owns this tip - // (EvmLeaderUpstream — typically the WS ingress that called - // SuggestLatestBlock). If TipHW advanced via Redis/WS while - // local pollers lag inside their debounce window, force-poll - // the leader once before deciding. Fall back to excluding the - // stale responder when no local poller has caught up yet. + // Prefer the upstream that delivered this TipHW via WS newHeads + // (tip-source pin). Fall back to EvmLeaderUpstream (poller max), + // then to excluding the stale responder. Tip-source matches the + // cross-node race: TipHW from reth-0 WS must not re-fetch on + // lagging reth-1. useUpstream := "" - if leader := network.EvmLeaderUpstream(ctx); leader != nil { - if eu, ok := leader.(common.EvmUpstream); ok { - if sp := eu.EvmStatePoller(); sp != nil && !sp.IsObjectNull() { - if sp.LatestBlock() < highestBlockNumber { - _, _ = sp.PollLatestBlockNumberNow(ctx) - } - if sp.LatestBlock() >= highestBlockNumber { - useUpstream = leader.Id() + if provider, ok := network.(tipSourceProvider); ok { + useUpstream = provider.EvmTipSourceUpstreamId(highestBlockNumber) + } + if useUpstream == "" { + if leader := network.EvmLeaderUpstream(ctx); leader != nil { + if eu, ok := leader.(common.EvmUpstream); ok { + if sp := eu.EvmStatePoller(); sp != nil && !sp.IsObjectNull() { + if sp.LatestBlock() < highestBlockNumber { + _, _ = sp.PollLatestBlockNumberNow(ctx) + } + if sp.LatestBlock() >= highestBlockNumber { + useUpstream = leader.Id() + } } } } diff --git a/erpc/http_server_ws_tip_leader_test.go b/erpc/http_server_ws_tip_leader_test.go index 59a3c226e..5400f29d6 100644 --- a/erpc/http_server_ws_tip_leader_test.go +++ b/erpc/http_server_ws_tip_leader_test.go @@ -272,6 +272,118 @@ func TestHttpServer_GetBlockByNumberLatest_RefusesStaleFailOpen(t *testing.T) { "expected error when tip re-fetch cannot reach TipHW, got status=%d body=%s", statusCode, body) } +// After WS ingest records a tip-source upstream, EnforceHighestBlock must pin +// the concrete tip re-fetch to that upstream — even when its poller was not +// bumped (partition cannot help) and a lagging sibling answers "latest" first. +func TestHttpServer_GetBlockByNumberLatest_PinsReFetchToTipSourceUpstream(t *testing.T) { + util.ResetGock() + defer util.ResetGock() + util.SetupMocksForEvmStatePoller() + defer util.AssertNoPendingMocks(t, 0) + + const tip = int64(0x33338889) + tipHex := "0x33338889" + var tipSourceHits atomic.Int64 + + gock.New("http://rpc2.localhost"). + Post(""). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + if !strings.Contains(body, "eth_getBlockByNumber") || !strings.Contains(body, tipHex) { + return false + } + tipSourceHits.Add(1) + return true + }). + 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, + Integrity: &common.EvmIntegrityConfig{ + EnforceHighestBlock: util.BoolPtr(true), + }, + }, + Failsafe: []*common.FailsafeConfig{ + { + Retry: &common.RetryPolicyConfig{MaxAttempts: 3}, + }, + }, + }, + }, + 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) + // Prefer lagging HTTP first so tip-source pin (not partition) is exercised. + 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) + + // TipHW + tip-source id WITHOUT bumping rpc2's poller — partition cannot + // reorder rpc2 ahead; pin must select it on EnforceHighestBlock re-fetch. + nw.noteTipSource(tip, "rpc2") + nw.NoteObservedLatestBlock(context.Background(), tip) + require.Equal(t, tip, nw.EvmHighestLatestBlockNumber(context.Background())) + require.Equal(t, "rpc2", nw.EvmTipSourceUpstreamId(tip)) + + statusCode, _, body := sendRequest(`{ + "jsonrpc": "2.0", + "id": 1, + "method": "eth_getBlockByNumber", + "params": ["latest", false] + }`, nil, nil) + + require.Equal(t, http.StatusOK, statusCode) + + var respObject map[string]interface{} + require.NoError(t, sonic.UnmarshalString(body, &respObject)) + result, ok := respObject["result"].(map[string]interface{}) + require.True(t, ok, "response should have a result object, got: %s", body) + assert.Equal(t, tipHex, result["number"]) + assert.GreaterOrEqual(t, tipSourceHits.Load(), int64(1), + "EnforceHighestBlock must pin tip re-fetch to the tip-source upstream") +} + // TipHW tip re-fetch must refuse-stale when primaries miss the concrete tip, // without escaping to tier:fallback pay-per-call upstreams. Goes through the // HTTP → project doForward → HandleNetworkPostForward path (not bare diff --git a/erpc/networks.go b/erpc/networks.go index fa97e91b9..c499796a2 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -68,6 +68,18 @@ type Network struct { // cannot regress below a head we have already delivered on the same pod. lastReturnedLatestBlock atomic.Int64 lastReturnedFinalizedBlock atomic.Int64 + + // lastTipSource is the upstream that most recently delivered TipHW via + // WS newHeads on this pod (set by networkHandle.SuggestLatestBlock). + // EnforceHighestBlock pins tip re-fetch to this upstream so HTTP does + // not land on a lagging sibling after TipHW advanced from another node. + lastTipSource atomic.Pointer[tipSourceMark] +} + +// tipSourceMark records which upstream delivered a given TipHW head. +type tipSourceMark struct { + blockNumber int64 + upstreamId string } // NoteObservedLatestBlock records that this Network has observed head @@ -107,6 +119,57 @@ func (n *Network) NoteObservedLatestBlock(ctx context.Context, blockNumber int64 } } +// noteTipSource records that upstreamId delivered head blockNumber via WS +// newHeads. Only advances when blockNumber is ≥ the previously recorded tip +// (equal tip updates the source to the most recent deliverer). +func (n *Network) noteTipSource(blockNumber int64, upstreamId string) { + if n == nil || blockNumber <= 0 || upstreamId == "" { + return + } + for { + cur := n.lastTipSource.Load() + if cur != nil && blockNumber < cur.blockNumber { + return + } + next := &tipSourceMark{blockNumber: blockNumber, upstreamId: upstreamId} + if n.lastTipSource.CompareAndSwap(cur, next) { + return + } + } +} + +// EvmTipSourceUpstreamId returns the upstream that delivered TipHW via WS +// newHeads on this pod when blockNumber matches that tip. Fallback-tier tip +// sources are ignored while any primary is up (same rule as EvmLeaderUpstream) +// so tip re-fetch does not pin UseUpstream to a cordoned Infura/etc. WS. +func (n *Network) EvmTipSourceUpstreamId(blockNumber int64) string { + if n == nil || blockNumber <= 0 { + return "" + } + mark := n.lastTipSource.Load() + if mark == nil || mark.blockNumber != blockNumber || mark.upstreamId == "" { + return "" + } + upsList := n.upstreamsRegistry.GetNetworkUpstreams(context.Background(), n.networkId) + var tipUp common.Upstream + anyPrimaryUp := false + for _, u := range upsList { + if u.Id() == mark.upstreamId { + tipUp = u + } + if u.Config() != nil && u.Config().HasTag(common.TagTierFallback) { + continue + } + if !u.IsDown() { + anyPrimaryUp = true + } + } + if tipUp != nil && tipUp.Config() != nil && tipUp.Config().HasTag(common.TagTierFallback) && anyPrimaryUp { + return "" + } + return mark.upstreamId +} + // EvmRefreshHighestLatestBlockNumber pulls TipHW from Redis once and returns // the network tip after adopting any higher remote value. Used when the local // TipHW cache would otherwise skip EnforceHighestBlock (false-negative under diff --git a/erpc/networks_ws_tip_test.go b/erpc/networks_ws_tip_test.go index 06ec4820b..c225bea67 100644 --- a/erpc/networks_ws_tip_test.go +++ b/erpc/networks_ws_tip_test.go @@ -190,6 +190,8 @@ func TestNetworkHandle_SuggestLatestBlock_AdvancesNetworkTipBeforeFanOut(t *test "network tip must advance before any client would see the WS head") assert.GreaterOrEqual(t, network.lastReturnedLatestBlock.Load(), int64(90677359), "process-local high-water mark must cover the delivered WS tip") + assert.Equal(t, "bor-1", network.EvmTipSourceUpstreamId(90677359), + "tip-source must be the WS ingress that delivered the head") } // Fallback WS tips must advance TipHW even while primaries are up: Ingest @@ -431,3 +433,196 @@ func TestEvmRefreshHighestLatestBlockNumber_PreservesObservedTip(t *testing.T) { "refresh after sync TipHW publish must keep the observed tip") assert.Equal(t, int64(1001), network.EvmHighestLatestBlockNumber(ctx)) } + +func TestHttpTwinUpstreamId(t *testing.T) { + assert.Equal(t, "internal-eth-mainnet-reth-0", httpTwinUpstreamId("internal-eth-mainnet-reth-ws-0")) + assert.Equal(t, "eth-mainnet-reth", httpTwinUpstreamId("eth-mainnet-reth-ws")) + assert.Equal(t, "", httpTwinUpstreamId("rpc1")) +} + +func TestHttpTwinEndpoint(t *testing.T) { + assert.Equal(t, + "https://eth-mainnet-reth.internal.linkpool.com/0", + httpTwinEndpoint("wss://eth-mainnet-reth.internal.linkpool.com/0/ws"), + ) + assert.Equal(t, "http://host/1", httpTwinEndpoint("ws://host/1/websocket")) + assert.Equal(t, "", httpTwinEndpoint("https://already-http.example/0")) +} + +// SuggestLatestBlock must record tip-source and bump the HTTP twin poller so +// partition/leader prefer the same physical node that delivered newHeads. +func TestNetworkHandle_SuggestLatestBlock_RecordsTipSourceAndBumpsHttpTwin(t *testing.T) { + util.ResetGock() + defer util.ResetGock() + util.SetupMocksForEvmStatePoller() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + wsUp := &common.UpstreamConfig{ + Type: common.UpstreamTypeEvm, + Id: "internal-eth-mainnet-reth-ws-0", + Endpoint: "wss://eth-mainnet-reth.internal.linkpool.com/0/ws", + Evm: &common.EvmUpstreamConfig{ChainId: 123}, + } + httpUp := &common.UpstreamConfig{ + Type: common.UpstreamTypeEvm, + Id: "internal-eth-mainnet-reth-0", + Endpoint: "https://eth-mainnet-reth.internal.linkpool.com/0", + Evm: &common.EvmUpstreamConfig{ChainId: 123}, + } + sibling := &common.UpstreamConfig{ + Type: common.UpstreamTypeEvm, + Id: "internal-eth-mainnet-reth-1", + Endpoint: "https://eth-mainnet-reth.internal.linkpool.com/1", + Evm: &common.EvmUpstreamConfig{ChainId: 123}, + } + + for _, host := range []string{ + "eth-mainnet-reth.internal.linkpool.com/0", + "eth-mainnet-reth.internal.linkpool.com/1", + } { + gock.New("https://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return strings.Contains(util.SafeReadBody(r), `eth_chainId`) + }). + Reply(200). + JSON([]byte(`{"result":"0x7b"}`)) + } + // WS upstream may still hit https after scheme normalize in some paths; + // chainId mocks above cover HTTP twins. WS client bootstrap is separate. + + rateLimitersRegistry, _ := upstream.NewRateLimitersRegistry(context.Background(), &common.RateLimiterConfig{}, &log.Logger) + metricsTracker := health.NewTracker(&log.Logger, "test", time.Minute) + vr := thirdparty.NewVendorsRegistry() + pr, err := thirdparty.NewProvidersRegistry(&log.Logger, vr, []*common.ProviderConfig{}, nil) + require.NoError(t, err) + ssr, err := data.NewSharedStateRegistry(ctx, &log.Logger, &common.SharedStateConfig{ + Connector: &common.ConnectorConfig{ + Driver: "memory", + Memory: &common.MemoryConnectorConfig{MaxItems: 100_000, MaxTotalSize: "1GB"}, + }, + }) + require.NoError(t, err) + + // Use http endpoints for all three so bootstrap does not need a live WS. + wsUp.Endpoint = "http://rpc-ws-twin.localhost" + httpUp.Endpoint = "http://rpc-http-twin.localhost" + sibling.Endpoint = "http://rpc-sibling.localhost" + for _, host := range []string{"rpc-ws-twin.localhost", "rpc-http-twin.localhost", "rpc-sibling.localhost"} { + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return strings.Contains(util.SafeReadBody(r), `eth_chainId`) + }). + Reply(200). + JSON([]byte(`{"result":"0x7b"}`)) + } + + upstreamsRegistry := upstream.NewUpstreamsRegistry( + ctx, &log.Logger, "test", + []*common.UpstreamConfig{wsUp, httpUp, sibling}, ssr, rateLimitersRegistry, vr, pr, nil, + metricsTracker, nil, + ) + networkConfig := &common.NetworkConfig{ + Architecture: common.ArchitectureEvm, + Evm: &common.EvmNetworkConfig{ChainId: 123}, + } + network, err := NewNetwork(ctx, &log.Logger, "test", networkConfig, + rateLimitersRegistry, upstreamsRegistry, metricsTracker, nil) + require.NoError(t, err) + + upstreamsRegistry.Bootstrap(ctx) + time.Sleep(200 * time.Millisecond) + require.NoError(t, upstreamsRegistry.GetInitializer().WaitForTasks(ctx)) + require.NoError(t, network.Bootstrap(ctx)) + time.Sleep(250 * time.Millisecond) + + upsList := upstreamsRegistry.GetNetworkUpstreams(ctx, util.EvmNetworkId(123)) + require.Len(t, upsList, 3) + byID := map[string]*upstream.Upstream{} + for _, u := range upsList { + byID[u.Id()] = u + u.EvmStatePoller().SuggestLatestBlock(1000) + } + time.Sleep(50 * time.Millisecond) + + handle := &networkHandle{nw: network} + handle.SuggestLatestBlock("ws:internal-eth-mainnet-reth-ws-0", 1001, []byte(`{"number":"0x3e9"}`)) + + assert.Equal(t, int64(1001), byID["internal-eth-mainnet-reth-ws-0"].EvmStatePoller().LatestBlock()) + assert.Equal(t, int64(1001), byID["internal-eth-mainnet-reth-0"].EvmStatePoller().LatestBlock(), + "HTTP twin poller must advance with WS tip (same physical node)") + assert.Equal(t, int64(1000), byID["internal-eth-mainnet-reth-1"].EvmStatePoller().LatestBlock(), + "lagging sibling must not be bumped") + assert.Equal(t, "internal-eth-mainnet-reth-ws-0", network.EvmTipSourceUpstreamId(1001)) + assert.Equal(t, int64(1001), network.EvmHighestLatestBlockNumber(ctx)) +} + +func TestEvmTipSourceUpstreamId_IgnoresFallbackWhilePrimaryUp(t *testing.T) { + util.ResetGock() + defer util.ResetGock() + util.SetupMocksForEvmStatePoller() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + primary := &common.UpstreamConfig{ + Type: common.UpstreamTypeEvm, + Id: "primary", + Endpoint: "http://primary.localhost", + Evm: &common.EvmUpstreamConfig{ChainId: 123}, + } + fallback := &common.UpstreamConfig{ + Type: common.UpstreamTypeEvm, + Id: "fallback-ws", + Endpoint: "http://fallback.localhost", + Tags: []string{common.TagTierFallback}, + Evm: &common.EvmUpstreamConfig{ChainId: 123}, + } + for _, host := range []string{"primary.localhost", "fallback.localhost"} { + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return strings.Contains(util.SafeReadBody(r), `eth_chainId`) + }). + Reply(200). + JSON([]byte(`{"result":"0x7b"}`)) + } + + rateLimitersRegistry, _ := upstream.NewRateLimitersRegistry(context.Background(), &common.RateLimiterConfig{}, &log.Logger) + metricsTracker := health.NewTracker(&log.Logger, "test", time.Minute) + vr := thirdparty.NewVendorsRegistry() + pr, err := thirdparty.NewProvidersRegistry(&log.Logger, vr, []*common.ProviderConfig{}, nil) + require.NoError(t, err) + ssr, err := data.NewSharedStateRegistry(ctx, &log.Logger, &common.SharedStateConfig{ + Connector: &common.ConnectorConfig{ + Driver: "memory", + Memory: &common.MemoryConnectorConfig{MaxItems: 100_000, MaxTotalSize: "1GB"}, + }, + }) + require.NoError(t, err) + + upstreamsRegistry := upstream.NewUpstreamsRegistry( + ctx, &log.Logger, "test", + []*common.UpstreamConfig{primary, fallback}, ssr, rateLimitersRegistry, vr, pr, nil, + metricsTracker, nil, + ) + network, err := NewNetwork(ctx, &log.Logger, "test", + &common.NetworkConfig{Architecture: common.ArchitectureEvm, Evm: &common.EvmNetworkConfig{ChainId: 123}}, + rateLimitersRegistry, upstreamsRegistry, metricsTracker, nil) + require.NoError(t, err) + upstreamsRegistry.Bootstrap(ctx) + time.Sleep(200 * time.Millisecond) + require.NoError(t, upstreamsRegistry.GetInitializer().WaitForTasks(ctx)) + require.NoError(t, network.Bootstrap(ctx)) + time.Sleep(250 * time.Millisecond) + + network.noteTipSource(2000, "fallback-ws") + assert.Equal(t, "", network.EvmTipSourceUpstreamId(2000), + "must not pin tip re-fetch to fallback WS while a primary is up") +} diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go index 2c05546ca..26a6dc017 100644 --- a/erpc/subscription_manager.go +++ b/erpc/subscription_manager.go @@ -579,9 +579,11 @@ func (h *networkHandle) FinalityDepth() int64 { // pod is already ≥ N. That invariant applies to every ingress source, // including tier:fallback: Ingest fans out all sources, so skipping TipHW // for fallback heads while still delivering them to clients causes -// MultiNode FOOS (WS tip ahead of HTTP TipHW). Tip re-fetch of a TipHW -// that came from a fallback must reach that fallback via the emptyish -// escape hatch instead. +// MultiNode FOOS (WS tip ahead of HTTP TipHW). +// +// Also records the tip-source upstream id (for EnforceHighestBlock pin) and +// bumps any HTTP twin's poller so partitionUpstreamsByLatestBlock prefers +// the same physical node for subsequent eth_getBlockByNumber tip reads. func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64, payload json.RawMessage) { _ = payload const prefix = "ws:" @@ -589,19 +591,90 @@ func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64, p return } upstreamID := sourceId[len(prefix):] - for _, u := range h.nw.upstreamsRegistry.GetNetworkUpstreams(context.Background(), h.nw.networkId) { + var wsUp *upstream.Upstream + ups := h.nw.upstreamsRegistry.GetNetworkUpstreams(context.Background(), h.nw.networkId) + for _, u := range ups { if u.Id() != upstreamID { continue } + wsUp = u poller := u.EvmStatePoller() if poller != nil && !poller.IsObjectNull() { poller.SuggestLatestBlock(blockNumber) } break } + h.nw.noteTipSource(blockNumber, upstreamID) + if wsUp != nil { + suggestHttpTwinLatestBlock(ups, wsUp, blockNumber) + } h.nw.NoteObservedLatestBlock(h.nw.appCtx, blockNumber) } +// suggestHttpTwinLatestBlock advances pollers for HTTP upstreams that are +// the same physical node as wsUp (id convention *-ws-* / endpoint /ws twin). +// Same-node live tests show WS newHeads implies HTTP getBlock is immediately +// available on that node; the lagging sibling is the real tip-race. +func suggestHttpTwinLatestBlock(ups []*upstream.Upstream, wsUp *upstream.Upstream, blockNumber int64) { + if wsUp == nil || blockNumber <= 0 { + return + } + twinID := httpTwinUpstreamId(wsUp.Id()) + wsEp := "" + if cfg := wsUp.Config(); cfg != nil { + wsEp = cfg.Endpoint + } + httpTwinEp := httpTwinEndpoint(wsEp) + for _, u := range ups { + if u == nil || u.Id() == wsUp.Id() { + continue + } + match := twinID != "" && u.Id() == twinID + if !match && httpTwinEp != "" { + if cfg := u.Config(); cfg != nil && cfg.Endpoint == httpTwinEp { + match = true + } + } + if !match { + continue + } + poller := u.EvmStatePoller() + if poller != nil && !poller.IsObjectNull() { + poller.SuggestLatestBlock(blockNumber) + } + } +} + +// httpTwinUpstreamId maps internal-eth-mainnet-reth-ws-0 → internal-eth-mainnet-reth-0. +func httpTwinUpstreamId(wsUpstreamId string) string { + if strings.Contains(wsUpstreamId, "-ws-") { + return strings.Replace(wsUpstreamId, "-ws-", "-", 1) + } + if strings.HasSuffix(wsUpstreamId, "-ws") { + return strings.TrimSuffix(wsUpstreamId, "-ws") + } + return "" +} + +// httpTwinEndpoint maps wss://host/0/ws → https://host/0 (and ws→http). +func httpTwinEndpoint(wsEndpoint string) string { + if wsEndpoint == "" { + return "" + } + ep := wsEndpoint + switch { + case strings.HasPrefix(ep, "wss://"): + ep = "https://" + strings.TrimPrefix(ep, "wss://") + case strings.HasPrefix(ep, "ws://"): + ep = "http://" + strings.TrimPrefix(ep, "ws://") + default: + return "" + } + ep = strings.TrimSuffix(ep, "/ws") + ep = strings.TrimSuffix(ep, "/websocket") + return ep +} + // Interface checks: fail the build if either contract drifts. var ( _ wsclient.NotificationWriter = (*WsConnection)(nil) From 75cbdf9657722e9331bf34bc2812e79b11c471f2 Mon Sep 17 00:00:00 2001 From: shpookas Date: Thu, 30 Jul 2026 15:26:27 +0200 Subject: [PATCH 3/7] fix(ws): slim tip fix to HTTP twin poller bump Drop tip-source registry. Bumping the *-ws-* HTTP twin on SuggestLatestBlock is enough for existing EvmLeaderUpstream / partition routing. Keep SkipFallbackEscape for Infura. Co-authored-by: Cursor --- architecture/evm/eth_getBlockByNumber.go | 39 ++--- erpc/http_server_ws_tip_leader_test.go | 112 --------------- erpc/networks.go | 63 -------- erpc/networks_ws_tip_test.go | 174 +++-------------------- erpc/subscription_manager.go | 91 ++---------- 5 files changed, 44 insertions(+), 435 deletions(-) diff --git a/architecture/evm/eth_getBlockByNumber.go b/architecture/evm/eth_getBlockByNumber.go index 6b4f268b0..ca21f02db 100644 --- a/architecture/evm/eth_getBlockByNumber.go +++ b/architecture/evm/eth_getBlockByNumber.go @@ -19,13 +19,6 @@ type tipRefresher interface { EvmRefreshHighestLatestBlockNumber(ctx context.Context) int64 } -// tipSourceProvider is implemented by *erpc.Network. Optional so test -// doubles keep compiling. Returns the WS tip-source upstream id for a -// TipHW block when known on this pod. -type tipSourceProvider interface { - EvmTipSourceUpstreamId(blockNumber int64) string -} - func refreshHighestLatestBlockNumber(ctx context.Context, network common.Network) int64 { if r, ok := network.(tipRefresher); ok { return r.EvmRefreshHighestLatestBlockNumber(ctx) @@ -232,25 +225,21 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common ).Inc() } - // Prefer the upstream that delivered this TipHW via WS newHeads - // (tip-source pin). Fall back to EvmLeaderUpstream (poller max), - // then to excluding the stale responder. Tip-source matches the - // cross-node race: TipHW from reth-0 WS must not re-fetch on - // lagging reth-1. + // Prefer the upstream whose poller already owns this tip + // (EvmLeaderUpstream — typically the WS ingress / HTTP twin that + // SuggestLatestBlock advanced). If TipHW advanced via Redis/WS while + // local pollers lag inside their debounce window, force-poll + // the leader once before deciding. Fall back to excluding the + // stale responder when no local poller has caught up yet. useUpstream := "" - if provider, ok := network.(tipSourceProvider); ok { - useUpstream = provider.EvmTipSourceUpstreamId(highestBlockNumber) - } - if useUpstream == "" { - if leader := network.EvmLeaderUpstream(ctx); leader != nil { - if eu, ok := leader.(common.EvmUpstream); ok { - if sp := eu.EvmStatePoller(); sp != nil && !sp.IsObjectNull() { - if sp.LatestBlock() < highestBlockNumber { - _, _ = sp.PollLatestBlockNumberNow(ctx) - } - if sp.LatestBlock() >= highestBlockNumber { - useUpstream = leader.Id() - } + if leader := network.EvmLeaderUpstream(ctx); leader != nil { + if eu, ok := leader.(common.EvmUpstream); ok { + if sp := eu.EvmStatePoller(); sp != nil && !sp.IsObjectNull() { + if sp.LatestBlock() < highestBlockNumber { + _, _ = sp.PollLatestBlockNumberNow(ctx) + } + if sp.LatestBlock() >= highestBlockNumber { + useUpstream = leader.Id() } } } diff --git a/erpc/http_server_ws_tip_leader_test.go b/erpc/http_server_ws_tip_leader_test.go index 5400f29d6..59a3c226e 100644 --- a/erpc/http_server_ws_tip_leader_test.go +++ b/erpc/http_server_ws_tip_leader_test.go @@ -272,118 +272,6 @@ func TestHttpServer_GetBlockByNumberLatest_RefusesStaleFailOpen(t *testing.T) { "expected error when tip re-fetch cannot reach TipHW, got status=%d body=%s", statusCode, body) } -// After WS ingest records a tip-source upstream, EnforceHighestBlock must pin -// the concrete tip re-fetch to that upstream — even when its poller was not -// bumped (partition cannot help) and a lagging sibling answers "latest" first. -func TestHttpServer_GetBlockByNumberLatest_PinsReFetchToTipSourceUpstream(t *testing.T) { - util.ResetGock() - defer util.ResetGock() - util.SetupMocksForEvmStatePoller() - defer util.AssertNoPendingMocks(t, 0) - - const tip = int64(0x33338889) - tipHex := "0x33338889" - var tipSourceHits atomic.Int64 - - gock.New("http://rpc2.localhost"). - Post(""). - Filter(func(r *http.Request) bool { - body := util.SafeReadBody(r) - if !strings.Contains(body, "eth_getBlockByNumber") || !strings.Contains(body, tipHex) { - return false - } - tipSourceHits.Add(1) - return true - }). - 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, - Integrity: &common.EvmIntegrityConfig{ - EnforceHighestBlock: util.BoolPtr(true), - }, - }, - Failsafe: []*common.FailsafeConfig{ - { - Retry: &common.RetryPolicyConfig{MaxAttempts: 3}, - }, - }, - }, - }, - 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) - // Prefer lagging HTTP first so tip-source pin (not partition) is exercised. - 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) - - // TipHW + tip-source id WITHOUT bumping rpc2's poller — partition cannot - // reorder rpc2 ahead; pin must select it on EnforceHighestBlock re-fetch. - nw.noteTipSource(tip, "rpc2") - nw.NoteObservedLatestBlock(context.Background(), tip) - require.Equal(t, tip, nw.EvmHighestLatestBlockNumber(context.Background())) - require.Equal(t, "rpc2", nw.EvmTipSourceUpstreamId(tip)) - - statusCode, _, body := sendRequest(`{ - "jsonrpc": "2.0", - "id": 1, - "method": "eth_getBlockByNumber", - "params": ["latest", false] - }`, nil, nil) - - require.Equal(t, http.StatusOK, statusCode) - - var respObject map[string]interface{} - require.NoError(t, sonic.UnmarshalString(body, &respObject)) - result, ok := respObject["result"].(map[string]interface{}) - require.True(t, ok, "response should have a result object, got: %s", body) - assert.Equal(t, tipHex, result["number"]) - assert.GreaterOrEqual(t, tipSourceHits.Load(), int64(1), - "EnforceHighestBlock must pin tip re-fetch to the tip-source upstream") -} - // TipHW tip re-fetch must refuse-stale when primaries miss the concrete tip, // without escaping to tier:fallback pay-per-call upstreams. Goes through the // HTTP → project doForward → HandleNetworkPostForward path (not bare diff --git a/erpc/networks.go b/erpc/networks.go index c499796a2..fa97e91b9 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -68,18 +68,6 @@ type Network struct { // cannot regress below a head we have already delivered on the same pod. lastReturnedLatestBlock atomic.Int64 lastReturnedFinalizedBlock atomic.Int64 - - // lastTipSource is the upstream that most recently delivered TipHW via - // WS newHeads on this pod (set by networkHandle.SuggestLatestBlock). - // EnforceHighestBlock pins tip re-fetch to this upstream so HTTP does - // not land on a lagging sibling after TipHW advanced from another node. - lastTipSource atomic.Pointer[tipSourceMark] -} - -// tipSourceMark records which upstream delivered a given TipHW head. -type tipSourceMark struct { - blockNumber int64 - upstreamId string } // NoteObservedLatestBlock records that this Network has observed head @@ -119,57 +107,6 @@ func (n *Network) NoteObservedLatestBlock(ctx context.Context, blockNumber int64 } } -// noteTipSource records that upstreamId delivered head blockNumber via WS -// newHeads. Only advances when blockNumber is ≥ the previously recorded tip -// (equal tip updates the source to the most recent deliverer). -func (n *Network) noteTipSource(blockNumber int64, upstreamId string) { - if n == nil || blockNumber <= 0 || upstreamId == "" { - return - } - for { - cur := n.lastTipSource.Load() - if cur != nil && blockNumber < cur.blockNumber { - return - } - next := &tipSourceMark{blockNumber: blockNumber, upstreamId: upstreamId} - if n.lastTipSource.CompareAndSwap(cur, next) { - return - } - } -} - -// EvmTipSourceUpstreamId returns the upstream that delivered TipHW via WS -// newHeads on this pod when blockNumber matches that tip. Fallback-tier tip -// sources are ignored while any primary is up (same rule as EvmLeaderUpstream) -// so tip re-fetch does not pin UseUpstream to a cordoned Infura/etc. WS. -func (n *Network) EvmTipSourceUpstreamId(blockNumber int64) string { - if n == nil || blockNumber <= 0 { - return "" - } - mark := n.lastTipSource.Load() - if mark == nil || mark.blockNumber != blockNumber || mark.upstreamId == "" { - return "" - } - upsList := n.upstreamsRegistry.GetNetworkUpstreams(context.Background(), n.networkId) - var tipUp common.Upstream - anyPrimaryUp := false - for _, u := range upsList { - if u.Id() == mark.upstreamId { - tipUp = u - } - if u.Config() != nil && u.Config().HasTag(common.TagTierFallback) { - continue - } - if !u.IsDown() { - anyPrimaryUp = true - } - } - if tipUp != nil && tipUp.Config() != nil && tipUp.Config().HasTag(common.TagTierFallback) && anyPrimaryUp { - return "" - } - return mark.upstreamId -} - // EvmRefreshHighestLatestBlockNumber pulls TipHW from Redis once and returns // the network tip after adopting any higher remote value. Used when the local // TipHW cache would otherwise skip EnforceHighestBlock (false-negative under diff --git a/erpc/networks_ws_tip_test.go b/erpc/networks_ws_tip_test.go index c225bea67..6243ef933 100644 --- a/erpc/networks_ws_tip_test.go +++ b/erpc/networks_ws_tip_test.go @@ -190,8 +190,6 @@ func TestNetworkHandle_SuggestLatestBlock_AdvancesNetworkTipBeforeFanOut(t *test "network tip must advance before any client would see the WS head") assert.GreaterOrEqual(t, network.lastReturnedLatestBlock.Load(), int64(90677359), "process-local high-water mark must cover the delivered WS tip") - assert.Equal(t, "bor-1", network.EvmTipSourceUpstreamId(90677359), - "tip-source must be the WS ingress that delivered the head") } // Fallback WS tips must advance TipHW even while primaries are up: Ingest @@ -434,24 +432,9 @@ func TestEvmRefreshHighestLatestBlockNumber_PreservesObservedTip(t *testing.T) { assert.Equal(t, int64(1001), network.EvmHighestLatestBlockNumber(ctx)) } -func TestHttpTwinUpstreamId(t *testing.T) { - assert.Equal(t, "internal-eth-mainnet-reth-0", httpTwinUpstreamId("internal-eth-mainnet-reth-ws-0")) - assert.Equal(t, "eth-mainnet-reth", httpTwinUpstreamId("eth-mainnet-reth-ws")) - assert.Equal(t, "", httpTwinUpstreamId("rpc1")) -} - -func TestHttpTwinEndpoint(t *testing.T) { - assert.Equal(t, - "https://eth-mainnet-reth.internal.linkpool.com/0", - httpTwinEndpoint("wss://eth-mainnet-reth.internal.linkpool.com/0/ws"), - ) - assert.Equal(t, "http://host/1", httpTwinEndpoint("ws://host/1/websocket")) - assert.Equal(t, "", httpTwinEndpoint("https://already-http.example/0")) -} - -// SuggestLatestBlock must record tip-source and bump the HTTP twin poller so -// partition/leader prefer the same physical node that delivered newHeads. -func TestNetworkHandle_SuggestLatestBlock_RecordsTipSourceAndBumpsHttpTwin(t *testing.T) { +// SuggestLatestBlock on *-ws-* must also advance the HTTP twin poller so +// EvmLeaderUpstream / partition prefer the same physical node. +func TestNetworkHandle_SuggestLatestBlock_BumpsHttpTwin(t *testing.T) { util.ResetGock() defer util.ResetGock() util.SetupMocksForEvmStatePoller() @@ -459,40 +442,16 @@ func TestNetworkHandle_SuggestLatestBlock_RecordsTipSourceAndBumpsHttpTwin(t *te ctx, cancel := context.WithCancel(context.Background()) defer cancel() - wsUp := &common.UpstreamConfig{ - Type: common.UpstreamTypeEvm, - Id: "internal-eth-mainnet-reth-ws-0", - Endpoint: "wss://eth-mainnet-reth.internal.linkpool.com/0/ws", - Evm: &common.EvmUpstreamConfig{ChainId: 123}, + cfgs := []*common.UpstreamConfig{ + {Type: common.UpstreamTypeEvm, Id: "internal-eth-mainnet-reth-ws-0", Endpoint: "http://rpc-ws.localhost", Evm: &common.EvmUpstreamConfig{ChainId: 123}}, + {Type: common.UpstreamTypeEvm, Id: "internal-eth-mainnet-reth-0", Endpoint: "http://rpc-0.localhost", Evm: &common.EvmUpstreamConfig{ChainId: 123}}, + {Type: common.UpstreamTypeEvm, Id: "internal-eth-mainnet-reth-1", Endpoint: "http://rpc-1.localhost", Evm: &common.EvmUpstreamConfig{ChainId: 123}}, } - httpUp := &common.UpstreamConfig{ - Type: common.UpstreamTypeEvm, - Id: "internal-eth-mainnet-reth-0", - Endpoint: "https://eth-mainnet-reth.internal.linkpool.com/0", - Evm: &common.EvmUpstreamConfig{ChainId: 123}, - } - sibling := &common.UpstreamConfig{ - Type: common.UpstreamTypeEvm, - Id: "internal-eth-mainnet-reth-1", - Endpoint: "https://eth-mainnet-reth.internal.linkpool.com/1", - Evm: &common.EvmUpstreamConfig{ChainId: 123}, - } - - for _, host := range []string{ - "eth-mainnet-reth.internal.linkpool.com/0", - "eth-mainnet-reth.internal.linkpool.com/1", - } { - gock.New("https://" + host). - Post(""). - Persist(). - Filter(func(r *http.Request) bool { - return strings.Contains(util.SafeReadBody(r), `eth_chainId`) - }). - Reply(200). - JSON([]byte(`{"result":"0x7b"}`)) + for _, host := range []string{"rpc-ws.localhost", "rpc-0.localhost", "rpc-1.localhost"} { + gock.New("http://" + host).Post("").Persist(). + Filter(func(r *http.Request) bool { return strings.Contains(util.SafeReadBody(r), `eth_chainId`) }). + Reply(200).JSON([]byte(`{"result":"0x7b"}`)) } - // WS upstream may still hit https after scheme normalize in some paths; - // chainId mocks above cover HTTP twins. WS client bootstrap is separate. rateLimitersRegistry, _ := upstream.NewRateLimitersRegistry(context.Background(), &common.RateLimiterConfig{}, &log.Logger) metricsTracker := health.NewTracker(&log.Logger, "test", time.Minute) @@ -500,129 +459,34 @@ func TestNetworkHandle_SuggestLatestBlock_RecordsTipSourceAndBumpsHttpTwin(t *te pr, err := thirdparty.NewProvidersRegistry(&log.Logger, vr, []*common.ProviderConfig{}, nil) require.NoError(t, err) ssr, err := data.NewSharedStateRegistry(ctx, &log.Logger, &common.SharedStateConfig{ - Connector: &common.ConnectorConfig{ - Driver: "memory", - Memory: &common.MemoryConnectorConfig{MaxItems: 100_000, MaxTotalSize: "1GB"}, - }, + Connector: &common.ConnectorConfig{Driver: "memory", Memory: &common.MemoryConnectorConfig{MaxItems: 100_000, MaxTotalSize: "1GB"}}, }) require.NoError(t, err) - // Use http endpoints for all three so bootstrap does not need a live WS. - wsUp.Endpoint = "http://rpc-ws-twin.localhost" - httpUp.Endpoint = "http://rpc-http-twin.localhost" - sibling.Endpoint = "http://rpc-sibling.localhost" - for _, host := range []string{"rpc-ws-twin.localhost", "rpc-http-twin.localhost", "rpc-sibling.localhost"} { - gock.New("http://" + host). - Post(""). - Persist(). - Filter(func(r *http.Request) bool { - return strings.Contains(util.SafeReadBody(r), `eth_chainId`) - }). - Reply(200). - JSON([]byte(`{"result":"0x7b"}`)) - } - upstreamsRegistry := upstream.NewUpstreamsRegistry( - ctx, &log.Logger, "test", - []*common.UpstreamConfig{wsUp, httpUp, sibling}, ssr, rateLimitersRegistry, vr, pr, nil, - metricsTracker, nil, + ctx, &log.Logger, "test", cfgs, ssr, rateLimitersRegistry, vr, pr, nil, metricsTracker, nil, ) - networkConfig := &common.NetworkConfig{ - Architecture: common.ArchitectureEvm, - Evm: &common.EvmNetworkConfig{ChainId: 123}, - } - network, err := NewNetwork(ctx, &log.Logger, "test", networkConfig, + network, err := NewNetwork(ctx, &log.Logger, "test", + &common.NetworkConfig{Architecture: common.ArchitectureEvm, Evm: &common.EvmNetworkConfig{ChainId: 123}}, rateLimitersRegistry, upstreamsRegistry, metricsTracker, nil) require.NoError(t, err) - upstreamsRegistry.Bootstrap(ctx) time.Sleep(200 * time.Millisecond) require.NoError(t, upstreamsRegistry.GetInitializer().WaitForTasks(ctx)) require.NoError(t, network.Bootstrap(ctx)) time.Sleep(250 * time.Millisecond) - upsList := upstreamsRegistry.GetNetworkUpstreams(ctx, util.EvmNetworkId(123)) - require.Len(t, upsList, 3) byID := map[string]*upstream.Upstream{} - for _, u := range upsList { + for _, u := range upstreamsRegistry.GetNetworkUpstreams(ctx, util.EvmNetworkId(123)) { byID[u.Id()] = u u.EvmStatePoller().SuggestLatestBlock(1000) } - time.Sleep(50 * time.Millisecond) - handle := &networkHandle{nw: network} - handle.SuggestLatestBlock("ws:internal-eth-mainnet-reth-ws-0", 1001, []byte(`{"number":"0x3e9"}`)) + (&networkHandle{nw: network}).SuggestLatestBlock("ws:internal-eth-mainnet-reth-ws-0", 1001, nil) assert.Equal(t, int64(1001), byID["internal-eth-mainnet-reth-ws-0"].EvmStatePoller().LatestBlock()) assert.Equal(t, int64(1001), byID["internal-eth-mainnet-reth-0"].EvmStatePoller().LatestBlock(), - "HTTP twin poller must advance with WS tip (same physical node)") + "HTTP twin must advance with WS tip") assert.Equal(t, int64(1000), byID["internal-eth-mainnet-reth-1"].EvmStatePoller().LatestBlock(), - "lagging sibling must not be bumped") - assert.Equal(t, "internal-eth-mainnet-reth-ws-0", network.EvmTipSourceUpstreamId(1001)) - assert.Equal(t, int64(1001), network.EvmHighestLatestBlockNumber(ctx)) -} - -func TestEvmTipSourceUpstreamId_IgnoresFallbackWhilePrimaryUp(t *testing.T) { - util.ResetGock() - defer util.ResetGock() - util.SetupMocksForEvmStatePoller() - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - primary := &common.UpstreamConfig{ - Type: common.UpstreamTypeEvm, - Id: "primary", - Endpoint: "http://primary.localhost", - Evm: &common.EvmUpstreamConfig{ChainId: 123}, - } - fallback := &common.UpstreamConfig{ - Type: common.UpstreamTypeEvm, - Id: "fallback-ws", - Endpoint: "http://fallback.localhost", - Tags: []string{common.TagTierFallback}, - Evm: &common.EvmUpstreamConfig{ChainId: 123}, - } - for _, host := range []string{"primary.localhost", "fallback.localhost"} { - gock.New("http://" + host). - Post(""). - Persist(). - Filter(func(r *http.Request) bool { - return strings.Contains(util.SafeReadBody(r), `eth_chainId`) - }). - Reply(200). - JSON([]byte(`{"result":"0x7b"}`)) - } - - rateLimitersRegistry, _ := upstream.NewRateLimitersRegistry(context.Background(), &common.RateLimiterConfig{}, &log.Logger) - metricsTracker := health.NewTracker(&log.Logger, "test", time.Minute) - vr := thirdparty.NewVendorsRegistry() - pr, err := thirdparty.NewProvidersRegistry(&log.Logger, vr, []*common.ProviderConfig{}, nil) - require.NoError(t, err) - ssr, err := data.NewSharedStateRegistry(ctx, &log.Logger, &common.SharedStateConfig{ - Connector: &common.ConnectorConfig{ - Driver: "memory", - Memory: &common.MemoryConnectorConfig{MaxItems: 100_000, MaxTotalSize: "1GB"}, - }, - }) - require.NoError(t, err) - - upstreamsRegistry := upstream.NewUpstreamsRegistry( - ctx, &log.Logger, "test", - []*common.UpstreamConfig{primary, fallback}, ssr, rateLimitersRegistry, vr, pr, nil, - metricsTracker, nil, - ) - network, err := NewNetwork(ctx, &log.Logger, "test", - &common.NetworkConfig{Architecture: common.ArchitectureEvm, Evm: &common.EvmNetworkConfig{ChainId: 123}}, - rateLimitersRegistry, upstreamsRegistry, metricsTracker, nil) - require.NoError(t, err) - upstreamsRegistry.Bootstrap(ctx) - time.Sleep(200 * time.Millisecond) - require.NoError(t, upstreamsRegistry.GetInitializer().WaitForTasks(ctx)) - require.NoError(t, network.Bootstrap(ctx)) - time.Sleep(250 * time.Millisecond) - - network.noteTipSource(2000, "fallback-ws") - assert.Equal(t, "", network.EvmTipSourceUpstreamId(2000), - "must not pin tip re-fetch to fallback WS while a primary is up") + "sibling must stay behind") } diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go index 26a6dc017..83771dee5 100644 --- a/erpc/subscription_manager.go +++ b/erpc/subscription_manager.go @@ -579,11 +579,13 @@ func (h *networkHandle) FinalityDepth() int64 { // pod is already ≥ N. That invariant applies to every ingress source, // including tier:fallback: Ingest fans out all sources, so skipping TipHW // for fallback heads while still delivering them to clients causes -// MultiNode FOOS (WS tip ahead of HTTP TipHW). +// MultiNode FOOS (WS tip ahead of HTTP TipHW). Tip re-fetch of a TipHW +// that came from a fallback must reach that fallback via the emptyish +// escape hatch instead. // -// Also records the tip-source upstream id (for EnforceHighestBlock pin) and -// bumps any HTTP twin's poller so partitionUpstreamsByLatestBlock prefers -// the same physical node for subsequent eth_getBlockByNumber tip reads. +// Also bumps the HTTP twin poller (id: *-ws-* → *) so existing +// partitionUpstreamsByLatestBlock / EvmLeaderUpstream prefer the same +// physical node that just delivered newHeads — the cross-node tip race. func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64, payload json.RawMessage) { _ = payload const prefix = "ws:" @@ -591,90 +593,19 @@ func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64, p return } upstreamID := sourceId[len(prefix):] - var wsUp *upstream.Upstream - ups := h.nw.upstreamsRegistry.GetNetworkUpstreams(context.Background(), h.nw.networkId) - for _, u := range ups { - if u.Id() != upstreamID { + twinID := strings.Replace(upstreamID, "-ws-", "-", 1) // no-op if no -ws- + for _, u := range h.nw.upstreamsRegistry.GetNetworkUpstreams(context.Background(), h.nw.networkId) { + id := u.Id() + if id != upstreamID && id != twinID { continue } - wsUp = u - poller := u.EvmStatePoller() - if poller != nil && !poller.IsObjectNull() { + if poller := u.EvmStatePoller(); poller != nil && !poller.IsObjectNull() { poller.SuggestLatestBlock(blockNumber) } - break - } - h.nw.noteTipSource(blockNumber, upstreamID) - if wsUp != nil { - suggestHttpTwinLatestBlock(ups, wsUp, blockNumber) } h.nw.NoteObservedLatestBlock(h.nw.appCtx, blockNumber) } -// suggestHttpTwinLatestBlock advances pollers for HTTP upstreams that are -// the same physical node as wsUp (id convention *-ws-* / endpoint /ws twin). -// Same-node live tests show WS newHeads implies HTTP getBlock is immediately -// available on that node; the lagging sibling is the real tip-race. -func suggestHttpTwinLatestBlock(ups []*upstream.Upstream, wsUp *upstream.Upstream, blockNumber int64) { - if wsUp == nil || blockNumber <= 0 { - return - } - twinID := httpTwinUpstreamId(wsUp.Id()) - wsEp := "" - if cfg := wsUp.Config(); cfg != nil { - wsEp = cfg.Endpoint - } - httpTwinEp := httpTwinEndpoint(wsEp) - for _, u := range ups { - if u == nil || u.Id() == wsUp.Id() { - continue - } - match := twinID != "" && u.Id() == twinID - if !match && httpTwinEp != "" { - if cfg := u.Config(); cfg != nil && cfg.Endpoint == httpTwinEp { - match = true - } - } - if !match { - continue - } - poller := u.EvmStatePoller() - if poller != nil && !poller.IsObjectNull() { - poller.SuggestLatestBlock(blockNumber) - } - } -} - -// httpTwinUpstreamId maps internal-eth-mainnet-reth-ws-0 → internal-eth-mainnet-reth-0. -func httpTwinUpstreamId(wsUpstreamId string) string { - if strings.Contains(wsUpstreamId, "-ws-") { - return strings.Replace(wsUpstreamId, "-ws-", "-", 1) - } - if strings.HasSuffix(wsUpstreamId, "-ws") { - return strings.TrimSuffix(wsUpstreamId, "-ws") - } - return "" -} - -// httpTwinEndpoint maps wss://host/0/ws → https://host/0 (and ws→http). -func httpTwinEndpoint(wsEndpoint string) string { - if wsEndpoint == "" { - return "" - } - ep := wsEndpoint - switch { - case strings.HasPrefix(ep, "wss://"): - ep = "https://" + strings.TrimPrefix(ep, "wss://") - case strings.HasPrefix(ep, "ws://"): - ep = "http://" + strings.TrimPrefix(ep, "ws://") - default: - return "" - } - ep = strings.TrimSuffix(ep, "/ws") - ep = strings.TrimSuffix(ep, "/websocket") - return ep -} - // Interface checks: fail the build if either contract drifts. var ( _ wsclient.NotificationWriter = (*WsConnection)(nil) From ca885898ecb6c1e85ac32a28b313e984be169fba Mon Sep 17 00:00:00 2001 From: shpookas Date: Thu, 30 Jul 2026 16:37:14 +0200 Subject: [PATCH 4/7] fix(ws): drop HTTP twin bump; keep SkipFallbackEscape Tip routing stays on EvmLeaderUpstream (PR11). This PR only stops tip re-fetch from escaping to tier:fallback when internals miss. Co-authored-by: Cursor --- architecture/evm/eth_getBlockByNumber.go | 4 +- erpc/networks_ws_tip_test.go | 59 ------------------------ erpc/subscription_manager.go | 12 ++--- 3 files changed, 6 insertions(+), 69 deletions(-) diff --git a/architecture/evm/eth_getBlockByNumber.go b/architecture/evm/eth_getBlockByNumber.go index ca21f02db..4bf0614f2 100644 --- a/architecture/evm/eth_getBlockByNumber.go +++ b/architecture/evm/eth_getBlockByNumber.go @@ -226,8 +226,8 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common } // Prefer the upstream whose poller already owns this tip - // (EvmLeaderUpstream — typically the WS ingress / HTTP twin that - // SuggestLatestBlock advanced). If TipHW advanced via Redis/WS while + // (EvmLeaderUpstream — typically the WS ingress that called + // SuggestLatestBlock). If TipHW advanced via Redis/WS while // local pollers lag inside their debounce window, force-poll // the leader once before deciding. Fall back to excluding the // stale responder when no local poller has caught up yet. diff --git a/erpc/networks_ws_tip_test.go b/erpc/networks_ws_tip_test.go index 6243ef933..06ec4820b 100644 --- a/erpc/networks_ws_tip_test.go +++ b/erpc/networks_ws_tip_test.go @@ -431,62 +431,3 @@ func TestEvmRefreshHighestLatestBlockNumber_PreservesObservedTip(t *testing.T) { "refresh after sync TipHW publish must keep the observed tip") assert.Equal(t, int64(1001), network.EvmHighestLatestBlockNumber(ctx)) } - -// SuggestLatestBlock on *-ws-* must also advance the HTTP twin poller so -// EvmLeaderUpstream / partition prefer the same physical node. -func TestNetworkHandle_SuggestLatestBlock_BumpsHttpTwin(t *testing.T) { - util.ResetGock() - defer util.ResetGock() - util.SetupMocksForEvmStatePoller() - - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - cfgs := []*common.UpstreamConfig{ - {Type: common.UpstreamTypeEvm, Id: "internal-eth-mainnet-reth-ws-0", Endpoint: "http://rpc-ws.localhost", Evm: &common.EvmUpstreamConfig{ChainId: 123}}, - {Type: common.UpstreamTypeEvm, Id: "internal-eth-mainnet-reth-0", Endpoint: "http://rpc-0.localhost", Evm: &common.EvmUpstreamConfig{ChainId: 123}}, - {Type: common.UpstreamTypeEvm, Id: "internal-eth-mainnet-reth-1", Endpoint: "http://rpc-1.localhost", Evm: &common.EvmUpstreamConfig{ChainId: 123}}, - } - for _, host := range []string{"rpc-ws.localhost", "rpc-0.localhost", "rpc-1.localhost"} { - gock.New("http://" + host).Post("").Persist(). - Filter(func(r *http.Request) bool { return strings.Contains(util.SafeReadBody(r), `eth_chainId`) }). - Reply(200).JSON([]byte(`{"result":"0x7b"}`)) - } - - rateLimitersRegistry, _ := upstream.NewRateLimitersRegistry(context.Background(), &common.RateLimiterConfig{}, &log.Logger) - metricsTracker := health.NewTracker(&log.Logger, "test", time.Minute) - vr := thirdparty.NewVendorsRegistry() - pr, err := thirdparty.NewProvidersRegistry(&log.Logger, vr, []*common.ProviderConfig{}, nil) - require.NoError(t, err) - ssr, err := data.NewSharedStateRegistry(ctx, &log.Logger, &common.SharedStateConfig{ - Connector: &common.ConnectorConfig{Driver: "memory", Memory: &common.MemoryConnectorConfig{MaxItems: 100_000, MaxTotalSize: "1GB"}}, - }) - require.NoError(t, err) - - upstreamsRegistry := upstream.NewUpstreamsRegistry( - ctx, &log.Logger, "test", cfgs, ssr, rateLimitersRegistry, vr, pr, nil, metricsTracker, nil, - ) - network, err := NewNetwork(ctx, &log.Logger, "test", - &common.NetworkConfig{Architecture: common.ArchitectureEvm, Evm: &common.EvmNetworkConfig{ChainId: 123}}, - rateLimitersRegistry, upstreamsRegistry, metricsTracker, nil) - require.NoError(t, err) - upstreamsRegistry.Bootstrap(ctx) - time.Sleep(200 * time.Millisecond) - require.NoError(t, upstreamsRegistry.GetInitializer().WaitForTasks(ctx)) - require.NoError(t, network.Bootstrap(ctx)) - time.Sleep(250 * time.Millisecond) - - byID := map[string]*upstream.Upstream{} - for _, u := range upstreamsRegistry.GetNetworkUpstreams(ctx, util.EvmNetworkId(123)) { - byID[u.Id()] = u - u.EvmStatePoller().SuggestLatestBlock(1000) - } - - (&networkHandle{nw: network}).SuggestLatestBlock("ws:internal-eth-mainnet-reth-ws-0", 1001, nil) - - assert.Equal(t, int64(1001), byID["internal-eth-mainnet-reth-ws-0"].EvmStatePoller().LatestBlock()) - assert.Equal(t, int64(1001), byID["internal-eth-mainnet-reth-0"].EvmStatePoller().LatestBlock(), - "HTTP twin must advance with WS tip") - assert.Equal(t, int64(1000), byID["internal-eth-mainnet-reth-1"].EvmStatePoller().LatestBlock(), - "sibling must stay behind") -} diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go index 83771dee5..2c05546ca 100644 --- a/erpc/subscription_manager.go +++ b/erpc/subscription_manager.go @@ -582,10 +582,6 @@ func (h *networkHandle) FinalityDepth() int64 { // MultiNode FOOS (WS tip ahead of HTTP TipHW). Tip re-fetch of a TipHW // that came from a fallback must reach that fallback via the emptyish // escape hatch instead. -// -// Also bumps the HTTP twin poller (id: *-ws-* → *) so existing -// partitionUpstreamsByLatestBlock / EvmLeaderUpstream prefer the same -// physical node that just delivered newHeads — the cross-node tip race. func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64, payload json.RawMessage) { _ = payload const prefix = "ws:" @@ -593,15 +589,15 @@ func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64, p return } upstreamID := sourceId[len(prefix):] - twinID := strings.Replace(upstreamID, "-ws-", "-", 1) // no-op if no -ws- for _, u := range h.nw.upstreamsRegistry.GetNetworkUpstreams(context.Background(), h.nw.networkId) { - id := u.Id() - if id != upstreamID && id != twinID { + if u.Id() != upstreamID { continue } - if poller := u.EvmStatePoller(); poller != nil && !poller.IsObjectNull() { + poller := u.EvmStatePoller() + if poller != nil && !poller.IsObjectNull() { poller.SuggestLatestBlock(blockNumber) } + break } h.nw.NoteObservedLatestBlock(h.nw.appCtx, blockNumber) } From a5ebf7089118f0cede790bf41523487b9a9ba0cf Mon Sep 17 00:00:00 2001 From: shpookas Date: Thu, 30 Jul 2026 17:17:53 +0200 Subject: [PATCH 5/7] fix(ws): re-resolve leader pin for second tip re-fetch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first tip re-fetch pins to EvmLeaderUpstream only when the leader poller already owns TipHW at resolve time. When TipHW was adopted from Redis before the local WS delivery, that resolve fails, the first re-fetch goes out with only a stale-responder exclusion, and the second re-fetch dropped the pin entirely — sweeping siblings that are provably one block behind (cross-node live test: 12/12 null at t=0, catch-up in 115-193ms; the announcing node serves its own head in ~0ms). Resolve the leader again before the second re-fetch when the first was not leader-pinned: the leader has had the first Forward's retry budget plus a forced poll (PollLatestBlockNumberNow bypasses debounce) to catch up, so the pin lands on the node that has the block. If the first re-fetch WAS leader-pinned and still missed, the leader cannot serve — keep the unpinned primary sweep as before. Fallback escape stays suppressed on both paths. Adds a discriminating test: the lagging stale responder must receive zero concrete-tip fetches once the caught-up leader is pinned on the second re-fetch (without the re-resolve, the unpinned sweep hits it first). Gock filter counters are gated on r.URL.Host because filters also run while matching requests bound for other hosts. --- architecture/evm/eth_getBlockByNumber.go | 50 +++-- erpc/http_server_ws_tip_leader_test.go | 223 +++++++++++++++++++++++ 2 files changed, 258 insertions(+), 15 deletions(-) diff --git a/architecture/evm/eth_getBlockByNumber.go b/architecture/evm/eth_getBlockByNumber.go index 4bf0614f2..bfb799290 100644 --- a/architecture/evm/eth_getBlockByNumber.go +++ b/architecture/evm/eth_getBlockByNumber.go @@ -231,19 +231,29 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common // local pollers lag inside their debounce window, force-poll // the leader once before deciding. Fall back to excluding the // stale responder when no local poller has caught up yet. - useUpstream := "" - if leader := network.EvmLeaderUpstream(ctx); leader != nil { - if eu, ok := leader.(common.EvmUpstream); ok { - if sp := eu.EvmStatePoller(); sp != nil && !sp.IsObjectNull() { - if sp.LatestBlock() < highestBlockNumber { - _, _ = sp.PollLatestBlockNumberNow(ctx) - } - if sp.LatestBlock() >= highestBlockNumber { - useUpstream = leader.Id() - } - } + resolveLeaderPin := func() string { + leader := network.EvmLeaderUpstream(ctx) + if leader == nil { + return "" + } + eu, ok := leader.(common.EvmUpstream) + if !ok { + return "" + } + sp := eu.EvmStatePoller() + if sp == nil || sp.IsObjectNull() { + return "" + } + if sp.LatestBlock() < highestBlockNumber { + _, _ = sp.PollLatestBlockNumberNow(ctx) + } + if sp.LatestBlock() >= highestBlockNumber { + return leader.Id() } + return "" } + useUpstream := resolveLeaderPin() + firstPinnedToLeader := useUpstream != "" if useUpstream == "" && respBlockNumber > 0 { useUpstream = fmt.Sprintf("!%s", nr.UpstreamId()) } @@ -264,10 +274,20 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common nnr.Release() } - // Pinned / excluded re-fetch missed the tip (sibling fullnode - // lag, WS JSON-RPC miss, etc.). Retry with no UseUpstream pin - // so remaining primaries can serve the concrete TipHW block. - nnr2, ferr2 := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, "", true) + // First re-fetch missed the tip. If it was NOT leader-pinned (the + // leader poller had not caught up to TipHW at resolve time — e.g. + // TipHW arrived via Redis before the local WS delivery), resolve + // the leader again: it has had the first Forward's retry budget + // plus a forced poll to catch up, and the node that delivered the + // head serves it immediately (same-node WS→HTTP gap is ~0ms). + // If the first re-fetch WAS leader-pinned and still missed, the + // leader genuinely cannot serve — sweep the remaining primaries + // unpinned instead (fallback escape stays suppressed either way). + pin2 := "" + if !firstPinnedToLeader { + pin2 = resolveLeaderPin() + } + nnr2, ferr2 := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, pin2, true) if meetsTipFloor(ctx, nnr2, highestBlockNumber) { if nr != nil { nr.Release() diff --git a/erpc/http_server_ws_tip_leader_test.go b/erpc/http_server_ws_tip_leader_test.go index 59a3c226e..ee9308c17 100644 --- a/erpc/http_server_ws_tip_leader_test.go +++ b/erpc/http_server_ws_tip_leader_test.go @@ -419,3 +419,226 @@ func TestHttpServer_GetBlockByNumberLatest_TipRefetchSkipsFallbackEscape(t *test // escape counter + refuse-stale response are the request-path proofs. _ = fallbackHits } + +// When the leader poller has not caught up to TipHW at first resolve (e.g. +// TipHW adopted from Redis before the local WS delivery), the SECOND tip +// re-fetch must re-resolve the leader — whose forced poll now returns the +// tip — and pin to it, instead of sweeping lagging siblings unpinned. +// +// Discriminator: rpc1 (the stale responder) must never receive the concrete +// tip fetch. Without the second-resolve pin, re-fetch #2 goes out unpinned +// and hits rpc1 first. +func TestHttpServer_GetBlockByNumberLatest_SecondRefetchResolvesLeaderPin(t *testing.T) { + util.ResetGock() + defer util.ResetGock() + + const tipHex = "0x3eb" // 1003 — TipHW + const rpc2Latest = "0x3ea" // 1002 — leader poller before catch-up + const rpc1Latest = "0x3e8" // 1000 — lagging stale responder + + var rpc1TipHits atomic.Int64 + var rpc2TipCalls atomic.Int64 + var leaderCaughtUp atomic.Bool + + for _, host := range []string{"rpc1.localhost", "rpc2.localhost"} { + h := host + gock.New("http://" + h). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return strings.Contains(util.SafeReadBody(r), "eth_chainId") + }). + Reply(200). + JSON([]byte(`{"result":"0x7b"}`)) + gock.New("http://" + h). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return strings.Contains(util.SafeReadBody(r), "eth_syncing") + }). + Reply(200). + JSON([]byte(`{"result":false}`)) + gock.New("http://" + h). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + return strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, "finalized") + }). + Reply(200). + JSON([]byte(`{"result":{"number":"0x300","timestamp":"0x6702a8e0"}}`)) + } + + // rpc1: "latest" polls and the client "latest" both see a lagging head. + gock.New("http://rpc1.localhost"). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + return strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, `"latest"`) + }). + Reply(200). + JSON([]byte(`{"result":{"number":"` + rpc1Latest + `","hash":"0x1111111111111111111111111111111111111111111111111111111111111111","parentHash":"0x2222222222222222222222222222222222222222222222222222222222222222","timestamp":"0x6702a8f0"}}`)) + // rpc1: concrete tip fetch — must never happen when re-fetch #2 is pinned. + 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+`"`) { + rpc1TipHits.Add(1) + return true + } + return false + }). + Reply(200). + JSON([]byte(`{"result":null}`)) + + // rpc2 "latest": stale until the leader "catches up" (flips after its + // first concrete-tip miss), then the forced poll returns the tip. + gock.New("http://rpc2.localhost"). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + return strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, `"latest"`) && !leaderCaughtUp.Load() + }). + Reply(200). + JSON([]byte(`{"result":{"number":"` + rpc2Latest + `","hash":"0x3333333333333333333333333333333333333333333333333333333333333333","parentHash":"0x4444444444444444444444444444444444444444444444444444444444444444","timestamp":"0x6702a8f0"}}`)) + gock.New("http://rpc2.localhost"). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + body := util.SafeReadBody(r) + return strings.Contains(body, "eth_getBlockByNumber") && strings.Contains(body, `"latest"`) && leaderCaughtUp.Load() + }). + Reply(200). + JSON([]byte(`{"result":{"number":"` + tipHex + `","hash":"0x5555555555555555555555555555555555555555555555555555555555555555","parentHash":"0x6666666666666666666666666666666666666666666666666666666666666666","timestamp":"0x6702a8f1"}}`)) + // rpc2 concrete tip: first call misses (and marks the node caught-up so + // the next forced poll sees the tip); subsequent calls serve the block. + gock.New("http://rpc2.localhost"). + Post(""). + Persist(). + 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+`"`) { + return false + } + if rpc2TipCalls.Add(1) == 1 { + // The miss itself marks the node caught-up: the next forced + // poll (second leader resolve) sees the tip. + leaderCaughtUp.Store(true) + return true + } + return false + }). + Reply(200). + JSON([]byte(`{"result":null}`)) + gock.New("http://rpc2.localhost"). + Post(""). + Persist(). + 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+`"`) { + return false + } + return rpc2TipCalls.Load() >= 1 + }). + Reply(200). + JSON([]byte(`{"result":{"number":"` + tipHex + `","hash":"0x7777777777777777777777777777777777777777777777777777777777777777","parentHash":"0x8888888888888888888888888888888888888888888888888888888888888888","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, + Integrity: &common.EvmIntegrityConfig{ + EnforceHighestBlock: util.BoolPtr(true), + }, + }, + SelectionPolicy: &common.SelectionPolicyConfig{ + EvalInterval: 0, + }, + Failsafe: []*common.FailsafeConfig{ + { + Retry: &common.RetryPolicyConfig{MaxAttempts: 1}, + }, + }, + }, + }, + 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) + nw, err := prj.GetNetwork(context.Background(), "evm:123") + require.NoError(t, err) + + time.Sleep(500 * time.Millisecond) + + policy.OverrideAllForTest(prj.policyEngine) + policy.OverrideOrderForTest(prj.policyEngine, "evm:123", "rpc1", "rpc2") + require.Equal(t, []string{"rpc1", "rpc2"}, nw.PolicyOrderedUpstreams("eth_getBlockByNumber"), + "rpc1 must serve the stale latest so it becomes the excluded responder") + + nw.NoteObservedLatestBlock(context.Background(), 1003) + require.Equal(t, int64(1003), nw.EvmHighestLatestBlockNumber(context.Background())) + + statusCode, _, body := sendRequest(`{ + "jsonrpc": "2.0", + "id": 1, + "method": "eth_getBlockByNumber", + "params": ["latest", false] + }`, nil, nil) + + require.Equal(t, http.StatusOK, statusCode, "expected tip served via leader-pinned second re-fetch; 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, "response should have a result object, got: %s", body) + assert.Equal(t, tipHex, result["number"], "must serve the TipHW block") + assert.Equal(t, int64(0), rpc1TipHits.Load(), + "second re-fetch must pin to the caught-up leader, not sweep the lagging stale responder") + assert.GreaterOrEqual(t, rpc2TipCalls.Load(), int64(2), + "leader must be re-tried via pin after its first tip miss") +} From 9ffb6f6a0ae8823d69e984a34c733ce099bcf7bc Mon Sep 17 00:00:00 2001 From: shpookas Date: Fri, 31 Jul 2026 10:57:53 +0200 Subject: [PATCH 6/7] chore(ws): log tip re-fetch leader pin miss details On first tip re-fetch miss, log tipHW, stale upstream, leader id/latest, whether pass-1 was leader-pinned, pin directives, and recovery/refuse outcome so canary can prove whether empties are pinned-leader misses or pin-empty Redis races. Co-authored-by: Cursor --- architecture/evm/eth_getBlockByNumber.go | 52 +++++++++++++++++++++++- 1 file changed, 51 insertions(+), 1 deletion(-) diff --git a/architecture/evm/eth_getBlockByNumber.go b/architecture/evm/eth_getBlockByNumber.go index bfb799290..e823e99d2 100644 --- a/architecture/evm/eth_getBlockByNumber.go +++ b/architecture/evm/eth_getBlockByNumber.go @@ -231,23 +231,31 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common // local pollers lag inside their debounce window, force-poll // the leader once before deciding. Fall back to excluding the // stale responder when no local poller has caught up yet. + var leaderId string + var leaderLatest int64 resolveLeaderPin := func() string { leader := network.EvmLeaderUpstream(ctx) if leader == nil { + leaderId = "" + leaderLatest = 0 return "" } + leaderId = leader.Id() eu, ok := leader.(common.EvmUpstream) if !ok { + leaderLatest = 0 return "" } sp := eu.EvmStatePoller() if sp == nil || sp.IsObjectNull() { + leaderLatest = 0 return "" } if sp.LatestBlock() < highestBlockNumber { _, _ = sp.PollLatestBlockNumberNow(ctx) } - if sp.LatestBlock() >= highestBlockNumber { + leaderLatest = sp.LatestBlock() + if leaderLatest >= highestBlockNumber { return leader.Id() } return "" @@ -270,7 +278,11 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common } return nnr, nil } + pin1Upstream := "" + pin1Empty := true if nnr != nil { + pin1Upstream = nnr.UpstreamId() + pin1Empty = nnr.IsResultEmptyish() nnr.Release() } @@ -287,14 +299,45 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common if !firstPinnedToLeader { pin2 = resolveLeaderPin() } + staleUpstream := "" + if nr != nil { + staleUpstream = nr.UpstreamId() + } + logger.Warn(). + Int64("tipHW", highestBlockNumber). + Int64("staleBlockNumber", respBlockNumber). + Str("staleUpstream", staleUpstream). + Str("leaderId", leaderId). + Int64("leaderLatest", leaderLatest). + Bool("firstPinnedToLeader", firstPinnedToLeader). + Str("pin1", useUpstream). + Str("pin1Upstream", pin1Upstream). + Bool("pin1Empty", pin1Empty). + Err(ferr). + Str("pin2", pin2). + Msg("tip re-fetch miss after first attempt") + nnr2, ferr2 := forwardGetBlockByNumber(ctx, network, nq, highestBlockNumber, itx, pin2, true) if meetsTipFloor(ctx, nnr2, highestBlockNumber) { + servedBy := "" + if nnr2 != nil { + servedBy = nnr2.UpstreamId() + } + logger.Warn(). + Int64("tipHW", highestBlockNumber). + Str("leaderId", leaderId). + Bool("firstPinnedToLeader", firstPinnedToLeader). + Str("pin2", pin2). + Str("servedBy", servedBy). + Msg("tip re-fetch recovered on second attempt") if nr != nil { nr.Release() } return nnr2, nil } + pin2Upstream := "" if nnr2 != nil { + pin2Upstream = nnr2.UpstreamId() nnr2.Release() } @@ -302,6 +345,13 @@ func enforceHighestBlock(ctx context.Context, network common.Network, nq *common logger.Warn(). Int64("highestBlockNumber", highestBlockNumber). Int64("staleBlockNumber", respBlockNumber). + Str("staleUpstream", staleUpstream). + Str("leaderId", leaderId). + Int64("leaderLatest", leaderLatest). + Bool("firstPinnedToLeader", firstPinnedToLeader). + Str("pin1", useUpstream). + Str("pin2", pin2). + Str("pin2Upstream", pin2Upstream). Err(ferr2). Msg("tip re-fetch could not reach TipHW; refusing stale latest") if nr != nil { From 93f5b00f864df698b6bfa0bfe1b932db3560bac9 Mon Sep 17 00:00:00 2001 From: shpookas Date: Fri, 31 Jul 2026 11:31:10 +0200 Subject: [PATCH 7/7] fix(ws): suppress fallback escape for tip-race misses on any request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Direct client requests for a concrete tip block are the majority of the fallback-escape volume (evm:1, 6h: 66k getBlockByNumber escapes vs at most 31k attributable to enforcement re-fetches) and were untouched by the SkipFallbackEscape directive, which only covers the TipHW re-fetch. Generalize the guard in the escape hatch: when the requested block is at or one ahead of the primary leader's poller, the miss is the sibling import race (measured 100-200ms) — primaries serve it within a block time and pay-per-call fallbacks share the same race, so escaping buys nothing. The failsafe retry (emptyResultDelay / blockUnavailableDelay) re-visits primaries instead. Scope guards: blocks further ahead of the leader (primaries genuinely stuck) still escape — that is the HA case, covered by the existing gate-skip subtests; older-block data gaps still escape; block-less methods (eth_getTransactionReceipt) still escape. Repurposes EscapesOnEmptyishGetBlockByNumber into NearTipEmptyishDoesNotEscape: same fixture (primaries at tip returning null for it) now asserts the escape counter does not move and the response is not served by a fallback. Note: TestFailover_GateSkipsAccumulateErrorRate is flaky on the base branch (fails 2/3 runs, errorRate lands exactly on the 0.7 threshold depending on poller timing) — unrelated to this change. --- erpc/networks.go | 18 ++++++++++++ erpc/networks_failover_escape_test.go | 41 ++++++++++++++------------- 2 files changed, 40 insertions(+), 19 deletions(-) diff --git a/erpc/networks.go b/erpc/networks.go index fa97e91b9..372bca50a 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -1092,6 +1092,24 @@ func (n *Network) Forward(ctx context.Context, req *common.NormalizedRequest) (* // out to pay-per-call tier:fallback upstreams. dirs := effectiveReq.Directives() skipFallbackEscape := dirs != nil && dirs.SkipFallbackEscape + if !skipFallbackEscape { + // Tip-race miss: the requested block is at or one ahead of the + // primary leader's poller — primaries import it within ~a block + // time (sibling lag, measured 100-200ms), so escaping to + // pay-per-call fallbacks buys nothing. Let the failsafe retry + // (emptyResultDelay / blockUnavailableDelay) re-visit primaries. + // Blocks further ahead (primaries stuck), older-block data gaps, + // and block-less methods (receipts) escape as before. + if bn, ok := effectiveReq.EvmBlockNumber().(int64); ok && bn > 0 { + if leader, ok2 := n.EvmLeaderUpstream(execSpanCtx).(common.EvmUpstream); ok2 && leader != nil { + if sp := leader.EvmStatePoller(); sp != nil && !sp.IsObjectNull() { + if l := sp.LatestBlock(); l > 0 && bn >= l && bn <= l+1 { + skipFallbackEscape = true + } + } + } + } + } bestRespEmptyish := bestResp != nil && bestResp.IsResultEmptyish() if (bestResp == nil || bestRespEmptyish) && !skipFallbackEscape && diff --git a/erpc/networks_failover_escape_test.go b/erpc/networks_failover_escape_test.go index 6aae47a6c..de6a7beab 100644 --- a/erpc/networks_failover_escape_test.go +++ b/erpc/networks_failover_escape_test.go @@ -627,15 +627,18 @@ func TestFailover_EscapeHatch(t *testing.T) { "escape hatch must fire exactly once for the non-retryable gate-skip case") }) - t.Run("EscapesOnEmptyishGetBlockByNumber", func(t *testing.T) { + t.Run("NearTipEmptyishDoesNotEscape", func(t *testing.T) { defer util.ResetGock() ctx, cancel := context.WithCancel(context.Background()) defer cancel() - // Primaries and fallbacks both report tip 1002 via poller, so the - // availability gate fails open / passes. Primaries return null for - // the concrete tip block (missing data); fallbacks return the header. - // Before the emptyish-escape fix, bestResp=null blocked escalation. + // Primaries and fallbacks both report tip 1002 via poller, and the + // request targets that tip. Primaries returning null for a block at + // (or one ahead of) the primary leader's poller is the sibling + // import race — primaries serve it within ~a block time, so the + // escape hatch must NOT burn pay-per-call fallbacks. The failsafe + // retry re-visits primaries instead. (Blocks further ahead — stuck + // primaries — still escape: see the gate-skip subtests above.) network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ primaryLatest: "0x3ea", // 1002 fallbackLatest: "0x3ea", // 1002 @@ -675,22 +678,22 @@ func TestFailover_EscapeHatch(t *testing.T) { )) req.SetNetwork(network) resp, err := network.Forward(ctx, req) - require.NoError(t, err, "null from primaries must escalate to fallbacks on the same request") - require.NotNil(t, resp) - defer resp.Release() - - jrr, err := resp.JsonRpcResponse() - require.NoError(t, err) - require.False(t, jrr.IsResultEmptyish(), "fallback must return a non-null block header") - num, err := jrr.PeekStringByPath(ctx, "number") - require.NoError(t, err) - assert.Equal(t, "0x3ea", num) + if resp != nil { + defer resp.Release() + } - assert.Contains(t, []string{"fallback-1", "fallback-2"}, resp.UpstreamId(), - "emptyish primary miss must be served by a fallback") + if err == nil { + require.NotNil(t, resp) + jrr, jerr := resp.JsonRpcResponse() + require.NoError(t, jerr) + require.True(t, jrr.IsResultEmptyish(), + "near-tip miss must stay emptyish from primaries, not be served by a fallback") + assert.NotContains(t, []string{"fallback-1", "fallback-2"}, resp.UpstreamId(), + "near-tip miss must not be served via fallback escape") + } after := promUtil.ToFloat64(counter) - assert.Equal(t, before+1, after, - "escape hatch must fire for emptyish eth_getBlockByNumber primary misses") + assert.Equal(t, before, after, + "escape hatch must NOT fire for near-tip emptyish primary misses") }) }