From 46c6a839bbb74b04a192b3f83aff465f8eb5a159 Mon Sep 17 00:00:00 2001
From: Jonny <jonny@linkpool.io>
Date: Tue, 6 Oct 2026 09:01:22 +0100
Subject: [PATCH 1/3] fix(network): don't let an all-missing hedge leg cancel
 its siblings

The hedge keeper kept an ErrUpstreamsExhausted whose causes were all
missing-data, on the premise that no sibling leg could do better. A
sibling may be on an upstream this leg never tried, e.g. a leg the
fallback escape sent to a slow fallback while another leg re-swept the
primaries, so the fast miss cancelled the leg about to succeed. Such a
result now keeps the race going; if every leg ends that way the hedge
still returns the last one to the retry layer.

Co-authored-by: Vitas Spokas <84668053+shpookas@users.noreply.github.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 erpc/network_executor.go              |  15 ---
 erpc/networks_failover_escape_test.go |  13 +-
 erpc/networks_failover_hedge_test.go  | 167 ++++++++++++++++++++++++++
 3 files changed, 179 insertions(+), 16 deletions(-)
 create mode 100644 erpc/networks_failover_hedge_test.go

diff --git a/erpc/network_executor.go b/erpc/network_executor.go
index cab3dd46d..8be2b55b2 100644
--- a/erpc/network_executor.go
+++ b/erpc/network_executor.go
@@ -656,21 +656,6 @@ func (e *networkExecutor) runHedge(
 				if uxe.Upstreams() == nil || len(uxe.Upstreams()) == 0 {
 					return false
 				}
-				// When every upstream returned -32004/missing-data, no sibling
-				// hedge leg can do better — they will all find the same upstreams
-				// consumed. Keep this result so the retry layer (shouldRetryWithReason)
-				// sees ErrUpstreamsExhausted directly and applies the 500ms delay.
-				causes := uxe.Errors()
-				allMissing := len(causes) > 0
-				for _, c := range causes {
-					if !common.HasErrorCode(c, common.ErrCodeEndpointMissingData) {
-						allMissing = false
-						break
-					}
-				}
-				if allMissing {
-					return true
-				}
 			}
 			// Underlying-retryable wrapped errors (e.g. ErrUpstreamsExhausted
 			// wrapping a 5xx) should continue racing for a healthier sibling.
diff --git a/erpc/networks_failover_escape_test.go b/erpc/networks_failover_escape_test.go
index 1f9f60cac..97434ac63 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 {
@@ -208,6 +211,10 @@ 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)
+	// network adjusts the network config before the network is built.
+	network func(cfg *common.NetworkConfig)
 }
 
 func setupFailoverFixture(
@@ -232,7 +239,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)
diff --git a/erpc/networks_failover_hedge_test.go b/erpc/networks_failover_hedge_test.go
new file mode 100644
index 000000000..cf6d1765f
--- /dev/null
+++ b/erpc/networks_failover_hedge_test.go
@@ -0,0 +1,167 @@
+package erpc
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"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"
+)
+
+// unboundedAll drops the availability bound from every upstream, so each is
+// called (and answers) for blocks above its polled head.
+func unboundedAll(cfgs []*common.UpstreamConfig) {
+	for _, cfg := range cfgs {
+		cfg.Evm.BlockAvailability = nil
+	}
+}
+
+// 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))
+}
+
+// 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)
+}
+
+const missingDataBody = `{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"header not found"}}`
+
+// hedgeFaster hedges well before a slow upstream 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)
+	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
+}
+
+// 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()
+
+	// 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:      unboundedAll,
+		failsafe:       hedgeFaster(),
+		mocks: func() {
+			ethCallMock("rpc1.localhost", 0, missingDataBody)
+			ethCallMock("rpc2.localhost", 0, missingDataBody)
+			ethCallMock("rpc3.localhost", 80*time.Millisecond, `{"jsonrpc":"2.0","id":1,"result":"0x3333"}`)
+			ethCallMock("rpc4.localhost", 80*time.Millisecond, `{"jsonrpc":"2.0","id":1,"result":"0x3333"}`)
+		},
+	})
+
+	escape := telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call")
+	escapeBefore := promUtil.ToFloat64(escape)
+	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(escape))
+}
+
+// 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
+		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)
+	}
+}
+
+// A fallback whose polled head is ahead does not take reads the primaries
+// serve: fallbacks are reached only through the escape. (The default
+// policy's probeExcluded may still mirror the request to them in the
+// background, so upstream hits are not counted here.)
+func TestFailover_FallbackAheadDoesNotTakeServedReads(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:      unboundedAll,
+	})
+
+	escape := telemetry.MetricNetworkFallbackEscapeTotal.WithLabelValues("main", "evm:999", "eth_call")
+	escapeBefore := promUtil.ToFloat64(escape)
+	for i := 0; i < 5; i++ {
+		result, err := forwardEthCall(t, ctx, network, i, "0x3ea")
+		require.NoError(t, err, "iter %d", i)
+		assert.Contains(t, []string{"0x1111", "0x2222"}, result, "iter %d", i)
+	}
+	assert.Equal(t, escapeBefore, promUtil.ToFloat64(escape))
+}

From aa0b3c90bd0f5d47c8e7117323542c8580d8e515 Mon Sep 17 00:00:00 2001
From: Jonny <jonny@linkpool.io>
Date: Tue, 6 Oct 2026 04:37:32 +0100
Subject: [PATCH 2/3] fix(ws): hold back fallback heads while a primary streams
 heads

newHeads is subscribed on every WebSocket upstream and each head goes
out from whichever source announces it first. With failover on, a
fallback-tier upstream whose WebSocket is faster told clients about a
block no primary had yet; their follow-up reads then missed on the
primaries, or escaped to the fallback while the primaries were healthy.

A fallback's head now reaches clients, and feeds the delivered-head
floor, only while no upstream outside that tier which the selection
policy routes to has a live newHeads subscription of its own. "Down"
is the policy's verdict, as for reads, so custom policies that keep
fallbacks in the ordered list behave the same. A network whose only
WebSocket upstreams are fallbacks, or whose primaries' subscriptions
are dead, keeps receiving fallback heads. A held-back head still
updates that fallback's state poller.

NetworkHandle.SuggestLatestBlock now reports whether the head may be
delivered; the indexer drops it before dedup otherwise.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/pages/operation/websocket.mdx            |  4 +-
 erpc/networks.go                              |  9 +-
 erpc/networks_ws_tip_test.go                  | 99 +++++++++++++++++++
 erpc/subscription_manager.go                  | 65 +++++++++---
 indexer/adapters/wsupstream/adapter.go        |  8 ++
 .../wsupstream/adapter_reconnect_test.go      |  6 +-
 indexer/indexer.go                            |  4 +-
 indexer/indexer_test.go                       | 34 ++++++-
 indexer/ingress.go                            |  9 +-
 indexer/integration_test.go                   |  3 +-
 10 files changed, 210 insertions(+), 31 deletions(-)

