diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md
index cace098f6..9e681801e 100644
--- a/docs/release-notes/change-log.md
+++ b/docs/release-notes/change-log.md
@@ -98,6 +98,49 @@ 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` -- `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`, 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
+  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 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.
+
+- `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.
+
+- **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, so an invocation passing
+  `--force-protocol-version 2` or `3` must drop the flag.
+
 ### 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..7cebb8e28
--- /dev/null
+++ b/tools/metrics.go
@@ -0,0 +1,141 @@
+package tools
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"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 (
+	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.
+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
+}
+
+// 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, connectFailed bool, err error) (failureReason, error) {
+	err = withDeadlineCause(ctx, err)
+	if connectFailed {
+		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.
+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..dcf3141d4
--- /dev/null
+++ b/tools/metrics_test.go
@@ -0,0 +1,165 @@
+package tools
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"slices"
+	"testing"
+	"time"
+
+	"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) })
+	}
+}
+
+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 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)
+	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 63790ad37..a9cbd8de0 100644
--- a/tools/prometheus-exporter.go
+++ b/tools/prometheus-exporter.go
@@ -2,19 +2,22 @@ package tools
 
 import (
 	"context"
+	"errors"
 	"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 +27,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 +53,49 @@ 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
+	blockAgeAboveHalf   bool
+	blockAgeStreak      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 +123,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 +198,30 @@ 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")
+
+	// 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")
-	forceProtocolVersion, err := client.ParseProtocolVersion(protocolVersionFlag)
+	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)
+	}
+
+	// The check above leaves only 0 and 4, both of which parse.
+	forceProtocolVersion, _ := client.ParseProtocolVersion(protocolVersionFlag)
 
 	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 +229,24 @@ 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))
+
+	for _, endpoint := range parsed {
+		endpointMap[endpoint.url] = endpointSpecs{
+			startBlock:  endpoint.startBlock,
+			labelValues: endpointLabelValues(endpoint.url, endpoint.params, labelNames),
+		}
 	}
 
-	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)
+	// 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 {
 		startBlock := blockNum
@@ -225,13 +268,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(collectors...)
 
 	handler := promhttp.HandlerFor(
 		promReg,
@@ -247,170 +290,318 @@ 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
+	trackBlockAge(endpoint, state, result)
+
+	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 {
+		// 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)),
+		)
+		zlog.Info("endpoint poll failed", fields...)
 	}
-	status.With(endpointMap[endpoint].labels).Set(0)
-	requestDurationMs.With(endpointMap[endpoint].labels).Set(float64(time.Since(begin).Milliseconds()))
+
+	trackBlockAge(endpoint, state, result)
+
+	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...)
+	}
+}
+
+// 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 {
+		// A poll that carries no age agrees with nothing, so a streak cannot span it.
+		state.blockAgeStreak = 0
+		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
+	maxFreshness    *time.Duration
+	reason          failureReason
+	err             error
 }
 
-func launchSubstreamsPoller(endpoint string, substreamsClientConfig *client.SubstreamsClientConfig, pkg *pbsubstreams.Package, outputStreamName string, blockNum int64, pollingInterval, pollingTimeout time.Duration, maxFreshness *time.Duration) {
+func (r *pollResult) totalDuration() time.Duration {
+	return r.connectDuration + r.streamDuration
+}
+
+// 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").
+//
+// 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.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))
+		}
+	}
+}
 
+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{maxFreshness: maxFreshness}
 
-		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 = reasonInvalidConfig, err
+		return
+	}
+	defer connClose()
 
-		var stopBlockNum uint64
-		if blockNum > 0 {
-			stopBlockNum = uint64(blockNum + 1)
+	connectCtx, cancelConnect := context.WithTimeoutCause(context.Background(), connectTimeout, fmt.Errorf("connect timeout of %s reached", connectTimeout))
+	defer cancelConnect()
+
+	var connectFailed bool
+	if err := waitForConnReady(connectCtx, conn); err != nil {
+		if !errors.Is(err, errConnDialFailed) {
+			result.connectDuration = time.Since(connectBegin)
+			result.reason, result.err = reasonConnectTimeout, err
+			return
 		}
-		subReq := &pbsubstreamsrpcv3.Request{
-			StartBlockNum: blockNum,
-			StopBlockNum:  stopBlockNum,
-			Package:       pkg,
-			OutputModule:  outputStreamName,
+
+		// 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.
+		connectFailed = true
+	}
+	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
+	}
+
+	// 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 = streamFailure(ctx, connectFailed, err)
+		return
+	}
+
+	for {
+		resp, err := streamClient.Recv()
+		if err != nil {
+			result.reason, result.err = streamFailure(ctx, connectFailed, 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
+		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
 		}
 
-	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
-			}
+		if maxFreshness == nil {
+			zlog.Debug("marking endpoint with success", zap.String("endpoint", endpoint), zap.Uint64("block_num", clock.Number))
+			return
+		}
+
+		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()
+		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..3a6586690 100644
--- a/tools/prometheus-exporter_test.go
+++ b/tools/prometheus-exporter_test.go
@@ -2,9 +2,10 @@ package tools
 
 import (
 	"testing"
+	"time"
 
-	"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 {
@@ -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)
+	})
+}
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)
+}