From 74e26e202172c4480d66a4bc0b8119a082583ee8 Mon Sep 17 00:00:00 2001 From: Krish-vemula Date: Thu, 5 Nov 2026 15:14:56 -0700 Subject: [PATCH 1/3] fix(multinode): report unhealthy fallback RPCs (PLEX-3543) --- multinode/mock_node_test.go | 45 +++++++++++++ multinode/multi_node.go | 3 +- multinode/multi_node_test.go | 37 +++++++++++ multinode/node.go | 7 +- multinode/node_fsm.go | 43 ++++++++++++ multinode/node_fsm_test.go | 27 ++++++++ multinode/node_lifecycle.go | 32 ++++++++- multinode/node_lifecycle_test.go | 98 +++++++++++++++++++++++++++- multinode/transaction_sender_test.go | 1 + 9 files changed, 287 insertions(+), 6 deletions(-) diff --git a/multinode/mock_node_test.go b/multinode/mock_node_test.go index 3ad4c11..78eb739 100644 --- a/multinode/mock_node_test.go +++ b/multinode/mock_node_test.go @@ -474,6 +474,51 @@ func (_c *mockNode_StateAndLatest_Call[CHAIN_ID, RPC]) RunAndReturn(run func() ( return _c } +// StateForMetrics provides a mock function with no fields +func (_m *mockNode[CHAIN_ID, RPC]) StateForMetrics() nodeState { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for StateForMetrics") + } + + var r0 nodeState + if rf, ok := ret.Get(0).(func() nodeState); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(nodeState) + } + + return r0 +} + +// mockNode_StateForMetrics_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'StateForMetrics' +type mockNode_StateForMetrics_Call[CHAIN_ID ID, RPC interface{}] struct { + *mock.Call +} + +// StateForMetrics is a helper method to define mock.On call +func (_e *mockNode_Expecter[CHAIN_ID, RPC]) StateForMetrics() *mockNode_StateForMetrics_Call[CHAIN_ID, RPC] { + return &mockNode_StateForMetrics_Call[CHAIN_ID, RPC]{Call: _e.mock.On("StateForMetrics")} +} + +func (_c *mockNode_StateForMetrics_Call[CHAIN_ID, RPC]) Run(run func()) *mockNode_StateForMetrics_Call[CHAIN_ID, RPC] { + _c.Call.Run(func(args mock.Arguments) { + run() + }) + return _c +} + +func (_c *mockNode_StateForMetrics_Call[CHAIN_ID, RPC]) Return(_a0 nodeState) *mockNode_StateForMetrics_Call[CHAIN_ID, RPC] { + _c.Call.Return(_a0) + return _c +} + +func (_c *mockNode_StateForMetrics_Call[CHAIN_ID, RPC]) RunAndReturn(run func() nodeState) *mockNode_StateForMetrics_Call[CHAIN_ID, RPC] { + _c.Call.Return(run) + return _c +} + // String provides a mock function with no fields func (_m *mockNode[CHAIN_ID, RPC]) String() string { ret := _m.Called() diff --git a/multinode/multi_node.go b/multinode/multi_node.go index 4c02b85..bdc9b21 100644 --- a/multinode/multi_node.go +++ b/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 diff --git a/multinode/multi_node_test.go b/multinode/multi_node_test.go index adb1158..3a8f16a 100644 --- a/multinode/multi_node_test.go +++ b/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) + } +} diff --git a/multinode/node.go b/multinode/node.go index 1db52b9..48f6d58 100644 --- a/multinode/node.go +++ b/multinode/node.go @@ -80,6 +80,8 @@ type Node[ // 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() 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 +126,9 @@ 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 poolInfoProvider PoolChainInfoProvider diff --git a/multinode/node_fsm.go b/multinode/node_fsm.go index cfec34a..e04b10d 100644 --- a/multinode/node_fsm.go +++ b/multinode/node_fsm.go @@ -98,6 +98,48 @@ 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 + healthCheckNoNewHead + healthCheckNoNewFinalizedHead + healthCheckPoolSync + healthCheckFinalizedState +) + +func (n *node[CHAIN_ID, HEAD, RPC]) setHealthCheckFailure(check healthCheckFailures, failed bool) { + n.stateMu.Lock() + defer n.stateMu.Unlock() + if failed { + n.stateHealthFailures |= check + } else { + n.stateHealthFailures &^= check + } +} + +// 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.stateHealthFailures != 0: + return nodeStateOutOfSync + default: + return state + } +} + func (n *node[CHAIN_ID, HEAD, RPC]) getCachedState() nodeState { n.stateMu.RLock() defer n.stateMu.RUnlock() @@ -187,6 +229,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToAlive(fn func()) { } switch n.state { case nodeStateDialed, nodeStateInvalidChainID, nodeStateSyncing, nodeStateFinalizedStateNotAvailable: + n.stateHealthFailures = 0 n.state = nodeStateAlive default: panic(transitionFail(n.state, nodeStateAlive)) diff --git a/multinode/node_fsm_test.go b/multinode/node_fsm_test.go index d513f30..0d571d4 100644 --- a/multinode/node_fsm_test.go +++ b/multinode/node_fsm_test.go @@ -134,3 +134,30 @@ 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}, + {healthCheckNoNewHead, nodeStateOutOfSync}, + {healthCheckNoNewFinalizedHead, nodeStateOutOfSync}, + {healthCheckPoolSync, nodeStateOutOfSync}, + {healthCheckFinalizedState, nodeStateFinalizedStateNotAvailable}, + } { + n.setHealthCheckFailure(tc.check, true) + assert.Equal(t, tc.want, n.StateForMetrics()) + assert.Equal(t, nodeStateAlive, n.State()) + n.setHealthCheckFailure(tc.check, false) + assert.Equal(t, nodeStateAlive, n.StateForMetrics()) + } + n.setHealthCheckFailure(healthCheckPoll|healthCheckNoNewHead, true) + n.setHealthCheckFailure(healthCheckPoll, false) + assert.Equal(t, nodeStateOutOfSync, n.StateForMetrics()) + n.setState(nodeStateClosed) + assert.Equal(t, nodeStateClosed, n.StateForMetrics()) +} diff --git a/multinode/node_lifecycle.go b/multinode/node_lifecycle.go index 46b09a7..d716ae6 100644 --- a/multinode/node_lifecycle.go +++ b/multinode/node_lifecycle.go @@ -143,6 +143,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { pollFailures-- } } + n.setHealthCheckFailure(healthCheckPoll, pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold) if pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold { lggr.Errorw(fmt.Sprintf("RPC endpoint failed to respond to %d consecutive polls", pollFailures), "pollFailures", pollFailures, "nodeState", n.getCachedState()) if n.poolInfoProvider != nil { @@ -154,7 +155,9 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { n.declareUnreachable() return } - if outOfSync, liveNodes := n.isOutOfSyncWithPool(); outOfSync { + outOfSync, liveNodes := n.isOutOfSyncWithPool() + n.setHealthCheckFailure(healthCheckPoolSync, outOfSync) + if outOfSync { // note: there must be another live node for us to be out of sync if liveNodes < 1 && !n.isLoadBalancedRPC { lggr.Criticalf("RPC endpoint has fallen behind; %s %s", msgCannotDisable, msgDegradedState) @@ -175,6 +178,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } lggr.Warnw("Finalized state not available", "err", stateErr, "failures", finalizedStateFailures, "threshold", finalizedStateCheckFailureThreshold) if finalizedStateFailures >= finalizedStateCheckFailureThreshold { + n.setHealthCheckFailure(healthCheckFinalizedState, true) lggr.Errorw("RPC node cannot serve finalized state after consecutive failures", "failures", finalizedStateFailures) if n.poolInfoProvider != nil { if l, _ := n.poolInfoProvider.LatestChainInfo(n.name); l < 2 && !n.isLoadBalancedRPC { @@ -194,6 +198,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } } else { finalizedStateFailures = 0 + n.setHealthCheckFailure(healthCheckFinalizedState, false) } } case bh, open := <-headsSub.Heads: @@ -203,6 +208,10 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { return } receivedNewHead := n.onNewHead(lggr, &localHighestChainInfo, bh) + if receivedNewHead { + n.setHealthCheckFailure(healthCheckNoNewHead, false) + n.refreshPoolSyncHealth() + } if receivedNewHead && noNewHeadsTimeoutThreshold > 0 { headsSub.ResetTimer(noNewHeadsTimeoutThreshold) } @@ -211,6 +220,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { n.declareUnreachable() return case <-headsSub.NoNewHeads: + n.setHealthCheckFailure(healthCheckNoNewHead, true) // We haven't received a head on the channel for at least the // threshold amount of time, mark it broken lggr.Errorw(fmt.Sprintf("RPC endpoint detected out of sync; no new heads received for %s (last head received was %v)", noNewHeadsTimeoutThreshold, localHighestChainInfo.BlockNumber), "nodeState", n.getCachedState(), "latestReceivedBlockNumber", localHighestChainInfo.BlockNumber, "noNewHeadsTimeoutThreshold", noNewHeadsTimeoutThreshold) @@ -235,10 +245,15 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } receivedNewHead := n.onNewFinalizedHead(lggr, &localHighestChainInfo, latestFinalized) + if receivedNewHead { + n.setHealthCheckFailure(healthCheckNoNewFinalizedHead, false) + n.refreshPoolSyncHealth() + } if receivedNewHead && noNewFinalizedBlocksTimeoutThreshold > 0 { finalizedHeadsSub.ResetTimer(noNewFinalizedBlocksTimeoutThreshold) } case <-finalizedHeadsSub.NoNewHeads: + n.setHealthCheckFailure(healthCheckNoNewFinalizedHead, true) // We haven't received a finalized head on the channel for at least the // threshold amount of time, mark it broken lggr.Errorw(fmt.Sprintf("RPC's finalized state is out of sync; no new finalized heads received for %s (last finalized head received was %v)", noNewFinalizedBlocksTimeoutThreshold, localHighestChainInfo.FinalizedBlockNumber), "latestReceivedBlockNumber", localHighestChainInfo.BlockNumber) @@ -263,6 +278,17 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } } +// A forcibly revived node can recover through its head subscription even when polling is disabled. +func (n *node[CHAIN_ID, HEAD, RPC]) refreshPoolSyncHealth() { + n.stateMu.RLock() + failed := n.stateHealthFailures&healthCheckPoolSync != 0 + n.stateMu.RUnlock() + if failed { + outOfSync, _ := n.isOutOfSyncWithPool() + n.setHealthCheckFailure(healthCheckPoolSync, outOfSync) + } +} + func (n *node[CHAIN_ID, HEAD, RPC]) unsubscribeHealthChecks() { n.stateMu.Lock() for _, sub := range n.healthCheckSubs { @@ -479,6 +505,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) outOfSyncLoop(syncIssues syncStatus) { _, localHighestChainInfo := n.rpc.GetInterceptedChainInfo() for { if syncIssues == syncStatusSynced { + n.setHealthCheckFailure(healthCheckNoNewHead|healthCheckNoNewFinalizedHead|healthCheckPoolSync, false) // back in-sync! flip back into alive loop lggr.Infow(fmt.Sprintf("%s: %s. Node was out-of-sync for %s", msgInSync, n.String(), time.Since(outOfSyncAt))) n.declareInSync() @@ -522,6 +549,9 @@ func (n *node[CHAIN_ID, HEAD, RPC]) outOfSyncLoop(syncIssues syncStatus) { n.declareOutOfSync(syncIssues) return } + n.setHealthCheckFailure(healthCheckNoNewHead, syncIssues&syncStatusNoNewHead != 0) + n.setHealthCheckFailure(healthCheckNoNewFinalizedHead, syncIssues&syncStatusNoNewFinalizedHead != 0) + n.setHealthCheckFailure(healthCheckPoolSync, syncIssues&syncStatusNotInSyncWithPool != 0) lggr.Criticalw("RPC endpoint is still out of sync, but there are no other available nodes. This RPC node will be forcibly moved back into the live pool in a degraded state", "syncIssues", syncIssues) n.declareInSync() return diff --git a/multinode/node_lifecycle_test.go b/multinode/node_lifecycle_test.go index 38c1deb..11f65d9 100644 --- a/multinode/node_lifecycle_test.go +++ b/multinode/node_lifecycle_test.go @@ -234,6 +234,7 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { node.declareAlive() tests.AssertLogEventually(t, observedLogs, fmt.Sprintf("RPC endpoint failed to respond to %d consecutive polls", pollFailureThreshold)) assert.Equal(t, nodeStateAlive, node.State()) + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateUnreachable }) }) t.Run("with threshold poll failures, we are the last node alive, but is a proxy, transitions to unreachable", func(t *testing.T) { t.Parallel() @@ -325,6 +326,8 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { node.SetPoolChainInfoProvider(poolInfo) node.declareAlive() tests.AssertLogEventually(t, observedLogs, fmt.Sprintf("RPC endpoint has fallen behind; %s %s", msgCannotDisable, msgDegradedState)) + assert.Equal(t, nodeStateAlive, node.State()) + assert.Equal(t, nodeStateOutOfSync, node.StateForMetrics()) }) t.Run("when behind more than SyncThreshold, we are the last live node, but is a proxy, transitions to out of sync -> unreachable", func(t *testing.T) { t.Parallel() @@ -430,6 +433,7 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { node.declareAlive() tests.AssertLogEventually(t, observedLogs, fmt.Sprintf("RPC endpoint detected out of sync; %s %s", msgCannotDisable, msgDegradedState)) assert.Equal(t, nodeStateAlive, node.State()) + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateOutOfSync }) }) t.Run("when no new heads received for threshold, we are the last live node, but is a proxy, transitions to out of sync -> unreachable", func(t *testing.T) { t.Parallel() @@ -679,6 +683,7 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { node.declareAlive() tests.AssertLogEventually(t, observed, fmt.Sprintf("RPC's finalized state is out of sync; %s %s", msgCannotDisable, msgDegradedState)) assert.Equal(t, nodeStateAlive, node.State()) + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateOutOfSync }) }) t.Run("when no new finalized heads received for threshold, we are the last live node, but is a proxy, transitions to out of sync -> unreachable", func(t *testing.T) { t.Parallel() @@ -1087,13 +1092,21 @@ func TestUnit_NodeLifecycle_outOfSyncLoop(t *testing.T) { outOfSyncSubscription.On("Err").Return((<-chan error)(nil)) outOfSyncSubscription.On("Unsubscribe").Once() rpc.On("SubscribeToHeads", mock.Anything).Return(make(<-chan Head), outOfSyncSubscription, nil).Once() - setupRPCForAliveLoop(t, rpc) + heads := make(chan Head) + rpc.On("SubscribeToHeads", mock.Anything).Return((<-chan Head)(heads), newSub(t), nil).Once() node.declareOutOfSync(syncStatusNoNewHead) tests.AssertLogEventually(t, observedLogs, "RPC endpoint is still out of sync, but there are no other available nodes. This RPC node will be forcibly moved back into the live pool in a degraded state") tests.AssertEventually(t, func() bool { - return node.State() == nodeStateAlive + return node.State() == nodeStateAlive && node.StateForMetrics() == nodeStateOutOfSync }) + // Forced revival preserves the failure until a newer head arrives, even without polling. + select { + case heads <- (head{BlockNumber: 1}).ToMockHead(t): + case <-time.After(tests.WaitTimeout(t)): + t.Fatal("head was not consumed") + } + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateAlive }) }) t.Run("becomes alive if there is no other nodes, unless proxy", func(t *testing.T) { t.Parallel() @@ -2511,3 +2524,84 @@ func TestUnit_NodeLifecycle_finalizedStateNotAvailableLoop(t *testing.T) { }) }) } + +// A responsive version endpoint does not prove that the ledger subscription recovered. +func TestUnit_NodeLifecycle_MetricHealthRecovery(t *testing.T) { + t.Parallel() + rpc := newMockRPCClient[ID, Head](t) + heads := make(chan Head) + rpc.On("SubscribeToHeads", mock.Anything).Return((<-chan Head)(heads), newSub(t), nil).Once() + rpc.On("GetInterceptedChainInfo").Return(ChainInfo{BlockNumber: 20}, ChainInfo{BlockNumber: 20}) + rpc.On("Close").Return(nil) + var pingHealthy atomic.Bool + rpc.On("ClientVersion", mock.Anything).Return(func(context.Context) (string, error) { + if pingHealthy.Load() { + return "version", nil + } + return "", errors.New("poll failed") + }) + node := newTestNode(t, testNodeOpts{ + rpc: rpc, + config: testNodeConfig{pollFailureThreshold: 1, pollInterval: tests.TestInterval}, + chainConfig: clientMocks.ChainConfig{NoNewHeadsThresholdVal: 10 * tests.TestInterval}, + }) + defer func() { assert.NoError(t, node.close()) }() + poolInfo := newMockPoolChainInfoProvider(t) + poolInfo.On("LatestChainInfo", mock.Anything).Return(0, ChainInfo{BlockNumber: 20}) + node.SetPoolChainInfoProvider(poolInfo) + node.setState(nodeStateDialed) + node.declareAlive() + tests.AssertEventually(t, func() bool { + node.stateMu.RLock() + defer node.stateMu.RUnlock() + return node.stateHealthFailures&(healthCheckPoll|healthCheckNoNewHead) == healthCheckPoll|healthCheckNoNewHead + }) + assert.Equal(t, nodeStateAlive, node.State()) + assert.Equal(t, nodeStateUnreachable, node.StateForMetrics()) + pingHealthy.Store(true) + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateOutOfSync }) + // A duplicate ledger is not evidence of progress. + select { + case heads <- (head{BlockNumber: 20}).ToMockHead(t): + case <-time.After(tests.WaitTimeout(t)): + t.Fatal("head was not consumed") + } + assert.Equal(t, nodeStateOutOfSync, node.StateForMetrics()) + select { + case heads <- (head{BlockNumber: 21}).ToMockHead(t): + case <-time.After(tests.WaitTimeout(t)): + t.Fatal("head was not consumed") + } + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateAlive }) + assert.Equal(t, nodeStateAlive, node.State()) +} + +func TestUnit_NodeLifecycle_FinalizedStateMetricRecovery(t *testing.T) { + t.Parallel() + rpc := newMockRPCClient[ID, Head](t) + rpc.On("SubscribeToHeads", mock.Anything).Return(make(<-chan Head), newSub(t), nil).Once() + rpc.On("GetInterceptedChainInfo").Return(ChainInfo{}, ChainInfo{}) + rpc.On("ClientVersion", mock.Anything).Return("version", nil) + rpc.On("Close").Return(nil) + var available atomic.Bool + rpc.On("CheckFinalizedStateAvailability", mock.Anything).Return(func(context.Context) error { + if available.Load() { + return nil + } + return ErrFinalizedStateUnavailable + }) + node := newTestNode(t, testNodeOpts{ + rpc: rpc, + config: testNodeConfig{pollInterval: tests.TestInterval, finalizedStateCheckFailureThreshold: 1}, + }) + defer func() { assert.NoError(t, node.close()) }() + poolInfo := newMockPoolChainInfoProvider(t) + poolInfo.On("LatestChainInfo", mock.Anything).Return(0, ChainInfo{}) + node.SetPoolChainInfoProvider(poolInfo) + node.setState(nodeStateDialed) + node.declareAlive() + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateFinalizedStateNotAvailable }) + assert.Equal(t, nodeStateAlive, node.State()) + available.Store(true) + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateAlive }) +} diff --git a/multinode/transaction_sender_test.go b/multinode/transaction_sender_test.go index 1c4bbce..66b2174 100644 --- a/multinode/transaction_sender_test.go +++ b/multinode/transaction_sender_test.go @@ -78,6 +78,7 @@ func TestTransactionSender_SendTransaction(t *testing.T) { node.On("String").Return("node name").Maybe() node.On("RPC").Return(rpc).Maybe() node.On("State").Return(state).Maybe() + node.On("StateForMetrics").Return(state).Maybe() node.On("Start", mock.Anything).Return(nil).Maybe() node.On("Close").Return(nil).Maybe() node.On("SetPoolChainInfoProvider", mock.Anything).Return(nil).Maybe() From b717bb3f70faa4d2682b63930aa0499ac0ff060c Mon Sep 17 00:00:00 2001 From: Krish-vemula Date: Fri, 6 Nov 2026 19:09:01 -0700 Subject: [PATCH 2/3] fix(multinode): refine health tracking and bound poll failures --- multinode/node.go | 10 +- multinode/node_fsm.go | 31 ++++-- multinode/node_fsm_test.go | 55 ++++++++-- multinode/node_lifecycle.go | 92 ++++++++++------- multinode/node_lifecycle_test.go | 166 ++++++++++++++++++++++++++++++- 5 files changed, 294 insertions(+), 60 deletions(-) diff --git a/multinode/node.go b/multinode/node.go index 48f6d58..2165d2d 100644 --- a/multinode/node.go +++ b/multinode/node.go @@ -76,9 +76,12 @@ 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 @@ -129,6 +132,7 @@ type node[ stateMu sync.RWMutex // protects state* fields state nodeState stateHealthFailures healthCheckFailures + stateSyncIssues syncStatus poolInfoProvider PoolChainInfoProvider diff --git a/multinode/node_fsm.go b/multinode/node_fsm.go index e04b10d..4594839 100644 --- a/multinode/node_fsm.go +++ b/multinode/node_fsm.go @@ -103,13 +103,10 @@ type healthCheckFailures uint8 const ( healthCheckPoll healthCheckFailures = 1 << iota - healthCheckNoNewHead - healthCheckNoNewFinalizedHead - healthCheckPoolSync healthCheckFinalizedState ) -func (n *node[CHAIN_ID, HEAD, RPC]) setHealthCheckFailure(check healthCheckFailures, failed bool) { +func (n *node[CHAIN_ID, HEAD, RPC]) modifyHealthCheckFailure(check healthCheckFailures, failed bool) { n.stateMu.Lock() defer n.stateMu.Unlock() if failed { @@ -119,6 +116,16 @@ func (n *node[CHAIN_ID, HEAD, RPC]) setHealthCheckFailure(check healthCheckFailu } } +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() @@ -133,7 +140,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) StateForMetrics() nodeState { return nodeStateUnreachable case n.stateHealthFailures&healthCheckFinalizedState != 0: return nodeStateFinalizedStateNotAvailable - case n.stateHealthFailures != 0: + case n.stateSyncIssues != syncStatusSynced: return nodeStateOutOfSync default: return state @@ -230,6 +237,7 @@ 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)) @@ -237,17 +245,17 @@ func (n *node[CHAIN_ID, HEAD, RPC]) transitionToAlive(fn func()) { 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) @@ -259,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)) diff --git a/multinode/node_fsm_test.go b/multinode/node_fsm_test.go index 0d571d4..02ddb9a 100644 --- a/multinode/node_fsm_test.go +++ b/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 @@ -144,20 +146,57 @@ func TestUnit_Node_StateForMetrics(t *testing.T) { want nodeState }{ {healthCheckPoll, nodeStateUnreachable}, - {healthCheckNoNewHead, nodeStateOutOfSync}, - {healthCheckNoNewFinalizedHead, nodeStateOutOfSync}, - {healthCheckPoolSync, nodeStateOutOfSync}, {healthCheckFinalizedState, nodeStateFinalizedStateNotAvailable}, } { - n.setHealthCheckFailure(tc.check, true) + n.modifyHealthCheckFailure(tc.check, true) assert.Equal(t, tc.want, n.StateForMetrics()) assert.Equal(t, nodeStateAlive, n.State()) - n.setHealthCheckFailure(tc.check, false) + n.modifyHealthCheckFailure(tc.check, false) assert.Equal(t, nodeStateAlive, n.StateForMetrics()) } - n.setHealthCheckFailure(healthCheckPoll|healthCheckNoNewHead, true) - n.setHealthCheckFailure(healthCheckPoll, false) + 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()) + }) + } +} diff --git a/multinode/node_lifecycle.go b/multinode/node_lifecycle.go index d716ae6..e8a596b 100644 --- a/multinode/node_lifecycle.go +++ b/multinode/node_lifecycle.go @@ -102,6 +102,18 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { // Get the latest chain info to use as local highest localHighestChainInfo, _ := n.rpc.GetInterceptedChainInfo() var pollFailures uint32 + // recordPollFailure caps pollFailures at pollFailureThreshold. Reaching the threshold normally + // ends this loop, but the last remaining node is kept alive past it; without the cap, a long + // outage leaves a backlog that keeps the endpoint marked unhealthy, and gets it ejected once + // a peer returns, long after it has recovered. + recordPollFailure := func() { + n.metrics.IncrementPollsFailed(ctx, n.name) + switch { + case pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold: + case pollFailures < math.MaxUint32: // prevent overflow + pollFailures++ + } + } // Finalized-state availability is for EVM-style stacks, but multinode is a shared framework package: mandatory RPCClient/NodeConfig APIs would make every consumer (e.g. Solana) // inherit stubs or real implementations trigger unrelated security review. Optional interfaces keep the feature opt-in—only callers that need it implement FinalizedStateCheckConfig and FinalizedStateChecker; everyone else stays untouched. @@ -128,11 +140,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { version, pingErr := n.RPC().ClientVersion(pollCtx) cancel() if pingErr != nil { - // prevent overflow - if pollFailures < math.MaxUint32 { - n.metrics.IncrementPollsFailed(ctx, n.name) - pollFailures++ - } + recordPollFailure() lggr.Warnw(fmt.Sprintf("Poll failure, RPC endpoint %s failed to respond properly", n.String()), "err", pingErr, "pollFailures", pollFailures, "nodeState", n.getCachedState()) } else { lggr.Debugw("Ping successful", "nodeState", n.State()) @@ -143,30 +151,43 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { pollFailures-- } } - n.setHealthCheckFailure(healthCheckPoll, pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold) - if pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold { + pollFailureThresholdReached := pollFailureThreshold > 0 && pollFailures >= pollFailureThreshold + // Hysteresis: report unhealthy at the threshold, healthy again only once the failure score + // drains to zero. The capped counter otherwise flaps around the threshold when failures + // outpace successes on a retained node. + if pollFailureThresholdReached { + n.modifyHealthCheckFailure(healthCheckPoll, true) + } else if pollFailures == 0 { + n.modifyHealthCheckFailure(healthCheckPoll, false) + } + if pollFailureThresholdReached { lggr.Errorw(fmt.Sprintf("RPC endpoint failed to respond to %d consecutive polls", pollFailures), "pollFailures", pollFailures, "nodeState", n.getCachedState()) + keepAlive := false if n.poolInfoProvider != nil { - if l, _ := n.poolInfoProvider.LatestChainInfo(n.name); l < 1 && !n.isLoadBalancedRPC { - lggr.Criticalf("RPC endpoint failed to respond to polls; %s %s", msgCannotDisable, msgDegradedState) - continue - } + liveNodes, _ := n.poolInfoProvider.LatestChainInfo(n.name) + keepAlive = liveNodes < 1 && !n.isLoadBalancedRPC } - n.declareUnreachable() - return + if !keepAlive { + n.declareUnreachable() + return + } + // Keep evaluating independent health checks while retaining this node for retries. + lggr.Criticalf("RPC endpoint failed to respond to polls; %s %s", msgCannotDisable, msgDegradedState) } outOfSync, liveNodes := n.isOutOfSyncWithPool() - n.setHealthCheckFailure(healthCheckPoolSync, outOfSync) + n.modifySyncIssue(syncStatusNotInSyncWithPool, outOfSync) if outOfSync { // note: there must be another live node for us to be out of sync if liveNodes < 1 && !n.isLoadBalancedRPC { lggr.Criticalf("RPC endpoint has fallen behind; %s %s", msgCannotDisable, msgDegradedState) - continue + } else { + n.declareOutOfSync(syncStatusNotInSyncWithPool) + return } - n.declareOutOfSync(syncStatusNotInSyncWithPool) - return } - if finalizedStateCheckFailureThreshold > 0 { + // Skip when the ping failed: an unresponsive endpoint says nothing about finalized state, + // and checking would block the loop for another pollInterval and count a second failure. + if finalizedStateCheckFailureThreshold > 0 && pingErr == nil { stateCheckCtx, stateCheckCancel := context.WithTimeout(ctx, pollInterval) stateErr := n.finalizedStateChecker.CheckFinalizedStateAvailability(stateCheckCtx) stateCheckCancel() @@ -178,7 +199,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } lggr.Warnw("Finalized state not available", "err", stateErr, "failures", finalizedStateFailures, "threshold", finalizedStateCheckFailureThreshold) if finalizedStateFailures >= finalizedStateCheckFailureThreshold { - n.setHealthCheckFailure(healthCheckFinalizedState, true) + n.modifyHealthCheckFailure(healthCheckFinalizedState, true) lggr.Errorw("RPC node cannot serve finalized state after consecutive failures", "failures", finalizedStateFailures) if n.poolInfoProvider != nil { if l, _ := n.poolInfoProvider.LatestChainInfo(n.name); l < 2 && !n.isLoadBalancedRPC { @@ -190,15 +211,12 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { return } } else { - if pollFailures < math.MaxUint32 { - n.metrics.IncrementPollsFailed(ctx, n.name) - pollFailures++ - } + recordPollFailure() lggr.Warnw("Finalized state check failed with RPC error", "err", stateErr, "pollFailures", pollFailures) } } else { finalizedStateFailures = 0 - n.setHealthCheckFailure(healthCheckFinalizedState, false) + n.modifyHealthCheckFailure(healthCheckFinalizedState, false) } } case bh, open := <-headsSub.Heads: @@ -209,7 +227,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } receivedNewHead := n.onNewHead(lggr, &localHighestChainInfo, bh) if receivedNewHead { - n.setHealthCheckFailure(healthCheckNoNewHead, false) + n.modifySyncIssue(syncStatusNoNewHead, false) n.refreshPoolSyncHealth() } if receivedNewHead && noNewHeadsTimeoutThreshold > 0 { @@ -220,7 +238,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { n.declareUnreachable() return case <-headsSub.NoNewHeads: - n.setHealthCheckFailure(healthCheckNoNewHead, true) + n.modifySyncIssue(syncStatusNoNewHead, true) // We haven't received a head on the channel for at least the // threshold amount of time, mark it broken lggr.Errorw(fmt.Sprintf("RPC endpoint detected out of sync; no new heads received for %s (last head received was %v)", noNewHeadsTimeoutThreshold, localHighestChainInfo.BlockNumber), "nodeState", n.getCachedState(), "latestReceivedBlockNumber", localHighestChainInfo.BlockNumber, "noNewHeadsTimeoutThreshold", noNewHeadsTimeoutThreshold) @@ -246,14 +264,13 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { receivedNewHead := n.onNewFinalizedHead(lggr, &localHighestChainInfo, latestFinalized) if receivedNewHead { - n.setHealthCheckFailure(healthCheckNoNewFinalizedHead, false) - n.refreshPoolSyncHealth() + n.modifySyncIssue(syncStatusNoNewFinalizedHead, false) } if receivedNewHead && noNewFinalizedBlocksTimeoutThreshold > 0 { finalizedHeadsSub.ResetTimer(noNewFinalizedBlocksTimeoutThreshold) } case <-finalizedHeadsSub.NoNewHeads: - n.setHealthCheckFailure(healthCheckNoNewFinalizedHead, true) + n.modifySyncIssue(syncStatusNoNewFinalizedHead, true) // We haven't received a finalized head on the channel for at least the // threshold amount of time, mark it broken lggr.Errorw(fmt.Sprintf("RPC's finalized state is out of sync; no new finalized heads received for %s (last finalized head received was %v)", noNewFinalizedBlocksTimeoutThreshold, localHighestChainInfo.FinalizedBlockNumber), "latestReceivedBlockNumber", localHighestChainInfo.BlockNumber) @@ -278,14 +295,17 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } } -// A forcibly revived node can recover through its head subscription even when polling is disabled. +// refreshPoolSyncHealth rechecks a pool-sync failure retained during forced revival. +// A new head proves progress, but the node may still be behind the pool. Recheck +// on head progress so the failure clears once it catches up, even when polling +// is disabled and the poll handler cannot clear it. func (n *node[CHAIN_ID, HEAD, RPC]) refreshPoolSyncHealth() { n.stateMu.RLock() - failed := n.stateHealthFailures&healthCheckPoolSync != 0 + failed := n.stateSyncIssues&syncStatusNotInSyncWithPool != 0 n.stateMu.RUnlock() if failed { outOfSync, _ := n.isOutOfSyncWithPool() - n.setHealthCheckFailure(healthCheckPoolSync, outOfSync) + n.modifySyncIssue(syncStatusNotInSyncWithPool, outOfSync) } } @@ -505,10 +525,9 @@ func (n *node[CHAIN_ID, HEAD, RPC]) outOfSyncLoop(syncIssues syncStatus) { _, localHighestChainInfo := n.rpc.GetInterceptedChainInfo() for { if syncIssues == syncStatusSynced { - n.setHealthCheckFailure(healthCheckNoNewHead|healthCheckNoNewFinalizedHead|healthCheckPoolSync, false) // back in-sync! flip back into alive loop lggr.Infow(fmt.Sprintf("%s: %s. Node was out-of-sync for %s", msgInSync, n.String(), time.Since(outOfSyncAt))) - n.declareInSync() + n.declareInSync(syncIssues) return } @@ -549,11 +568,8 @@ func (n *node[CHAIN_ID, HEAD, RPC]) outOfSyncLoop(syncIssues syncStatus) { n.declareOutOfSync(syncIssues) return } - n.setHealthCheckFailure(healthCheckNoNewHead, syncIssues&syncStatusNoNewHead != 0) - n.setHealthCheckFailure(healthCheckNoNewFinalizedHead, syncIssues&syncStatusNoNewFinalizedHead != 0) - n.setHealthCheckFailure(healthCheckPoolSync, syncIssues&syncStatusNotInSyncWithPool != 0) lggr.Criticalw("RPC endpoint is still out of sync, but there are no other available nodes. This RPC node will be forcibly moved back into the live pool in a degraded state", "syncIssues", syncIssues) - n.declareInSync() + n.declareInSync(syncIssues) return } } diff --git a/multinode/node_lifecycle_test.go b/multinode/node_lifecycle_test.go index 11f65d9..0233c2e 100644 --- a/multinode/node_lifecycle_test.go +++ b/multinode/node_lifecycle_test.go @@ -2554,7 +2554,7 @@ func TestUnit_NodeLifecycle_MetricHealthRecovery(t *testing.T) { tests.AssertEventually(t, func() bool { node.stateMu.RLock() defer node.stateMu.RUnlock() - return node.stateHealthFailures&(healthCheckPoll|healthCheckNoNewHead) == healthCheckPoll|healthCheckNoNewHead + return node.stateHealthFailures&healthCheckPoll != 0 && node.stateSyncIssues&syncStatusNoNewHead != 0 }) assert.Equal(t, nodeStateAlive, node.State()) assert.Equal(t, nodeStateUnreachable, node.StateForMetrics()) @@ -2605,3 +2605,167 @@ func TestUnit_NodeLifecycle_FinalizedStateMetricRecovery(t *testing.T) { available.Store(true) tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateAlive }) } + +func TestUnit_NodeLifecycle_DegradedNodeUpdatesIndependentHealth(t *testing.T) { + t.Parallel() + // Each phase changes checks independently while the last node stays selectable. + var phase atomic.Uint32 + rpc := newMockRPCClient[ID, Head](t) + rpc.On("SubscribeToHeads", mock.Anything).Return(make(<-chan Head), newSub(t), nil).Once() + rpc.On("GetInterceptedChainInfo").Return(ChainInfo{BlockNumber: 20}, ChainInfo{BlockNumber: 20}) + rpc.On("Close").Return(nil) + rpc.On("ClientVersion", mock.Anything).Return(func(context.Context) (string, error) { + if phase.Load() == 1 { + return "", errors.New("poll failed") + } + return "version", nil + }) + rpc.On("CheckFinalizedStateAvailability", mock.Anything).Return(func(context.Context) error { + switch phase.Load() { + case 0, 2: + return ErrFinalizedStateUnavailable + default: + return nil + } + }) + pool := newMockPoolChainInfoProvider(t) + pool.On("LatestChainInfo", mock.Anything).Return(func(string) (int, ChainInfo) { + block := int64(20) + if p := phase.Load(); p >= 1 && p <= 3 { + block = 30 + } + return 0, ChainInfo{BlockNumber: block} + }) + node := newTestNode(t, testNodeOpts{ + rpc: rpc, + config: testNodeConfig{ + pollInterval: tests.TestInterval, pollFailureThreshold: 1, + finalizedStateCheckFailureThreshold: 1, + syncThreshold: 1, selectionMode: NodeSelectionModeRoundRobin, + }, + }) + defer func() { assert.NoError(t, node.close()) }() + node.SetPoolChainInfoProvider(pool) + node.setState(nodeStateDialed) + node.declareAlive() + for index, expected := range []struct { + failures healthCheckFailures + syncIssues syncStatus + state nodeState + }{ + {healthCheckFinalizedState, syncStatusSynced, nodeStateFinalizedStateNotAvailable}, + // Finalized state is not checked while pings fail, so its last result is retained. + {healthCheckPoll | healthCheckFinalizedState, syncStatusNotInSyncWithPool, nodeStateUnreachable}, + {healthCheckFinalizedState, syncStatusNotInSyncWithPool, nodeStateFinalizedStateNotAvailable}, + {0, syncStatusNotInSyncWithPool, nodeStateOutOfSync}, + {0, syncStatusSynced, nodeStateAlive}, + } { + phase.Store(uint32(index)) + tests.AssertEventually(t, func() bool { + node.stateMu.RLock() + defer node.stateMu.RUnlock() + return node.stateHealthFailures == expected.failures && node.stateSyncIssues == expected.syncIssues + }) + assert.Equal(t, nodeStateAlive, node.State()) + assert.Equal(t, expected.state, node.StateForMetrics()) + } +} + +func TestUnit_NodeLifecycle_RetainedNodePollFailuresAreCapped(t *testing.T) { + t.Parallel() + const pollFailureThreshold = 2 + const outagePolls = 20 + rpc := newMockRPCClient[ID, Head](t) + rpc.On("SubscribeToHeads", mock.Anything).Return(make(<-chan Head), newSub(t), nil).Once() + rpc.On("GetInterceptedChainInfo").Return(ChainInfo{BlockNumber: 20}, ChainInfo{BlockNumber: 20}) + rpc.On("Close").Return(nil) + var healthy atomic.Bool + var failedPolls, successfulPolls atomic.Uint32 + rpc.On("ClientVersion", mock.Anything).Return(func(context.Context) (string, error) { + if healthy.Load() { + successfulPolls.Add(1) + return "version", nil + } + failedPolls.Add(1) + return "", errors.New("poll failed") + }) + var liveNodes atomic.Int32 + pool := newMockPoolChainInfoProvider(t) + pool.On("LatestChainInfo", mock.Anything).Return(func(string) (int, ChainInfo) { + return int(liveNodes.Load()), ChainInfo{BlockNumber: 20} + }) + node := newTestNode(t, testNodeOpts{ + rpc: rpc, + config: testNodeConfig{pollInterval: tests.TestInterval, pollFailureThreshold: pollFailureThreshold}, + }) + defer func() { assert.NoError(t, node.close()) }() + node.SetPoolChainInfoProvider(pool) + node.setState(nodeStateDialed) + node.declareAlive() + + // Long outage on the last node: it is retained for retries but reported unreachable. + tests.AssertEventually(t, func() bool { return failedPolls.Load() >= outagePolls }) + assert.Equal(t, nodeStateAlive, node.State()) + assert.Equal(t, nodeStateUnreachable, node.StateForMetrics()) + + // Capped, the score drains in pollFailureThreshold successes; uncapped it would take ~outagePolls. + // The bound is loose so slow runners observing the transition late do not fail the test. + healthy.Store(true) + tests.AssertEventually(t, func() bool { return node.StateForMetrics() == nodeStateAlive }) + assert.Less(t, successfulPolls.Load(), uint32(outagePolls/2)) + + // A peer returning must not eject the recovered node for its past outage. + liveNodes.Store(1) + polled := successfulPolls.Load() + tests.AssertEventually(t, func() bool { return successfulPolls.Load() >= polled+3 }) + assert.Equal(t, nodeStateAlive, node.State()) + assert.Equal(t, nodeStateAlive, node.StateForMetrics()) +} + +func TestUnit_NodeLifecycle_RetainedNodePollHealthDoesNotFlap(t *testing.T) { + t.Parallel() + const pollFailureThreshold = 2 + const observedPolls = 15 + var node testNode + var polls, observed, clearedAfterSet atomic.Uint32 + var reported atomic.Bool + rpc := newMockRPCClient[ID, Head](t) + rpc.On("SubscribeToHeads", mock.Anything).Return(make(<-chan Head), newSub(t), nil).Once() + rpc.On("GetInterceptedChainInfo").Return(ChainInfo{BlockNumber: 20}, ChainInfo{BlockNumber: 20}) + rpc.On("Close").Return(nil) + // Two failures per success: failures outpace successes, so the node stays unhealthy. + rpc.On("ClientVersion", mock.Anything).Return(func(context.Context) (string, error) { + // Runs on the alive loop between ticks, so it sees the flag left by the previous tick. + node.stateMu.RLock() + unhealthy := node.stateHealthFailures&healthCheckPoll != 0 + node.stateMu.RUnlock() + if unhealthy { + reported.Store(true) + } + if reported.Load() { + observed.Add(1) + if !unhealthy { + clearedAfterSet.Add(1) + } + } + if polls.Add(1)%3 == 0 { + return "version", nil + } + return "", errors.New("poll failed") + }) + pool := newMockPoolChainInfoProvider(t) + pool.On("LatestChainInfo", mock.Anything).Return(0, ChainInfo{BlockNumber: 20}) + node = newTestNode(t, testNodeOpts{ + rpc: rpc, + config: testNodeConfig{pollInterval: tests.TestInterval, pollFailureThreshold: pollFailureThreshold}, + }) + defer func() { assert.NoError(t, node.close()) }() + node.SetPoolChainInfoProvider(pool) + node.setState(nodeStateDialed) + node.declareAlive() + + tests.AssertEventually(t, func() bool { return observed.Load() >= observedPolls }) + assert.Zero(t, clearedAfterSet.Load(), "poll health flag cleared while failures outpace successes") + assert.Equal(t, nodeStateAlive, node.State()) + assert.Equal(t, nodeStateUnreachable, node.StateForMetrics()) +} From 2ce0777aaad832a9ff2cb11cd4e54980268f36a0 Mon Sep 17 00:00:00 2001 From: Krish-vemula Date: Mon, 9 Nov 2026 13:02:13 -0700 Subject: [PATCH 3/3] fix nits --- multinode/node_lifecycle.go | 8 +++----- multinode/node_lifecycle_test.go | 4 ++-- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/multinode/node_lifecycle.go b/multinode/node_lifecycle.go index e8a596b..68858ef 100644 --- a/multinode/node_lifecycle.go +++ b/multinode/node_lifecycle.go @@ -157,10 +157,6 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { // outpace successes on a retained node. if pollFailureThresholdReached { n.modifyHealthCheckFailure(healthCheckPoll, true) - } else if pollFailures == 0 { - n.modifyHealthCheckFailure(healthCheckPoll, false) - } - if pollFailureThresholdReached { lggr.Errorw(fmt.Sprintf("RPC endpoint failed to respond to %d consecutive polls", pollFailures), "pollFailures", pollFailures, "nodeState", n.getCachedState()) keepAlive := false if n.poolInfoProvider != nil { @@ -173,6 +169,8 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { } // Keep evaluating independent health checks while retaining this node for retries. lggr.Criticalf("RPC endpoint failed to respond to polls; %s %s", msgCannotDisable, msgDegradedState) + } else if pollFailures == 0 { + n.modifyHealthCheckFailure(healthCheckPoll, false) } outOfSync, liveNodes := n.isOutOfSyncWithPool() n.modifySyncIssue(syncStatusNotInSyncWithPool, outOfSync) @@ -202,7 +200,7 @@ func (n *node[CHAIN_ID, HEAD, RPC]) aliveLoop() { n.modifyHealthCheckFailure(healthCheckFinalizedState, true) lggr.Errorw("RPC node cannot serve finalized state after consecutive failures", "failures", finalizedStateFailures) if n.poolInfoProvider != nil { - if l, _ := n.poolInfoProvider.LatestChainInfo(n.name); l < 2 && !n.isLoadBalancedRPC { + if l, _ := n.poolInfoProvider.LatestChainInfo(n.name); l < 1 && !n.isLoadBalancedRPC { lggr.Criticalf("RPC endpoint cannot serve finalized state; %s %s", msgCannotDisable, msgDegradedState) continue } diff --git a/multinode/node_lifecycle_test.go b/multinode/node_lifecycle_test.go index 0233c2e..1a1e12e 100644 --- a/multinode/node_lifecycle_test.go +++ b/multinode/node_lifecycle_test.go @@ -2472,7 +2472,7 @@ func TestUnit_NodeLifecycle_finalizedStateNotAvailableLoop(t *testing.T) { }) }) - t.Run("transitions from alive to finalizedStateNotAvailable and back", func(t *testing.T) { + t.Run("with one other alive node, transitions to finalizedStateNotAvailable and back", func(t *testing.T) { t.Parallel() rpc := newMockRPCClient[ID, Head](t) nodeChainID := RandomID() @@ -2506,7 +2506,7 @@ func TestUnit_NodeLifecycle_finalizedStateNotAvailableLoop(t *testing.T) { ).Times(3) poolInfo := newMockPoolChainInfoProvider(t) - poolInfo.EXPECT().LatestChainInfo(mock.Anything).Return(5, ChainInfo{}).Maybe() + poolInfo.EXPECT().LatestChainInfo(mock.Anything).Return(1, ChainInfo{}).Maybe() poolInfo.EXPECT().HighestUserObservations().Return(ChainInfo{}).Maybe() node.SetPoolChainInfoProvider(poolInfo)