Merge branch 'main' into fix-latency-scale

Author: vlfig

metrics/client.go

@@ -39,6 +39,7 @@ const rpcCallLatencyMetricName = "rpc_call_latency"
 type RPCClientMetrics interface {
 	// RecordRequest records latency for an RPC call (observed in nanoseconds for Prometheus and Beholder).
 	// Failures use success="false"; derive error rate from rpc_call_latency_count{success="false"} (or equivalent).
+	// rpcURL is sanitized before export (userinfo and query removed; path hashed).
 	RecordRequest(ctx context.Context, rpcURL string, isSendOnly bool, callName string, latency time.Duration, err error)
 }
 
@@ -77,15 +78,16 @@ func (m *rpcClientMetrics) RecordRequest(ctx context.Context, rpcURL string, isS
 	successStr := strconv.FormatBool(err != nil)
 	sendStr := strconv.FormatBool(isSendOnly)
 	latencyNs := float64(latency)
+	safeRPCURL := SanitizeRPCURL(rpcURL)
 
 	RPCCallLatency.WithLabelValues(
-		m.chainFamily, m.chainID, rpcURL, sendStr, successStr, callName,
+		m.chainFamily, m.chainID, safeRPCURL, sendStr, successStr, callName,
 	).Observe(latencyNs)
 
 	m.latencyHis.Record(ctx, latencyNs/float64(time.Millisecond), metric.WithAttributes(
 		attribute.String("chainFamily", m.chainFamily),
 		attribute.String("chainID", m.chainID),
-		attribute.String("rpcUrl", rpcURL),
+		attribute.String("rpcUrl", safeRPCURL),
 		attribute.String("isSendOnly", sendStr),
 		attribute.String("success", successStr),
 		attribute.String("rpcCallName", callName),

metrics/sanitize_rpc_url.go

@@ -0,0 +1,43 @@
+package metrics
+
+import (
+	"crypto/sha1" //nolint:gosec // sha1 used only for URL anonymisation, not security
+	"fmt"
+	"net/url"
+	"strings"
+)
+
+// SanitizeRPCURL either strips user:passwd or replaces path and params with their sha1-hex, excluding leading / if present
+func SanitizeRPCURL(raw string) string {
+	u, err := url.Parse(raw)
+	if err != nil {
+		return "invalid_rpc_url"
+	}
+
+	if u.User != nil {
+		// Strip credentials and leave everything else intact.
+		u.User = nil
+		return u.String()
+	}
+
+	// Build the sensitive portion: path (without leading /) plus optional query.
+	sensitive := strings.TrimPrefix(u.Path, "/")
+	if u.RawQuery != "" {
+		if sensitive != "" {
+			sensitive += "?" + u.RawQuery
+		} else {
+			sensitive = u.RawQuery
+		}
+	}
+
+	if sensitive == "" {
+		// Nothing to redact.
+		return u.String()
+	}
+
+	//nolint:gosec
+	h := sha1.Sum([]byte(sensitive))
+	u.Path = "/" + fmt.Sprintf("%x", h)
+	u.RawQuery = ""
+	return u.String()
+}

metrics/sanitize_rpc_url_test.go

@@ -0,0 +1,64 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSanitizeRPCURL_RedactsSecrets(t *testing.T) {
+	cases := []struct {
+		input string
+		want  string
+	}{
+		// simple path
+		{"https://bsc-mainnet.core.chainstack.com/MjcwMDk3ZGFhMDA5NjJjMDM1", "https://bsc-mainnet.core.chainstack.com/b0b99e8b33b401b05251f91aec08b6a9581c86dd"},
+		// less simple path
+		{"http://172.16.156.14:8000/MmZmNTJmOWRiNzg0NTgxNDYyNzJjMTYzNDlmNGJ/iYWEwOTVmYWE0OQ/bsc/mainnet/", "http://172.16.156.14:8000/bd161d616d46b90a248de0f0a3ebc2daf2b8bb20"},
+		// path with no / is excluded from sha
+		{"https://anyblocks-01.mainnet.bnb.bdnodes.net?auth=MDcwMTgzODk3NzIyMjU4YzY2MTQzNGMyNTU2OWE2NGEzYjhlODM0NA", "https://anyblocks-01.mainnet.bnb.bdnodes.net/7c87697e63d8f9c049183bb4f8c171af40715b2b"},
+		// path with leading / is included in sha
+		{"https://anyblocks-02.mainnet.bnb.bdnodes.net/somepath/?auth=2Dc8bNAqCC0X74zZfi_4ra6XzuBY8lmXcTE1ic9EO5o", "https://anyblocks-02.mainnet.bnb.bdnodes.net/22c028ea2d53fd106e2bb93bc61d838ed6b01c19"},
+		// strip creds keep path
+		{"https://myLittleNop:YjY5MjAwOGJkMzBjNW@broadcast-mirror.fiews.io/?chain_id=56", "https://broadcast-mirror.fiews.io/?chain_id=56"},
+		// even if no creds, sacrifice path for uniformity
+		{"https://eu-bsc.rpc.linkriver.internal/rpc", "https://eu-bsc.rpc.linkriver.internal/e64b40f2bd5c8a9560773d16476a86ede7e7c1ba"},
+		// keeps protocol too
+		{"wss://bsc-mainnet-proxy.internal.linkpool.io/ws", "wss://bsc-mainnet-proxy.internal.linkpool.io/1457b75dc8c5500c0f1d4503cf801b60deb045a4"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.input, func(t *testing.T) {
+			assert.Equal(t, tc.want, SanitizeRPCURL(tc.input))
+		})
+	}
+}
+
+func TestSanitizeRPCURL_AlreadySanitized(t *testing.T) {
+	urls := []string{
+		"http://10.0.1.191:8545",
+		"http://144.178.241.22:8545",
+		"http://222.106.187.14:12001",
+		"http://at2-bsc-main03.blockchain.fiews.net:8545",
+		"http://berlioz.stakesystems.io:8745",
+		"http://blockchains-1.shultzpro.com:8545",
+		"http://dfw3-bsc-main01.blockchain.fiews.net:8545",
+		"http://sylvester.stakesystems.io:8745",
+		"https://bsc-dataseed.binance.org/",
+		"https://chainlink-bsc.rpc.blxrbdn.com",
+		"https://puissant-builder.48.club",
+		"ws://10.0.1.191:8546",
+		"ws://144.76.108.206:8546",
+		"ws://172.16.152.140:8546",
+		"ws://bsc-rpc-2.piertwo.prod:8546",
+		"ws://bsc.rpc.cinternal.com",
+		"ws://sylvester.stakesystems.io:8746",
+		"wss://bsc-rpc.o1.wtf",
+	}
+
+	for _, u := range urls {
+		t.Run(u, func(t *testing.T) {
+			assert.Equal(t, u, SanitizeRPCURL(u))
+		})
+	}
+}

multinode/README.md

@@ -19,4 +19,32 @@ Manages all nodes performing node selection and load balancing, health checks an
 Used to poll for new heads and finalized heads within subscriptions.
 
 ### Transaction Sender
-Used to send transactions to all healthy RPCs and aggregate the results.
+Used to send transactions to all healthy RPCs and aggregate the results.
+
+## States diagram
+
+```mermaid
+graph TD
+	Undialed     --> Dialed
+	Undialed     --> Unreachable
+	Dialed       --> Alive
+	Dialed       --> InvalidChainID
+	Dialed       --> Syncing
+	Dialed       --> Unreachable
+	Alive        --> OutOfSync
+	Alive        --> Unreachable
+	OutOfSync    --> Alive
+	OutOfSync    --> InvalidChainID
+	OutOfSync    --> Syncing
+	OutOfSync    --> Unreachable
+	InvalidChainID --> Alive
+	InvalidChainID --> Syncing
+	InvalidChainID --> Unreachable
+	Syncing      --> Alive
+	Syncing      --> OutOfSync
+	Syncing      --> InvalidChainID
+	Syncing      --> Unreachable
+	Unreachable  --> Dialed
+	Unusable:::terminal
+	Closed:::terminal
+```

multinode/config/config.go

@@ -44,6 +44,10 @@ func (c *MultiNodeConfig) PollFailureThreshold() uint32 {
 	return *c.MultiNode.PollFailureThreshold
 }
 
+func (c *MultiNodeConfig) PollSuccessThreshold() uint32 {
+	return 0 // retaining source compat for -solana; -evm sets via NodePoolConfig
+}
+
 func (c *MultiNodeConfig) PollInterval() time.Duration {
 	return c.MultiNode.PollInterval.Duration()
 }

multinode/ctx_test.go

@@ -4,12 +4,10 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-
-	"github.com/smartcontractkit/chainlink-common/pkg/utils/tests"
 )
 
 func TestContext(t *testing.T) {
-	ctx := tests.Context(t)
+	ctx := t.Context()
 	assert.False(t, CtxIsHealthCheckRequest(ctx), "expected false for test context")
 	ctx = CtxAddHealthCheckFlag(ctx)
 	assert.True(t, CtxIsHealthCheckRequest(ctx), "expected context to contain the healthcheck flag")

multinode/multi_node_test.go

@@ -81,7 +81,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			selectionMode: NodeSelectionModeRoundRobin,
 			chainID:       RandomID(),
 		})
-		err := mn.Start(tests.Context(t))
+		err := mn.Start(t.Context())
 		assert.ErrorContains(t, err, fmt.Sprintf("no available nodes for chain %s", mn.chainID))
 	})
 	t.Run("Fails with wrong node's chainID", func(t *testing.T) {
@@ -97,7 +97,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			chainID:       multiNodeChainID,
 			nodes:         []Node[ID, multiNodeRPCClient]{node},
 		})
