From 44c481de7f5ac8e8c7f876cf1636036aaa346e22 Mon Sep 17 00:00:00 2001
From: Matthieu Vachon <matt@streamingfast.io>
Date: Tue, 1 Sep 2026 14:10:20 -0400
Subject: [PATCH 1/8] Explain why prometheus-exporter marks an endpoint
 unavailable

Classify every failed poll into a reason (connect, connect_timeout,
invalid_request, request_timeout, stream_error, stale_block, no_data),
exposed on a new substreams_healthcheck_failure_count{reason,grpc_code}
counter and carried in the logs.

Give connection establishment its own --connect-timeout budget, separate
from --timeout: gRPC dials lazily, so DNS, TLS and load-balancer
resolution used to be charged to the request timeout and a slow
connection was reported as an endpoint failure. Report the two phases
separately as connect_duration_ms and stream_duration_ms.

Log every failed poll, not only the transition into unavailable, and log
a block age above half of --max-freshness so an alert on block_age_ms is
no longer silent. Add a consecutive_failures gauge and reset block_age_ms
to NaN when a poll returns no block.

Speak sf.substreams.rpc.v4.Stream/Blocks only, dropping the v3-to-v2
fallback that closed the connection and then kept reading from it.
--force-protocol-version now accepts only v4 and is validated at startup.

Fix a panic on inconsistent label cardinality when endpoints carry
different sets of query-parameter labels.

# Conflicts:
#	docs/release-notes/change-log.md
---
 docs/release-notes/change-log.md  |  37 +++
 tools/log_test.go                 |   9 +
 tools/metrics.go                  | 129 +++++++++
 tools/metrics_test.go             | 125 +++++++++
 tools/prometheus-exporter.go      | 443 +++++++++++++++++++-----------
 tools/prometheus-exporter_test.go |   4 +-
 6 files changed, 580 insertions(+), 167 deletions(-)
 create mode 100644 tools/log_test.go
 create mode 100644 tools/metrics.go
 create mode 100644 tools/metrics_test.go

diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md
index cace098f6..4617ff6a5 100644
--- a/docs/release-notes/change-log.md
+++ b/docs/release-notes/change-log.md
@@ -98,6 +98,43 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 - `substreams-tier1` now names the usage marker it writes in every module cache folder after the request's plan tier: `last_used_<plan>` (lowercase, e.g. `last_used_pro`), still plain `last_used` when unauthenticated. `firecore tools substreams purge` reads the plan back from that name to apply a retention per plan.
 
+### Tools
+
+- `substreams tools prometheus-exporter` now says *why* an endpoint is down. Every failure is classified into a
+  `reason` -- `connect`, `connect_timeout`, `invalid_request`, `request_timeout`, `stream_error`, `stale_block` or
+  `no_data` -- exposed on the new `substreams_healthcheck_failure_count{reason,grpc_code}` counter and included in
+  the logs. An alert firing on `substreams_healthcheck_status` no longer requires guessing whether the endpoint was
+  unreachable, unauthenticated, overloaded or merely late.
+
+- Connection establishment gets its own budget, `--connect-timeout` (default 10s), separate from `--timeout`, which
+  now covers the `Blocks` request alone. gRPC dials lazily, so DNS, TLS and load-balancer resolution used to be
+  charged to the request timeout and a slow connection was reported as an endpoint failure -- this is what produced
+  the `received context error while waiting for new LB policy update: context deadline exceeded` errors. The exporter
+  now waits for the channel to be `READY` before issuing the request, and reports the two phases separately as
+  `substreams_healthcheck_connect_duration_ms` and `substreams_healthcheck_stream_duration_ms`.
+  `substreams_healthcheck_duration_ms` keeps its previous meaning of the two combined.
+
+- Every failed poll is logged, not just the transition into `unavailable`. An endpoint that fails repeatedly, or one
+  that flaps between two Prometheus scrapes, previously produced a single line and then nothing. Failure logs carry
+  the reason, the gRPC code, both durations and the consecutive failure count; the recovery log carries how long the
+  endpoint was down and how many polls failed meanwhile. A block age above half of `--max-freshness` is logged too,
+  so an alert on `substreams_healthcheck_block_age_ms` is no longer silent.
+
+- New `substreams_healthcheck_consecutive_failures` gauge, meant to be alerted on instead of
+  `substreams_healthcheck_status` when single-poll hiccups should be ignored.
+
+- `substreams_healthcheck_block_age_ms` is reset to `NaN` when a poll returns no block, instead of keeping the age of
+  the last block ever seen -- which silently under-reported staleness for as long as an endpoint stayed broken.
+
+- Fixed: endpoints configured with different sets of query-parameter labels (e.g. one with `?namespace=x&region=y`
+  and one with only `?namespace=z`) made the exporter panic on inconsistent label cardinality. Missing labels are now
+  filled with an empty value.
+
+- The exporter now speaks `sf.substreams.rpc.v4.Stream/Blocks` only. The v3-to-v2 fallback is gone -- it closed the
+  connection and then kept reading from it, double-counting the failure -- and `--force-protocol-version` accepts
+  only `4` (or `0`), the flag being kept for the protocol versions to come. An invalid value used to be parsed and
+  then silently ignored, it is now rejected at startup.
+
 ### Dependencies
 
 - `google.golang.org/grpc` is at v1.83.1, which clears GHSA-vp52-pcj8-j9qc, reported as HIGH: a peer could exhaust