diff --git a/docs/pages/operation/websocket.mdx b/docs/pages/operation/websocket.mdx
index 3408a61b4..6fd0aa9a4 100644
--- a/docs/pages/operation/websocket.mdx
+++ b/docs/pages/operation/websocket.mdx
@@ -137,7 +137,7 @@ ws.send(JSON.stringify({
 
 ### Subscription features
 
-- **Resilient delivery** — `newHeads` is subscribed on every connected WebSocket upstream of the network, and each block is delivered from whichever upstream announces it first. `logs` and `newPendingTransactions` are subscribed on every WebSocket upstream whose circuit breaker is closed; with [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, upstreams tagged `tier:fallback` are used only when every other upstream rejects the subscribe.
+- **Resilient delivery** — `newHeads` is subscribed on every connected WebSocket upstream of the network, and each block is delivered from whichever upstream announces it first. With [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, heads from upstreams tagged `tier:fallback` are delivered only while no other upstream that the selection policy routes to has a live `newHeads` subscription, so clients are not told of a block only a fallback has while the primaries are up. A held-back head still updates that fallback's tip tracker. How quickly fallback heads take over when the primaries fail follows the selection policy's `evalInterval`. `logs` and `newPendingTransactions` are subscribed on every WebSocket upstream whose circuit breaker is closed; with [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, upstreams tagged `tier:fallback` are used only when every other upstream rejects the subscribe.
 - **No surprise disconnects** — upstream reconnects, scoring changes and brief outages never close your connection; eRPC re-subscribes upstream behind the scenes. eRPC closes a connection only on shutdown (`1001`), when the client stops answering pings, or when a `logs` subscription overflows its buffer (see below).
 - **Deduplication** — even though multiple upstreams are subscribed in parallel, you only receive each block, log or pending tx once. Logs are identified by `blockHash`, `transactionHash` and `logIndex`; a log missing any of them is passed through without deduplication.
 - **Shared subscriptions** — clients subscribing to the same event on the same network share one upstream subscription per upstream. Each client gets its own subscription ID and receives every notification independently.
@@ -329,7 +329,7 @@ eRPC minimises the gap in three ways:
 
 - **Per-block cache keying for moving tags.** The realtime cache entry for a `latest` / `finalized` request is keyed by the block number the tag currently resolves to, so each tip advance is a distinct key rather than pinning every call within the TTL to the same cached payload.
 - **`enforceHighestBlock` response rewrite.** When the upstream returns a block number below the network's tracked tip, eRPC's [integrity check](/config/failsafe/integrity) fetches the specific higher block from another upstream and returns that instead (falling back to the highest valid block it has if none can serve it).
-- **Delivered-head floor.** When an upstream WebSocket delivers a `newHeads` notification, eRPC updates that upstream's tip tracker before fanning the head out to clients, and records it as a floor for the network's `latest`. Every head a subscriber can receive feeds the floor once the delivering upstream's tip tracker has accepted it, and the floor is bounded: a value no live upstream comes close to (a stale or poisoned entry) is ignored rather than trusted, so it cannot pin the served tip. It never lifts `latest` past [`servedTip.guaranteedFor`](/config/projects/networks) / `guaranteedMethods`, and it does not apply to requests scoped with `use-upstream`. With shared state configured the floor is also propagated to other instances in the background, so they converge within the propagation delay rather than instantly.
+- **Delivered-head floor.** When an upstream WebSocket delivers a `newHeads` notification, eRPC updates that upstream's tip tracker before fanning the head out to clients, and records it as a floor for the network's `latest`. Every head a subscriber can receive (so not a held-back fallback head) feeds the floor once the delivering upstream's tip tracker has accepted it, and the floor is bounded: a value no live upstream comes close to (a stale or poisoned entry) is ignored rather than trusted, so it cannot pin the served tip. It never lifts `latest` past [`servedTip.guaranteedFor`](/config/projects/networks) / `guaranteedMethods`, and it does not apply to requests scoped with `use-upstream`. With shared state configured the floor is also propagated to other instances in the background, so they converge within the propagation delay rather than instantly.
 
 Even so, a transient single-block lag can reach a client when the chain produces a block faster than an in-flight HTTP round-trip completes. If your consumer treats single-block HTTP-vs-WS lag as a hard failure, widen its tolerance to match the chain's block time rather than expect eRPC (or any proxy) to eliminate it.
 
diff --git a/erpc/networks.go b/erpc/networks.go
index be977e951..d1ad6ebd6 100644
--- a/erpc/networks.go
+++ b/erpc/networks.go
@@ -3627,9 +3627,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
diff --git a/erpc/networks_ws_tip_test.go b/erpc/networks_ws_tip_test.go
index d86d83723..bd3c69cf1 100644
--- a/erpc/networks_ws_tip_test.go
+++ b/erpc/networks_ws_tip_test.go
@@ -10,6 +10,7 @@ import (
 	"github.com/erpc/erpc/common"
 	"github.com/erpc/erpc/data"
 	"github.com/erpc/erpc/health"
+	"github.com/erpc/erpc/internal/policy"
 	"github.com/erpc/erpc/thirdparty"
 	"github.com/erpc/erpc/upstream"
 	"github.com/erpc/erpc/util"
@@ -154,6 +155,104 @@ func TestNetworkHandle_SuggestLatestBlock_EveryKnownSourceLiftsLatest(t *testing
 		"a head a client can receive from the cordoned fallback floors latest")
 }
 
+type fakeHeadSource bool
+
+func (f fakeHeadSource) HeadsLive() bool { return bool(f) }
+
+// With failover on, a fallback's head reaches clients (and lifts latest) only
+// when no primary the policy routes to is streaming heads itself. The
+// fallback's own poller sees every head regardless.
+func TestNetworkHandle_FallbackHeadsHeldWhilePrimariesStream(t *testing.T) {
+	setup := func(t *testing.T, ctx context.Context) (*Network, map[string]*upstream.Upstream) {
+		network, ups, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{
+			primaryLatest:  "0x3e8", // 1000
+			fallbackLatest: "0x3e8",
+			enableFailover: true,
+		})
+		require.Equal(t, int64(1000), network.EvmHighestLatestBlockNumber(ctx))
+		byId := make(map[string]*upstream.Upstream, len(ups))
+		for _, u := range ups {
+			byId[u.Id()] = u
+		}
+		return network, byId
+	}
+
+	t.Run("HeldWhileAPrimaryStreams", func(t *testing.T) {
+		defer util.ResetGock()
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		network, ups := setup(t, ctx)
+
+		handle := &networkHandle{nw: network, heads: map[string]headSource{
+			"primary-1":  fakeHeadSource(false),
+			"primary-2":  fakeHeadSource(true),
+			"fallback-1": fakeHeadSource(true),
+		}}
+		assert.False(t, handle.SuggestLatestBlock("ws:fallback-1", 1010))
+		assert.Equal(t, int64(1010), ups["fallback-1"].EvmStatePoller().LatestBlock(),
+			"the fallback's poller still sees its head")
+		assert.Equal(t, int64(1000), network.EvmHighestLatestBlockNumber(ctx),
+			"a held head must not lift latest")
+
+		assert.True(t, handle.SuggestLatestBlock("ws:primary-2", 1001))
+		assert.Equal(t, int64(1001), network.EvmHighestLatestBlockNumber(ctx))
+	})
+
+	t.Run("DeliveredWhenNoPrimaryStreams", func(t *testing.T) {
+		defer util.ResetGock()
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		network, _ := setup(t, ctx)
+
+		// Primaries connected over HTTP only, or with no live subscription.
+		for _, heads := range []map[string]headSource{
+			{"fallback-1": fakeHeadSource(true)},
+			{"primary-1": fakeHeadSource(false), "fallback-1": fakeHeadSource(true)},
+		} {
+			handle := &networkHandle{nw: network, heads: heads}
+			assert.True(t, handle.SuggestLatestBlock("ws:fallback-1", 1010))
+		}
+		assert.Equal(t, int64(1010), network.EvmHighestLatestBlockNumber(ctx))
+	})
+
+	t.Run("DeliveredWhenPolicyExcludesPrimaries", func(t *testing.T) {
+		defer util.ResetGock()
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		network, ups := setup(t, ctx)
+
+		handle := &networkHandle{nw: network, heads: map[string]headSource{
+			"primary-1":  fakeHeadSource(true),
+			"primary-2":  fakeHeadSource(true),
+			"fallback-1": fakeHeadSource(true),
+		}}
+		require.False(t, handle.SuggestLatestBlock("ws:fallback-1", 1010))
+
+		ups["primary-1"].Cordon("*", "test")
+		ups["primary-2"].Cordon("*", "test")
+		policy.TickForTest(network.policyEngine, network.networkId, "*")
+
+		assert.True(t, handle.SuggestLatestBlock("ws:fallback-1", 1011))
+		assert.Equal(t, int64(1011), network.EvmHighestLatestBlockNumber(ctx))
+	})
+
+	t.Run("DeliveredWithFailoverOff", func(t *testing.T) {
+		defer util.ResetGock()
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		network, _, _ := setupFailoverFixture(t, ctx, failoverFixtureOpts{
+			primaryLatest:  "0x3e8",
+			fallbackLatest: "0x3e8",
+		})
+
+		handle := &networkHandle{nw: network, heads: map[string]headSource{
+			"primary-1":  fakeHeadSource(true),
+			"fallback-1": fakeHeadSource(true),
+		}}
+		assert.True(t, handle.SuggestLatestBlock("ws:fallback-1", 1010))
+	})
+}
+
 // The delivered-head floor is network-wide: a use-upstream-scoped request is
 // answered from its group's own head.
 func TestDeliveredHeadFloor_SkipsSelectorScopedRequests(t *testing.T) {
diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go
index b4e0ac8d2..251e185b8 100644
--- a/erpc/subscription_manager.go
+++ b/erpc/subscription_manager.go
@@ -259,14 +259,22 @@ func (sm *SubscriptionManager) bootstrapNetwork(ctx context.Context, nw *Network
 		return common.NewErrNoWsUpstreamAvailable(networkID)
 	}
 
-	sm.idx.RegisterNetwork(&networkHandle{nw: nw})
-	sm.idx.RegisterNetworkSelector(networkID, &subIngressSelector{nw: nw})
 	var adapterOpts wsupstream.Options
 	if cfg := nw.cfg; cfg != nil && cfg.Evm != nil && cfg.Evm.StripSubscribeFromBlockZero != nil {
 		adapterOpts.StripSubscribeFromBlockZero = *cfg.Evm.StripSubscribeFromBlockZero
 	}
+	adapters := make(map[string]*wsupstream.Adapter, len(wsUpstreams))
+	heads := make(map[string]headSource, len(wsUpstreams))
 	for _, up := range wsUpstreams {
-		adapter := wsupstream.New(up, networkID, sm.logger, adapterOpts)
+		if adapter := wsupstream.New(up, networkID, sm.logger, adapterOpts); adapter != nil {
+			adapters[up.Id()] = adapter
+			heads[up.Id()] = adapter
+		}
+	}
+	sm.idx.RegisterNetwork(&networkHandle{nw: nw, heads: heads})
+	sm.idx.RegisterNetworkSelector(networkID, &subIngressSelector{nw: nw})
+	for _, up := range wsUpstreams {
+		adapter := adapters[up.Id()]
 		if adapter == nil {
 			continue
 		}
@@ -416,38 +424,65 @@ func (sm *SubscriptionManager) recordFailureMetrics(
 // networkHandle adapts *Network to indexer.NetworkHandle.
 type networkHandle struct {
 	nw *Network
+	// heads are the network's WebSocket ingresses by upstream id; fixed once
+	// the network is registered.
+	heads map[string]headSource
+}
+
+// headSource is an ingress that streams newHeads.
+type headSource interface {
+	// HeadsLive reports whether its newHeads subscription is live.
+	HeadsLive() bool
 }
 
 func (h *networkHandle) Id() string { return h.nw.networkId }
 
 // SuggestLatestBlock passes a head from the ingress "ws:<upstreamId>" to
-// that upstream's state poller. Once the poller has accepted it (a major
-// jump is verified asynchronously first), a head from a tip candidate also
-// advances the network's delivered-head floor before clients see it; a
-// fallback-tier or policy-excluded upstream cannot lift "latest".
-func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64) {
+// that upstream's state poller and reports whether clients may receive it
+// (see deliversHeadsFrom). Once the poller has accepted a deliverable head (a
+// major jump is verified asynchronously first), it also advances the
+// network's delivered-head floor before clients see it.
+func (h *networkHandle) SuggestLatestBlock(sourceId string, blockNumber int64) bool {
 	upstreamID, ok := strings.CutPrefix(sourceId, "ws:")
 	if !ok {
-		return
+		return true
 	}
 	ctx := context.Background()
 	for _, u := range h.nw.upstreamsRegistry.GetNetworkUpstreams(ctx, h.nw.networkId) {
 		if u.Id() != upstreamID {
 			continue
 		}
+		deliver := h.deliversHeadsFrom(ctx, u)
 		poller := u.EvmStatePoller()
 		if poller == nil || poller.IsObjectNull() {
-			return
+			return deliver
 		}
 		poller.SuggestLatestBlock(blockNumber)
-		// Every source's heads reach clients (whichever delivers first), so
-		// every source feeds the floor — once its poller accepted the head,
-		// which excludes jumps still pending chain-id verification.
-		if poller.LatestBlock() >= blockNumber {
+		if deliver && poller.LatestBlock() >= blockNumber {
 			h.nw.NoteObservedLatestBlock(h.nw.appCtx, blockNumber)
 		}
-		return
+		return deliver
+	}
+	return true
+}
+
+// deliversHeadsFrom reports whether clients may receive u's heads. With
+// failover on, a fallback-tier upstream's heads are held back while some
+// upstream outside that tier, which the selection policy routes to, has a
+// live newHeads subscription of its own: clients are not told of a block
+// only a fallback has while the primaries are up. Whether they are up is the
+// policy's verdict, as for reads. Without such a primary (none eligible, or
+// none streaming heads) the fallbacks' heads are delivered.
+func (h *networkHandle) deliversHeadsFrom(ctx context.Context, u common.Upstream) bool {
+	if h.nw.cfg == nil || !h.nw.cfg.Failover.Enabled() || !isFallbackTier(u) {
+		return true
+	}
+	for _, c := range h.nw.tipCandidateUpstreams(ctx, "*") {
+		if s := h.heads[c.Id()]; s != nil && !isFallbackTier(c) && s.HeadsLive() {
+			return false
+		}
 	}
+	return true
 }
 
 var (
diff --git a/indexer/adapters/wsupstream/adapter.go b/indexer/adapters/wsupstream/adapter.go
index d0814ea9b..0f3caa545 100644
--- a/indexer/adapters/wsupstream/adapter.go
+++ b/indexer/adapters/wsupstream/adapter.go
@@ -116,6 +116,14 @@ func New(up *upstream.Upstream, networkID string, logger *zerolog.Logger, opts O
 
 func (a *Adapter) Name() string { return "ws:" + a.upstreamID }
 
+// HeadsLive reports whether the newHeads subscription is live on the
+// current connection.
+func (a *Adapter) HeadsLive() bool {
+	a.subsMu.Lock()
+	defer a.subsMu.Unlock()
+	return a.heads.id != ""
+}
+
 // Start registers the connection hooks and subscribes newHeads in the
 // background, detached from ctx.
 func (a *Adapter) Start(_ context.Context, nw indexer.NetworkHandle, sink indexer.Sink) error {
diff --git a/indexer/adapters/wsupstream/adapter_reconnect_test.go b/indexer/adapters/wsupstream/adapter_reconnect_test.go
index 672e152c0..ec46a84c7 100644
--- a/indexer/adapters/wsupstream/adapter_reconnect_test.go
+++ b/indexer/adapters/wsupstream/adapter_reconnect_test.go
@@ -26,8 +26,8 @@ import (
 
 type fakeNetworkHandle struct{}
 
-func (fakeNetworkHandle) Id() string                       { return "evm:123" }
-func (fakeNetworkHandle) SuggestLatestBlock(string, int64) {}
+func (fakeNetworkHandle) Id() string                            { return "evm:123" }
+func (fakeNetworkHandle) SuggestLatestBlock(string, int64) bool { return true }
 
 type fakeSink struct {
 	events chan indexer.StreamEvent
@@ -235,10 +235,12 @@ func TestAdapterResubscribesWithRetryAfterReconnect(t *testing.T) {
 		return send(ctx, nq, bypass)
 	}
 
+	require.False(t, a.HeadsLive(), "no heads before the subscribe succeeds")
 	require.NoError(t, a.Start(context.Background(), fakeNetworkHandle{}, sink))
 
 	require.Eventually(t, subscribedHeads(a), 3*time.Second, 10*time.Millisecond,
 		"adapter never subscribed newHeads despite the failures stopping after %d", failuresPerEpoch)
+	require.True(t, a.HeadsLive())
 	require.GreaterOrEqual(t, forwardCalls.Load(), int64(failuresPerEpoch+1),
 		"expected the subscribe to be retried through failures")
 
diff --git a/indexer/indexer.go b/indexer/indexer.go
index e382a5d88..11a592f3e 100644
--- a/indexer/indexer.go
+++ b/indexer/indexer.go
@@ -303,7 +303,9 @@ func (i *Indexer) Ingest(ev StreamEvent) {
 	// Before dedup, so every source's head counts even if another source
 	// already delivered it.
 	if ev.Kind == KindNewHead && !ev.Block.Zero() && ev.SourceId != "" {
-		ns.handle.SuggestLatestBlock(ev.SourceId, ev.Block.Number)
+		if !ns.handle.SuggestLatestBlock(ev.SourceId, ev.Block.Number) {
+			return
+		}
 	}
 
 	// The upstream's removed flag is trusted as-is.
diff --git a/indexer/indexer_test.go b/indexer/indexer_test.go
index fcbe50949..3d0866c88 100644
--- a/indexer/indexer_test.go
+++ b/indexer/indexer_test.go
@@ -18,7 +18,8 @@ type fakeNetwork struct {
 	id string
 
 	mu          sync.Mutex
-	suggestedBy map[string][]int64 // sourceId -> block nums seen
+	suggestedBy map[string][]int64  // sourceId -> block nums seen
+	held        map[string]struct{} // sourceIds whose heads are not delivered
 }
 
 func newFakeNetwork(id string) *fakeNetwork {
@@ -29,10 +30,12 @@ func newFakeNetwork(id string) *fakeNetwork {
 }
 
 func (n *fakeNetwork) Id() string { return n.id }
-func (n *fakeNetwork) SuggestLatestBlock(sourceId string, block int64) {
+func (n *fakeNetwork) SuggestLatestBlock(sourceId string, block int64) bool {
 	n.mu.Lock()
+	defer n.mu.Unlock()
 	n.suggestedBy[sourceId] = append(n.suggestedBy[sourceId], block)
-	n.mu.Unlock()
+	_, held := n.held[sourceId]
+	return !held
 }
 
 type fakeEgress struct {
@@ -271,6 +274,31 @@ func TestIndexer_NewHead_StalerDroppedKeepsStatePollerFed(t *testing.T) {
 	}
 }
 
+// A head the network does not deliver is still observed, and does not
+// advance the dedup marker past the heads the other sources deliver.
+func TestIndexer_NewHead_HeldSourceObservedNotDelivered(t *testing.T) {
+	idx := newIndexer(t)
+	nw := newFakeNetwork("evm:1")
+	nw.held = map[string]struct{}{"ws:fb": {}}
+	idx.RegisterNetwork(nw)
+	eg := &fakeEgress{name: "eg1", acceptAllHeads: true}
+	idx.Attach(eg)
+
+	idx.Ingest(StreamEvent{Kind: KindNewHead, NetworkId: "evm:1", SourceId: "ws:fb", Block: BlockRef{Number: 101, Hash: "0xBBB"}})
+	if got := eg.count(); got != 0 {
+		t.Fatalf("held head delivered: got %d", got)
+	}
+	idx.Ingest(StreamEvent{Kind: KindNewHead, NetworkId: "evm:1", SourceId: "ws:up1", Block: BlockRef{Number: 101, Hash: "0xBBB"}})
+	if got := eg.count(); got != 1 {
+		t.Fatalf("the delivering source's head must go out: want 1, got %d", got)
+	}
+	nw.mu.Lock()
+	defer nw.mu.Unlock()
+	if len(nw.suggestedBy["ws:fb"]) != 1 {
+		t.Fatalf("held source must still be observed, got %v", nw.suggestedBy)
+	}
+}
+
 func TestIndexer_Log_RefcountFanOutAndTeardown(t *testing.T) {
 	idx := newIndexer(t)
 	nw := newFakeNetwork("evm:1")
diff --git a/indexer/ingress.go b/indexer/ingress.go
index 3df2d4086..26338e672 100644
--- a/indexer/ingress.go
+++ b/indexer/ingress.go
@@ -11,10 +11,11 @@ type Sink interface {
 // NetworkHandle is the part of a network an ingress may touch.
 type NetworkHandle interface {
 	Id() string
-	// SuggestLatestBlock reports a head observed by source. It is called
-	// before dedup and fan-out, so every source's observation counts and
-	// the network tip moves before clients see the head.
-	SuggestLatestBlock(sourceId string, blockNumber int64)
+	// SuggestLatestBlock reports a head observed by source and whether
+	// clients may receive it. It is called before dedup and fan-out, so
+	// every source's observation counts and the network tip moves before
+	// clients see the head; a head it does not deliver is dropped.
+	SuggestLatestBlock(sourceId string, blockNumber int64) (deliver bool)
 }
 
 // EventIngress turns a transport-specific subscription into StreamEvents
diff --git a/indexer/integration_test.go b/indexer/integration_test.go
index 976144854..ebecd78a8 100644
--- a/indexer/integration_test.go
+++ b/indexer/integration_test.go
@@ -66,13 +66,14 @@ type stubNetwork struct {
 }
 
 func (s *stubNetwork) Id() string { return s.id }
-func (s *stubNetwork) SuggestLatestBlock(sourceID string, block int64) {
+func (s *stubNetwork) SuggestLatestBlock(sourceID string, block int64) bool {
 	s.mu.Lock()
 	if s.suggestions == nil {
 		s.suggestions = make(map[string][]int64)
 	}
 	s.suggestions[sourceID] = append(s.suggestions[sourceID], block)
 	s.mu.Unlock()
+	return true
 }
 
 type recordingEgress struct {

From 361a84af84043f557e5ffaa0b9f9796baebe93ac Mon Sep 17 00:00:00 2001
From: Jonny <jonny@linkpool.io>
Date: Tue, 6 Oct 2026 09:01:10 +0100
Subject: [PATCH 3/3] fix(ws): keep filter subscriptions on the same tiers as
 heads

logs and newPendingTransactions picked their upstreams once, when the
first client subscribed: on every primary whose eth_subscribe breaker
was closed, on the fallbacks only if every primary rejected the
subscribe. Nothing revisited that for the filter's lifetime, so a
filter went silent when its primaries dropped it, stayed on the
fallbacks after the primaries recovered, and never reached a primary
whose first subscribe failed.

A filter now belongs on every primary the selection policy routes to,
and on the fallback tier while none of those has it live; the indexer
rechecks this on every head, adding and removing the filter as that
changes. The fallbacks hand a filter back only once a primary has it
live, so there is no gap. An upstream whose subscribe fails keeps the
filter and retries it in the background, like its other subscriptions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/pages/operation/websocket.mdx            |   2 +-
 erpc/networks_ws_tip_test.go                  |  24 ++++
 erpc/subscription_manager.go                  |  53 ++++---
 indexer/adapters/nullingress/adapter.go       |   2 +
 indexer/adapters/wsupstream/adapter.go        |  50 +++++--
 .../wsupstream/adapter_reconnect_test.go      |  40 +++++-
 indexer/indexer.go                            | 136 +++++++++++++++---
 indexer/indexer_test.go                       | 121 ++++++++++++++++
 indexer/ingress.go                            |   9 +-
 9 files changed, 377 insertions(+), 60 deletions(-)

diff --git a/docs/pages/operation/websocket.mdx b/docs/pages/operation/websocket.mdx
index 6fd0aa9a4..5b9b0290e 100644
--- a/docs/pages/operation/websocket.mdx
+++ b/docs/pages/operation/websocket.mdx
@@ -137,7 +137,7 @@ ws.send(JSON.stringify({
 
 ### Subscription features
 
-- **Resilient delivery** — `newHeads` is subscribed on every connected WebSocket upstream of the network, and each block is delivered from whichever upstream announces it first. With [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, heads from upstreams tagged `tier:fallback` are delivered only while no other upstream that the selection policy routes to has a live `newHeads` subscription, so clients are not told of a block only a fallback has while the primaries are up. A held-back head still updates that fallback's tip tracker. How quickly fallback heads take over when the primaries fail follows the selection policy's `evalInterval`. `logs` and `newPendingTransactions` are subscribed on every WebSocket upstream whose circuit breaker is closed; with [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, upstreams tagged `tier:fallback` are used only when every other upstream rejects the subscribe.
+- **Resilient delivery** — `newHeads` is subscribed on every connected WebSocket upstream of the network, and each block is delivered from whichever upstream announces it first. With [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, heads from upstreams tagged `tier:fallback` are delivered only while no other upstream that the selection policy routes to has a live `newHeads` subscription, so clients are not told of a block only a fallback has while the primaries are up. A held-back head still updates that fallback's tip tracker. How quickly fallback heads take over when the primaries fail follows the selection policy's `evalInterval`. `logs` and `newPendingTransactions` are subscribed on every WebSocket upstream; with [`failover.onDefaultsExhausted`](/config/projects/networks#failover) enabled, on every one outside `tier:fallback` that the selection policy routes to, and on the `tier:fallback` upstreams only while none of those has the subscription live. This is rechecked on every head, so the fallbacks take over when the primaries drop a subscription (or the policy excludes them) and hand it back once a primary has it live again; a subscribe that fails is retried in the background.
 - **No surprise disconnects** — upstream reconnects, scoring changes and brief outages never close your connection; eRPC re-subscribes upstream behind the scenes. eRPC closes a connection only on shutdown (`1001`), when the client stops answering pings, or when a `logs` subscription overflows its buffer (see below).
 - **Deduplication** — even though multiple upstreams are subscribed in parallel, you only receive each block, log or pending tx once. Logs are identified by `blockHash`, `transactionHash` and `logIndex`; a log missing any of them is passed through without deduplication.
 - **Shared subscriptions** — clients subscribing to the same event on the same network share one upstream subscription per upstream. Each client gets its own subscription ID and receives every notification independently.
diff --git a/erpc/networks_ws_tip_test.go b/erpc/networks_ws_tip_test.go
index bd3c69cf1..275147ce3 100644
--- a/erpc/networks_ws_tip_test.go
+++ b/erpc/networks_ws_tip_test.go
@@ -300,3 +300,27 @@ func TestDeliveredHeadFloor_IsScopedPerProject(t *testing.T) {
 	assert.Equal(t, int64(1001), network.latestBlockShared.GetValue())
 	assert.Equal(t, int64(0), sibling.latestBlockShared.GetValue())
 }
+
+// Filter subscriptions follow the same tiers as heads: eligible primaries
+// carry them, the fallback tier stands in, and without failover every
+// WebSocket upstream is a default.
+func TestTierWsIngresses(t *testing.T) {
+	ws := []common.Upstream{
+		common.NewFakeUpstream("p1"),
+		common.NewFakeUpstream("p2"),
+		common.NewFakeUpstream("fb", common.WithTags(common.TagTierFallback)),
+	}
+	eligible := map[string]struct{}{"p1": {}, "fb": {}}
+
+	d, f := tierWsIngresses(ws, eligible, true)
+	assert.Equal(t, []string{"ws:p1"}, d, "an ineligible primary does not carry filters")
+	assert.Equal(t, []string{"ws:fb"}, f)
+
+	d, f = tierWsIngresses(ws, nil, true)
+	assert.Empty(t, d, "no eligible primary: only the fallbacks remain")
+	assert.Equal(t, []string{"ws:fb"}, f)
+
+	d, f = tierWsIngresses(ws, nil, false)
+	assert.Equal(t, []string{"ws:p1", "ws:p2", "ws:fb"}, d)
+	assert.Empty(t, f)
+}
diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go
index 251e185b8..49b5c1dff 100644
--- a/erpc/subscription_manager.go
+++ b/erpc/subscription_manager.go
@@ -13,7 +13,6 @@ import (
 	"github.com/erpc/erpc/indexer/adapters/wsclient"
 	"github.com/erpc/erpc/indexer/adapters/wsupstream"
 	"github.com/erpc/erpc/telemetry"
-	"github.com/erpc/erpc/upstream"
 	"github.com/rs/zerolog"
 )
 
@@ -287,36 +286,48 @@ func (sm *SubscriptionManager) bootstrapNetwork(ctx context.Context, nw *Network
 	return nil
 }
 
-// subIngressSelector picks the ingresses of a filter subscribe the way the
-// HTTP path picks upstreams: by score for eth_subscribe, skipping upstreams
-// whose circuit breaker is open. Fallback-tier upstreams form the fallback
-// tier only when the network's failover is enabled, as over HTTP.
+// subIngressSelector picks the ingresses that carry a filter subscription.
+// With failover on, the defaults are the upstreams outside the fallback tier
+// that the selection policy routes to, and the fallback-tier upstreams stand
+// in for them (see indexer.IngressSelector): the same rule newHeads follows
+// (see networkHandle.deliversHeadsFrom). Without failover every WebSocket
+// upstream is a default.
 type subIngressSelector struct {
 	nw *Network
 }
 
 func (s *subIngressSelector) Select(_, _ string, _ []interface{}) (defaults, fallbacks []string) {
-	ups, err := s.nw.upstreamsRegistry.GetSortedUpstreams(context.Background(), s.nw.networkId, MethodEthSubscribe)
-	if err != nil {
-		return nil, nil
-	}
-
+	ctx := context.Background()
 	failoverOn := s.nw.cfg != nil && s.nw.cfg.Failover.Enabled()
-	for _, u := range ups {
-		up, ok := u.(*upstream.Upstream)
-		if !ok {
-			continue
-		}
-		cfg := up.Config()
-		if cfg == nil || !upstream.IsWsEndpoint(cfg.Endpoint) || up.IsDown(MethodEthSubscribe) {
-			continue
+	eligible := make(map[string]struct{})
+	if failoverOn {
+		for _, u := range s.nw.tipCandidateUpstreams(ctx, "*") {
+			eligible[u.Id()] = struct{}{}
 		}
+	}
+	var ws []common.Upstream
+	for _, up := range s.nw.upstreamsRegistry.GetWsUpstreams(ctx, s.nw.networkId) {
+		ws = append(ws, up)
+	}
+	return tierWsIngresses(ws, eligible, failoverOn)
+}
+
+// tierWsIngresses names the ingresses of the WebSocket upstreams ws: with
+// failover on, the eligible ones outside the fallback tier are the defaults
+// and the fallback tier the fallbacks; otherwise all are defaults.
+func tierWsIngresses(ws []common.Upstream, eligible map[string]struct{}, failoverOn bool) (defaults, fallbacks []string) {
+	for _, up := range ws {
 		name := "ws:" + up.Id()
-		if failoverOn && cfg.HasTag(common.TagTierFallback) {
+		switch {
+		case !failoverOn:
+			defaults = append(defaults, name)
+		case isFallbackTier(up):
 			fallbacks = append(fallbacks, name)
-			continue
+		default:
+			if _, ok := eligible[up.Id()]; ok {
+				defaults = append(defaults, name)
+			}
 		}
-		defaults = append(defaults, name)
 	}
 	return defaults, fallbacks
 }
diff --git a/indexer/adapters/nullingress/adapter.go b/indexer/adapters/nullingress/adapter.go
index 475dfbe84..77ccbfd48 100644
--- a/indexer/adapters/nullingress/adapter.go
+++ b/indexer/adapters/nullingress/adapter.go
@@ -60,6 +60,8 @@ func (a *Adapter) RemoveFilter(context.Context, string, string) error {
 	return nil
 }
 
+func (a *Adapter) FilterLive(string, string) bool { return true }
+
 // Push injects an event. It blocks when the buffer is full, so a stalled
 // indexer shows up as a test timeout rather than silent drops.
 func (a *Adapter) Push(ev indexer.StreamEvent) {
diff --git a/indexer/adapters/wsupstream/adapter.go b/indexer/adapters/wsupstream/adapter.go
index 0f3caa545..08b42742a 100644
--- a/indexer/adapters/wsupstream/adapter.go
+++ b/indexer/adapters/wsupstream/adapter.go
@@ -61,7 +61,11 @@ type Adapter struct {
 	// resubCancel cancels the resubscribe loop of connection resubEpoch.
 	resubCancel context.CancelFunc
 	resubEpoch  uint64
-	heads       upstreamSub
+	// resubDone is set once that loop has established everything;
+	// resubAgain asks a running loop for another pass before it finishes.
+	resubDone  bool
+	resubAgain bool
+	heads      upstreamSub
 	// filters (by filterKey) survive disconnects to be resubscribed.
 	filters map[string]*filterSub
 }
@@ -143,6 +147,7 @@ func (a *Adapter) Start(_ context.Context, nw indexer.NetworkHandle, sink indexe
 		if a.resubCancel != nil {
 			a.resubCancel()
 		}
+		a.resubDone = false
 		a.heads.id = ""
 		for _, sub := range a.filters {
 			sub.id = ""
@@ -160,7 +165,11 @@ func (a *Adapter) startResubscribe() {
 	epoch := a.wsClient.Epoch()
 	a.subsMu.Lock()
 	defer a.subsMu.Unlock()
-	if epoch == 0 || epoch == a.resubEpoch {
+	if epoch == 0 {
+		return
+	}
+	if epoch == a.resubEpoch && !a.resubDone {
+		a.resubAgain = true
 		return
 	}
 	if a.resubCancel != nil {
@@ -168,12 +177,13 @@ func (a *Adapter) startResubscribe() {
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	a.resubCancel, a.resubEpoch = cancel, epoch
+	a.resubDone, a.resubAgain = false, false
 	go a.resubscribeWithRetry(ctx)
 }
 
-// EnsureFilter subscribes a filter; a no-op if it is already subscribed. A
-// filter this call added is dropped again on failure, so it isn't
-// resubscribed on every reconnect.
+// EnsureFilter subscribes a filter; a no-op if it is already subscribed. On
+// failure the filter is kept and retried in the background, like every
+// other subscription of this upstream, until RemoveFilter.
 func (a *Adapter) EnsureFilter(ctx context.Context, subType, paramsHash string, params []interface{}) error {
 	key := filterKey(subType, paramsHash)
 
@@ -186,19 +196,23 @@ func (a *Adapter) EnsureFilter(ctx context.Context, subType, paramsHash string,
 	a.subsMu.Unlock()
 
 	sub.mu.Lock()
-	defer sub.mu.Unlock()
 	err := a.subscribeFilterLocked(ctx, sub)
-	if err != nil && !exists {
-		a.subsMu.Lock()
-		if a.filters[key] == sub {
-			delete(a.filters, key)
-		}
-		a.subsMu.Unlock()
-		a.drop(ctx, &sub.upstreamSub)
+	sub.mu.Unlock()
+	if err != nil {
+		a.startResubscribe()
 	}
 	return err
 }
 
+// FilterLive reports whether the filter's subscription is live on the
+// current connection.
+func (a *Adapter) FilterLive(subType, paramsHash string) bool {
+	a.subsMu.Lock()
+	defer a.subsMu.Unlock()
+	sub, ok := a.filters[filterKey(subType, paramsHash)]
+	return ok && sub.id != ""
+}
+
 // RemoveFilter unsubscribes a filter and stops resubscribing it.
 func (a *Adapter) RemoveFilter(ctx context.Context, subType, paramsHash string) error {
 	key := filterKey(subType, paramsHash)
@@ -257,6 +271,16 @@ func (a *Adapter) resubscribeWithRetry(ctx context.Context) {
 			}
 		}
 		if done {
+			a.subsMu.Lock()
+			again := a.resubAgain && ctx.Err() == nil
+			a.resubAgain = false
+			if !again && ctx.Err() == nil {
+				a.resubDone = true
+			}
+			a.subsMu.Unlock()
+			if again {
+				continue
+			}
 			a.logger.Info().Msg("all upstream subscriptions (re)established")
 			return
 		}
diff --git a/indexer/adapters/wsupstream/adapter_reconnect_test.go b/indexer/adapters/wsupstream/adapter_reconnect_test.go
index ec46a84c7..054bcc991 100644
--- a/indexer/adapters/wsupstream/adapter_reconnect_test.go
+++ b/indexer/adapters/wsupstream/adapter_reconnect_test.go
@@ -353,7 +353,8 @@ func TestAdapterRemoveFilterDuringSubscribe(t *testing.T) {
 }
 
 // TestAdapterEnsureFilterWhileDisconnected: EnsureFilter must report the
-// failure (so the indexer tries other ingresses) and not keep the filter.
+// failure (so the indexer tries other ingresses) and keep the filter, to
+// subscribe it once connected.
 func TestAdapterEnsureFilterWhileDisconnected(t *testing.T) {
 	server := newNotifyServer(t)
 	u := server.wsURL(t)
@@ -361,9 +362,42 @@ func TestAdapterEnsureFilterWhileDisconnected(t *testing.T) {
 	a, _ := newTestAdapter(t, u)
 
 	params := logsParams("0xabc")
-	err := a.EnsureFilter(context.Background(), indexer.SubTypeLogs, indexer.BuildParamsKey(params), params)
+	hash := indexer.BuildParamsKey(params)
+	err := a.EnsureFilter(context.Background(), indexer.SubTypeLogs, hash, params)
 	require.ErrorIs(t, err, errNotConnected)
+	assert.False(t, a.FilterLive(indexer.SubTypeLogs, hash))
 	a.subsMu.Lock()
-	assert.Empty(t, a.filters)
+	assert.Len(t, a.filters, 1)
 	a.subsMu.Unlock()
 }
+
+// TestAdapterRetriesFailedEnsureFilter: a filter whose subscribe failed on
+// a live connection is retried in the background until it is live, and
+// RemoveFilter ends that.
+func TestAdapterRetriesFailedEnsureFilter(t *testing.T) {
+	compressResubRetry(t)
+	server := newNotifyServer(t)
+	a, sink := newTestAdapter(t, server.wsURL(t))
+	<-server.newConn
+	require.NoError(t, a.Start(context.Background(), fakeNetworkHandle{}, sink))
+	require.Eventually(t, subscribedHeads(a), 3*time.Second, 10*time.Millisecond)
+
+	var failuresLeft atomic.Int64
+	failuresLeft.Store(2)
+	send := a.forward
+	a.forward = func(ctx context.Context, nq *common.NormalizedRequest, bypass bool) (*common.NormalizedResponse, error) {
+		if m, _ := nq.Method(); m == methodEthSubscribe && failuresLeft.Add(-1) >= 0 {
+			return nil, errors.New("circuit breaker is open on upstream-level")
+		}
+		return send(ctx, nq, bypass)
+	}
+
+	params := logsParams("0xabc")
+	hash := indexer.BuildParamsKey(params)
+	require.Error(t, a.EnsureFilter(context.Background(), indexer.SubTypeLogs, hash, params))
+	require.Eventually(t, func() bool { return a.FilterLive(indexer.SubTypeLogs, hash) },
+		3*time.Second, 10*time.Millisecond, "the failed filter must be retried until live")
+
+	require.NoError(t, a.RemoveFilter(context.Background(), indexer.SubTypeLogs, hash))
+	assert.False(t, a.FilterLive(indexer.SubTypeLogs, hash))
+}
diff --git a/indexer/indexer.go b/indexer/indexer.go
index 11a592f3e..2b408c79b 100644
--- a/indexer/indexer.go
+++ b/indexer/indexer.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 
 	"github.com/rs/zerolog"
 )
@@ -50,6 +51,9 @@ type networkState struct {
 
 	filterMu sync.RWMutex
 	filters  map[string]*filterState // paramsHash -> state
+
+	// reconciling is set while a reconcile of the filters runs.
+	reconciling atomic.Bool
 }
 
 // filterState is one filter subscription shared by every client with the
@@ -61,6 +65,12 @@ type filterState struct {
 	subscribed bool // guarded by mu
 	refs       int  // guarded by networkState.filterMu
 	dedup      *DedupWindow
+
+	// subType and params are the filter's; on names the ingresses it was
+	// put on and not removed from since. Guarded by mu.
+	subType string
+	params  []interface{}
+	on      map[string]struct{}
 }
 
 // New returns an empty Indexer. Events for networks not registered with
@@ -147,7 +157,8 @@ func (i *Indexer) EnsureFilter(ctx context.Context, networkId, subType string, p
 	if f.subscribed {
 		return paramsHash, nil
 	}
-	if err := i.subscribe(ctx, ns, subType, paramsHash, params); err != nil {
+	f.subType, f.params, f.on = subType, params, make(map[string]struct{})
+	if err := i.subscribe(ctx, ns, f, paramsHash); err != nil {
 		// Ingresses may keep a filter whose subscribe failed (to retry on
 		// reconnect), so remove it everywhere.
 		i.removeFromIngresses(ctx, ns, subType, paramsHash)
@@ -163,30 +174,24 @@ func (i *Indexer) EnsureFilter(ctx context.Context, networkId, subType string, p
 	return paramsHash, nil
 }
 
-// subscribe tries the selector's defaults, then its fallbacks only if every
-// default failed. It fails when no ingress subscribed, including when none
-// was selected.
-func (i *Indexer) subscribe(ctx context.Context, ns *networkState, subType, paramsHash string, params []interface{}) error {
+// subscribe puts a new filter on the selector's defaults, and on its
+// fallbacks if no default took it. It fails when no ingress subscribed,
+// including when none was selected. An ingress that failed keeps retrying
+// the filter until it is removed. The caller holds f.mu.
+func (i *Indexer) subscribe(ctx context.Context, ns *networkState, f *filterState, paramsHash string) error {
 	networkId := ns.handle.Id()
-
-	ns.ingressMu.RLock()
-	ings := make(map[string]EventIngress, len(ns.ingresses))
-	for name, ing := range ns.ingresses {
-		ings[name] = ing
-	}
-	sel := ns.selector
-	ns.ingressMu.RUnlock()
-
-	defaults, fallbacks := partitionIngresses(sel, ings, networkId, subType, params)
+	ings, sel := ns.snapshotIngresses()
+	defaults, fallbacks := partitionIngresses(sel, ings, networkId, f.subType, f.params)
 
 	var errs []error
 	for _, tier := range [][]EventIngress{defaults, fallbacks} {
 		subscribed := false
 		for _, ing := range tier {
-			if err := ing.EnsureFilter(ctx, subType, paramsHash, params); err != nil {
+			f.on[ing.Name()] = struct{}{}
+			if err := ing.EnsureFilter(ctx, f.subType, paramsHash, f.params); err != nil {
 				errs = append(errs, err)
 				i.logger.Warn().Err(err).Str("ingress", ing.Name()).Str("networkId", networkId).
-					Str("subType", subType).Str("paramsHash", paramsHash).
+					Str("subType", f.subType).Str("paramsHash", paramsHash).
 					Msg("ingress EnsureFilter failed")
 				continue
 			}
@@ -197,11 +202,99 @@ func (i *Indexer) subscribe(ctx context.Context, ns *networkState, subType, para
 		}
 	}
 	if len(errs) == 0 {
-		return fmt.Errorf("indexer: no ingress selected for %s filter on network %q", subType, networkId)
+		return fmt.Errorf("indexer: no ingress selected for %s filter on network %q", f.subType, networkId)
 	}
 	return errors.Join(errs...)
 }
 
+// snapshotIngresses copies the network's ingresses and selector.
+func (ns *networkState) snapshotIngresses() (map[string]EventIngress, IngressSelector) {
+	ns.ingressMu.RLock()
+	defer ns.ingressMu.RUnlock()
+	ings := make(map[string]EventIngress, len(ns.ingresses))
+	for name, ing := range ns.ingresses {
+		ings[name] = ing
+	}
+	return ings, ns.selector
+}
+
+// triggerReconcile starts a reconcile of ns's filters unless one is running
+// or there are none. Triggers it skips are covered by the next one.
+func (i *Indexer) triggerReconcile(ns *networkState) {
+	ns.filterMu.RLock()
+	none := len(ns.filters) == 0
+	ns.filterMu.RUnlock()
+	if none || !ns.reconciling.CompareAndSwap(false, true) {
+		return
+	}
+	go func() {
+		defer ns.reconciling.Store(false)
+		i.reconcile(context.Background(), ns)
+	}()
+}
+
+// reconcile moves every subscribed filter of ns onto the ingresses the
+// selector picks now: all its defaults, plus its fallbacks while no default
+// has the filter live; it is removed from every other ingress it is on.
+func (i *Indexer) reconcile(ctx context.Context, ns *networkState) {
+	ns.filterMu.RLock()
+	filters := make(map[string]*filterState, len(ns.filters))
+	for hash, f := range ns.filters {
+		filters[hash] = f
+	}
+	ns.filterMu.RUnlock()
+	ings, sel := ns.snapshotIngresses()
+
+	for hash, f := range filters {
+		f.mu.Lock()
+		if f.subscribed {
+			i.reconcileFilter(ctx, ns, ings, sel, hash, f)
+		}
+		f.mu.Unlock()
+	}
+}
+
+// reconcileFilter is reconcile for one filter; the caller holds f.mu.
+func (i *Indexer) reconcileFilter(ctx context.Context, ns *networkState, ings map[string]EventIngress, sel IngressSelector, hash string, f *filterState) {
+	defaults, fallbacks := partitionIngresses(sel, ings, ns.handle.Id(), f.subType, f.params)
+	want := make(map[string]EventIngress, len(defaults)+len(fallbacks))
+	covered := false
+	for _, ing := range defaults {
+		want[ing.Name()] = ing
+		covered = covered || ing.FilterLive(f.subType, hash)
+	}
+	if !covered {
+		for _, ing := range fallbacks {
+			want[ing.Name()] = ing
+		}
+	}
+
+	for name, ing := range want {
+		if _, ok := f.on[name]; ok {
+			continue
+		}
+		f.on[name] = struct{}{}
+		if err := ing.EnsureFilter(ctx, f.subType, hash, f.params); err != nil {
+			i.logger.Warn().Err(err).Str("ingress", name).Str("networkId", ns.handle.Id()).
+				Str("subType", f.subType).Str("paramsHash", hash).
+				Msg("ingress EnsureFilter failed, it keeps retrying")
+		}
+	}
+	for name := range f.on {
+		if _, ok := want[name]; ok {
+			continue
+		}
+		delete(f.on, name)
+		if ing := ings[name]; ing != nil {
+			if err := ing.RemoveFilter(ctx, f.subType, hash); err != nil {
+				i.logger.Warn().Err(err).Str("ingress", name).Str("networkId", ns.handle.Id()).
+					Str("subType", f.subType).Str("paramsHash", hash).
+					Msg("ingress RemoveFilter failed")
+			}
+		}
+	}
+}
+
 // partitionIngresses resolves the selector's tiers to registered ingresses,
 // dropping unknown and repeated names. Without a selector every ingress is
 // a default.
@@ -264,7 +357,7 @@ func (i *Indexer) ReleaseFilter(ctx context.Context, networkId, subType, paramsH
 		return
 	}
 	i.removeFromIngresses(ctx, ns, subType, paramsHash)
-	f.subscribed = false
+	f.subscribed, f.on = false, nil
 	ns.filterMu.Lock()
 	if f.refs == 0 {
 		delete(ns.filters, paramsHash)
@@ -300,6 +393,11 @@ func (i *Indexer) Ingest(ev StreamEvent) {
 	}
 	ns := nsRaw.(*networkState)
 
+	// Each head, delivered or not, rechecks where the filters belong.
+	if ev.Kind == KindNewHead {
+		i.triggerReconcile(ns)
+	}
+
 	// Before dedup, so every source's head counts even if another source
 	// already delivered it.
 	if ev.Kind == KindNewHead && !ev.Block.Zero() && ev.SourceId != "" {
diff --git a/indexer/indexer_test.go b/indexer/indexer_test.go
index 3d0866c88..08c98851c 100644
--- a/indexer/indexer_test.go
+++ b/indexer/indexer_test.go
@@ -114,6 +114,9 @@ func (i *fakeIngress) EnsureFilter(_ context.Context, _ string, paramsHash strin
 	i.active.Store(true)
 	return i.getErr()
 }
+func (i *fakeIngress) FilterLive(_, _ string) bool {
+	return i.active.Load() && i.getErr() == nil
+}
 func (i *fakeIngress) RemoveFilter(_ context.Context, _, _ string) error {
 	if i.hook != nil {
 		i.hook("remove")
@@ -682,6 +685,124 @@ func TestIndexer_Selector_UnnamedIngressIsExcluded(t *testing.T) {
 	}
 }
 
+// reconcileNow runs a reconcile of evm:1 synchronously.
+func reconcileNow(t *testing.T, idx *Indexer) {
+	t.Helper()
+	nsRaw, ok := idx.networks.Load("evm:1")
+	if !ok {
+		t.Fatal("evm:1 not registered")
+	}
+	idx.reconcile(context.Background(), nsRaw.(*networkState))
+}
+
+// The fallbacks carry a filter only while no default has it live: they take
+// over when the defaults lose it, and hand it back once one has it again.
+func TestIndexer_Reconcile_FallbackTakesOverAndHandsBack(t *testing.T) {
+	idx := newIndexer(t)
+	a, b, c := registerThreeIngresses(t, idx)
+	idx.RegisterNetworkSelector("evm:1", &fakeSelector{defaults: []string{"a", "b"}, fallbacks: []string{"c"}})
+	if _, err := idx.EnsureFilter(context.Background(), "evm:1", "logs", []interface{}{"logs"}); err != nil {
+		t.Fatal(err)
+	}
+
+	reconcileNow(t, idx)
+	if c.ensureCalls.Load() != 0 {
+		t.Fatalf("fallback must stay idle while a default is live, got %d calls", c.ensureCalls.Load())
+	}
+
+	a.setErr(errTest("a lost it"))
+	b.setErr(errTest("b lost it"))
+	reconcileNow(t, idx)
+	if c.ensureCalls.Load() != 1 || !c.FilterLive("logs", "") {
+		t.Fatalf("fallback must take over once no default is live, got %d calls", c.ensureCalls.Load())
+	}
+	if a.removeCalls.Load() != 0 || b.removeCalls.Load() != 0 {
+		t.Fatal("defaults keep the filter, to retry it")
+	}
+
+	reconcileNow(t, idx)
+	if c.ensureCalls.Load() != 1 {
+		t.Fatalf("an ingress already carrying the filter must not be asked again, got %d", c.ensureCalls.Load())
+	}
+
+	b.setErr(nil)
+	reconcileNow(t, idx)
+	if c.removeCalls.Load() != 1 || c.active.Load() {
+		t.Fatalf("fallback must hand the filter back once a default is live, removes=%d", c.removeCalls.Load())
+	}
+}
+
+// A default the selector no longer names (e.g. the policy excluded it) gives
+// the filter up; if that leaves no default, the fallbacks take it.
+func TestIndexer_Reconcile_FollowsSelector(t *testing.T) {
+	idx := newIndexer(t)
+	a, b, c := registerThreeIngresses(t, idx)
+	sel := &fakeSelector{defaults: []string{"a", "b"}, fallbacks: []string{"c"}}
+	idx.RegisterNetworkSelector("evm:1", sel)
+	if _, err := idx.EnsureFilter(context.Background(), "evm:1", "logs", []interface{}{"logs"}); err != nil {
+		t.Fatal(err)
+	}
+
+	sel.defaults = []string{"b"}
+	reconcileNow(t, idx)
+	if a.removeCalls.Load() != 1 || c.ensureCalls.Load() != 0 {
+		t.Fatalf("a must give the filter up while b still carries it: a.removes=%d c.ensures=%d",
+			a.removeCalls.Load(), c.ensureCalls.Load())
+	}
+
+	sel.defaults = nil
+	reconcileNow(t, idx)
+	if b.removeCalls.Load() != 1 || c.ensureCalls.Load() != 1 {
+		t.Fatalf("with no default left the fallback must carry it: b.removes=%d c.ensures=%d",
+			b.removeCalls.Load(), c.ensureCalls.Load())
+	}
+
+	// Returning defaults take it back; the fallback keeps it until one of
+	// them has it live, so there is no gap.
+	sel.defaults = []string{"a", "b"}
+	reconcileNow(t, idx)
+	if a.ensureCalls.Load() != 2 || b.ensureCalls.Load() != 2 || c.removeCalls.Load() != 0 {
+		t.Fatalf("returning defaults must take it back alongside the fallback: a=%d b=%d c.removes=%d",
+			a.ensureCalls.Load(), b.ensureCalls.Load(), c.removeCalls.Load())
+	}
+	reconcileNow(t, idx)
+	if c.removeCalls.Load() != 1 {
+		t.Fatalf("the fallback must hand back once a default is live, removes=%d", c.removeCalls.Load())
+	}
+}
+
+// Every head triggers the recheck, and a released filter is left alone.
+func TestIndexer_Reconcile_RunsOnHeadsAndSkipsReleased(t *testing.T) {
+	idx := newIndexer(t)
+	a, _, c := registerThreeIngresses(t, idx)
+	idx.RegisterNetworkSelector("evm:1", &fakeSelector{defaults: []string{"a"}, fallbacks: []string{"c"}})
+	h, err := idx.EnsureFilter(context.Background(), "evm:1", "logs", []interface{}{"logs"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	a.setErr(errTest("a lost it"))
+
+	idx.Ingest(StreamEvent{Kind: KindNewHead, NetworkId: "evm:1", SourceId: "ws:c", Block: BlockRef{Number: 1, Hash: "0x1"}})
+	deadline := time.Now().Add(2 * time.Second)
+	for c.ensureCalls.Load() != 1 {
+		if time.Now().After(deadline) {
+			t.Fatal("a head must trigger the recheck")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	nsRaw, _ := idx.networks.Load("evm:1")
+	ns := nsRaw.(*networkState)
+	for ns.reconciling.Load() {
+		time.Sleep(time.Millisecond)
+	}
+	idx.ReleaseFilter(context.Background(), "evm:1", "logs", h)
+	reconcileNow(t, idx)
+	if a.ensureCalls.Load() != 1 || c.ensureCalls.Load() != 1 {
+		t.Fatalf("a released filter must not be resubscribed: a=%d c=%d", a.ensureCalls.Load(), c.ensureCalls.Load())
+	}
+}
+
 func TestIndexer_Selector_NothingSelectedFails(t *testing.T) {
 	idx := newIndexer(t)
 	registerThreeIngresses(t, idx)
diff --git a/indexer/ingress.go b/indexer/ingress.go
index 26338e672..ec72629ff 100644
--- a/indexer/ingress.go
+++ b/indexer/ingress.go
@@ -32,12 +32,15 @@ type EventIngress interface {
 	// RemoveFilter unsubscribes the filter; a no-op if it was never
 	// subscribed. Called when its last client leaves or its subscribe failed.
 	RemoveFilter(ctx context.Context, subType string, paramsHash string) error
+	// FilterLive reports whether the filter is subscribed and live now.
+	FilterLive(subType string, paramsHash string) bool
 }
 
 // IngressSelector chooses which ingresses carry a filter subscription, by
-// Name(). Defaults are tried first and at least one must succeed; fallbacks
-// are tried only if every default failed. Ingresses named in neither are
-// excluded. Without a selector every ingress is a default.
+// Name(). A filter is kept on every default, and on the fallbacks while no
+// default has it live (or, when it is first subscribed, if every default
+// failed). Ingresses named in neither do not carry it. The choice is
+// rechecked on every head. Without a selector every ingress is a default.
 type IngressSelector interface {
 	Select(networkId, subType string, params []interface{}) (defaults, fallbacks []string)
 }