-		err := mn.Start(tests.Context(t))
+		err := mn.Start(t.Context())
 		assert.ErrorContains(t, err, fmt.Sprintf("node %s has configured chain ID %s which does not match multinode configured chain ID of %s", nodeName, nodeChainID, mn.chainID))
 	})
 	t.Run("Fails if node fails", func(t *testing.T) {
@@ -113,7 +113,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			chainID:       chainID,
 			nodes:         []Node[ID, multiNodeRPCClient]{node},
 		})
-		err := mn.Start(tests.Context(t))
+		err := mn.Start(t.Context())
 		assert.ErrorIs(t, err, expectedError)
 	})
 
@@ -132,7 +132,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			chainID:       chainID,
 			nodes:         []Node[ID, multiNodeRPCClient]{node1, node2},
 		})
-		err := mn.Start(tests.Context(t))
+		err := mn.Start(t.Context())
 		assert.ErrorIs(t, err, expectedError)
 	})
 	t.Run("Fails with wrong send only node's chainID", func(t *testing.T) {
@@ -151,7 +151,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			nodes:         []Node[ID, multiNodeRPCClient]{node},
 			sendonlys:     []SendOnlyNode[ID, multiNodeRPCClient]{sendOnly},
 		})
-		err := mn.Start(tests.Context(t))
+		err := mn.Start(t.Context())
 		assert.ErrorContains(t, err, fmt.Sprintf("sendonly node %s has configured chain ID %s which does not match multinode configured chain ID of %s", sendOnlyName, sendOnlyChainID, mn.chainID))
 	})
 
