DF-23489 multinode: make PollFailure hysteretic

Successful polls now decrement the failure count, don't fully reset it.
So that nodes (RPCs) are eventually declared unreachable if they sustain poll error rates above 1:1.

Author: vlfig

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
 }

multinode/node_lifecycle_test.go

@@ -110,7 +110,7 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
 		tests.AssertLogEventually(t, observedLogs, "Polling disabled")
 		assert.Equal(t, nodeStateAlive, node.State())
 	})
-	t.Run("stays alive while below pollFailureThreshold and resets counter on success", func(t *testing.T) {
+	t.Run("stays alive while below pollFailureThreshold, success decrements failure count", func(t *testing.T) {
 		t.Parallel()
 		rpc := newMockRPCClient[ID, Head](t)
 		rpc.On("GetInterceptedChainInfo").Return(ChainInfo{}, ChainInfo{})
@@ -132,9 +132,9 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
 			// stays healthy while below threshold
 			assert.Equal(t, nodeStateAlive, node.State())
 		}).Times(pollFailureThreshold - 1)
-		// 2. Successful call that is expected to reset counter
+		// 2. Successful call that is expected to decrement the counter (counter: 2 → 1)
 		rpc.On("ClientVersion", mock.Anything).Return("", nil).Once()
-		// 3. Return error. If we have not reset the timer, we'll transition to nonAliveState
+		// 3. Return error. Counter was decremented (not reset), so it reaches 2 — still below threshold.
 		rpc.On("ClientVersion", mock.Anything).Return("", pollError).Once()
 		// 4. Once during the call, check if node is alive
 		var ensuredAlive atomic.Bool
@@ -176,6 +176,37 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
 			return nodeStateUnreachable == node.State()
 		})
 	})
+	t.Run("transitions to unreachable when net poll failures accumulate despite intermittent successes", func(t *testing.T) {
+		t.Parallel()
+		rpc := newMockRPCClient[ID, Head](t)
+		rpc.On("GetInterceptedChainInfo").Return(ChainInfo{}, ChainInfo{})
+		const pollFailureThreshold = 3
+		node := newSubscribedNode(t, testNodeOpts{
+			config: testNodeConfig{
+				pollFailureThreshold: pollFailureThreshold,
+				pollInterval:         tests.TestInterval,
+			},
+			rpc: rpc,
+		})
+		defer func() { assert.NoError(t, node.close()) }()
+
+		pollError := errors.New("failed to get ClientVersion")
+		// Pattern F·F·S·F·F: with the decay counter the net failure debt reaches
+		// threshold=3 at the 5th poll (counter: 1→2→1→2→3). With the old
+		// reset-on-success behaviour the counter resets to 0 at S and peaks at only
+		// 2 before the next success, never tripping.
+		rpc.On("ClientVersion", mock.Anything).Return("", pollError).Times(2)
+		rpc.On("ClientVersion", mock.Anything).Return("", nil).Once()
+		rpc.On("ClientVersion", mock.Anything).Return("", pollError).Times(2)
+		// Unlimited successes after: ensures old code stays alive indefinitely so
+		// the test correctly fails (times out) when run against the old behaviour.
+		rpc.On("ClientVersion", mock.Anything).Return("", nil)
+		rpc.On("Dial", mock.Anything).Return(errors.New("failed to dial")).Maybe()
+		node.declareAlive()
+		tests.AssertEventually(t, func() bool {
+			return node.State() == nodeStateUnreachable
+		})
+	})
 	t.Run("with threshold poll failures, but we are the last node alive, forcibly keeps it alive", func(t *testing.T) {
 		t.Parallel()
 		rpc := newMockRPCClient[ID, Head](t)