diff --git a/tools/log_test.go b/tools/log_test.go
new file mode 100644
index 000000000..dedf5918c
--- /dev/null
+++ b/tools/log_test.go
@@ -0,0 +1,9 @@
+package tools
+
+import (
+	"github.com/streamingfast/logging"
+)
+
+func init() {
+	logging.InstantiateLoggers()
+}
diff --git a/tools/metrics.go b/tools/metrics.go
new file mode 100644
index 000000000..a6dce171f
--- /dev/null
+++ b/tools/metrics.go
@@ -0,0 +1,129 @@
+package tools
+
+import (
+	"context"
+	"errors"
+	"io"
+	"slices"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/streamingfast/dgrpc"
+	"github.com/streamingfast/dmetrics"
+	"google.golang.org/grpc/codes"
+)
+
+// This file holds the metrics of the `prometheus-exporter` command: the failure taxonomy
+// that gives them their labels, their declaration and the label plumbing. The polling loop
+// that feeds them lives in `prometheus-exporter.go`.
+
+// failureReason categorizes *where* a poll failed, so that an alert firing on
+// `substreams_healthcheck_status == 0` can be traced back to a cause without
+// having to correlate it with the logs.
+type failureReason string
+
+const (
+	// reasonConnect means the gRPC client could not even be constructed.
+	reasonConnect failureReason = "connect"
+	// reasonConnectTimeout means the gRPC channel never reached the READY state within
+	// the connect timeout: DNS, TCP, TLS or load-balancer resolution is the problem, the
+	// Substreams backend was never reached.
+	reasonConnectTimeout failureReason = "connect_timeout"
+	// reasonInvalidRequest means the request could not be built or validated, this is a
+	// configuration problem on our side, never an endpoint problem.
+	reasonInvalidRequest failureReason = "invalid_request"
+	// reasonRequestTimeout means the endpoint accepted the request but did not deliver a
+	// block within the request timeout.
+	reasonRequestTimeout failureReason = "request_timeout"
+	// reasonStreamError means the endpoint returned a gRPC error, see the `grpc_code` label.
+	reasonStreamError failureReason = "stream_error"
+	// reasonStaleBlock means the endpoint answered correctly but the block it returned is
+	// older than --max-freshness.
+	reasonStaleBlock failureReason = "stale_block"
+	// reasonNoData means the stream completed without ever returning block data.
+	reasonNoData failureReason = "no_data"
+)
+
+// noGRPCCode is the value of the `grpc_code` label for failures that did not carry a gRPC status.
+const noGRPCCode = "none"
+
+const (
+	// endpointLabel is the one label every endpoint always carries, the others come from the
+	// query parameters of the endpoint specification.
+	endpointLabel = "endpoint"
+	// reasonLabel and grpcCodeLabel are carried by the failure counter only.
+	reasonLabel   = "reason"
+	grpcCodeLabel = "grpc_code"
+)
+
+var healthcheckMetrics = dmetrics.NewSet(dmetrics.PrefixNameWith("substreams_healthcheck"))
+
+var (
+	status              *dmetrics.GaugeVec
+	requestDurationMs   *dmetrics.GaugeVec
+	connectDurationMs   *dmetrics.GaugeVec
+	streamDurationMs    *dmetrics.GaugeVec
+	blockAgeMs          *dmetrics.GaugeVec
+	consecutiveFailures *dmetrics.GaugeVec
+	failureCount        *dmetrics.CounterVec
+)
+
+// initHealthcheckMetrics declares every healthcheck metric over the given label names, which
+// are the union of the labels across all the polled endpoints, and returns the collectors so
+// that the caller can register them on its own registry.
+func initHealthcheckMetrics(labelNames []string) []prometheus.Collector {
+	failureLabelNames := append(slices.Clone(labelNames), reasonLabel, grpcCodeLabel)
+
+	status = healthcheckMetrics.NewGaugeVec("status", labelNames, "Either 1 for successful subtreams request, or 0 for failure")
+	requestDurationMs = healthcheckMetrics.NewGaugeVec("duration_ms", labelNames, "Request full processing time in millisecond, connection establishment included")
+	connectDurationMs = healthcheckMetrics.NewGaugeVec("connect_duration_ms", labelNames, "Time spent establishing the gRPC connection in millisecond, this excludes the Blocks request itself")
+	streamDurationMs = healthcheckMetrics.NewGaugeVec("stream_duration_ms", labelNames, "Time spent on the Blocks request in millisecond, from an established connection to the first block")
+	blockAgeMs = healthcheckMetrics.NewGaugeVec("block_age_ms", labelNames, "Age of returned block, NaN when the last poll did not return a block")
+	consecutiveFailures = healthcheckMetrics.NewGaugeVec("consecutive_failures", labelNames, "Number of consecutive failed polls, 0 when the last poll succeeded. Alert on this instead of 'status' to ignore single-poll hiccups")
+	failureCount = healthcheckMetrics.NewCounterVec("failure_count", failureLabelNames, "Number of failed polls, broken down by 'reason' and by gRPC status code")
+
+	return []prometheus.Collector{status, requestDurationMs, connectDurationMs, streamDurationMs, blockAgeMs, consecutiveFailures, failureCount}
+}
+
+// endpointLabelValues orders an endpoint's parameters along labelNames, filling in an empty
+// value for the labels this endpoint was not given.
+func endpointLabelValues(url string, params map[string]string, labelNames []string) []string {
+	values := make([]string, len(labelNames))
+	for i, name := range labelNames {
+		if name == endpointLabel {
+			values[i] = url
+			continue
+		}
+		values[i] = params[name]
+	}
+	return values
+}
+
+// failureLabelValues returns the endpoint label values augmented with the failure
+// classification, used by the `substreams_healthcheck_failure_count` counter only.
+func failureLabelValues(endpoint string, reason failureReason, grpcCode string) []string {
+	return append(slices.Clone(endpointMap[endpoint].labelValues), string(reason), grpcCode)
+}
+
+// grpcCodeOf returns the gRPC status code carried by err, or `noGRPCCode` when there is none.
+func grpcCodeOf(err error) string {
+	if grpcError := dgrpc.AsGRPCError(err); grpcError != nil {
+		return grpcError.Code().String()
+	}
+	return noGRPCCode
+}
+
+// classifyStreamError maps an error returned while talking to the endpoint onto a
+// failureReason, distinguishing a timeout from an outright refusal, and an empty
+// stream from a stream that errored out.
+func classifyStreamError(err error) failureReason {
+	if errors.Is(err, io.EOF) {
+		return reasonNoData
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return reasonRequestTimeout
+	}
+	if grpcError := dgrpc.AsGRPCError(err); grpcError != nil && grpcError.Code() == codes.DeadlineExceeded {
+		return reasonRequestTimeout
+	}
+	return reasonStreamError
+}
diff --git a/tools/metrics_test.go b/tools/metrics_test.go
new file mode 100644
index 000000000..0dd935c71
--- /dev/null
+++ b/tools/metrics_test.go
@@ -0,0 +1,125 @@
+package tools
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	grpcstatus "google.golang.org/grpc/status"
+)
+
+func TestClassifyStreamError(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want failureReason
+	}{
+		{
+			name: "stream completed without any block",
+			err:  io.EOF,
+			want: reasonNoData,
+		},
+		{
+			name: "wrapped end of stream",
+			err:  fmt.Errorf("receiving message: %w", io.EOF),
+			want: reasonNoData,
+		},
+		{
+			name: "local context deadline",
+			err:  context.DeadlineExceeded,
+			want: reasonRequestTimeout,
+		},
+		{
+			name: "grpc deadline exceeded",
+			err:  grpcstatus.Error(codes.DeadlineExceeded, "context deadline exceeded"),
+			want: reasonRequestTimeout,
+		},
+		{
+			name: "endpoint unavailable",
+			err:  grpcstatus.Error(codes.Unavailable, "no healthy upstream"),
+			want: reasonStreamError,
+		},
+		{
+			name: "endpoint refused our credentials",
+			err:  grpcstatus.Error(codes.Unauthenticated, "invalid token"),
+			want: reasonStreamError,
+		},
+		{
+			name: "plain error",
+			err:  errors.New("boom"),
+			want: reasonStreamError,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, classifyStreamError(tt.err))
+		})
+	}
+}
+
+func TestGRPCCodeOf(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want string
+	}{
+		{"no error", nil, noGRPCCode},
+		{"not a grpc error", errors.New("boom"), noGRPCCode},
+		{"grpc error", grpcstatus.Error(codes.ResourceExhausted, "overloaded"), "ResourceExhausted"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, grpcCodeOf(tt.err))
+		})
+	}
+}
+
+// Endpoints configured with different query parameters must all produce the same number of
+// label values, otherwise the prometheus `WithLabelValues` calls panic on inconsistent cardinality.
+func TestEndpointLabelValues(t *testing.T) {
+	endpoints := []string{
+		"a.domain:443?namespace=eth-mainnet",
+		"b.domain:443?namespace=sol-mainnet&region=us-east",
+		"c.domain:443",
+	}
+
+	allLabels := map[string]bool{endpointLabel: true}
+	params := map[string]map[string]string{}
+
+	for _, endpoint := range endpoints {
+		url, _, endpointParams, err := parseEndpoint(endpoint)
+		require.NoError(t, err)
+
+		for k := range endpointParams {
+			allLabels[k] = true
+		}
+		params[url] = endpointParams
+	}
+
+	labelNames := slices.Sorted(maps.Keys(allLabels))
+	require.Equal(t, []string{"endpoint", "namespace", "region"}, labelNames)
+
+	values := map[string][]string{}
+	for url, endpointParams := range params {
+		values[url] = endpointLabelValues(url, endpointParams, labelNames)
+	}
+
+	assert.Equal(t, []string{"a.domain:443", "eth-mainnet", ""}, values["a.domain:443"])
+	assert.Equal(t, []string{"b.domain:443", "sol-mainnet", "us-east"}, values["b.domain:443"])
+	assert.Equal(t, []string{"c.domain:443", "", ""}, values["c.domain:443"])
+
+	gauge := prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "test_status"}, labelNames)
+	for _, labelValues := range values {
+		assert.NotPanics(t, func() { gauge.WithLabelValues(labelValues...).Set(1) })
+	}
+}
diff --git a/tools/prometheus-exporter.go b/tools/prometheus-exporter.go
index 63790ad37..fd9e2b230 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -3,18 +3,20 @@ package tools
 import (
 	"context"
 	"fmt"
+	"maps"
+	"math"
 	"net/http"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/metadata"
 
 	"github.com/streamingfast/cli"
 	"github.com/streamingfast/cli/sflags"
-	"github.com/streamingfast/dgrpc"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -24,15 +26,11 @@ import (
 
 	"github.com/streamingfast/substreams/client"
 	"github.com/streamingfast/substreams/manifest"
-	pbsubstreamsrpc "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2"
-	pbsubstreamsrpcv2 "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v2"
 	pbsubstreamsrpcv3 "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v3"
+	pbsubstreamsrpcv4 "github.com/streamingfast/substreams/pb/sf/substreams/rpc/v4"
 	pbsubstreams "github.com/streamingfast/substreams/pb/sf/substreams/v1"
 )
 
-var lastStatus = map[string]bool{}
-var lock = &sync.Mutex{}
-
 var prometheusCmd = &cobra.Command{
 	Use:   "prometheus-exporter <endpoint[,endpoint[,endpoint[@<block_height>]],[,...]]> <manifest> <module_name>",
 	Short: "run substreams client periodically on a single block, exporting the values in prometheus format",
@@ -54,24 +52,48 @@ func init() {
 	prometheusCmd.Flags().String("substreams-api-key-envvar", "SUBSTREAMS_API_KEY", "Name of variable containing Substreams Api Key")
 	prometheusCmd.Flags().BoolP("insecure", "k", false, "Skip certificate validation on GRPC connection")
 	prometheusCmd.Flags().BoolP("plaintext", "p", false, "Establish GRPC connection in plaintext")
-	prometheusCmd.Flags().Int("force-protocol-version", 0, "Force the use of a specific protocol version (0=unset/auto, 2=v2, 3=v3)")
+	prometheusCmd.Flags().Int("force-protocol-version", 0, "Force the use of a specific protocol version (0=unset, 4=v4), only v4 is accepted for now, the flag is kept for the next protocol versions")
 	prometheusCmd.Flags().Int64("block-height", -1, "Block number to request (defaults to -1, which means the HEAD)")
 	prometheusCmd.Flags().Duration("max-freshness", time.Minute*2, "(only used if block-height is relative, i.e. below 0) check the age of the received blocks, fail an endpoint if it is older than this duration")
 	prometheusCmd.Flags().Duration("interval", time.Second*20, "endpoints will be polled at this interval")
-	prometheusCmd.Flags().Duration("timeout", time.Second*10, "endpoints will be considered 'failing' if they don't complete in that duration")
+	prometheusCmd.Flags().Duration("connect-timeout", time.Second*10, "endpoints will be considered 'failing' (with reason 'connect_timeout') if the gRPC connection does not become ready in that duration, this budget is separate from --timeout")
+	prometheusCmd.Flags().Duration("timeout", time.Second*10, "endpoints will be considered 'failing' if the Blocks request does not complete in that duration, this excludes the time spent establishing the connection (see --connect-timeout)")
 
 	Cmd.AddCommand(prometheusCmd)
 }
 
-var status *prometheus.GaugeVec
-var requestDurationMs *prometheus.GaugeVec
-var blockAgeMs *prometheus.GaugeVec
 var endpointMap = make(map[string]endpointSpecs)
 
 type endpointSpecs struct {
 	url        string
 	startBlock *int
-	labels     prometheus.Labels
+	// labelValues holds one value per metric label name, in the same order. Prometheus
+	// requires a value for every declared label, so an endpoint given fewer query parameters
+	// than another one gets an empty value for the labels it is missing.
+	labelValues []string
+}
+
+// endpointState tracks what we already reported about an endpoint so that we can log
+// both the state transitions and the individual failures, with enough context to tell a
+// single hiccup apart from a sustained outage.
+type endpointState struct {
+	known               bool
+	available           bool
+	since               time.Time
+	consecutiveFailures int
+}
+
+var endpointStates = map[string]*endpointState{}
+var lock = &sync.Mutex{}
+
+// stateFor must be called with `lock` held.
+func stateFor(endpoint string) *endpointState {
+	state, found := endpointStates[endpoint]
+	if !found {
+		state = &endpointState{since: time.Now()}
+		endpointStates[endpoint] = state
+	}
+	return state
 }
 
 func extractStartblock(in string) (prefix string, startBlock *int, err error) {
@@ -99,7 +121,7 @@ func extractParams(in string) (params map[string]string, err error) {
 	}
 
 	params = make(map[string]string)
-	for _, part := range strings.Split(in, "&") {
+	for part := range strings.SplitSeq(in, "&") {
 		parts := strings.SplitN(part, "=", 2)
 		switch len(parts) {
 		case 0:
@@ -174,13 +196,28 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 	insecure := sflags.MustGetBool(cmd, "insecure")
 	plaintext := sflags.MustGetBool(cmd, "plaintext")
 	interval := sflags.MustGetDuration(cmd, "interval")
+	connectTimeout := sflags.MustGetDuration(cmd, "connect-timeout")
 	timeout := sflags.MustGetDuration(cmd, "timeout")
+
 	protocolVersionFlag := sflags.MustGetInt(cmd, "force-protocol-version")
 	forceProtocolVersion, err := client.ParseProtocolVersion(protocolVersionFlag)
+	if err != nil {
+		return fmt.Errorf("invalid --force-protocol-version: %w", err)
+	}
+	if !forceProtocolVersion.IsUnset() && !forceProtocolVersion.IsV4() {
+		return fmt.Errorf("invalid --force-protocol-version %d: the prometheus exporter only speaks %s for now", protocolVersionFlag, client.ProtocolVersionV4)
+	}
 
 	maxFreshness := sflags.MustGetDuration(cmd, "max-freshness")
 
-	allLabels := map[string]bool{"endpoint": true}
+	type parsedEndpoint struct {
+		url        string
+		startBlock *int
+		params     map[string]string
+	}
+
+	allLabels := map[string]bool{endpointLabel: true}
+	parsed := make([]parsedEndpoint, 0, len(endpoints))
 
 	for _, endpoint := range endpoints {
 		url, startBlock, params, err := parseEndpoint(endpoint)
@@ -188,22 +225,21 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 			return fmt.Errorf("invalid endpoint %q: %w", endpoint, err)
 		}
 
-		labels := prometheus.Labels{"endpoint": url}
-		for k, v := range params {
-			labels[k] = v
+		for k := range params {
 			allLabels[k] = true
 		}
-		endpointMap[url] = endpointSpecs{url: url, startBlock: startBlock, labels: labels}
+		parsed = append(parsed, parsedEndpoint{url: url, startBlock: startBlock, params: params})
 	}
 
-	allLabelsSlice := make([]string, 0, len(allLabels))
-	for k := range allLabels {
-		allLabelsSlice = append(allLabelsSlice, k)
-	}
+	labelNames := slices.Sorted(maps.Keys(allLabels))
 
-	status = prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "substreams_healthcheck_status", Help: "Either 1 for successful subtreams request, or 0 for failure"}, allLabelsSlice)
-	requestDurationMs = prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "substreams_healthcheck_duration_ms", Help: "Request full processing time in millisecond"}, allLabelsSlice)
-	blockAgeMs = prometheus.NewGaugeVec(prometheus.GaugeOpts{Name: "substreams_healthcheck_block_age_ms", Help: "Age of returned block"}, allLabelsSlice)
+	for _, endpoint := range parsed {
+		endpointMap[endpoint.url] = endpointSpecs{
+			url:         endpoint.url,
+			startBlock:  endpoint.startBlock,
+			labelValues: endpointLabelValues(endpoint.url, endpoint.params, labelNames),
+		}
+	}
 
 	for endpoint := range endpointMap {
 		startBlock := blockNum
@@ -225,13 +261,13 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 			fresh = &maxFreshness
 		}
 
-		go launchSubstreamsPoller(endpoint, substreamsClientConfig, pkgBundle.Package, outputStreamName, startBlock, interval, timeout, fresh)
+		go launchSubstreamsPoller(endpoint, substreamsClientConfig, pkgBundle.Package, outputStreamName, startBlock, interval, connectTimeout, timeout, fresh)
 	}
 
+	// The exporter serves only its own metrics, so the collectors go to a dedicated registry
+	// instead of the global one that `dmetrics.Set.Register` would use.
 	promReg := prometheus.NewRegistry()
-	promReg.MustRegister(status)
-	promReg.MustRegister(requestDurationMs)
-	promReg.MustRegister(blockAgeMs)
+	promReg.MustRegister(initHealthcheckMetrics(labelNames)...)
 
 	handler := promhttp.HandlerFor(
 		promReg,
@@ -247,170 +283,247 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-func markSuccess(endpoint string, begin time.Time) {
+func markSuccess(endpoint string, result *pollResult) {
 	lock.Lock()
 	defer lock.Unlock()
-	if !lastStatus[endpoint] {
-		zlog.Info("endpoint now marked as available", zap.String("endpoint", endpoint))
+
+	state := stateFor(endpoint)
+	if !state.known || !state.available {
+		fields := []zap.Field{
+			zap.String("endpoint", endpoint),
+			zap.Duration("connect_duration", result.connectDuration),
+			zap.Duration("stream_duration", result.streamDuration),
+		}
+		if state.known {
+			fields = append(fields,
+				zap.Duration("unavailable_for", time.Since(state.since)),
+				zap.Int("failed_polls", state.consecutiveFailures),
+			)
+		}
+		zlog.Info("endpoint now marked as available", fields...)
+
+		state.known = true
+		state.available = true
+		state.since = time.Now()
+	}
+	state.consecutiveFailures = 0
+
+	labelValues := endpointMap[endpoint].labelValues
+	status.SetInt(1, labelValues...)
+	consecutiveFailures.SetInt(0, labelValues...)
+	requestDurationMs.SetInt64(result.totalDuration().Milliseconds(), labelValues...)
+	connectDurationMs.SetInt64(result.connectDuration.Milliseconds(), labelValues...)
+	streamDurationMs.SetInt64(result.streamDuration.Milliseconds(), labelValues...)
+	if result.blockAge != nil {
+		blockAgeMs.SetInt64(result.blockAge.Milliseconds(), labelValues...)
 	}
-	lastStatus[endpoint] = true
-	status.With(endpointMap[endpoint].labels).Set(1)
-	requestDurationMs.With(endpointMap[endpoint].labels).Set(float64(time.Since(begin).Milliseconds()))
 }
 
-func markFailure(endpoint string, begin time.Time, err error) {
+func markFailure(endpoint string, result *pollResult) {
 	lock.Lock()
 	defer lock.Unlock()
-	if val, ok := lastStatus[endpoint]; !ok || val {
-		zlog.Info("endpoint now marked as unavailable", zap.String("endpoint", endpoint), zap.Error(err))
-		lastStatus[endpoint] = false
+
+	state := stateFor(endpoint)
+	state.consecutiveFailures++
+
+	grpcCode := grpcCodeOf(result.err)
+	fields := []zap.Field{
+		zap.String("endpoint", endpoint),
+		zap.String("reason", string(result.reason)),
+		zap.String("grpc_code", grpcCode),
+		zap.Duration("connect_duration", result.connectDuration),
+		zap.Duration("stream_duration", result.streamDuration),
+		zap.Error(result.err),
+	}
+
+	if !state.known || state.available {
+		if state.known {
+			fields = append(fields, zap.Duration("available_for", time.Since(state.since)))
+		}
+		zlog.Info("endpoint now marked as unavailable", fields...)
+
+		state.known = true
+		state.available = false
+		state.since = time.Now()
+	} else {
+		// Logged on every single failure, not only on the transition: an endpoint that fails
+		// repeatedly, or one that flaps between two scrapes, is otherwise invisible in the logs.
+		fields = append(fields,
+			zap.Int("consecutive_failures", state.consecutiveFailures),
+			zap.Duration("unavailable_for", time.Since(state.since)),
+		)
+		zlog.Info("endpoint poll failed", fields...)
+	}
+
+	labelValues := endpointMap[endpoint].labelValues
+	status.SetInt(0, labelValues...)
+	consecutiveFailures.SetInt(state.consecutiveFailures, labelValues...)
+	requestDurationMs.SetInt64(result.totalDuration().Milliseconds(), labelValues...)
+	connectDurationMs.SetInt64(result.connectDuration.Milliseconds(), labelValues...)
+	streamDurationMs.SetInt64(result.streamDuration.Milliseconds(), labelValues...)
+	failureCount.Inc(failureLabelValues(endpoint, result.reason, grpcCode)...)
+
+	if result.blockAge != nil {
+		blockAgeMs.SetInt64(result.blockAge.Milliseconds(), labelValues...)
+	} else {
+		// Without this, the gauge keeps reporting the age of the last block we ever saw, which
+		// silently gets younger than reality the longer the endpoint stays broken.
+		blockAgeMs.SetFloat64(math.NaN(), labelValues...)
 	}
-	status.With(endpointMap[endpoint].labels).Set(0)
-	requestDurationMs.With(endpointMap[endpoint].labels).Set(float64(time.Since(begin).Milliseconds()))
 }
 
-func launchSubstreamsPoller(endpoint string, substreamsClientConfig *client.SubstreamsClientConfig, pkg *pbsubstreams.Package, outputStreamName string, blockNum int64, pollingInterval, pollingTimeout time.Duration, maxFreshness *time.Duration) {
+// pollResult is the outcome of a single poll, `err` being nil means the endpoint is healthy.
+type pollResult struct {
+	connectDuration time.Duration
+	streamDuration  time.Duration
+	blockAge        *time.Duration
+	blockNum        uint64
+	reason          failureReason
+	err             error
+}
+
+func (r *pollResult) totalDuration() time.Duration {
+	return r.connectDuration + r.streamDuration
+}
 
+// waitForConnReady blocks until the gRPC channel is usable. gRPC dials lazily, so without
+// this the DNS resolution, the TLS handshake and the load-balancer setup would all be
+// charged to the Blocks request budget, and every slow connection would be reported as an
+// endpoint failure ("waiting for new LB policy update: context deadline exceeded").
+func waitForConnReady(ctx context.Context, conn *grpc.ClientConn) error {
+	conn.Connect()
+	for {
+		state := conn.GetState()
+		switch state {
+		case connectivity.Ready:
+			return nil
+		case connectivity.Shutdown:
+			return fmt.Errorf("connection shut down before becoming ready")
+		}
+
+		if !conn.WaitForStateChange(ctx, state) {
+			return fmt.Errorf("connection stuck in state %q: %w", state, context.Cause(ctx))
+		}
+	}
+}
+
+func launchSubstreamsPoller(endpoint string, substreamsClientConfig *client.SubstreamsClientConfig, pkg *pbsubstreams.Package, outputStreamName string, blockNum int64, pollingInterval, connectTimeout, pollingTimeout time.Duration, maxFreshness *time.Duration) {
 	sleep := time.Duration(0)
 	for {
 		time.Sleep(sleep)
 		sleep = pollingInterval
 
-		ctx, cancel := context.WithTimeout(context.Background(), pollingTimeout)
-		begin := time.Now()
-		conn, connClose, callOpts, headers, err := client.NewSubstreamsClientConn(substreamsClientConfig)
-		if err != nil {
-			zlog.Error("substreams client connection setup", zap.Error(err))
-			markFailure(endpoint, begin, err)
-			cancel()
+		result := pollEndpoint(endpoint, substreamsClientConfig, pkg, outputStreamName, blockNum, connectTimeout, pollingTimeout, maxFreshness)
+		if result.err != nil {
+			markFailure(endpoint, result)
 			continue
 		}
+		markSuccess(endpoint, result)
+	}
+}
 
-		ssClientV2 := pbsubstreamsrpcv2.NewStreamClient(conn)
-		ssClientV3 := pbsubstreamsrpcv3.NewStreamClient(conn)
+func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClientConfig, pkg *pbsubstreams.Package, outputStreamName string, blockNum int64, connectTimeout, pollingTimeout time.Duration, maxFreshness *time.Duration) (result *pollResult) {
+	result = &pollResult{}
 
-		if headers.IsSet() {
-			ctx = metadata.AppendToOutgoingContext(ctx, headers.ToArray()...)
-		}
+	connectBegin := time.Now()
+	conn, connClose, callOpts, headers, err := client.NewSubstreamsClientConn(substreamsClientConfig)
+	if err != nil {
+		result.connectDuration = time.Since(connectBegin)
+		result.reason, result.err = reasonConnect, err
+		return
+	}
+	defer connClose()
 
-		var stopBlockNum uint64
-		if blockNum > 0 {
-			stopBlockNum = uint64(blockNum + 1)
-		}
-		subReq := &pbsubstreamsrpcv3.Request{
-			StartBlockNum: blockNum,
-			StopBlockNum:  stopBlockNum,
-			Package:       pkg,
-			OutputModule:  outputStreamName,
+	connectCtx, cancelConnect := context.WithTimeoutCause(context.Background(), connectTimeout, fmt.Errorf("connect timeout of %s reached", connectTimeout))
+	defer cancelConnect()
+
+	if err := waitForConnReady(connectCtx, conn); err != nil {
+		result.connectDuration = time.Since(connectBegin)
+		result.reason, result.err = reasonConnectTimeout, err
+		return
+	}
+	result.connectDuration = time.Since(connectBegin)
+
+	streamBegin := time.Now()
+	defer func() { result.streamDuration = time.Since(streamBegin) }()
+
+	ctx, cancel := context.WithTimeoutCause(context.Background(), pollingTimeout, fmt.Errorf("request timeout of %s reached", pollingTimeout))
+	defer cancel()
+
+	if headers.IsSet() {
+		ctx = metadata.AppendToOutgoingContext(ctx, headers.ToArray()...)
+	}
+
+	var stopBlockNum uint64
+	if blockNum > 0 {
+		stopBlockNum = uint64(blockNum + 1)
+	}
+
+	// `sf.substreams.rpc.v4.Stream/Blocks` takes a v3 request and answers with v4 responses.
+	subReq := &pbsubstreamsrpcv3.Request{
+		StartBlockNum: blockNum,
+		StopBlockNum:  stopBlockNum,
+		Package:       pkg,
+		OutputModule:  outputStreamName,
+	}
+
+	if err := subReq.Validate(); err != nil {
+		result.reason, result.err = reasonInvalidRequest, err
+		return
+	}
+
+	// The connection is already READY, so a failure here is the endpoint refusing us, never
+	// a connection still being established.
+	callOpts = append(callOpts, grpc.WaitForReady(false))
+	zlog.Debug("calling sf.substreams.rpc.v4.Stream/Blocks", zap.String("endpoint", endpoint), zap.String("output_module", outputStreamName), zap.Int64("start_block", blockNum), zap.Uint64("stop_block", stopBlockNum), zap.Duration("connect_duration", result.connectDuration))
+
+	streamClient, err := pbsubstreamsrpcv4.NewStreamClient(conn).Blocks(ctx, subReq, callOpts...)
+	if err != nil {
+		result.reason, result.err = classifyStreamError(err), err
+		return
+	}
+
+	for {
+		resp, err := streamClient.Recv()
+		if err != nil {
+			result.reason, result.err = classifyStreamError(err), err
+			return
 		}
 
-		if err := subReq.Validate(); err != nil {
-			zlog.Error("validate request", zap.Error(err))
-			markFailure(endpoint, begin, err)
-			connClose()
-			cancel()
+		data, ok := resp.Message.(*pbsubstreamsrpcv4.Response_BlockScopedDatas)
+		if !ok || len(data.BlockScopedDatas.Items) == 0 {
 			continue
 		}
-		callOpts = append(callOpts, grpc.WaitForReady(false))
-		zlog.Debug("calling sf.substreams.rpc.v2.Stream/Blocks", zap.String("endpoint", endpoint), zap.String("output_module", outputStreamName), zap.Int64("start_block", blockNum), zap.Uint64("stop_block", stopBlockNum))
 
-		var isRunningV2 bool
-		var cli grpc.ServerStreamingClient[pbsubstreamsrpc.Response]
-		if substreamsClientConfig.ForceProtocolVersion().IsV2() {
-			reqV2, err := subReq.ToV2()
-			if err != nil {
-				zlog.Error("call sf.substreams.rpc.v2.Stream/Blocks", zap.String("endpoint", endpoint), zap.Error(err))
-				markFailure(endpoint, begin, err)
-				connClose()
-				cancel()
-				continue
-			}
-			cli, err = ssClientV2.Blocks(ctx, reqV2, callOpts...)
-			if err != nil {
-				zlog.Error("call sf.substreams.rpc.v2.Stream/Blocks", zap.String("endpoint", endpoint), zap.Error(err))
-				markFailure(endpoint, begin, err)
-				connClose()
-				cancel()
-				continue
-			}
-			isRunningV2 = true
-		} else {
-			cli, err = ssClientV3.Blocks(ctx, subReq, callOpts...)
-			if err != nil {
-				zlog.Error("call sf.substreams.rpc.v2.Stream/Blocks", zap.String("endpoint", endpoint), zap.Error(err))
-				markFailure(endpoint, begin, err)
-				connClose()
-				cancel()
-				continue
-			}
+		// Items are ordered by block number ascending, the last one is the freshest, which is
+		// what a HEAD healthcheck cares about.
+		clock := data.BlockScopedDatas.Items[len(data.BlockScopedDatas.Items)-1].Clock
+		result.blockNum = clock.Number
+		if maxFreshness == nil {
+			zlog.Debug("marking endpoint with success", zap.String("endpoint", endpoint), zap.Uint64("block_num", clock.Number))
+			return
 		}
 
-	forloop:
-		for {
-			resp, err := cli.Recv()
-			if resp != nil {
-				switch resp.Message.(type) {
-				case *pbsubstreamsrpc.Response_BlockScopedData:
-					if maxFreshness == nil {
-						zlog.Debug("marking endpoint with success",
-							zap.String("endpoint", endpoint),
-							zap.Duration("duration", time.Since(begin)),
-							zap.Uint64("block_num", resp.Message.(*pbsubstreamsrpc.Response_BlockScopedData).BlockScopedData.Clock.Number),
-						)
-						markSuccess(endpoint, begin)
-						break forloop
-					}
-					blockTime := resp.Message.(*pbsubstreamsrpc.Response_BlockScopedData).BlockScopedData.Clock.Timestamp.AsTime()
-					blockAgeMs.With(endpointMap[endpoint].labels).Set(float64(time.Since(blockTime).Milliseconds()))
-					if age := time.Since(blockTime); age > *maxFreshness {
-						markFailure(endpoint, begin, fmt.Errorf("block is too old: %s", age))
-						zlog.Debug("marking endpoint with failure because of freshness", zap.String("endpoint", endpoint), zap.Duration("duration", time.Since(begin)), zap.Duration("block_age", time.Since(blockTime)))
-					} else {
-						markSuccess(endpoint, begin)
-						zlog.Debug("marking endpoint with success",
-							zap.String("endpoint", endpoint),
-							zap.Duration("duration", time.Since(begin)),
-							zap.Duration("block_age", time.Since(blockTime)),
-							zap.Uint64("block_num", resp.Message.(*pbsubstreamsrpc.Response_BlockScopedData).BlockScopedData.Clock.Number),
-						)
-					}
-					break forloop
-				}
-			}
-			if err != nil {
-				if substreamsClientConfig.ForceProtocolVersion().IsUnset() && !isRunningV2 {
-					if dgrpcError := dgrpc.AsGRPCError(err); dgrpcError != nil {
-						switch dgrpcError.Code() {
-						case codes.Unimplemented, codes.NotFound:
-
-							zlog.Debug("server does not implement sf.substreams.rpc.v3.Stream/Blocks, trying v2")
-							reqV2, err := subReq.ToV2()
-							if err != nil {
-								zlog.Error("call sf.substreams.rpc.v2.Stream/Blocks", zap.String("endpoint", endpoint), zap.Error(err))
-								markFailure(endpoint, begin, err)
-								connClose()
-								cancel()
-								break
-							}
-							cli, err = ssClientV2.Blocks(ctx, reqV2, callOpts...)
-							if err != nil {
-								zlog.Error("call sf.substreams.rpc.v2.Stream/Blocks", zap.String("endpoint", endpoint), zap.Error(err))
-								markFailure(endpoint, begin, err)
-								connClose()
-								cancel()
-								continue
-							}
-							isRunningV2 = true
-						}
-					}
-				}
-
-				markFailure(endpoint, begin, err)
-				break
-			}
+		age := time.Since(clock.Timestamp.AsTime())
+		result.blockAge = &age
+		if age > *maxFreshness {
+			result.reason = reasonStaleBlock
+			result.err = fmt.Errorf("block %d is too old: %s, above the %s max freshness", clock.Number, age, *maxFreshness)
+			return
 		}
 
-		connClose()
-		cancel()
+		// A block age climbing towards the threshold is what precedes a `stale_block`
+		// failure, reporting it here is what makes an alert on `block_age_ms` explainable.
+		if age > *maxFreshness/2 {
+			zlog.Info("endpoint block age is above half of the max freshness",
+				zap.String("endpoint", endpoint),
+				zap.Uint64("block_num", clock.Number),
+				zap.Duration("block_age", age),
+				zap.Duration("max_freshness", *maxFreshness),
+			)
+		}
+
+		zlog.Debug("marking endpoint with success", zap.String("endpoint", endpoint), zap.Uint64("block_num", clock.Number), zap.Duration("block_age", age))
+		return
 	}
 }
diff --git a/tools/prometheus-exporter_test.go b/tools/prometheus-exporter_test.go
index dd5c679ec..61a4d97f9 100644
--- a/tools/prometheus-exporter_test.go
+++ b/tools/prometheus-exporter_test.go
@@ -3,8 +3,8 @@ package tools
 import (
 	"testing"
 
-	"github.com/test-go/testify/assert"
-	"github.com/test-go/testify/require"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func intPtr(i int) *int {

From 062ee00c47ce29a83874d4546fbbcedbfbb3cf65 Mon Sep 17 00:00:00 2001
From: Matthieu Vachon <matt@streamingfast.io>
Date: Wed, 2 Sep 2026 09:18:29 -0400
Subject: [PATCH 2/8] Address review feedback on the prometheus-exporter

Declare the healthcheck metrics before starting the pollers. Moving them
into initHealthcheckMetrics left the seven package-level pointers nil
while the poller goroutines were already running, which is a data race
and a nil dereference for any endpoint that fails on its first attempt.

Fail fast when the gRPC channel reaches TRANSIENT_FAILURE instead of
waiting out --connect-timeout. gRPC re-dials on its own backoff, so a
refused connection or a DNS failure used to burn the whole budget and
then report connect_timeout with the real dial error discarded. The
request is now issued anyway, since that is what surfaces the dial error,
and the failure is reported as connect_failed. connect_timeout is left to
mean a connection that is merely slow.

Rename the reason for a client that could not be constructed from
connect to invalid_config: it never described the endpoint.

Read the request deadline cause. gRPC answers with its own
DeadlineExceeded status and drops the cause, so the log never named which
budget expired.

Report the block age on crossings of half of --max-freshness rather than
on every poll, and only once three consecutive polls agree. A chain whose
block interval straddles the threshold otherwise reports a healthy
endpoint forever.

Reject --force-protocol-version before ParseProtocolVersion so that an
operator passing 2 or 3 is not first told those versions are supported.

Guard the Clock dereference, reported as invalid_response, and drop the
write-only pollResult.blockNum and endpointSpecs.url fields.

Flag the --force-protocol-version break as an explicit operator step in
the changelog, and drop comments that restated their identifier.
---
 docs/release-notes/change-log.md  |  26 +++---
 tools/metrics.go                  |  52 +++++++-----
 tools/metrics_test.go             |  40 +++++++++
 tools/prometheus-exporter.go      | 132 +++++++++++++++++++++++-------
 tools/prometheus-exporter_test.go |  43 ++++++++++
 5 files changed, 234 insertions(+), 59 deletions(-)

diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md
index 4617ff6a5..89ed7fe7f 100644
--- a/docs/release-notes/change-log.md
+++ b/docs/release-notes/change-log.md
@@ -101,10 +101,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Tools
 
 - `substreams tools prometheus-exporter` now says *why* an endpoint is down. Every failure is classified into a
-  `reason` -- `connect`, `connect_timeout`, `invalid_request`, `request_timeout`, `stream_error`, `stale_block` or
-  `no_data` -- exposed on the new `substreams_healthcheck_failure_count{reason,grpc_code}` counter and included in
-  the logs. An alert firing on `substreams_healthcheck_status` no longer requires guessing whether the endpoint was
-  unreachable, unauthenticated, overloaded or merely late.
+  `reason` -- `invalid_config`, `connect_failed`, `connect_timeout`, `invalid_request`, `request_timeout`,
+  `stream_error`, `stale_block`, `invalid_response` or `no_data` -- exposed on the new
+  `substreams_healthcheck_failure_count{reason,grpc_code}` counter and included in the logs. An alert firing on
+  `substreams_healthcheck_status` no longer requires guessing whether the endpoint was unreachable, unauthenticated,
+  overloaded or merely late. A dial that fails outright is reported as `connect_failed` within milliseconds, carrying
+  the dial error (`connection refused`, DNS failure); `connect_timeout` is reserved for a connection that is merely
+  slow to come up.
 
 - Connection establishment gets its own budget, `--connect-timeout` (default 10s), separate from `--timeout`, which
   now covers the `Blocks` request alone. gRPC dials lazily, so DNS, TLS and load-balancer resolution used to be
@@ -117,8 +120,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Every failed poll is logged, not just the transition into `unavailable`. An endpoint that fails repeatedly, or one
   that flaps between two Prometheus scrapes, previously produced a single line and then nothing. Failure logs carry
   the reason, the gRPC code, both durations and the consecutive failure count; the recovery log carries how long the
-  endpoint was down and how many polls failed meanwhile. A block age above half of `--max-freshness` is logged too,
-  so an alert on `substreams_healthcheck_block_age_ms` is no longer silent.
+  endpoint was down and how many polls failed meanwhile. A block age crossing half of `--max-freshness` is reported
+  too, so an alert on `substreams_healthcheck_block_age_ms` is no longer silent. That one is edge-triggered and only
+  after three consecutive polls agree, so a chain whose block interval straddles the threshold stays quiet.
 
 - New `substreams_healthcheck_consecutive_failures` gauge, meant to be alerted on instead of
   `substreams_healthcheck_status` when single-poll hiccups should be ignored.
@@ -130,10 +134,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
   and one with only `?namespace=z`) made the exporter panic on inconsistent label cardinality. Missing labels are now
   filled with an empty value.
 
-- The exporter now speaks `sf.substreams.rpc.v4.Stream/Blocks` only. The v3-to-v2 fallback is gone -- it closed the
-  connection and then kept reading from it, double-counting the failure -- and `--force-protocol-version` accepts
-  only `4` (or `0`), the flag being kept for the protocol versions to come. An invalid value used to be parsed and
-  then silently ignored, it is now rejected at startup.
+- **Breaking** The exporter now speaks `sf.substreams.rpc.v4.Stream/Blocks` only. The v3-to-v2 fallback is gone --
+  it closed the connection and then kept reading from it, double-counting the failure -- and
+  `--force-protocol-version` accepts only `4` (or `0`), the flag being kept for the protocol versions to come. An
+  invalid value used to be parsed and then silently ignored, it is now rejected at startup, which means a deployment
+  currently passing `--force-protocol-version 2` or `3` will refuse to start: **unset the flag before rolling the
+  new image**, otherwise the exporter crash-loops and the fleet loses its healthcheck metrics entirely.
 
 ### Dependencies
 
diff --git a/tools/metrics.go b/tools/metrics.go
index a6dce171f..025bda951 100644
--- a/tools/metrics.go
+++ b/tools/metrics.go
@@ -3,6 +3,7 @@ package tools
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"slices"
 
@@ -22,25 +23,15 @@ import (
 type failureReason string
 
 const (
-	// reasonConnect means the gRPC client could not even be constructed.
-	reasonConnect failureReason = "connect"
-	// reasonConnectTimeout means the gRPC channel never reached the READY state within
-	// the connect timeout: DNS, TCP, TLS or load-balancer resolution is the problem, the
-	// Substreams backend was never reached.
-	reasonConnectTimeout failureReason = "connect_timeout"
-	// reasonInvalidRequest means the request could not be built or validated, this is a
-	// configuration problem on our side, never an endpoint problem.
-	reasonInvalidRequest failureReason = "invalid_request"
-	// reasonRequestTimeout means the endpoint accepted the request but did not deliver a
-	// block within the request timeout.
-	reasonRequestTimeout failureReason = "request_timeout"
-	// reasonStreamError means the endpoint returned a gRPC error, see the `grpc_code` label.
-	reasonStreamError failureReason = "stream_error"
-	// reasonStaleBlock means the endpoint answered correctly but the block it returned is
-	// older than --max-freshness.
-	reasonStaleBlock failureReason = "stale_block"
-	// reasonNoData means the stream completed without ever returning block data.
-	reasonNoData failureReason = "no_data"
+	reasonInvalidConfig   failureReason = "invalid_config"
+	reasonConnectFailed   failureReason = "connect_failed"
+	reasonConnectTimeout  failureReason = "connect_timeout"
+	reasonInvalidRequest  failureReason = "invalid_request"
+	reasonRequestTimeout  failureReason = "request_timeout"
+	reasonStreamError     failureReason = "stream_error"
+	reasonStaleBlock      failureReason = "stale_block"
+	reasonInvalidResponse failureReason = "invalid_response"
+	reasonNoData          failureReason = "no_data"
 )
 
 // noGRPCCode is the value of the `grpc_code` label for failures that did not carry a gRPC status.
@@ -112,6 +103,29 @@ func grpcCodeOf(err error) string {
 	return noGRPCCode
 }
 
+// streamFailure attributes an error returned by the Blocks call. `connect_failed` means the
+// dial itself failed, `connect_timeout` (set by the caller) means it was merely slow and the
+// backend was never reached at all. When the channel never
+// became ready, gRPC answers with the dial error it was holding, so the failure belongs to
+// the connection rather than to the endpoint's own answer.
+func streamFailure(ctx context.Context, connectFailedFast bool, err error) (failureReason, error) {
+	err = withDeadlineCause(ctx, err)
+	if connectFailedFast {
+		return reasonConnectFailed, err
+	}
+	return classifyStreamError(err), err
+}
+
+// withDeadlineCause appends the reason ctx was cut short. gRPC answers a request that
+// outlived its deadline with its own "context deadline exceeded" status and drops the cause
+// attached to the context, so without this the error never names which budget expired.
+func withDeadlineCause(ctx context.Context, err error) error {
+	if cause := context.Cause(ctx); cause != nil {
+		return fmt.Errorf("%w: %s", err, cause)
+	}
+	return err
+}
+
 // classifyStreamError maps an error returned while talking to the endpoint onto a
 // failureReason, distinguishing a timeout from an outright refusal, and an empty
 // stream from a stream that errored out.
diff --git a/tools/metrics_test.go b/tools/metrics_test.go
index 0dd935c71..c711cb02b 100644
--- a/tools/metrics_test.go
+++ b/tools/metrics_test.go
@@ -8,6 +8,7 @@ import (
 	"maps"
 	"slices"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
@@ -123,3 +124,42 @@ func TestEndpointLabelValues(t *testing.T) {
 		assert.NotPanics(t, func() { gauge.WithLabelValues(labelValues...).Set(1) })
 	}
 }
+
+func TestWithDeadlineCause(t *testing.T) {
+	streamErr := grpcstatus.Error(codes.DeadlineExceeded, "context deadline exceeded")
+
+	t.Run("live context leaves the error alone", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		assert.Equal(t, streamErr, withDeadlineCause(ctx, streamErr))
+	})
+
+	t.Run("expired context names the budget that ran out", func(t *testing.T) {
+		ctx, cancel := context.WithTimeoutCause(context.Background(), time.Nanosecond, errors.New("request timeout of 15s reached"))
+		defer cancel()
+		<-ctx.Done()
+
+		err := withDeadlineCause(ctx, streamErr)
+		assert.ErrorContains(t, err, "request timeout of 15s reached")
+		assert.ErrorIs(t, err, streamErr)
+	})
+}
+
+func TestStreamFailure(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// A dial that failed fast answers through the request, so the gRPC error carries the dial
+	// message and belongs to the connection, not to the endpoint's own answer.
+	refused := grpcstatus.Error(codes.Unavailable, `connection error: desc = "transport: Error while dialing: dial tcp 127.0.0.1:19999: connect: connection refused"`)
+	reason, err := streamFailure(ctx, true, refused)
+	assert.Equal(t, reasonConnectFailed, reason)
+	assert.ErrorContains(t, err, "connection refused")
+
+	reason, _ = streamFailure(ctx, false, refused)
+	assert.Equal(t, reasonStreamError, reason)
+
+	reason, _ = streamFailure(ctx, false, grpcstatus.Error(codes.DeadlineExceeded, "context deadline exceeded"))
+	assert.Equal(t, reasonRequestTimeout, reason)
+}
diff --git a/tools/prometheus-exporter.go b/tools/prometheus-exporter.go
index fd9e2b230..7a27e83dc 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -2,6 +2,7 @@ package tools
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"maps"
 	"math"
@@ -65,7 +66,6 @@ func init() {
 var endpointMap = make(map[string]endpointSpecs)
 
 type endpointSpecs struct {
-	url        string
 	startBlock *int
 	// labelValues holds one value per metric label name, in the same order. Prometheus
 	// requires a value for every declared label, so an endpoint given fewer query parameters
@@ -81,6 +81,8 @@ type endpointState struct {
 	available           bool
 	since               time.Time
 	consecutiveFailures int
+	blockAgeAboveHalf   bool
+	blockAgeStreak      int
 }
 
 var endpointStates = map[string]*endpointState{}
@@ -199,14 +201,18 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 	connectTimeout := sflags.MustGetDuration(cmd, "connect-timeout")
 	timeout := sflags.MustGetDuration(cmd, "timeout")
 
+	// Checked before `ParseProtocolVersion` so that an operator passing 2 or 3 reads the one
+	// message that applies here, rather than being told v2 and v3 are supported and then
+	// refused them on the next line.
 	protocolVersionFlag := sflags.MustGetInt(cmd, "force-protocol-version")
+	if protocolVersionFlag != 0 && protocolVersionFlag != int(client.ProtocolVersionV4) {
+		return fmt.Errorf("invalid --force-protocol-version %d: the prometheus exporter only speaks %s for now, leave the flag unset or pass 4", protocolVersionFlag, client.ProtocolVersionV4)
+	}
+
 	forceProtocolVersion, err := client.ParseProtocolVersion(protocolVersionFlag)
 	if err != nil {
 		return fmt.Errorf("invalid --force-protocol-version: %w", err)
 	}
-	if !forceProtocolVersion.IsUnset() && !forceProtocolVersion.IsV4() {
-		return fmt.Errorf("invalid --force-protocol-version %d: the prometheus exporter only speaks %s for now", protocolVersionFlag, client.ProtocolVersionV4)
-	}
 
 	maxFreshness := sflags.MustGetDuration(cmd, "max-freshness")
 
@@ -235,12 +241,16 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 
 	for _, endpoint := range parsed {
 		endpointMap[endpoint.url] = endpointSpecs{
-			url:         endpoint.url,
 			startBlock:  endpoint.startBlock,
 			labelValues: endpointLabelValues(endpoint.url, endpoint.params, labelNames),
 		}
 	}
 
+	// Declared before the pollers start: a poller that fails on its very first attempt reports
+	// through these, and a nil one is both a data race and a nil dereference that takes the
+	// whole process down.
+	collectors := initHealthcheckMetrics(labelNames)
+
 	for endpoint := range endpointMap {
 		startBlock := blockNum
 		if endpointMap[endpoint].startBlock != nil {
@@ -267,7 +277,7 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 	// The exporter serves only its own metrics, so the collectors go to a dedicated registry
 	// instead of the global one that `dmetrics.Set.Register` would use.
 	promReg := prometheus.NewRegistry()
-	promReg.MustRegister(initHealthcheckMetrics(labelNames)...)
+	promReg.MustRegister(collectors...)
 
 	handler := promhttp.HandlerFor(
 		promReg,
@@ -307,6 +317,7 @@ func markSuccess(endpoint string, result *pollResult) {
 		state.since = time.Now()
 	}
 	state.consecutiveFailures = 0
+	trackBlockAge(endpoint, state, result)
 
 	labelValues := endpointMap[endpoint].labelValues
 	status.SetInt(1, labelValues...)
@@ -346,8 +357,8 @@ func markFailure(endpoint string, result *pollResult) {
 		state.available = false
 		state.since = time.Now()
 	} else {
-		// Logged on every single failure, not only on the transition: an endpoint that fails
-		// repeatedly, or one that flaps between two scrapes, is otherwise invisible in the logs.
+		// An endpoint that fails repeatedly, or one that flaps between two Prometheus scrapes,
+		// stays invisible in the logs unless every failure is reported, not just the first one.
 		fields = append(fields,
 			zap.Int("consecutive_failures", state.consecutiveFailures),
 			zap.Duration("unavailable_for", time.Since(state.since)),
@@ -355,6 +366,8 @@ func markFailure(endpoint string, result *pollResult) {
 		zlog.Info("endpoint poll failed", fields...)
 	}
 
+	trackBlockAge(endpoint, state, result)
+
 	labelValues := endpointMap[endpoint].labelValues
 	status.SetInt(0, labelValues...)
 	consecutiveFailures.SetInt(state.consecutiveFailures, labelValues...)
@@ -372,12 +385,56 @@ func markFailure(endpoint string, result *pollResult) {
 	}
 }
 
+// blockAgeFlipPolls is how many consecutive polls must agree before the block-age report
+// flips. A chain whose block interval straddles half of --max-freshness crosses the threshold
+// on almost every poll, so a bare edge trigger reports a healthy endpoint forever, just in
+// pairs of lines instead of one. Requiring a streak reports only a drift that persists.
+const blockAgeFlipPolls = 3
+
+// trackBlockAge reports the block age on the crossings rather than on every poll, the way the
+// availability transitions are reported: an age sitting just above the threshold is the normal
+// state of a chain whose block interval is close to it, and saying so once per poll drowns the
+// signal. Must be called with `lock` held.
+func trackBlockAge(endpoint string, state *endpointState, result *pollResult) {
+	if result.blockAge == nil || result.maxFreshness == nil {
+		return
+	}
+
+	aboveHalf := *result.blockAge > *result.maxFreshness/2
+	if aboveHalf == state.blockAgeAboveHalf {
+		state.blockAgeStreak = 0
+		return
+	}
+
+	state.blockAgeStreak++
+	if state.blockAgeStreak < blockAgeFlipPolls {
+		return
+	}
+
+	state.blockAgeAboveHalf = aboveHalf
+	state.blockAgeStreak = 0
+
+	message := "endpoint block age fell back below half of the max freshness"
+	if aboveHalf {
+		// This is what precedes a `stale_block` failure, and what makes an alert on
+		// `block_age_ms` explainable.
+		message = "endpoint block age climbed above half of the max freshness"
+	}
+
+	zlog.Info(message,
+		zap.String("endpoint", endpoint),
+		zap.Duration("block_age", *result.blockAge),
+		zap.Duration("max_freshness", *result.maxFreshness),
+		zap.Int("confirmed_over_polls", blockAgeFlipPolls),
+	)
+}
+
 // pollResult is the outcome of a single poll, `err` being nil means the endpoint is healthy.
 type pollResult struct {
 	connectDuration time.Duration
 	streamDuration  time.Duration
 	blockAge        *time.Duration
-	blockNum        uint64
+	maxFreshness    *time.Duration
 	reason          failureReason
 	err             error
 }
@@ -386,10 +443,19 @@ func (r *pollResult) totalDuration() time.Duration {
 	return r.connectDuration + r.streamDuration
 }
 
+// errConnFailedFast reports that the channel reached TRANSIENT_FAILURE: the dial failed
+// outright rather than being slow. The connectivity API never hands over the dial error, so
+// the caller issues the request anyway and lets gRPC answer with it.
+var errConnFailedFast = errors.New("connection failed to establish")
+
 // waitForConnReady blocks until the gRPC channel is usable. gRPC dials lazily, so without
 // this the DNS resolution, the TLS handshake and the load-balancer setup would all be
 // charged to the Blocks request budget, and every slow connection would be reported as an
 // endpoint failure ("waiting for new LB policy update: context deadline exceeded").
+//
+// It only ever waits on a connection that is still making progress. gRPC re-dials on its own
+// backoff, so waiting through TRANSIENT_FAILURE would burn the whole connect budget on an
+// endpoint that answered "connection refused" in a microsecond, and report it as a timeout.
 func waitForConnReady(ctx context.Context, conn *grpc.ClientConn) error {
 	conn.Connect()
 	for {
@@ -397,6 +463,8 @@ func waitForConnReady(ctx context.Context, conn *grpc.ClientConn) error {
 		switch state {
 		case connectivity.Ready:
 			return nil
+		case connectivity.TransientFailure:
+			return errConnFailedFast
 		case connectivity.Shutdown:
 			return fmt.Errorf("connection shut down before becoming ready")
 		}
@@ -423,13 +491,13 @@ func launchSubstreamsPoller(endpoint string, substreamsClientConfig *client.Subs
 }
 
 func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClientConfig, pkg *pbsubstreams.Package, outputStreamName string, blockNum int64, connectTimeout, pollingTimeout time.Duration, maxFreshness *time.Duration) (result *pollResult) {
-	result = &pollResult{}
+	result = &pollResult{maxFreshness: maxFreshness}
 
 	connectBegin := time.Now()
 	conn, connClose, callOpts, headers, err := client.NewSubstreamsClientConn(substreamsClientConfig)
 	if err != nil {
 		result.connectDuration = time.Since(connectBegin)
-		result.reason, result.err = reasonConnect, err
+		result.reason, result.err = reasonInvalidConfig, err
 		return
 	}
 	defer connClose()
@@ -437,10 +505,18 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 	connectCtx, cancelConnect := context.WithTimeoutCause(context.Background(), connectTimeout, fmt.Errorf("connect timeout of %s reached", connectTimeout))
 	defer cancelConnect()
 
+	var connectFailedFast bool
 	if err := waitForConnReady(connectCtx, conn); err != nil {
-		result.connectDuration = time.Since(connectBegin)
-		result.reason, result.err = reasonConnectTimeout, err
-		return
+		if !errors.Is(err, errConnFailedFast) {
+			result.connectDuration = time.Since(connectBegin)
+			result.reason, result.err = reasonConnectTimeout, err
+			return
+		}
+
+		// The dial failed, and only the request will say why: gRPC answers it immediately with
+		// the dial error it is holding ("connection refused", "no such host"), which is exactly
+		// the information an operator needs and which the connectivity API does not expose.
+		connectFailedFast = true
 	}
 	result.connectDuration = time.Since(connectBegin)
 
@@ -472,21 +548,21 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 		return
 	}
 
-	// The connection is already READY, so a failure here is the endpoint refusing us, never
-	// a connection still being established.
+	// Fail fast: the connection is either READY or known-broken by now, so waiting for it here
+	// would only re-do what the connect phase already decided.
 	callOpts = append(callOpts, grpc.WaitForReady(false))
 	zlog.Debug("calling sf.substreams.rpc.v4.Stream/Blocks", zap.String("endpoint", endpoint), zap.String("output_module", outputStreamName), zap.Int64("start_block", blockNum), zap.Uint64("stop_block", stopBlockNum), zap.Duration("connect_duration", result.connectDuration))
 
 	streamClient, err := pbsubstreamsrpcv4.NewStreamClient(conn).Blocks(ctx, subReq, callOpts...)
 	if err != nil {
-		result.reason, result.err = classifyStreamError(err), err
+		result.reason, result.err = streamFailure(ctx, connectFailedFast, err)
 		return
 	}
 
 	for {
 		resp, err := streamClient.Recv()
 		if err != nil {
-			result.reason, result.err = classifyStreamError(err), err
+			result.reason, result.err = streamFailure(ctx, connectFailedFast, err)
 			return
 		}
 
@@ -498,7 +574,14 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 		// Items are ordered by block number ascending, the last one is the freshest, which is
 		// what a HEAD healthcheck cares about.
 		clock := data.BlockScopedDatas.Items[len(data.BlockScopedDatas.Items)-1].Clock
-		result.blockNum = clock.Number
+		if clock == nil {
+			// Nothing recovers a poller, so an unguarded dereference here would take down the
+			// exporter for every other endpoint too.
+			result.reason = reasonInvalidResponse
+			result.err = fmt.Errorf("endpoint returned block data without a clock")
+			return
+		}
+
 		if maxFreshness == nil {
 			zlog.Debug("marking endpoint with success", zap.String("endpoint", endpoint), zap.Uint64("block_num", clock.Number))
 			return
@@ -512,17 +595,6 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 			return
 		}
 
-		// A block age climbing towards the threshold is what precedes a `stale_block`
-		// failure, reporting it here is what makes an alert on `block_age_ms` explainable.
-		if age > *maxFreshness/2 {
-			zlog.Info("endpoint block age is above half of the max freshness",
-				zap.String("endpoint", endpoint),
-				zap.Uint64("block_num", clock.Number),
-				zap.Duration("block_age", age),
-				zap.Duration("max_freshness", *maxFreshness),
-			)
-		}
-
 		zlog.Debug("marking endpoint with success", zap.String("endpoint", endpoint), zap.Uint64("block_num", clock.Number), zap.Duration("block_age", age))
 		return
 	}
diff --git a/tools/prometheus-exporter_test.go b/tools/prometheus-exporter_test.go
index 61a4d97f9..3a6586690 100644
--- a/tools/prometheus-exporter_test.go
+++ b/tools/prometheus-exporter_test.go
@@ -2,6 +2,7 @@ package tools
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -171,3 +172,45 @@ func TestExtractParams(t *testing.T) {
 		})
 	}
 }
+
+func TestTrackBlockAge(t *testing.T) {
+	maxFreshness := 40 * time.Second
+	poll := func(age time.Duration) *pollResult {
+		return &pollResult{blockAge: &age, maxFreshness: &maxFreshness}
+	}
+
+	t.Run("flips only once the streak confirms it", func(t *testing.T) {
+		state := &endpointState{}
+
+		for range blockAgeFlipPolls - 1 {
+			trackBlockAge("endpoint:443", state, poll(30*time.Second))
+			assert.False(t, state.blockAgeAboveHalf, "flipped before the streak completed")
+		}
+
+		trackBlockAge("endpoint:443", state, poll(30*time.Second))
+		assert.True(t, state.blockAgeAboveHalf)
+	})
+
+	t.Run("an age straddling the threshold never flips", func(t *testing.T) {
+		state := &endpointState{}
+
+		// This is the chain whose block interval sits near half of --max-freshness: without the
+		// streak it would report a crossing on every single poll, forever, while healthy.
+		for range 20 {
+			trackBlockAge("endpoint:443", state, poll(25*time.Second))
+			trackBlockAge("endpoint:443", state, poll(15*time.Second))
+		}
+
+		assert.False(t, state.blockAgeAboveHalf)
+	})
+
+	t.Run("no block age is not a crossing", func(t *testing.T) {
+		state := &endpointState{blockAgeAboveHalf: true}
+
+		for range blockAgeFlipPolls + 2 {
+			trackBlockAge("endpoint:443", state, &pollResult{maxFreshness: &maxFreshness})
+		}
+
+		assert.True(t, state.blockAgeAboveHalf)
+	})
+}

From 06d6569b01aa9f5ab5cbb0186c2eb7e7fe482768 Mon Sep 17 00:00:00 2001
From: Matthieu Vachon <matt@streamingfast.io>
Date: Wed, 2 Sep 2026 09:34:47 -0400
Subject: [PATCH 3/8] Drop the prometheus-exporter deploy-order warning

--force-protocol-version is not passed by any production deployment, so
the flag becoming v4-only needs no operator step. Keep the breaking
marker, drop the crash-loop warning.
---
 docs/release-notes/change-log.md | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md
index 89ed7fe7f..b7c8cc180 100644
--- a/docs/release-notes/change-log.md
+++ b/docs/release-notes/change-log.md
@@ -137,9 +137,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - **Breaking** The exporter now speaks `sf.substreams.rpc.v4.Stream/Blocks` only. The v3-to-v2 fallback is gone --
   it closed the connection and then kept reading from it, double-counting the failure -- and
   `--force-protocol-version` accepts only `4` (or `0`), the flag being kept for the protocol versions to come. An
-  invalid value used to be parsed and then silently ignored, it is now rejected at startup, which means a deployment
-  currently passing `--force-protocol-version 2` or `3` will refuse to start: **unset the flag before rolling the
-  new image**, otherwise the exporter crash-loops and the fleet loses its healthcheck metrics entirely.
+  invalid value used to be parsed and then silently ignored, it is now rejected at startup, so an invocation passing
+  `--force-protocol-version 2` or `3` must drop the flag.
 
 ### Dependencies
 

From 769d70a0619e1c0c6dd86a9193253d858cc5965f Mon Sep 17 00:00:00 2001
From: 0xGabey <cartier.gabriel@gmail.com>
Date: Thu, 3 Sep 2026 15:37:46 +0700
Subject: [PATCH 4/8] test(tools): cover waitForConnReady connect budget

TRANSIENT_FAILURE is not terminal: gRPC re-dials on its own backoff,
so an endpoint that refuses one dial and accepts the next (rolling
restart, DNS blip, load balancer with no healthy backend) is reported
unavailable on the first refused SYN, and --connect-timeout is never
spent on the case it exists for.

Both tests fail at 06d6569b. They pass once waitForConnReady waits
through TRANSIENT_FAILURE while recording that a dial failed, so
connect_failed keeps the real dial error and connect_timeout keeps
its meaning.

Refs #916
---
 tools/prometheus_exporter_connready_test.go | 91 +++++++++++++++++++++
 1 file changed, 91 insertions(+)
 create mode 100644 tools/prometheus_exporter_connready_test.go

diff --git a/tools/prometheus_exporter_connready_test.go b/tools/prometheus_exporter_connready_test.go
new file mode 100644
index 000000000..f59d888e4
--- /dev/null
+++ b/tools/prometheus_exporter_connready_test.go
@@ -0,0 +1,91 @@
+package tools
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// reserveAddr returns an address nothing is listening on, so that the first dial to it is
+// refused rather than merely slow.
+func reserveAddr(t *testing.T) string {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	addr := listener.Addr().String()
+	require.NoError(t, listener.Close())
+
+	return addr
+}
+
+// An endpoint that refuses the first dial and accepts a moment later is a rolling restart, not
+// a dead endpoint. The connect budget exists to cover exactly that window, so the poll must
+// succeed as long as the backend comes up inside it.
+func TestWaitForConnReady_RecoversWithinConnectBudget(t *testing.T) {
+	addr := reserveAddr(t)
+
+	const backendDownFor = 500 * time.Millisecond
+	const connectTimeout = 5 * time.Second
+
+	server := grpc.NewServer()
+	t.Cleanup(server.Stop)
+
+	serving := make(chan struct{})
+	go func() {
+		time.Sleep(backendDownFor)
+
+		listener, err := net.Listen("tcp", addr)
+		if err != nil {
+			close(serving)
+			return
+		}
+
+		close(serving)
+		_ = server.Serve(listener)
+	}()
+
+	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+
+	ctx, cancel := context.WithTimeout(context.Background(), connectTimeout)
+	defer cancel()
+
+	begin := time.Now()
+	err = waitForConnReady(ctx, conn)
+	elapsed := time.Since(begin)
+
+	<-serving
+
+	require.NoError(t, err, "endpoint came up after %s, well inside the %s connect budget, but the poll gave up after %s", backendDownFor, connectTimeout, elapsed)
+	require.Greater(t, elapsed, backendDownFor, "returned before the backend was listening, so readiness was never actually established")
+}
+
+// A backend that stays down for longer than the connect budget must still be reported, and the
+// budget must be spent waiting rather than returned immediately.
+func TestWaitForConnReady_GivesUpAfterConnectBudget(t *testing.T) {
+	addr := reserveAddr(t)
+
+	const connectTimeout = 500 * time.Millisecond
+
+	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+
+	ctx, cancel := context.WithTimeoutCause(context.Background(), connectTimeout, context.DeadlineExceeded)
+	defer cancel()
+
+	begin := time.Now()
+	err = waitForConnReady(ctx, conn)
+	elapsed := time.Since(begin)
+
+	require.Error(t, err)
+	require.GreaterOrEqual(t, elapsed, connectTimeout, "gave up after %s without spending the %s connect budget", elapsed, connectTimeout)
+}

From 6686ef61359551996dafc06178e4af7cbed2e944 Mon Sep 17 00:00:00 2001
From: 0xGabey <cartier.gabriel@gmail.com>
Date: Thu, 3 Sep 2026 15:43:03 +0700
Subject: [PATCH 5/8] fix(tools): reset block age streak on polls without an
 age

trackBlockAge returned early without clearing the streak, so a poll
carrying no block age did not break it. Two polls above half, an
outage, then one more above half reported confirmed_over_polls: 3 for
a window that was never consecutive.

Refs #916
---
 tools/prometheus-exporter.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/tools/prometheus-exporter.go b/tools/prometheus-exporter.go
index 7a27e83dc..85c60d2dc 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -397,6 +397,8 @@ const blockAgeFlipPolls = 3
 // signal. Must be called with `lock` held.
 func trackBlockAge(endpoint string, state *endpointState, result *pollResult) {
 	if result.blockAge == nil || result.maxFreshness == nil {
+		// A poll that carries no age agrees with nothing, so a streak cannot span it.
+		state.blockAgeStreak = 0
 		return
 	}
 

From dae4ae6cda62d5380e273a34f0599d9a7802cb5c Mon Sep 17 00:00:00 2001
From: 0xGabey <cartier.gabriel@gmail.com>
Date: Thu, 3 Sep 2026 15:43:03 +0700
Subject: [PATCH 6/8] refactor(tools): drop unreachable protocol version error

The check above it rejects everything but 0 and 4, and both parse, so
the error branch could not be taken.

Refs #916
---
 tools/prometheus-exporter.go | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/tools/prometheus-exporter.go b/tools/prometheus-exporter.go
index 85c60d2dc..de94da8c7 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -209,10 +209,8 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("invalid --force-protocol-version %d: the prometheus exporter only speaks %s for now, leave the flag unset or pass 4", protocolVersionFlag, client.ProtocolVersionV4)
 	}
 
-	forceProtocolVersion, err := client.ParseProtocolVersion(protocolVersionFlag)
-	if err != nil {
-		return fmt.Errorf("invalid --force-protocol-version: %w", err)
-	}
+	// The check above leaves only 0 and 4, both of which parse.
+	forceProtocolVersion, _ := client.ParseProtocolVersion(protocolVersionFlag)
 
 	maxFreshness := sflags.MustGetDuration(cmd, "max-freshness")
 

From 1118bbbbcfb6b1351ffffc21a95baa22c7107b88 Mon Sep 17 00:00:00 2001
From: 0xGabey <cartier.gabriel@gmail.com>
Date: Thu, 3 Sep 2026 15:43:13 +0700
Subject: [PATCH 7/8] docs(tools): correct the connect failure story

The changelog said connect_failed carries the DNS failure. It does
not: the roundrobin balancer replaces the resolver error, so a
hostname that does not resolve surfaces as "no children to pick from"
and no part of the message names DNS.

streamFailure's doc comment broke mid-sentence and documented
connect_timeout, which it never returns.

Refs #916
---
 docs/release-notes/change-log.md | 7 ++++---
 tools/metrics.go                 | 8 +++-----
 tools/prometheus-exporter.go     | 5 ++---
 3 files changed, 9 insertions(+), 11 deletions(-)

diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md
index b7c8cc180..9e681801e 100644
--- a/docs/release-notes/change-log.md
+++ b/docs/release-notes/change-log.md
@@ -105,9 +105,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
   `stream_error`, `stale_block`, `invalid_response` or `no_data` -- exposed on the new
   `substreams_healthcheck_failure_count{reason,grpc_code}` counter and included in the logs. An alert firing on
   `substreams_healthcheck_status` no longer requires guessing whether the endpoint was unreachable, unauthenticated,
-  overloaded or merely late. A dial that fails outright is reported as `connect_failed` within milliseconds, carrying
-  the dial error (`connection refused`, DNS failure); `connect_timeout` is reserved for a connection that is merely
-  slow to come up.
+  overloaded or merely late. A dial that fails outright is reported as `connect_failed`, carrying the dial
+  error where gRPC exposes it (`connection refused`); a hostname that does not resolve surfaces as the balancer's
+  own `no children to pick from`, since it replaces the resolver error. `connect_timeout` is reserved for a
+  connection that is merely slow to come up and never failed a dial.
 
 - Connection establishment gets its own budget, `--connect-timeout` (default 10s), separate from `--timeout`, which
   now covers the `Blocks` request alone. gRPC dials lazily, so DNS, TLS and load-balancer resolution used to be
diff --git a/tools/metrics.go b/tools/metrics.go
index 025bda951..4fd7ebac0 100644
--- a/tools/metrics.go
+++ b/tools/metrics.go
@@ -103,11 +103,9 @@ func grpcCodeOf(err error) string {
 	return noGRPCCode
 }
 
-// streamFailure attributes an error returned by the Blocks call. `connect_failed` means the
-// dial itself failed, `connect_timeout` (set by the caller) means it was merely slow and the
-// backend was never reached at all. When the channel never
-// became ready, gRPC answers with the dial error it was holding, so the failure belongs to
-// the connection rather than to the endpoint's own answer.
+// streamFailure attributes an error returned by the Blocks call. A channel that never became
+// ready makes gRPC answer with the dial error it was holding, so the failure belongs to the
+// connection rather than to the endpoint's own answer, and `connect_failed` says so.
 func streamFailure(ctx context.Context, connectFailedFast bool, err error) (failureReason, error) {
 	err = withDeadlineCause(ctx, err)
 	if connectFailedFast {
diff --git a/tools/prometheus-exporter.go b/tools/prometheus-exporter.go
index de94da8c7..f0dc22328 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -244,9 +244,8 @@ func runPrometheus(cmd *cobra.Command, args []string) error {
 		}
 	}
 
-	// Declared before the pollers start: a poller that fails on its very first attempt reports
-	// through these, and a nil one is both a data race and a nil dereference that takes the
-	// whole process down.
+	// Declared before the pollers start: a poller that fails on its very first attempt already
+	// reports through these, and they are read without synchronisation.
 	collectors := initHealthcheckMetrics(labelNames)
 
 	for endpoint := range endpointMap {

From ebbe729420372d2c3587e0bac21d928e6827f176 Mon Sep 17 00:00:00 2001
From: Matthieu Vachon <matt@streamingfast.io>
Date: Thu, 3 Sep 2026 09:05:10 -0400
Subject: [PATCH 8/8] Spend the connect budget before reporting a failed dial

The connect budget exists to cover a backend that is restarting, so a
refused dial is not the end of the attempt: gRPC re-dials on its own
backoff and an endpoint that comes back inside the budget is healthy.
Returning on the first TRANSIENT_FAILURE reported those as connect_failed
and broke TestWaitForConnReady_RecoversWithinConnectBudget.

waitForConnReady goes back to waiting for the deadline, but remembers
whether a dial failed along the way. That is what still separates the two
timeouts: a channel that failed a dial reports connect_failed and lets the
request surface the actual dial error, while one that never left
CONNECTING reports connect_timeout. Renamed errConnFailedFast to
errConnDialFailed, which is what it now means.
---
 tools/metrics.go             |  4 ++--
 tools/metrics_test.go        |  4 ++--
 tools/prometheus-exporter.go | 35 +++++++++++++++++++++--------------
 3 files changed, 25 insertions(+), 18 deletions(-)

diff --git a/tools/metrics.go b/tools/metrics.go
index 4fd7ebac0..7cebb8e28 100644
--- a/tools/metrics.go
+++ b/tools/metrics.go
@@ -106,9 +106,9 @@ func grpcCodeOf(err error) string {
 // streamFailure attributes an error returned by the Blocks call. A channel that never became
 // ready makes gRPC answer with the dial error it was holding, so the failure belongs to the
 // connection rather than to the endpoint's own answer, and `connect_failed` says so.
-func streamFailure(ctx context.Context, connectFailedFast bool, err error) (failureReason, error) {
+func streamFailure(ctx context.Context, connectFailed bool, err error) (failureReason, error) {
 	err = withDeadlineCause(ctx, err)
-	if connectFailedFast {
+	if connectFailed {
 		return reasonConnectFailed, err
 	}
 	return classifyStreamError(err), err
diff --git a/tools/metrics_test.go b/tools/metrics_test.go
index c711cb02b..dcf3141d4 100644
--- a/tools/metrics_test.go
+++ b/tools/metrics_test.go
@@ -150,8 +150,8 @@ func TestStreamFailure(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	// A dial that failed fast answers through the request, so the gRPC error carries the dial
-	// message and belongs to the connection, not to the endpoint's own answer.
+	// A channel that never became ready answers through the request, so the gRPC error carries
+	// the dial message and belongs to the connection, not to the endpoint's own answer.
 	refused := grpcstatus.Error(codes.Unavailable, `connection error: desc = "transport: Error while dialing: dial tcp 127.0.0.1:19999: connect: connection refused"`)
 	reason, err := streamFailure(ctx, true, refused)
 	assert.Equal(t, reasonConnectFailed, reason)
diff --git a/tools/prometheus-exporter.go b/tools/prometheus-exporter.go
index f0dc22328..a9cbd8de0 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -442,33 +442,40 @@ func (r *pollResult) totalDuration() time.Duration {
 	return r.connectDuration + r.streamDuration
 }
 
-// errConnFailedFast reports that the channel reached TRANSIENT_FAILURE: the dial failed
-// outright rather than being slow. The connectivity API never hands over the dial error, so
-// the caller issues the request anyway and lets gRPC answer with it.
-var errConnFailedFast = errors.New("connection failed to establish")
+// errConnDialFailed reports that the channel failed a dial before the connect budget ran out.
+// The connectivity API never hands over the dial error, so the caller issues the request
+// anyway and lets gRPC answer with it.
+var errConnDialFailed = errors.New("connection failed to establish")
 
 // waitForConnReady blocks until the gRPC channel is usable. gRPC dials lazily, so without
 // this the DNS resolution, the TLS handshake and the load-balancer setup would all be
 // charged to the Blocks request budget, and every slow connection would be reported as an
 // endpoint failure ("waiting for new LB policy update: context deadline exceeded").
 //
-// It only ever waits on a connection that is still making progress. gRPC re-dials on its own
-// backoff, so waiting through TRANSIENT_FAILURE would burn the whole connect budget on an
-// endpoint that answered "connection refused" in a microsecond, and report it as a timeout.
+// A failed dial does not end the attempt. gRPC re-dials on its own backoff and the connect
+// budget exists to cover a backend that is restarting, so an endpoint that comes back inside
+// the budget is healthy and must be reported as such. What the failed dial does change is how
+// the timeout is described: `errConnDialFailed` says the endpoint refused us rather than
+// being slow, which the caller turns into the real dial error.
 func waitForConnReady(ctx context.Context, conn *grpc.ClientConn) error {
 	conn.Connect()
+
+	dialFailed := false
 	for {
 		state := conn.GetState()
 		switch state {
 		case connectivity.Ready:
 			return nil
-		case connectivity.TransientFailure:
-			return errConnFailedFast
 		case connectivity.Shutdown:
 			return fmt.Errorf("connection shut down before becoming ready")
+		case connectivity.TransientFailure:
+			dialFailed = true
 		}
 
 		if !conn.WaitForStateChange(ctx, state) {
+			if dialFailed {
+				return errConnDialFailed
+			}
 			return fmt.Errorf("connection stuck in state %q: %w", state, context.Cause(ctx))
 		}
 	}
@@ -504,9 +511,9 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 	connectCtx, cancelConnect := context.WithTimeoutCause(context.Background(), connectTimeout, fmt.Errorf("connect timeout of %s reached", connectTimeout))
 	defer cancelConnect()
 
-	var connectFailedFast bool
+	var connectFailed bool
 	if err := waitForConnReady(connectCtx, conn); err != nil {
-		if !errors.Is(err, errConnFailedFast) {
+		if !errors.Is(err, errConnDialFailed) {
 			result.connectDuration = time.Since(connectBegin)
 			result.reason, result.err = reasonConnectTimeout, err
 			return
@@ -515,7 +522,7 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 		// The dial failed, and only the request will say why: gRPC answers it immediately with
 		// the dial error it is holding ("connection refused", "no such host"), which is exactly
 		// the information an operator needs and which the connectivity API does not expose.
-		connectFailedFast = true
+		connectFailed = true
 	}
 	result.connectDuration = time.Since(connectBegin)
 
@@ -554,14 +561,14 @@ func pollEndpoint(endpoint string, substreamsClientConfig *client.SubstreamsClie
 
 	streamClient, err := pbsubstreamsrpcv4.NewStreamClient(conn).Blocks(ctx, subReq, callOpts...)
 	if err != nil {
-		result.reason, result.err = streamFailure(ctx, connectFailedFast, err)
+		result.reason, result.err = streamFailure(ctx, connectFailed, err)
 		return
 	}
 
 	for {
 		resp, err := streamClient.Recv()
 		if err != nil {
-			result.reason, result.err = streamFailure(ctx, connectFailedFast, err)
+			result.reason, result.err = streamFailure(ctx, connectFailed, err)
 			return
 		}