PLEX-3543: Adjust MultiNode RPC health metric

multinode/multi_node.go

@@ -361,7 +361,8 @@ func (c *MultiNode[CHAIN_ID, RPC]) report(nodesStateInfo []nodeWithState) {
 	counts := make(map[nodeState]int)
 	for i, n := range c.primaryNodes {
 		state := n.State()
-		counts[state]++
+		// Routing can retain the last unhealthy node for optimistic retries.
+		counts[n.StateForMetrics()]++
 		nodesStateInfo[i].State = state.String()
 		if state == nodeStateAlive {
 			nodesStateInfo[i].DeadSince = nil

multinode/multi_node_test.go

@@ -61,6 +61,7 @@ func newHealthyNode(t *testing.T, chainID ID) *mockNode[ID, multiNodeRPCClient]
 func newNodeWithState(t *testing.T, chainID ID, state nodeState) *mockNode[ID, multiNodeRPCClient] {
 	node := newStatelessNode(t, chainID)
 	node.On("State").Return(state).Maybe()
+	node.On("StateForMetrics").Return(state).Maybe()
 	return node
 }
 
@@ -83,6 +84,7 @@ func newNodeWithOrder(t *testing.T, chainID ID, order int32) (*mockNode[ID, mult
 	node.On("UnsubscribeAllExceptAliveLoop").Maybe()
 	var alive atomic.Bool
 	alive.Store(true)
+	node.On("StateForMetrics").Return(func() nodeState { return node.State() }).Maybe()
 	node.On("State").Return(func() nodeState {
 		if alive.Load() {
 			return nodeStateAlive
@@ -653,3 +655,38 @@ func TestMultiNode_ChainInfo(t *testing.T) {
 		})
 	}
 }
+
+func TestMultiNode_ReportUsesHealthStateForMetrics(t *testing.T) {
+	t.Parallel()
+	node := newMockNode[ID, multiNodeRPCClient](t)
+	node.On("State").Return(nodeStateAlive)
+	var healthy bool
+	node.On("StateForMetrics").Return(func() nodeState {
+		if healthy {
+			return nodeStateAlive
+		}
+		return nodeStateOutOfSync
+	})
+	mn := newTestMultiNode(t, multiNodeOpts{
+		chainID: RandomID(), selectionMode: NodeSelectionModeRoundRobin,
+		nodes: []Node[ID, multiNodeRPCClient]{node},
+	})
+	info := make([]nodeWithState, 1)
+	for _, recovered := range []bool{false, true} {
+		healthy = recovered
+		metrics := newMockMultiNodeMetrics(t)
+		for _, state := range allNodeStates {
+			var count int64
+			if (!healthy && state == nodeStateOutOfSync) || (healthy && state == nodeStateAlive) {
+				count = 1
+			}
+			metrics.On("RecordNodeStates", mock.Anything, state.String(), count).Once()
+		}
+		mn.metrics = metrics
+		mn.report(info)
+		// The retained node is still available for optimistic routing.
+		assert.Equal(t, "Alive", info[0].State)
+		assert.Nil(t, info[0].DeadSince)
+		metrics.AssertExpectations(t)
+	}
+}

multinode/node.go

@@ -76,10 +76,15 @@ type Node[
 	CHAIN_ID ID,
 	RPC any,
 ] interface {
-	// State returns most accurate state of the Node on the moment of call.
-	// While some of the checks may be performed in the background and State may return cached value, critical, like
-	// `FinalizedBlockOutOfSync`, must be executed upon every call.
+	// State returns the node's current routing state.
+	// If the last remaining Alive node fails soft health checks, it may be forcibly
+	// kept Alive for optimistic retries. State then returns Alive, while
+	// StateForMetrics reports its unhealthy state.
+	// Background checks may return cached results; critical checks such as
+	// FinalizedBlockOutOfSync are evaluated on every call.
 	State() nodeState
+	// StateForMetrics reports health even when an unhealthy node remains selectable for retries.
+	StateForMetrics() nodeState
 	// StateAndLatest returns nodeState with the latest ChainInfo observed by Node during current lifecycle.
 	StateAndLatest() (nodeState, ChainInfo)
 	// HighestUserObservations - returns highest ChainInfo ever observed by underlying RPC excluding results of health check requests
@@ -124,8 +129,10 @@ type node[
 	finalizedStateChecker FinalizedStateChecker // set in NewNode when rpc implements FinalizedStateChecker
 	isLoadBalancedRPC     bool
 
-	stateMu sync.RWMutex // protects state* fields
-	state   nodeState
+	stateMu             sync.RWMutex // protects state* fields
+	state               nodeState
+	stateHealthFailures healthCheckFailures
+	stateSyncIssues     syncStatus
 
 	poolInfoProvider PoolChainInfoProvider
 

multinode/node_fsm.go

@@ -98,6 +98,55 @@ func (n *node[CHAIN_ID, HEAD, RPC]) State() nodeState {
 	return n.recalculateState()
 }
 
+// Health checks are independent: recovery of one must not hide another failure.
+type healthCheckFailures uint8
+
+const (
+	healthCheckPoll healthCheckFailures = 1 << iota
+	healthCheckFinalizedState
+)
+
+func (n *node[CHAIN_ID, HEAD, RPC]) modifyHealthCheckFailure(check healthCheckFailures, failed bool) {
+	n.stateMu.Lock()
+	defer n.stateMu.Unlock()
+	if failed {
+		n.stateHealthFailures |= check
+	} else {
+		n.stateHealthFailures &^= check
+	}
+}
+
+func (n *node[CHAIN_ID, HEAD, RPC]) modifySyncIssue(issue syncStatus, failed bool) {
+	n.stateMu.Lock()
+	defer n.stateMu.Unlock()
+	if failed {
+		n.stateSyncIssues |= issue
+	} else {
+		n.stateSyncIssues &^= issue
+	}
+}
+
+// StateForMetrics preserves routing state while exposing failures suppressed by last-node fallback.
+func (n *node[CHAIN_ID, HEAD, RPC]) StateForMetrics() nodeState {
+	n.stateMu.RLock()
+	defer n.stateMu.RUnlock()
+	state := n.recalculateState()
+	if state != nodeStateAlive {
+		return state
+	}
+	// An endpoint contributes to one state; prefer transport failure over synchronization failures.
+	switch {
+	case n.stateHealthFailures&healthCheckPoll != 0:
+		return nodeStateUnreachable
+	case n.stateHealthFailures&healthCheckFinalizedState != 0:
+		return nodeStateFinalizedStateNotAvailable
+	case n.stateSyncIssues != syncStatusSynced:
+		return nodeStateOutOfSync
+	default:
+		return state
+	}
+}
+
 func (n *node[CHAIN_ID, HEAD, RPC]) getCachedState() nodeState {
 	n.stateMu.RLock()
 	defer n.stateMu.RUnlock()
@@ -187,24 +236,26 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToAlive(fn func()) {
 	}
 	switch n.state {
 	case nodeStateDialed, nodeStateInvalidChainID, nodeStateSyncing, nodeStateFinalizedStateNotAvailable:
+		n.stateHealthFailures = 0
+		n.stateSyncIssues = syncStatusSynced
 		n.state = nodeStateAlive
 	default:
 		panic(transitionFail(n.state, nodeStateAlive))
 	}
 	fn()
 }
 
-// declareInSync puts a node back into Alive state, allowing it to be used by
-// pool consumers again
-func (n *node[CHAIN_ID, HEAD, RPC]) declareInSync() {
-	n.transitionToInSync(func() {
+// declareInSync returns a node to the live pool, retaining any remaining
+// synchronization issues for metric reporting during forced revival.
+func (n *node[CHAIN_ID, HEAD, RPC]) declareInSync(syncIssues syncStatus) {
+	n.transitionToInSync(syncIssues, func() {
 		n.lfcLog.Infow("RPC Node is back in sync", "nodeState", n.state)
 		n.wg.Add(1)
 		go n.aliveLoop()
 	})
 }
 
-func (n *node[CHAIN_ID, HEAD, RPC]) transitionToInSync(fn func()) {
+func (n *node[CHAIN_ID, HEAD, RPC]) transitionToInSync(syncIssues syncStatus, fn func()) {
 	ctx, cancel := n.stopCh.NewCtx()
 	defer cancel()
 	n.metrics.IncrementNodeTransitionsToAlive(ctx, n.name)
@@ -216,6 +267,9 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToInSync(fn func()) {
 	}
 	switch n.state {
 	case nodeStateOutOfSync, nodeStateSyncing:
+		// The new aliveLoop starts fresh polling and finalized-state failure counters.
+		n.stateHealthFailures = 0
+		n.stateSyncIssues = syncIssues
 		n.state = nodeStateAlive
 	default:
 		panic(transitionFail(n.state, nodeStateAlive))

multinode/node_fsm_test.go

@@ -45,7 +45,9 @@ func TestUnit_Node_StateTransitions(t *testing.T) {
 		const destinationState = nodeStateAlive
 		allowedStates := []nodeState{nodeStateOutOfSync, nodeStateSyncing}
 		rpc := newMockRPCClient[ID, Head](t)
-		testTransition(t, rpc, testNode.transitionToInSync, destinationState, allowedStates...)
+		testTransition(t, rpc, func(node testNode, fn func()) {
+			node.transitionToInSync(syncStatusSynced, fn)
+		}, destinationState, allowedStates...)
 	})
 	t.Run("transitionToOutOfSync", func(t *testing.T) {
 		const destinationState = nodeStateOutOfSync
@@ -134,3 +136,67 @@ func TestNodeState_String(t *testing.T) {
 		}
 	})
 }
+
+func TestUnit_Node_StateForMetrics(t *testing.T) {
+	t.Parallel()
+	n := newTestNode(t, testNodeOpts{})
+	n.setState(nodeStateAlive)
+	for _, tc := range []struct {
+		check healthCheckFailures
+		want  nodeState
+	}{
+		{healthCheckPoll, nodeStateUnreachable},
+		{healthCheckFinalizedState, nodeStateFinalizedStateNotAvailable},
+	} {
+		n.modifyHealthCheckFailure(tc.check, true)
+		assert.Equal(t, tc.want, n.StateForMetrics())
+		assert.Equal(t, nodeStateAlive, n.State())
+		n.modifyHealthCheckFailure(tc.check, false)
+		assert.Equal(t, nodeStateAlive, n.StateForMetrics())
+	}
+	for _, issue := range []syncStatus{syncStatusNoNewHead, syncStatusNoNewFinalizedHead, syncStatusNotInSyncWithPool} {
+		n.modifySyncIssue(issue, true)
+		assert.Equal(t, nodeStateOutOfSync, n.StateForMetrics())
+		assert.Equal(t, nodeStateAlive, n.State())
+		n.modifySyncIssue(issue, false)
+		assert.Equal(t, nodeStateAlive, n.StateForMetrics())
+	}
+	n.modifySyncIssue(syncStatusNoNewHead|syncStatusNoNewFinalizedHead, true)
+	n.modifySyncIssue(syncStatusNoNewHead, false)
+	assert.Equal(t, nodeStateOutOfSync, n.StateForMetrics())
+	n.modifyHealthCheckFailure(healthCheckPoll, true)
+	n.modifyHealthCheckFailure(healthCheckPoll, false)
+	assert.Equal(t, nodeStateOutOfSync, n.StateForMetrics())
+	n.modifySyncIssue(syncStatusNoNewHead|syncStatusNoNewFinalizedHead|syncStatusNotInSyncWithPool, false)
+	assert.Equal(t, nodeStateAlive, n.StateForMetrics())
+	n.modifyHealthCheckFailure(healthCheckPoll, true)
+	n.modifySyncIssue(syncStatusNoNewHead, true)
+	n.modifySyncIssue(syncStatusNoNewHead|syncStatusNoNewFinalizedHead|syncStatusNotInSyncWithPool, false)
+	assert.Equal(t, nodeStateUnreachable, n.StateForMetrics())
+	n.setState(nodeStateClosed)
+	assert.Equal(t, nodeStateClosed, n.StateForMetrics())
+}
+
+func TestUnit_Node_TransitionToInSyncHealth(t *testing.T) {
+	t.Parallel()
+	for _, issues := range []syncStatus{syncStatusSynced, syncStatusNoNewHead | syncStatusNotInSyncWithPool} {
+		t.Run(issues.String(), func(t *testing.T) {
+			n := newTestNode(t, testNodeOpts{})
+			n.setState(nodeStateOutOfSync)
+			n.modifySyncIssue(syncStatusNoNewFinalizedHead, true)
+			n.modifyHealthCheckFailure(healthCheckPoll|healthCheckFinalizedState, true)
+			n.transitionToInSync(issues, func() {
+				// The transition callback runs under stateMu after routing and health are updated.
+				assert.Equal(t, nodeStateAlive, n.state)
+				assert.Equal(t, issues, n.stateSyncIssues)
+				assert.Zero(t, n.stateHealthFailures)
+			})
+			assert.Equal(t, nodeStateAlive, n.State())
+			want := nodeStateAlive
+			if issues != syncStatusSynced {
+				want = nodeStateOutOfSync
+			}
+			assert.Equal(t, want, n.StateForMetrics())
+		})
+	}
+}