From 6a7dfd1d7006756ea7b4888855e6921c704a42ea Mon Sep 17 00:00:00 2001 From: shpookas Date: Tue, 29 Sep 2026 18:23:27 +0200 Subject: [PATCH 1/4] fix(network): route block-pinned requests to a fallback that has the block When a request is pinned to a block above every routed upstream's known head, try the tier:fallback upstreams whose head already reached it first, then the routed list. A fallback's newHeads subscription keeps its poller current, so it is often the source that announced the block to clients. Without this the routed upstreams answer missing data first and the request only reaches the fallback through the escape, if at all. Uses the per-request escalation, so the escape does not fire again. No-op when any routed head is unknown or already at the block, for consensus, and when failover is off. Counted in erpc_network_tip_leader_route_total. --- docs/pages/config/projects/networks.mdx | 2 + docs/pages/reference/metrics.mdx | 1 + erpc/networks.go | 76 +++++++++- erpc/networks_failover_escape_test.go | 27 +++- erpc/networks_tip_leader_test.go | 186 ++++++++++++++++++++++++ telemetry/metrics.go | 9 ++ 6 files changed, 295 insertions(+), 6 deletions(-) create mode 100644 erpc/networks_tip_leader_test.go diff --git a/docs/pages/config/projects/networks.mdx b/docs/pages/config/projects/networks.mdx index 33e967ce3..d21a8f72c 100644 --- a/docs/pages/config/projects/networks.mdx +++ b/docs/pages/config/projects/networks.mdx @@ -147,6 +147,8 @@ The escape: - fires at most once per request, and never for consensus requests; - is counted in `erpc_network_fallback_escape_total{project,network,category}` — expect zero in steady state. +With `onDefaultsExhausted` on, a request pinned to a block above every routed upstream's known head is sent first to the `tier:fallback` upstreams whose head has already reached that block (for example because their `newHeads` subscription announced it first), then to the routed upstreams. This uses the same once-per-request escalation, so the escape above does not fire again for that request. It does nothing when any routed upstream's head is unknown or already at the block, and never applies to consensus requests. It is counted in `erpc_network_tip_leader_route_total{project,network,category}`. + The validation report warns when `onDefaultsExhausted` is enabled but no upstream is tagged `tier:fallback`. ## Agent reference diff --git a/docs/pages/reference/metrics.mdx b/docs/pages/reference/metrics.mdx index b6e97c800..9d0e830b8 100644 --- a/docs/pages/reference/metrics.mdx +++ b/docs/pages/reference/metrics.mdx @@ -312,6 +312,7 @@ All metric names carry the `erpc_` prefix. Full definitions: 0 && !evm.EmptyResultBeyondConfidence(ctx, effectiveReq) } + // Tip-leader routing: a request pinned to a block above every routed + // upstream's head goes first to the fallbacks that already have it + // (a fallback's WS heads keep its poller current), then to the routed + // list. It spends the per-request escalation, so the escape below + // does not fire again. + if !oneUpstreamOnly && n.cfg.Failover.Enabled() && !failsafeExecutor.HasConsensus() { + if leaders := n.tipLeaderFallbacks(execSpanCtx, effectiveReq, method, upsList); len(leaders) > 0 && + effectiveReq.MarkEscalatedToFallbacks() { + routed := effectiveReq.NextUpstream + maxLoopIterations += len(leaders) + nextUpstream = func() (common.Upstream, error) { + for len(leaders) > 0 { + fb := leaders[0] + leaders = leaders[1:] + if _, loaded := effectiveReq.ConsumedUpstreams.LoadOrStore(fb, true); !loaded { + return fb, nil + } + } + return routed() + } + + telemetry.MetricNetworkTipLeaderRouteTotal.WithLabelValues( + n.projectId, n.Label(), method, + ).Inc() + lg.Debug().Int("leaders", len(leaders)).Msg("routing to fallback upstreams ahead of the routed tip") + } + } + escalationLoop: for { for loopIteration := 0; loopIteration < maxLoopIterations; loopIteration++ { @@ -3627,9 +3655,12 @@ func (n *Network) acquireRateLimitPermit(ctx context.Context, req *common.Normal // tierUpstreamsByGroup moves fallback-tier upstreams behind the rest, // preserving order within each tier. func tierUpstreamsByGroup(ups []common.Upstream) []common.Upstream { - return stablePartition(ups, func(u common.Upstream) bool { - return u.Config() != nil && u.Config().HasTag(common.TagTierFallback) - }) + return stablePartition(ups, isFallbackTier) +} + +// isFallbackTier reports whether u is tagged tier:fallback. +func isFallbackTier(u common.Upstream) bool { + return u.Config() != nil && u.Config().HasTag(common.TagTierFallback) } // partitionUpstreamsByLatestBlock moves upstreams whose polled head is known @@ -3690,6 +3721,45 @@ func preferTipLeaderForNearTipGetBlock(ups []common.Upstream, method string, bn return out } +// tipLeaderFallbacks returns the fallback-tier upstreams, outside the routed +// list, whose head has reached the block req is pinned to, when every routed +// default-tier upstream's head is known and below it. Nil otherwise, +// including when a routed head is unknown. +func (n *Network) tipLeaderFallbacks(ctx context.Context, req *common.NormalizedRequest, method string, routed []common.Upstream) []common.Upstream { + if n.Architecture() != common.ArchitectureEvm { + return nil + } + bn := requestBlockNumber(ctx, req) + if bn <= 0 { + return nil + } + routedIds := make(map[string]struct{}, len(routed)) + defaults := 0 + for _, u := range routed { + routedIds[u.Id()] = struct{}{} + if isFallbackTier(u) { + continue + } + defaults++ + if lb := upstreamLatestBlock(u); lb <= 0 || lb >= bn { + return nil + } + } + if defaults == 0 { + return nil + } + var leaders []common.Upstream + for _, fb := range n.upstreamsRegistry.GetFallbackEscapeUpstreams(ctx, n.networkId, method) { + if _, ok := routedIds[fb.Id()]; ok { + continue + } + if upstreamLatestBlock(fb) >= bn { + leaders = append(leaders, fb) + } + } + return leaders +} + // upstreamLatestBlock is u's polled head, or 0 when unknown. func upstreamLatestBlock(u common.Upstream) int64 { if eu, ok := u.(common.EvmUpstream); ok { diff --git a/erpc/networks_failover_escape_test.go b/erpc/networks_failover_escape_test.go index 1f9f60cac..91343c2f4 100644 --- a/erpc/networks_failover_escape_test.go +++ b/erpc/networks_failover_escape_test.go @@ -208,6 +208,8 @@ type failoverFixtureOpts struct { // mocks registers test-specific mocks ahead of the standard ones, before // any poller starts. mocks func() + // configure adjusts the upstream configs before the network is built. + configure func(cfgs []*common.UpstreamConfig) } func setupFailoverFixture( @@ -232,7 +234,11 @@ func setupFailoverFixture( mockEthCallReturning("rpc3.localhost", "0x3333") mockEthCallReturning("rpc4.localhost", "0x4444") - network, upr, mt := buildFailoverNetwork(t, ctx, failoverUpstreamConfigs(), opts) + cfgs := failoverUpstreamConfigs() + if opts.configure != nil { + opts.configure(cfgs) + } + network, upr, mt := buildFailoverNetwork(t, ctx, cfgs, opts) upsList := upr.GetNetworkUpstreams(ctx, util.EvmNetworkId(999)) require.Len(t, upsList, 4) @@ -265,11 +271,26 @@ func TestFailover_EscapeHatch(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - // Primaries at 1000 skip block 1002; the fallbacks at 1002 serve it. + // Primaries at 1002 fail eth_call with a retryable error; the + // fallbacks at 1002 serve it. (Primaries below the block would take + // tip-leader routing instead, see TestFailover_TipLeaderRouting.) network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ - primaryLatest: "0x3e8", // 1000 + primaryLatest: "0x3ea", // 1002 fallbackLatest: "0x3ea", // 1002 enableFailover: true, + mocks: func() { + for _, host := range []string{"rpc1.localhost", "rpc2.localhost"} { + host := host + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return r.URL.Host == host && strings.Contains(util.SafeReadBody(r), "eth_call") + }). + Reply(200). + JSON([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"internal error"}}`)) + } + }, }) counter := telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call") diff --git a/erpc/networks_tip_leader_test.go b/erpc/networks_tip_leader_test.go new file mode 100644 index 000000000..4812ff1bd --- /dev/null +++ b/erpc/networks_tip_leader_test.go @@ -0,0 +1,186 @@ +package erpc + +import ( + "context" + "net/http" + "strings" + "sync/atomic" + "testing" + + "github.com/erpc/erpc/common" + "github.com/erpc/erpc/telemetry" + "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" +) + +// unboundedPrimaries drops the availability bound from the primaries, so they +// are called (and answer) for blocks above their polled head, like a local +// node whose poller trails the head a fallback just announced. +func unboundedPrimaries(cfgs []*common.UpstreamConfig) { + for _, cfg := range cfgs { + if !cfg.HasTag(common.TagTierFallback) { + cfg.Evm.BlockAvailability = nil + } + } +} + +// countEthCalls counts eth_call requests reaching host, ahead of its standard +// mock. +func countEthCalls(host string, hits *atomic.Int64) { + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + // Filters run before host matching. + if r.URL.Host == host && strings.Contains(util.SafeReadBody(r), "eth_call") { + hits.Add(1) + } + return false + }). + Reply(200) +} + +func forwardEthCall(t *testing.T, ctx context.Context, network *Network, id int, blockHex string) (string, error) { + t.Helper() + req := ethCallRequest(id, blockHex) + req.SetNetwork(network) + resp, err := network.Forward(ctx, req) + if err != nil { + return "", err + } + require.NotNil(t, resp) + defer resp.Release() + jrr, err := resp.JsonRpcResponse() + require.NoError(t, err) + return strings.Trim(jrr.GetResultString(), `"`), nil +} + +func TestFailover_TipLeaderRouting(t *testing.T) { + leaderCounter := func() float64 { + return promUtil.ToFloat64(telemetry.MetricNetworkTipLeaderRouteTotal.WithLabelValues("main", "evm:999", "eth_call")) + } + escapeCounter := func() float64 { + return promUtil.ToFloat64(telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call")) + } + + t.Run("RoutesToFallbackThatHasTheBlock", func(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Primaries polled at 1000 would still answer for 1002; the fallbacks + // already have 1002, so they go first. + var primaryHits atomic.Int64 + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // 1002 + enableFailover: true, + configure: unboundedPrimaries, + mocks: func() { + countEthCalls("rpc1.localhost", &primaryHits) + countEthCalls("rpc2.localhost", &primaryHits) + }, + }) + + leaderBefore, escapeBefore := leaderCounter(), escapeCounter() + result, err := forwardEthCall(t, ctx, network, 1, "0x3ea") + require.NoError(t, err) + assert.Contains(t, []string{"0x3333", "0x4444"}, result, "a fallback that has the block must serve it") + assert.Equal(t, int64(0), primaryHits.Load(), "primaries must not be tried before the leader") + assert.Equal(t, leaderBefore+1, leaderCounter()) + assert.Equal(t, escapeBefore, escapeCounter(), "leader routing is not an escape") + }) + + t.Run("NoLeaderWhenRoutedHasTheBlock", func(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3ea", // 1002 + fallbackLatest: "0x3eb", // 1003 + enableFailover: true, + }) + + leaderBefore := leaderCounter() + for i := 0; i < 10; i++ { + result, err := forwardEthCall(t, ctx, network, i, "0x3ea") + require.NoError(t, err) + assert.Contains(t, []string{"0x1111", "0x2222"}, result, "a primary that has the block must serve it (iter %d)", i) + } + assert.Equal(t, leaderBefore, leaderCounter()) + }) + + t.Run("NoLeaderWhenFallbackIsBehindToo", func(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3e9", // 1001 + enableFailover: true, + configure: unboundedPrimaries, + }) + + leaderBefore := leaderCounter() + result, err := forwardEthCall(t, ctx, network, 1, "0x3ea") + require.NoError(t, err) + assert.Contains(t, []string{"0x1111", "0x2222"}, result) + assert.Equal(t, leaderBefore, leaderCounter()) + }) + + t.Run("NoLeaderWhenFailoverDisabled", func(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // 1002 + enableFailover: false, + configure: unboundedPrimaries, + }) + + leaderBefore := leaderCounter() + _, _ = forwardEthCall(t, ctx, network, 1, "0x3ea") + assert.Equal(t, leaderBefore, leaderCounter()) + }) + + t.Run("FallsBackToRoutedWhenLeaderFails", func(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // 1002 + enableFailover: true, + configure: unboundedPrimaries, + mocks: func() { + for _, host := range []string{"rpc3.localhost", "rpc4.localhost"} { + host := host + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return r.URL.Host == host && strings.Contains(util.SafeReadBody(r), "eth_call") + }). + Reply(200). + JSON([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"internal error"}}`)) + } + }, + }) + + leaderBefore, escapeBefore := leaderCounter(), escapeCounter() + result, err := forwardEthCall(t, ctx, network, 1, "0x3ea") + require.NoError(t, err) + assert.Contains(t, []string{"0x1111", "0x2222"}, result, "routed upstreams must still be tried after the leaders fail") + assert.Equal(t, leaderBefore+1, leaderCounter()) + assert.Equal(t, escapeBefore, escapeCounter(), "the escalation is already spent") + }) + +} diff --git a/telemetry/metrics.go b/telemetry/metrics.go index 033c525e5..73fc3968e 100644 --- a/telemetry/metrics.go +++ b/telemetry/metrics.go @@ -759,6 +759,15 @@ var ( Help: "Total number of times the per-request fallback escape hatch fired because the primary upstream set was exhausted with retryable errors.", }, []string{"project", "network", "category"}) + // MetricNetworkTipLeaderRouteTotal counts block-pinned requests sent to a + // fallback-tier upstream first because it already had the block while + // every routed upstream's head was below it. + MetricNetworkTipLeaderRouteTotal = DefineCounter(prometheus.CounterOpts{ + Namespace: "erpc", + Name: "network_tip_leader_route_total", + Help: "Total number of block-pinned requests routed to a fallback-tier upstream first because it had the block and no routed upstream did.", + }, []string{"project", "network", "category"}) + // MetricCacheExecutorAttempt counts every attempt a cache-connector // failsafe executor governs, keyed by the executor's identity (the // matchMethod / matchFinality it was configured with) and the outcome From 0767204685aa7f5064b47acf0df860b9a4e4e6f1 Mon Sep 17 00:00:00 2001 From: shpookas Date: Tue, 29 Sep 2026 18:23:35 +0200 Subject: [PATCH 2/4] fix(network): don't let an all-missing hedge leg cancel a fallback escalation The hedge keeper kept an ErrUpstreamsExhausted whose causes are all missing data, on the basis that no sibling leg can do better. Once the request has escalated to the fallbacks that no longer holds: the escalated leg may still be waiting on a fallback that has the data, while a later leg, with the escalation already spent, only re-sweeps the routed upstreams, misses again, and cancels it. Seen as "missing data" errors with hedges:1 on chains where a fallback announces heads first. Keep racing when the request escalated and no fallback appears among the causes yet. If every leg finishes this way the hedge still returns the last result. --- common/request.go | 6 +++ erpc/network_executor.go | 7 +++ erpc/networks_tip_leader_test.go | 93 ++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+) diff --git a/common/request.go b/common/request.go index a68358e14..092c63e83 100644 --- a/common/request.go +++ b/common/request.go @@ -1281,6 +1281,12 @@ func (r *NormalizedRequest) MarkEscalatedToFallbacks() bool { return r.escalatedToFallbacks.CompareAndSwap(false, true) } +// EscalatedToFallbacks reports whether the request has spent its fallback +// escalation. +func (r *NormalizedRequest) EscalatedToFallbacks() bool { + return r != nil && r.escalatedToFallbacks.Load() +} + // UserId returns the user ID from the user object, or "n/a" if not available func (r *NormalizedRequest) UserId() string { if r == nil { diff --git a/erpc/network_executor.go b/erpc/network_executor.go index cab3dd46d..f5b2a7a95 100644 --- a/erpc/network_executor.go +++ b/erpc/network_executor.go @@ -3,6 +3,7 @@ package erpc import ( "context" "errors" + "slices" "strings" "sync/atomic" "time" @@ -668,7 +669,13 @@ func (e *networkExecutor) runHedge( break } } + // Unless the request escalated to the fallbacks and none has + // answered yet: a sibling leg may still be waiting on one that + // has the data, so keep racing. if allMissing { + if req.EscalatedToFallbacks() && !slices.ContainsFunc(uxe.Upstreams(), isFallbackTier) { + return false + } return true } } diff --git a/erpc/networks_tip_leader_test.go b/erpc/networks_tip_leader_test.go index 4812ff1bd..ea512631b 100644 --- a/erpc/networks_tip_leader_test.go +++ b/erpc/networks_tip_leader_test.go @@ -6,6 +6,7 @@ import ( "strings" "sync/atomic" "testing" + "time" "github.com/erpc/erpc/common" "github.com/erpc/erpc/telemetry" @@ -43,6 +44,44 @@ func countEthCalls(host string, hits *atomic.Int64) { Reply(200) } +// missingOnPrimariesSlowOnFallbacks makes the primaries answer eth_call with +// missing data and the fallbacks serve it after delay. +func missingOnPrimariesSlowOnFallbacks(delay time.Duration) func() { + return func() { + for _, host := range []string{"rpc1.localhost", "rpc2.localhost"} { + host := host + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return r.URL.Host == host && strings.Contains(util.SafeReadBody(r), "eth_call") + }). + Reply(200). + JSON([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"header not found"}}`)) + } + for _, host := range []string{"rpc3.localhost", "rpc4.localhost"} { + host := host + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return r.URL.Host == host && strings.Contains(util.SafeReadBody(r), "eth_call") + }). + Reply(200). + Delay(delay). + JSON([]byte(`{"jsonrpc":"2.0","id":1,"result":"0x3333"}`)) + } + } +} + +// hedgeFaster hedges well before a fallback answers. +func hedgeFaster() []*common.FailsafeConfig { + return []*common.FailsafeConfig{{ + MatchMethod: "*", + Hedge: &common.HedgePolicyConfig{Delay: common.NewStaticDuration(20 * time.Millisecond), MaxCount: 1}, + }} +} + func forwardEthCall(t *testing.T, ctx context.Context, network *Network, id int, blockHex string) (string, error) { t.Helper() req := ethCallRequest(id, blockHex) @@ -183,4 +222,58 @@ func TestFailover_TipLeaderRouting(t *testing.T) { assert.Equal(t, escapeBefore, escapeCounter(), "the escalation is already spent") }) + t.Run("HedgeDoesNotCancelSlowLeader", func(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // The hedge leg sweeps the primaries (missing data) while the leader + // leg still waits on the fallback; the leader must win. + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // 1002 + enableFailover: true, + configure: unboundedPrimaries, + failsafe: hedgeFaster(), + mocks: missingOnPrimariesSlowOnFallbacks(80 * time.Millisecond), + }) + + for i := 0; i < 5; i++ { + result, err := forwardEthCall(t, ctx, network, i, "0x3ea") + require.NoError(t, err, "iter %d", i) + assert.Equal(t, "0x3333", result, "iter %d", i) + } + }) +} + +// A hedge leg that finds every routed upstream missing the data must not +// cancel a sibling leg that escalated to the fallbacks and is still waiting. +func TestFailover_HedgeKeepsEscalatedSibling(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Nobody's head reaches 1002, so there is no tip leader: the primaries + // miss, the escape sends one leg to the (slow) fallbacks, and the hedge + // leg misses on the primaries again. + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3e8", // 1000 + enableFailover: true, + configure: func(cfgs []*common.UpstreamConfig) { + for _, cfg := range cfgs { + cfg.Evm.BlockAvailability = nil + } + }, + failsafe: hedgeFaster(), + mocks: missingOnPrimariesSlowOnFallbacks(80 * time.Millisecond), + }) + + escapeBefore := promUtil.ToFloat64(telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call")) + for i := 0; i < 5; i++ { + result, err := forwardEthCall(t, ctx, network, i, "0x3ea") + require.NoError(t, err, "iter %d", i) + assert.Equal(t, "0x3333", result, "iter %d", i) + } + assert.Equal(t, escapeBefore+5, promUtil.ToFloat64(telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call"))) } From 4f7dcffa09e408e377e01785d71760a051207251 Mon Sep 17 00:00:00 2001 From: shpookas Date: Tue, 29 Sep 2026 18:50:20 +0200 Subject: [PATCH 3/4] fix(network): address review of tip-leader routing and hedge keeper - Hedge keeper: once the request escalated, never keep an all-missing result. ErrorsByUpstream is request-wide, so a fallback that already missed does not mean none is still working; the hedge still returns the last result if every leg misses. - Tip-leader routing no longer closes the escape: the sweep that took the escalation for its leaders still escapes, once, to the fallbacks it has not tried (e.g. an HTTP fallback whose 5m poller trails the block). - Check every routed upstream's head, fallback-tier ones included, and drop the default-tier count guard. - Only fallbacks allowed by the request's use-upstream selector can lead. --- erpc/network_executor.go | 12 +-- erpc/networks.go | 37 +++++----- erpc/networks_tip_leader_test.go | 123 +++++++++++++++++++++++++++++++ 3 files changed, 147 insertions(+), 25 deletions(-) diff --git a/erpc/network_executor.go b/erpc/network_executor.go index f5b2a7a95..57aecf8d6 100644 --- a/erpc/network_executor.go +++ b/erpc/network_executor.go @@ -3,7 +3,6 @@ package erpc import ( "context" "errors" - "slices" "strings" "sync/atomic" "time" @@ -669,14 +668,11 @@ func (e *networkExecutor) runHedge( break } } - // Unless the request escalated to the fallbacks and none has - // answered yet: a sibling leg may still be waiting on one that - // has the data, so keep racing. + // Unless the request escalated to the fallbacks: a sibling leg + // may still be waiting on one that has the data, so keep racing + // (if every leg ends like this the hedge returns the last one). if allMissing { - if req.EscalatedToFallbacks() && !slices.ContainsFunc(uxe.Upstreams(), isFallbackTier) { - return false - } - return true + return !req.EscalatedToFallbacks() } } // Underlying-retryable wrapped errors (e.g. ErrUpstreamsExhausted diff --git a/erpc/networks.go b/erpc/networks.go index 88a5c5e13..f300ea02e 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -2297,11 +2297,14 @@ func (n *Network) Forward(ctx context.Context, req *common.NormalizedRequest) (* // Tip-leader routing: a request pinned to a block above every routed // upstream's head goes first to the fallbacks that already have it // (a fallback's WS heads keep its poller current), then to the routed - // list. It spends the per-request escalation, so the escape below - // does not fire again. + // list. It takes the per-request escalation, so no other leg or + // attempt escapes; this sweep keeps its escape to the fallbacks it + // has not tried. + leaderRouted := false if !oneUpstreamOnly && n.cfg.Failover.Enabled() && !failsafeExecutor.HasConsensus() { if leaders := n.tipLeaderFallbacks(execSpanCtx, effectiveReq, method, upsList); len(leaders) > 0 && effectiveReq.MarkEscalatedToFallbacks() { + leaderRouted = true routed := effectiveReq.NextUpstream maxLoopIterations += len(leaders) nextUpstream = func() (common.Upstream, error) { @@ -2520,7 +2523,8 @@ func (n *Network) Forward(ctx context.Context, req *common.NormalizedRequest) (* fallbacks = append(fallbacks, fb) } } - if len(fallbacks) > 0 && effectiveReq.MarkEscalatedToFallbacks() { + if len(fallbacks) > 0 && (leaderRouted || effectiveReq.MarkEscalatedToFallbacks()) { + leaderRouted = false if bestResp != nil { bestResp.Release() bestResp = nil @@ -3722,9 +3726,9 @@ func preferTipLeaderForNearTipGetBlock(ups []common.Upstream, method string, bn } // tipLeaderFallbacks returns the fallback-tier upstreams, outside the routed -// list, whose head has reached the block req is pinned to, when every routed -// default-tier upstream's head is known and below it. Nil otherwise, -// including when a routed head is unknown. +// list and allowed by the request's upstream selector, whose head has reached +// the block req is pinned to, when every routed upstream's head is known and +// below it. Nil otherwise. func (n *Network) tipLeaderFallbacks(ctx context.Context, req *common.NormalizedRequest, method string, routed []common.Upstream) []common.Upstream { if n.Architecture() != common.ArchitectureEvm { return nil @@ -3734,28 +3738,27 @@ func (n *Network) tipLeaderFallbacks(ctx context.Context, req *common.Normalized return nil } routedIds := make(map[string]struct{}, len(routed)) - defaults := 0 for _, u := range routed { - routedIds[u.Id()] = struct{}{} - if isFallbackTier(u) { - continue - } - defaults++ if lb := upstreamLatestBlock(u); lb <= 0 || lb >= bn { return nil } + routedIds[u.Id()] = struct{}{} } - if defaults == 0 { - return nil + selector := "" + if d := req.Directives(); d != nil { + selector = d.UseUpstream } var leaders []common.Upstream for _, fb := range n.upstreamsRegistry.GetFallbackEscapeUpstreams(ctx, n.networkId, method) { - if _, ok := routedIds[fb.Id()]; ok { + if _, ok := routedIds[fb.Id()]; ok || upstreamLatestBlock(fb) < bn { continue } - if upstreamLatestBlock(fb) >= bn { - leaders = append(leaders, fb) + if selector != "" { + if match, err := common.UpstreamMatchesSelector(selector, fb); err != nil || !match { + continue + } } + leaders = append(leaders, fb) } return leaders } diff --git a/erpc/networks_tip_leader_test.go b/erpc/networks_tip_leader_test.go index ea512631b..77eb45299 100644 --- a/erpc/networks_tip_leader_test.go +++ b/erpc/networks_tip_leader_test.go @@ -277,3 +277,126 @@ func TestFailover_HedgeKeepsEscalatedSibling(t *testing.T) { } assert.Equal(t, escapeBefore+5, promUtil.ToFloat64(telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call"))) } + +// ethCallMock answers eth_call on host with body after delay. +func ethCallMock(host string, delay time.Duration, body string) { + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + return r.URL.Host == host && strings.Contains(util.SafeReadBody(r), "eth_call") + }). + Reply(200). + Delay(delay). + JSON([]byte(body)) +} + +// latestHeadMock pins host's polled latest block, ahead of its standard mock. +func latestHeadMock(host, latestHex string) { + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + b := util.SafeReadBody(r) + return r.URL.Host == host && strings.Contains(b, "eth_getBlockByNumber") && strings.Contains(b, `"latest"`) + }). + Reply(200). + JSON([]byte(`{"result":{"number":"` + latestHex + `","timestamp":"0x6702a8f0"}}`)) +} + +func unboundedAll(cfgs []*common.UpstreamConfig) { + for _, cfg := range cfgs { + cfg.Evm.BlockAvailability = nil + } +} + +const missingDataBody = `{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"header not found"}}` + +// A fallback that already answered missing data must not let a hedge leg +// cancel another fallback that is still working on the request. +func TestFailover_HedgeKeepsSlowFallbackAfterFastFallbackMiss(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3e8", // 1000, no tip leader + enableFailover: true, + configure: unboundedAll, + failsafe: hedgeFaster(), + mocks: func() { + ethCallMock("rpc1.localhost", 0, missingDataBody) + ethCallMock("rpc2.localhost", 0, missingDataBody) + ethCallMock("rpc3.localhost", 0, missingDataBody) + ethCallMock("rpc4.localhost", 80*time.Millisecond, `{"jsonrpc":"2.0","id":1,"result":"0x4444"}`) + }, + }) + + for i := 0; i < 5; i++ { + result, err := forwardEthCall(t, ctx, network, i, "0x3ea") + require.NoError(t, err, "iter %d", i) + assert.Equal(t, "0x4444", result, "iter %d", i) + } +} + +// When the tip leader fails, the sweep still escapes to the fallbacks it has +// not tried, even one whose polled head trails the block. +func TestFailover_TipLeaderKeepsEscapeToOtherFallbacks(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // fallback-1 at 1002 (leader) + enableFailover: true, + configure: unboundedAll, + mocks: func() { + latestHeadMock("rpc4.localhost", "0x3e8") // fallback-2 at 1000 + ethCallMock("rpc1.localhost", 0, missingDataBody) + ethCallMock("rpc2.localhost", 0, missingDataBody) + ethCallMock("rpc3.localhost", 0, `{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"internal error"}}`) + }, + }) + + leader := telemetry.MetricNetworkTipLeaderRouteTotal.WithLabelValues("main", "evm:999", "eth_call") + escape := telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call") + leaderBefore, escapeBefore := promUtil.ToFloat64(leader), promUtil.ToFloat64(escape) + + result, err := forwardEthCall(t, ctx, network, 1, "0x3ea") + require.NoError(t, err) + assert.Equal(t, "0x4444", result, "the untried fallback must still serve the request") + assert.Equal(t, leaderBefore+1, promUtil.ToFloat64(leader)) + assert.Equal(t, escapeBefore+1, promUtil.ToFloat64(escape)) +} + +// A request pinned to an upstream by the use-upstream directive never takes +// the tip-leader route to a fallback outside it. +func TestFailover_TipLeaderRespectsUseUpstream(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // 1002 + enableFailover: true, + configure: unboundedPrimaries, + }) + + leader := telemetry.MetricNetworkTipLeaderRouteTotal.WithLabelValues("main", "evm:999", "eth_call") + before := promUtil.ToFloat64(leader) + + req := ethCallRequest(1, "0x3ea") + req.SetDirectives(&common.RequestDirectives{UseUpstream: "primary-1"}) + req.SetNetwork(network) + resp, err := network.Forward(ctx, req) + require.NoError(t, err) + require.NotNil(t, resp) + defer resp.Release() + jrr, err := resp.JsonRpcResponse() + require.NoError(t, err) + assert.Equal(t, "0x1111", strings.Trim(jrr.GetResultString(), `"`)) + assert.Equal(t, before, promUtil.ToFloat64(leader)) +} From 11a0cb6bb7b684d7be662eb545022f909e47b89e Mon Sep 17 00:00:00 2001 From: shpookas Date: Tue, 29 Sep 2026 19:04:33 +0200 Subject: [PATCH 4/4] fix(network): don't null out a future block a reachable fallback has With served-tip on, tryShortCircuitFutureBlock answers a numbered eth_getBlockByNumber above every eligible head with null before the sweep. A fallback the policy keeps out of that set can already have the block (the tip-leader case), so with failover on, skip the short-circuit when a selector-allowed fallback's head has reached it. Shares the fallback lookup with tip-leader routing. Docs: the sweep that routes to tip leaders still escapes once to untried fallbacks; only other legs and retries do not. --- docs/pages/config/projects/networks.mdx | 2 +- erpc/networks.go | 19 ++++++++--- erpc/networks_failover_escape_test.go | 5 +++ erpc/networks_tip_leader_test.go | 42 +++++++++++++++++++++++++ 4 files changed, 63 insertions(+), 5 deletions(-) diff --git a/docs/pages/config/projects/networks.mdx b/docs/pages/config/projects/networks.mdx index d21a8f72c..b69dfde43 100644 --- a/docs/pages/config/projects/networks.mdx +++ b/docs/pages/config/projects/networks.mdx @@ -147,7 +147,7 @@ The escape: - fires at most once per request, and never for consensus requests; - is counted in `erpc_network_fallback_escape_total{project,network,category}` — expect zero in steady state. -With `onDefaultsExhausted` on, a request pinned to a block above every routed upstream's known head is sent first to the `tier:fallback` upstreams whose head has already reached that block (for example because their `newHeads` subscription announced it first), then to the routed upstreams. This uses the same once-per-request escalation, so the escape above does not fire again for that request. It does nothing when any routed upstream's head is unknown or already at the block, and never applies to consensus requests. It is counted in `erpc_network_tip_leader_route_total{project,network,category}`. +With `onDefaultsExhausted` on, a request pinned to a block above every routed upstream's known head is sent first to the `tier:fallback` upstreams whose head has already reached that block (for example because their `newHeads` subscription announced it first), then to the routed upstreams. This uses the same once-per-request escalation: other hedge legs and retries of that request do not escape, while the sweep that routed to the leaders still escapes once, to the fallbacks it has not tried, if the leaders and the routed upstreams all fail (counted in `erpc_network_fallback_escape_total` as usual). It does nothing when any routed upstream's head is unknown or already at the block, and never applies to consensus requests. It is counted in `erpc_network_tip_leader_route_total{project,network,category}`. The validation report warns when `onDefaultsExhausted` is enabled but no upstream is tagged `tier:fallback`. diff --git a/erpc/networks.go b/erpc/networks.go index f300ea02e..c974fee8d 100644 --- a/erpc/networks.go +++ b/erpc/networks.go @@ -915,6 +915,11 @@ func (n *Network) tryShortCircuitFutureBlock(ctx context.Context, req *common.No // Unknown head (fail open) or block within reach of some upstream. return nil, false } + // A fallback the sweep can still reach may already have it while the + // policy keeps it out of the eligible set. + if n.cfg.Failover.Enabled() && len(n.fallbacksAtBlock(ctx, req, method, bn, nil)) > 0 { + return nil, false + } jrr, err := common.NewJsonRpcResponse(req.ID(), nil, nil) if err != nil { return nil, false @@ -3744,13 +3749,19 @@ func (n *Network) tipLeaderFallbacks(ctx context.Context, req *common.Normalized } routedIds[u.Id()] = struct{}{} } + return n.fallbacksAtBlock(ctx, req, method, bn, routedIds) +} + +// fallbacksAtBlock returns the fallback-escape upstreams, not in skip and +// allowed by the request's upstream selector, whose head has reached bn. +func (n *Network) fallbacksAtBlock(ctx context.Context, req *common.NormalizedRequest, method string, bn int64, skip map[string]struct{}) []common.Upstream { selector := "" if d := req.Directives(); d != nil { selector = d.UseUpstream } - var leaders []common.Upstream + var out []common.Upstream for _, fb := range n.upstreamsRegistry.GetFallbackEscapeUpstreams(ctx, n.networkId, method) { - if _, ok := routedIds[fb.Id()]; ok || upstreamLatestBlock(fb) < bn { + if _, ok := skip[fb.Id()]; ok || upstreamLatestBlock(fb) < bn { continue } if selector != "" { @@ -3758,9 +3769,9 @@ func (n *Network) tipLeaderFallbacks(ctx context.Context, req *common.Normalized continue } } - leaders = append(leaders, fb) + out = append(out, fb) } - return leaders + return out } // upstreamLatestBlock is u's polled head, or 0 when unknown. diff --git a/erpc/networks_failover_escape_test.go b/erpc/networks_failover_escape_test.go index 91343c2f4..945a59372 100644 --- a/erpc/networks_failover_escape_test.go +++ b/erpc/networks_failover_escape_test.go @@ -169,6 +169,9 @@ func buildFailoverNetwork( networkConfig.Failover = &common.FailoverConfig{OnDefaultsExhausted: util.BoolPtr(true)} } networkConfig.Failsafe = opts.failsafe + if opts.network != nil { + opts.network(networkConfig) + } var policyEngine *policy.Engine if !opts.noPolicy { @@ -210,6 +213,8 @@ type failoverFixtureOpts struct { mocks func() // configure adjusts the upstream configs before the network is built. configure func(cfgs []*common.UpstreamConfig) + // network adjusts the network config before the network is built. + network func(cfg *common.NetworkConfig) } func setupFailoverFixture( diff --git a/erpc/networks_tip_leader_test.go b/erpc/networks_tip_leader_test.go index 77eb45299..1133cce99 100644 --- a/erpc/networks_tip_leader_test.go +++ b/erpc/networks_tip_leader_test.go @@ -400,3 +400,45 @@ func TestFailover_TipLeaderRespectsUseUpstream(t *testing.T) { assert.Equal(t, "0x1111", strings.Trim(jrr.GetResultString(), `"`)) assert.Equal(t, before, promUtil.ToFloat64(leader)) } + +// With served-tip on, a numbered eth_getBlockByNumber above every eligible +// head is short-circuited to null, unless a reachable fallback already has +// the block. +func TestFailover_FutureBlockShortCircuitSparesFallbackThatHasIt(t *testing.T) { + defer util.ResetGock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{ + primaryLatest: "0x3e8", // 1000 + fallbackLatest: "0x3ea", // 1002 + enableFailover: true, + network: func(cfg *common.NetworkConfig) { + cfg.Evm.ServedTip = &common.EvmServedTipConfig{EnabledFor: []string{"latest"}} + }, + mocks: func() { + for _, host := range []string{"rpc3.localhost", "rpc4.localhost"} { + host := host + gock.New("http://" + host). + Post(""). + Persist(). + Filter(func(r *http.Request) bool { + b := util.SafeReadBody(r) + return r.URL.Host == host && strings.Contains(b, "eth_getBlockByNumber") && strings.Contains(b, `"0x3ea"`) + }). + Reply(200). + JSON([]byte(`{"jsonrpc":"2.0","id":1,"result":{"number":"0x3ea","hash":"0xfb","timestamp":"0x6702a8f2"}}`)) + } + }, + }) + + req := common.NewNormalizedRequest([]byte(`{"jsonrpc":"2.0","id":1,"method":"eth_getBlockByNumber","params":["0x3ea",false]}`)) + req.SetNetwork(network) + resp, err := network.Forward(ctx, req) + require.NoError(t, err) + require.NotNil(t, resp) + defer resp.Release() + jrr, err := resp.JsonRpcResponse() + require.NoError(t, err) + assert.Contains(t, jrr.GetResultString(), `"0x3ea"`, "the fallback's block must be returned, not a synthetic null") +}