From 822d183b0cfd46b7d1235d8dd681a7db0022a5e6 Mon Sep 17 00:00:00 2001 From: snowkide Date: Wed, 22 Jul 2026 12:25:10 +0200 Subject: [PATCH] fix(websocket): refuse newHeads when no live ingress Restore only the fail-closed gate dropped in the post-merge slim: ErrNoLiveSubscriptionSource + IngressHealth/HealthReporter + a short bootstrap wait. Head-less pods return retryable 503 instead of zombie subscription IDs. Align ClientStaysConnectedOnUpstreamDrop with self-heal (1001 GoingAway remains shutdown-only). Skip LastHead metric and healthcheck subscriptions surface. Co-authored-by: Cursor --- common/errors.go | 26 ++++++ erpc/subscription_manager.go | 48 +++++++++++ erpc/subscription_manager_health_test.go | 100 +++++++++++++++++++++++ erpc/ws_server_test.go | 39 +++++---- indexer/health_test.go | 59 +++++++++++++ indexer/indexer.go | 34 ++++++++ 6 files changed, 292 insertions(+), 14 deletions(-) create mode 100644 erpc/subscription_manager_health_test.go create mode 100644 indexer/health_test.go diff --git a/common/errors.go b/common/errors.go index 3073feb31..a8c4bcf34 100644 --- a/common/errors.go +++ b/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" diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go index 54b78f152..baf5c5b40 100644 --- a/erpc/subscription_manager.go +++ b/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 diff --git a/erpc/subscription_manager_health_test.go b/erpc/subscription_manager_health_test.go new file mode 100644 index 000000000..42d3be6cb --- /dev/null +++ b/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)) + }) +} diff --git a/erpc/ws_server_test.go b/erpc/ws_server_test.go index 94e4d569a..489fc4842 100644 --- a/erpc/ws_server_test.go +++ b/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) + } }) } diff --git a/indexer/health_test.go b/indexer/health_test.go new file mode 100644 index 000000000..2d2d26606 --- /dev/null +++ b/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) + }) +} diff --git a/indexer/indexer.go b/indexer/indexer.go index a686fecad..07c6a519d 100644 --- a/indexer/indexer.go +++ b/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 {