fix(websocket): restore newHeads refusal gate + align upstream-drop test with self-heal

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/healthcheck.go

@@ -48,6 +48,11 @@ type NetworkHealthData struct {
 	Status      string                         `json:"status"`
 	Message     string                         `json:"message,omitempty"`
 	Upstreams   map[string]*UpstreamHealthData `json:"upstreams"`
+
+	// Subscriptions reports WS head-delivery liveness for this network.
+	// Present only when at least one client has subscribed on the network
+	// since the process started.
+	Subscriptions *NetworkSubscriptionHealth `json:"subscriptions,omitempty"`
 }
 
 type UpstreamHealthData struct {
@@ -245,6 +250,13 @@ func (s *HttpServer) handleHealthCheck(
 					ms := float64(bt.Milliseconds())
 					networkHealth.BlockTimeMs = &ms
 				}
+				// Subscription head-liveness: nil unless a client has
+				// subscribed on this network at least once. Lets load
+				// balancers and operators see "this pod delivers no heads
+				// for network X" without an active client subscription.
+				if s.subscriptionManager != nil {
+					networkHealth.Subscriptions = s.subscriptionManager.SubscriptionHealth(networkId)
+				}
 				projectHealth.Networks[networkId] = networkHealth
 			}
 

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,57 @@ func (sm *SubscriptionManager) CleanupConnection(wsc *WsConnection, _ *PreparedP
 	lg.Debug().Msg("cleaned up all subscriptions for connection")
 }
 
+// NetworkSubscriptionHealth summarizes a network's head-delivery liveness
+// for the health endpoint. Nil/absent when the network has never been
+// bootstrapped (no client ever subscribed on it).
+type NetworkSubscriptionHealth struct {
+	LiveIngresses  int    `json:"liveIngresses"`
+	TotalIngresses int    `json:"totalIngresses"`
+	LastHeadNumber int64  `json:"lastHeadNumber,omitempty"`
+	LastHeadAt     string `json:"lastHeadAt,omitempty"`
+	LastHeadAgeSec int64  `json:"lastHeadAgeSeconds,omitempty"`
+}
+
+// SubscriptionHealth reports the network's subscription liveness, or nil
+// when the network was never bootstrapped for subscriptions.
+func (sm *SubscriptionManager) SubscriptionHealth(networkId string) *NetworkSubscriptionHealth {
+	if _, ok := sm.networks.Load(networkId); !ok {
+		return nil
+	}
+	live, total := sm.idx.IngressHealth(networkId)
+	out := &NetworkSubscriptionHealth{LiveIngresses: live, TotalIngresses: total}
+	if block, at, ok := sm.idx.LastHead(networkId); ok {
+		out.LastHeadNumber = block.Number
+		out.LastHeadAt = at.UTC().Format(time.RFC3339)
+		out.LastHeadAgeSec = int64(time.Since(at).Seconds())
+	}
+	return out
+}
+
+// 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,131 @@
+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 incident-driven contract: a pod with
+// zero live head sources must refuse newHeads subscriptions (retryable
+// error) instead of handing out a subscription ID that never delivers —
+// the silent failure mode that hid the 2026-06-12 zkSync outage for hours.
+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))
+	})
+}
+
+func TestSubscriptionHealth(t *testing.T) {
+	const networkID = "evm:324"
+	sm, idx := newTestSubscriptionManager(t)
+
+	assert.Nil(t, sm.SubscriptionHealth(networkID), "nil before the network is bootstrapped")
+
+	idx.RegisterNetwork(stubNetworkHandle{id: networkID})
+	require.NoError(t, idx.AddIngress(context.Background(), networkID, &stubIngress{name: "ws:a", healthy: true}))
+	require.NoError(t, idx.AddIngress(context.Background(), networkID, &stubIngress{name: "ws:b", healthy: false}))
+	sm.networks.Store(networkID, struct{}{})
+
+	h := sm.SubscriptionHealth(networkID)
+	require.NotNil(t, h)
+	assert.Equal(t, 1, h.LiveIngresses)
+	assert.Equal(t, 2, h.TotalIngresses)
+	assert.Empty(t, h.LastHeadAt, "no head delivered yet")
+
+	idx.Ingest(indexer.StreamEvent{
+		Kind:      indexer.KindNewHead,
+		NetworkId: networkID,
+		SourceId:  "ws:a",
+		Block:     indexer.BlockRef{Number: 99, Hash: "0xaa", ParentHash: "0x98"},
+	})
+
+	h = sm.SubscriptionHealth(networkID)
+	require.NotNil(t, h)
+	assert.Equal(t, int64(99), h.LastHeadNumber)
+	assert.NotEmpty(t, h.LastHeadAt)
+}

erpc/ws_server_test.go

@@ -902,10 +902,13 @@ 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) {
+	// Verifies that when the upstream WS connection drops, eRPC does NOT
+	// close the client connection with CloseGoingAway. Self-heal re-dials
+	// the upstream and keeps the client subscription alive; 1001 is
+	// reserved for process shutdown (see TestWebSocket_GracefulShutdown).
+	// The pre-self-heal test expected 1001 here and later became a false
+	// positive (pass-on-read-timeout without asserting close code).
+	t.Run("ClientStaysConnectedOnUpstreamDrop", func(t *testing.T) {
 		closeUpstream := make(chan struct{})
 
 		mockUpstream := mockWsUpstream(t, func(conn *websocket.Conn) {
@@ -959,16 +962,18 @@ 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 — a GoingAway close would force
+		// MultiNode to mark the RPC unreachable. 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 +1000,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)
+		}
 	})
 }