fix(websocket): refuse newHeads when no live ingress

common/errors.go

@@ -2824,6 +2824,32 @@ func (e *ErrNoWsUpstreamAvailable) ErrorStatusCode() int {
 	return http.StatusBadRequest
 }
 
+type ErrNoLiveSubscriptionSource struct{ BaseError }
+
+const ErrCodeNoLiveSubscriptionSource ErrorCode = "ErrNoLiveSubscriptionSource"
+
+// NewErrNoLiveSubscriptionSource is returned when WS upstreams are
+// configured for the network but none currently has a live connection with
+// an active newHeads subscription. Refusing the subscription (HTTP 503 /
+// retryable) lets clients fail over to another node instead of holding a
+// subscription ID that will never deliver.
+var NewErrNoLiveSubscriptionSource = func(networkId string, totalIngresses int) error {
+	return &ErrNoLiveSubscriptionSource{
+		BaseError{
+			Code:    ErrCodeNoLiveSubscriptionSource,
+			Message: fmt.Sprintf("no upstream is currently able to deliver subscription events for network %s; refusing subscription so the client can fail over", networkId),
+			Details: map[string]interface{}{
+				"networkId":      networkId,
+				"totalIngresses": totalIngresses,
+			},
+		},
+	}
+}
+
+func (e *ErrNoLiveSubscriptionSource) ErrorStatusCode() int {
+	return http.StatusServiceUnavailable
+}
+
 type ErrSubscriptionLimitExceeded struct{ BaseError }
 
 const ErrCodeSubscriptionLimitExceeded ErrorCode = "ErrSubscriptionLimitExceeded"

erpc/subscription_manager.go

@@ -38,8 +38,19 @@ const (
 	// unsubscribeTimeout is the deadline for best-effort upstream
 	// unsubscribe calls during connection cleanup.
 	unsubscribeTimeout = 5 * time.Second
+
+	// liveHeadSourcePollEvery is how often waitForLiveHeadSource re-checks
+	// ingress health while waiting out the bootstrap race.
+	liveHeadSourcePollEvery = 100 * time.Millisecond
 )
 
+// liveHeadSourceWaitMax bounds how long a newHeads subscribe waits for at
+// least one ingress to come alive before refusing the subscription. Long
+// enough to cover the initial bootstrap (adapter connect + eth_subscribe
+// round-trip), short enough that a client talking to a head-less pod fails
+// over quickly. Var so tests can compress time.
+var liveHeadSourceWaitMax = 3 * time.Second
+
 // SubscriptionManager is the client-facing egress layer. It owns
 // per-connection *wsclient.Adapter instances, lazily registers networks +
 // ingresses with the indexer the first time a client subscribes on a
@@ -167,6 +178,19 @@ func (sm *SubscriptionManager) Subscribe(
 		return nil, fmt.Errorf("failed to generate subscription ID: %w", err)
 	}
 
+	// newHeads is fan-out only — no per-filter EnsureFilter ever touches an
+	// upstream for it, so without this check a pod whose WS upstreams are
+	// all down (or resubscribing) would happily return a subscription ID
+	// that never delivers a single head. Refuse instead so the client can
+	// retry/fail over. Filter subs get equivalent protection from
+	// EnsureFilter, which errors when every ingress fails.
+	if subType == SubTypeNewHeads {
+		if err := sm.waitForLiveHeadSource(ctx, networkId); err != nil {
+			sm.recordFailureMetrics(project, nw, method, reqFinality, start, nq, err)
+			return nil, err
+		}
+	}
+
 	kind, filterHash, err := sm.resolveSubscription(ctx, networkId, subType, jrReq.Params)
 	if err != nil {
 		sm.recordFailureMetrics(project, nw, method, reqFinality, start, nq, err)
@@ -300,6 +324,30 @@ func (sm *SubscriptionManager) CleanupConnection(wsc *WsConnection, _ *PreparedP
 	lg.Debug().Msg("cleaned up all subscriptions for connection")
 }
 
+// waitForLiveHeadSource returns nil as soon as at least one of the
+// network's ingresses reports it can deliver heads. The bounded wait
+// covers the bootstrap race where adapters' initial eth_subscribe calls
+// are still in flight; after that it refuses with a retryable error.
+func (sm *SubscriptionManager) waitForLiveHeadSource(ctx context.Context, networkId string) error {
+	deadline := time.Now().Add(liveHeadSourceWaitMax)
+	for {
+		live, total := sm.idx.IngressHealth(networkId)
+		if live > 0 {
+			return nil
+		}
+		if ctx.Err() != nil || time.Now().After(deadline) {
+			sm.logger.Warn().Str("networkId", networkId).Int("totalIngresses", total).
+				Msg("refusing newHeads subscription: no live head source on this instance")
+			return common.NewErrNoLiveSubscriptionSource(networkId, total)
+		}
+		select {
+		case <-ctx.Done():
+			return common.NewErrNoLiveSubscriptionSource(networkId, total)
+		case <-time.After(liveHeadSourcePollEvery):
+		}
+	}
+}
+
 // --- internals --------------------------------------------------------
 
 // buildWsAdapterOptions resolves network-level toggles that the wsupstream

erpc/subscription_manager_health_test.go

@@ -0,0 +1,100 @@
+package erpc
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/erpc/erpc/common"
+	"github.com/erpc/erpc/indexer"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type stubNetworkHandle struct{ id string }
+
+func (h stubNetworkHandle) Id() string                       { return h.id }
+func (h stubNetworkHandle) FinalityDepth() int64             { return 0 }
+func (h stubNetworkHandle) SuggestLatestBlock(string, int64) {}
+
+// stubIngress implements indexer.EventIngress plus indexer.HealthReporter
+// with a flippable health flag.
+type stubIngress struct {
+	name    string
+	healthy bool
+}
+
+func (i *stubIngress) Name() string { return i.name }
+func (i *stubIngress) Start(context.Context, indexer.NetworkHandle, indexer.Sink) error {
+	return nil
+}
+func (i *stubIngress) EnsureFilter(context.Context, string, string, []interface{}) error { return nil }
+func (i *stubIngress) RemoveFilter(context.Context, string, string) error                { return nil }
+func (i *stubIngress) Stop(context.Context) error                                        { return nil }
+func (i *stubIngress) Healthy() bool                                                     { return i.healthy }
+
+func newTestSubscriptionManager(t *testing.T) (*SubscriptionManager, *indexer.Indexer) {
+	t.Helper()
+	logger := zerolog.New(zerolog.NewTestWriter(t)).Level(zerolog.ErrorLevel)
+	idx := indexer.New(&logger, indexer.Options{})
+	return NewSubscriptionManager(&logger, idx), idx
+}
+
+// TestWaitForLiveHeadSource pins the contract: a pod with zero live head
+// sources must refuse newHeads (retryable ErrNoLiveSubscriptionSource)
+// instead of handing out a subscription ID that never delivers.
+func TestWaitForLiveHeadSource(t *testing.T) {
+	origWait := liveHeadSourceWaitMax
+	liveHeadSourceWaitMax = 300 * time.Millisecond
+	t.Cleanup(func() { liveHeadSourceWaitMax = origWait })
+
+	const networkID = "evm:324"
+
+	t.Run("refuses when no ingress is live", func(t *testing.T) {
+		sm, idx := newTestSubscriptionManager(t)
+		idx.RegisterNetwork(stubNetworkHandle{id: networkID})
+		require.NoError(t, idx.AddIngress(context.Background(), networkID, &stubIngress{name: "ws:a", healthy: false}))
+
+		err := sm.waitForLiveHeadSource(context.Background(), networkID)
+		require.Error(t, err)
+		assert.True(t, common.HasErrorCode(err, common.ErrCodeNoLiveSubscriptionSource),
+			"expected ErrNoLiveSubscriptionSource, got: %v", err)
+	})
+
+	t.Run("passes immediately when an ingress is live", func(t *testing.T) {
+		sm, idx := newTestSubscriptionManager(t)
+		idx.RegisterNetwork(stubNetworkHandle{id: networkID})
+		require.NoError(t, idx.AddIngress(context.Background(), networkID, &stubIngress{name: "ws:a", healthy: true}))
+
+		start := time.Now()
+		require.NoError(t, sm.waitForLiveHeadSource(context.Background(), networkID))
+		assert.Less(t, time.Since(start), liveHeadSourceWaitMax/2,
+			"a live source must not incur the bootstrap grace wait")
+	})
+
+	t.Run("passes when an ingress becomes live during the grace wait", func(t *testing.T) {
+		sm, idx := newTestSubscriptionManager(t)
+		idx.RegisterNetwork(stubNetworkHandle{id: networkID})
+		ing := &stubIngress{name: "ws:a", healthy: false}
+		require.NoError(t, idx.AddIngress(context.Background(), networkID, ing))
+
+		go func() {
+			time.Sleep(120 * time.Millisecond)
+			ing.healthy = true
+		}()
+		require.NoError(t, sm.waitForLiveHeadSource(context.Background(), networkID))
+	})
+
+	t.Run("honours caller context cancellation", func(t *testing.T) {
+		sm, idx := newTestSubscriptionManager(t)
+		idx.RegisterNetwork(stubNetworkHandle{id: networkID})
+		require.NoError(t, idx.AddIngress(context.Background(), networkID, &stubIngress{name: "ws:a", healthy: false}))
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+		err := sm.waitForLiveHeadSource(ctx, networkID)
+		require.Error(t, err)
+		assert.True(t, common.HasErrorCode(err, common.ErrCodeNoLiveSubscriptionSource))
+	})
+}

erpc/ws_server_test.go

@@ -902,10 +902,10 @@ func TestWebSocket_UpstreamClient(t *testing.T) {
 //
 
 func TestWebSocket_SubscriptionRecovery(t *testing.T) {
-	// Verifies that when the upstream WS connection drops, eRPC closes the
-	// client connection with CloseGoingAway (1001) so the client can reconnect
-	// and re-subscribe cleanly instead of holding a zombie subscription.
-	t.Run("ClientDisconnectedOnUpstreamDrop", func(t *testing.T) {
+	// Self-heal keeps the client WS up when the upstream drops; 1001
+	// GoingAway is reserved for process shutdown. A GoingAway close here
+	// would force MultiNode to mark the RPC unreachable.
+	t.Run("ClientStaysConnectedOnUpstreamDrop", func(t *testing.T) {
 		closeUpstream := make(chan struct{})
 
 		mockUpstream := mockWsUpstream(t, func(conn *websocket.Conn) {
@@ -959,16 +959,17 @@ func TestWebSocket_SubscriptionRecovery(t *testing.T) {
 		// Kill the upstream WS connection
 		close(closeUpstream)
 
-		// Client should receive a close frame with GoingAway (1001)
-		conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+		// Client must stay connected — expect a read timeout (no frames)
+		// or a non-GoingAway error, never 1001.
+		conn.SetReadDeadline(time.Now().Add(3 * time.Second))
 		_, _, err := conn.ReadMessage()
-		require.Error(t, err, "client should be disconnected")
-		closeErr, ok := err.(*websocket.CloseError)
-		if ok {
-			assert.Equal(t, websocket.CloseGoingAway, closeErr.Code, "close code should be 1001 GoingAway")
-			t.Logf("client received close frame: code=%d reason=%q", closeErr.Code, closeErr.Text)
+		require.Error(t, err, "expected no spontaneous client close frame")
+		if closeErr, ok := err.(*websocket.CloseError); ok {
+			assert.NotEqual(t, websocket.CloseGoingAway, closeErr.Code,
+				"upstream drop must not close the client with GoingAway (1001); got code=%d reason=%q",
+				closeErr.Code, closeErr.Text)
 		} else {
-			t.Logf("client disconnected with error: %v", err)
+			t.Logf("client stayed connected (read ended with: %v)", err)
 		}
 	})
 
@@ -995,8 +996,18 @@ func TestWebSocket_SubscriptionRecovery(t *testing.T) {
 		defer conn.Close()
 
 		resp := sendAndReceive(t, conn, `{"jsonrpc":"2.0","id":1,"method":"eth_subscribe","params":["newHeads"]}`)
-		assert.NotNil(t, resp["error"], "should return error when upstream WS is not connected")
-		t.Logf("got expected error: %v", resp["error"])
+		require.NotNil(t, resp["error"], "should return error when upstream WS is not connected")
+		errObj, _ := resp["error"].(map[string]interface{})
+		t.Logf("got expected error: %v", errObj)
+		// Prefer the loud refusal (ErrNoLiveSubscriptionSource). ErrNoWsUpstreamAvailable
+		// is also acceptable if bootstrap never registered a WS ingress.
+		if data, ok := errObj["data"].(map[string]interface{}); ok {
+			code, _ := data["code"].(string)
+			assert.Contains(t, []string{
+				"ErrNoLiveSubscriptionSource",
+				"ErrNoWsUpstreamAvailable",
+			}, code, "unexpected error code: %v", errObj)
+		}
 	})
 }
 

indexer/health_test.go

@@ -0,0 +1,59 @@
+package indexer
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// healthyIngress wraps fakeIngress with a controllable HealthReporter
+// implementation.
+type healthyIngress struct {
+	fakeIngress
+	healthy bool
+}
+
+func (i *healthyIngress) Healthy() bool { return i.healthy }
+
+func TestIndexer_IngressHealth(t *testing.T) {
+	idx := newIndexer(t)
+	nw := newFakeNetwork("evm:324", 0)
+	idx.RegisterNetwork(nw)
+
+	t.Run("unknown network", func(t *testing.T) {
+		live, total := idx.IngressHealth("evm:999")
+		assert.Equal(t, 0, live)
+		assert.Equal(t, 0, total)
+	})
+
+	t.Run("no ingresses yet", func(t *testing.T) {
+		live, total := idx.IngressHealth("evm:324")
+		assert.Equal(t, 0, live)
+		assert.Equal(t, 0, total)
+	})
+
+	up := &healthyIngress{fakeIngress: fakeIngress{name: "ws:up"}, healthy: true}
+	down := &healthyIngress{fakeIngress: fakeIngress{name: "ws:down"}, healthy: false}
+	// An ingress that doesn't implement HealthReporter counts as live —
+	// the indexer can't assess transports it doesn't understand.
+	opaque := &fakeIngress{name: "kafka:topic"}
+
+	require.NoError(t, idx.AddIngress(context.Background(), "evm:324", up))
+	require.NoError(t, idx.AddIngress(context.Background(), "evm:324", down))
+	require.NoError(t, idx.AddIngress(context.Background(), "evm:324", opaque))
+
+	t.Run("mixed health", func(t *testing.T) {
+		live, total := idx.IngressHealth("evm:324")
+		assert.Equal(t, 2, live, "healthy reporter + opaque ingress")
+		assert.Equal(t, 3, total)
+	})
+
+	t.Run("all reporters down", func(t *testing.T) {
+		up.healthy = false
+		live, total := idx.IngressHealth("evm:324")
+		assert.Equal(t, 1, live, "only the opaque ingress remains assumed-live")
+		assert.Equal(t, 3, total)
+	})
+}

indexer/indexer.go

@@ -528,6 +528,40 @@ func (i *Indexer) classify(ns *networkState, ev StreamEvent) Lifecycle {
 	return LifeSoft
 }
 
+// HealthReporter is an optional interface an EventIngress can implement to
+// report whether it can currently deliver events (e.g. a WS upstream
+// adapter with a live connection and an active newHeads subscription).
+// Ingresses that don't implement it are assumed live — the indexer can't
+// assess transports it doesn't understand.
+type HealthReporter interface {
+	Healthy() bool
+}
+
+// IngressHealth returns how many of the network's registered ingresses
+// currently report themselves able to deliver events, alongside the total
+// registered count. (0, 0) means the network is unknown or has no
+// ingresses yet.
+func (i *Indexer) IngressHealth(networkId string) (live, total int) {
+	nsRaw, ok := i.networks.Load(networkId)
+	if !ok {
+		return 0, 0
+	}
+	ns := nsRaw.(*networkState)
+	ns.ingressMu.RLock()
+	defer ns.ingressMu.RUnlock()
+	for _, ing := range ns.ingresses {
+		total++
+		if hr, ok := ing.(HealthReporter); ok {
+			if hr.Healthy() {
+				live++
+			}
+		} else {
+			live++
+		}
+	}
+	return live, total
+}
+
 // fanOut dispatches to every registered egress whose InterestedIn matches.
 func (i *Indexer) fanOut(ev IndexedEvent) {
 	i.egresses.Range(func(_, v any) bool {