@@ -178,7 +178,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			nodes:         []Node[ID, multiNodeRPCClient]{node},
 			sendonlys:     []SendOnlyNode[ID, multiNodeRPCClient]{sendOnly1, sendOnly2},
 		})
-		err := mn.Start(tests.Context(t))
+		err := mn.Start(t.Context())
 		assert.ErrorIs(t, err, expectedError)
 	})
 	t.Run("Starts successfully with healthy nodes", func(t *testing.T) {
@@ -192,7 +192,7 @@ func TestMultiNode_Dial(t *testing.T) {
 			sendonlys:     []SendOnlyNode[ID, multiNodeRPCClient]{newHealthySendOnly(t, chainID)},
 		})
 		servicetest.Run(t, mn)
-		selectedNode, err := mn.selectNode(tests.Context(t))
+		selectedNode, err := mn.selectNode(t.Context())
 		require.NoError(t, err)
 		assert.Equal(t, node, selectedNode)
 	})
@@ -336,7 +336,7 @@ func TestMultiNode_selectNode(t *testing.T) {
 	t.Parallel()
 	t.Run("Returns same node, if it's still healthy", func(t *testing.T) {
 		t.Parallel()
-		ctx := tests.Context(t)
+		ctx := t.Context()
 		chainID := RandomID()
 		node1 := newMockNode[ID, multiNodeRPCClient](t)
 		node1.On("State").Return(nodeStateAlive).Once()
@@ -360,7 +360,7 @@ func TestMultiNode_selectNode(t *testing.T) {
 	})
 	t.Run("Updates node if active is not healthy", func(t *testing.T) {
 		t.Parallel()
-		ctx := tests.Context(t)
+		ctx := t.Context()
 		chainID := RandomID()
 		oldBest := newMockNode[ID, multiNodeRPCClient](t)
 		oldBest.On("String").Return("oldBest").Maybe()
@@ -387,7 +387,7 @@ func TestMultiNode_selectNode(t *testing.T) {
 	})
 	t.Run("No active nodes - reports critical error", func(t *testing.T) {
 		t.Parallel()
-		ctx := tests.Context(t)
+		ctx := t.Context()
 		chainID := RandomID()
 		lggr, observedLogs := logger.TestObserved(t, zap.InfoLevel)
 		mn := newTestMultiNode(t, multiNodeOpts{

multinode/node.go

@@ -18,6 +18,7 @@ var errInvalidChainID = errors.New("invalid chain id")
 
 type NodeConfig interface {
 	PollFailureThreshold() uint32
+	PollSuccessThreshold() uint32
 	PollInterval() time.Duration
 	SelectionMode() string
 	SyncThreshold() uint32

multinode/node_lifecycle.go

@@ -123,7 +123,10 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() {
 				lggr.Debugw("Ping successful", "nodeState", n.State())
 				n.metrics.RecordNodeClientVersion(ctx, n.name, version)
 				n.metrics.IncrementPollsSuccess(ctx, n.name)
-				pollFailures = 0
+				// Decay rather than reset; detects sustained failure rates above 1:1
+				if pollFailures > 0 {
+					pollFailures--
+				}
 			}
 			if pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold {
 				lggr.Errorw(fmt.Sprintf("RPC endpoint failed to respond to %d consecutive polls", pollFailures), "pollFailures", pollFailures, "nodeState", n.getCachedState())
@@ -356,7 +359,13 @@ func (n *node[CHAIN_ID, HEAD, RPC]) isOutOfSyncWithPool() (outOfSync bool, liveN
 	}
 
 	if outOfSync && n.getCachedState() == nodeStateAlive {
-		n.lfcLog.Errorw("RPC endpoint has fallen behind", "blockNumber", localChainInfo.BlockNumber, "bestLatestBlockNumber", ci.BlockNumber, "totalDifficulty", localChainInfo.TotalDifficulty)
+		n.lfcLog.Errorw(
+			"RPC endpoint has fallen behind",
+			"blockNumber", localChainInfo.BlockNumber,
+			"bestLatestBlockNumber", ci.BlockNumber,
+			"totalDifficulty", localChainInfo.TotalDifficulty,
+			"blockDifference", localChainInfo.BlockNumber-ci.BlockNumber,
+		)
 	}
 	return outOfSync, ln
 }
@@ -518,6 +527,39 @@ func (n *node[CHAIN_ID, HEAD, RPC]) outOfSyncLoop(syncIssues syncStatus) {
 	}
 }
 
+// probeUntilStable polls the node PollSuccessThreshold consecutive times before allowing it back into
+// the alive pool. Returns true if all probes pass, false if any probe fails or ctx is cancelled.
+// When threshold is 0 the probe is disabled and the function returns true immediately.
+func (n *node[CHAIN_ID, HEAD, RPC]) probeUntilStable(ctx context.Context, lggr logger.Logger) bool {
+	threshold := n.nodePoolCfg.PollSuccessThreshold()
+	pollInterval := n.nodePoolCfg.PollInterval()
+	if threshold == 0 || pollInterval <= 0 {
+		return true
+	}
+	var successes uint32
+	for successes < threshold {
+		select {
+		case <-ctx.Done():
+			return false
+		case <-time.After(pollInterval):
+		}
+		n.metrics.IncrementPolls(ctx, n.name)
+		pollCtx, cancel := context.WithTimeout(ctx, pollInterval)
+		version, err := n.RPC().ClientVersion(pollCtx)
+		cancel()
+		if err != nil {
+			n.metrics.IncrementPollsFailed(ctx, n.name)
+			lggr.Warnw("Recovery probe poll failed; restarting redial", "err", err, "successesSoFar", successes, "threshold", threshold)
+			return false
+		}
+		n.metrics.IncrementPollsSuccess(ctx, n.name)
+		n.metrics.RecordNodeClientVersion(ctx, n.name, version)
+		successes++
+		lggr.Debugw("Recovery probe poll succeeded", "successes", successes, "threshold", threshold)
+	}
+	return true
+}
+
 func (n *node[CHAIN_ID, HEAD, RPC]) unreachableLoop() {
 	defer n.wg.Done()
 	ctx, cancel := n.newCtx()
@@ -563,6 +605,11 @@ func (n *node[CHAIN_ID, HEAD, RPC]) unreachableLoop() {
 				n.setState(nodeStateUnreachable)
 				continue
 			case nodeStateAlive:
+				if !n.probeUntilStable(ctx, lggr) {
+					n.rpc.Close()
+					n.setState(nodeStateUnreachable)
+					continue
+				}
 				lggr.Infow(fmt.Sprintf("Successfully redialled and verified RPC node %s. Node was offline for %s", n.String(), time.Since(unreachableAt)), "nodeState", n.getCachedState())
 				fallthrough
 			default: