From 98fbe986cfcc03e1484e6b8d5dcfe6238749b60d Mon Sep 17 00:00:00 2001 From: snowkide Date: Wed, 22 Jul 2026 12:19:16 +0200 Subject: [PATCH] fix(websocket): restore newHeads refusal gate and align drop test with self-heal The post-PR1 slim dropped waitForLiveHeadSource / ErrNoLiveSubscriptionSource, so head-less pods again hand out zombie newHeads IDs and MultiNode cannot fail over cleanly. Restore IngressHealth + last-head observability and replace the false-positive ClientDisconnectedOnUpstreamDrop expectation (1001 on upstream drop) with ClientStaysConnectedOnUpstreamDrop. Co-authored-by: Cursor --- common/errors.go | 26 +++++ erpc/healthcheck.go | 12 +++ erpc/subscription_manager.go | 75 +++++++++++++ erpc/subscription_manager_health_test.go | 131 +++++++++++++++++++++++ erpc/ws_server_test.go | 43 +++++--- indexer/health_test.go | 109 +++++++++++++++++++ indexer/indexer.go | 71 ++++++++++++ telemetry/metrics.go | 6 ++ 8 files changed, 459 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/healthcheck.go b/erpc/healthcheck.go index 85e4953b5..bbf03f4bb 100644 --- a/erpc/healthcheck.go +++ b/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 } diff --git a/erpc/subscription_manager.go b/erpc/subscription_manager.go index 54b78f152..d5a5c0e1b 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,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 diff --git a/erpc/subscription_manager_health_test.go b/erpc/subscription_manager_health_test.go new file mode 100644 index 000000000..2a0580591 --- /dev/null +++ b/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) +} diff --git a/erpc/ws_server_test.go b/erpc/ws_server_test.go index 94e4d569a..e27ac3bba 100644 --- a/erpc/ws_server_test.go +++ b/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) + } }) } diff --git a/indexer/health_test.go b/indexer/health_test.go new file mode 100644 index 000000000..ac72fb934 --- /dev/null +++ b/indexer/health_test.go @@ -0,0 +1,109 @@ +package indexer + +import ( + "context" + "testing" + "time" + + "github.com/rs/zerolog" + "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) + }) +} + +func TestIndexer_LastHead(t *testing.T) { + now := time.Date(2026, 6, 12, 9, 30, 0, 0, time.UTC) + logger := zerolog.New(zerolog.NewTestWriter(t)) + idx := New(&logger, Options{Now: func() time.Time { return now }}) + nw := newFakeNetwork("evm:324", 0) + idx.RegisterNetwork(nw) + + _, _, ok := idx.LastHead("evm:324") + assert.False(t, ok, "no head delivered yet") + + _, _, ok = idx.LastHead("evm:999") + assert.False(t, ok, "unknown network") + + idx.Ingest(StreamEvent{ + Kind: KindNewHead, + NetworkId: "evm:324", + SourceId: "ws:up", + Block: BlockRef{Number: 42, Hash: "0xaa", ParentHash: "0x99"}, + }) + + block, at, ok := idx.LastHead("evm:324") + require.True(t, ok) + assert.Equal(t, int64(42), block.Number) + assert.True(t, now.Equal(at), "expected %s got %s", now, at) + + // A duplicate head must not move the liveness timestamp (it was + // deduped, not delivered) — but a NEW head must. + now = now.Add(10 * time.Second) + idx.Ingest(StreamEvent{ + Kind: KindNewHead, + NetworkId: "evm:324", + SourceId: "ws:up", + Block: BlockRef{Number: 42, Hash: "0xaa", ParentHash: "0x99"}, + }) + _, at, _ = idx.LastHead("evm:324") + assert.True(t, now.Add(-10*time.Second).Equal(at), "deduped head must not refresh liveness") + + idx.Ingest(StreamEvent{ + Kind: KindNewHead, + NetworkId: "evm:324", + SourceId: "ws:up", + Block: BlockRef{Number: 43, Hash: "0xbb", ParentHash: "0xaa"}, + }) + block, at, _ = idx.LastHead("evm:324") + assert.Equal(t, int64(43), block.Number) + assert.True(t, now.Equal(at), "expected %s got %s", now, at) +} diff --git a/indexer/indexer.go b/indexer/indexer.go index a686fecad..a15da3bd0 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -7,6 +7,7 @@ import ( "sync/atomic" "time" + "github.com/erpc/erpc/telemetry" "github.com/rs/zerolog" ) @@ -76,6 +77,13 @@ type networkState struct { lastHead atomic.Pointer[headMarker] headFallback *DedupWindow + // lastHeadAt is the UnixNano timestamp of the most recent delivered + // (post-dedup) newHeads event. Zero until the first head arrives. + // Drives the per-network head-liveness metric and health endpoint so + // a silent head stall is observable instead of only visible to + // subscribed clients. + lastHeadAt atomic.Int64 + // Per-filter dedup windows: filterHash -> *DedupWindow. filterMu sync.RWMutex filterDedup map[string]*DedupWindow @@ -369,6 +377,16 @@ func (i *Indexer) Ingest(ev StreamEvent) { return } + // Head-liveness bookkeeping: record when this network last delivered a + // head so operators can alert on "no heads for network X in Y seconds" + // (time() - gauge) instead of relying on clients to notice silence. + if ev.Kind == KindNewHead && !ev.Block.Zero() { + now := i.opts.Now() + ns.lastHeadAt.Store(now.UnixNano()) + telemetry.GaugeHandle(telemetry.MetricNetworkSubscriptionLastHeadTimestamp, ev.NetworkId). + Set(float64(now.Unix())) + } + // Detect and emit reorg invalidations BEFORE delivering the new // head. Consumers see: (removed logs) → reorg summary → new head. if ev.Kind == KindNewHead && !ev.Block.Zero() { @@ -528,6 +546,59 @@ 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 +} + +// LastHead returns the most recent delivered head for the network and when +// it was delivered. ok is false when the network is unknown or no head has +// been delivered yet. +func (i *Indexer) LastHead(networkId string) (block BlockRef, at time.Time, ok bool) { + nsRaw, found := i.networks.Load(networkId) + if !found { + return BlockRef{}, time.Time{}, false + } + ns := nsRaw.(*networkState) + nanos := ns.lastHeadAt.Load() + if nanos == 0 { + return BlockRef{}, time.Time{}, false + } + if head := ns.lastHead.Load(); head != nil { + block = BlockRef{Number: head.num, Hash: head.hash} + } + return block, time.Unix(0, nanos), true +} + // fanOut dispatches to every registered egress whose InterestedIn matches. func (i *Indexer) fanOut(ev IndexedEvent) { i.egresses.Range(func(_, v any) bool { diff --git a/telemetry/metrics.go b/telemetry/metrics.go index e7e05f56e..7fb347190 100644 --- a/telemetry/metrics.go +++ b/telemetry/metrics.go @@ -88,6 +88,12 @@ var ( Help: "Whether the upstream WebSocket connection is currently established (1) or down/wedged (0).", }, []string{"project", "vendor", "network", "upstream"}) + MetricNetworkSubscriptionLastHeadTimestamp = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: "erpc", + Name: "network_subscription_last_head_timestamp_seconds", + Help: "Unix timestamp of the last newHeads event delivered by the subscription indexer for a network. Alert on time() - this > N to catch silent head stalls.", + }, []string{"network"}) + MetricUpstreamCordoned = promauto.NewGaugeVec(prometheus.GaugeOpts{ Namespace: "erpc", Name: "upstream_cordoned",