DF-23489 unreachable node recovery depends on successful polls

Author: vlfig

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

@@ -17,6 +17,7 @@ type MultiNode struct {
 
 	// Node Configs
 	PollFailureThreshold       *uint32
+	PollSuccessThreshold       *uint32
 	PollInterval               *config.Duration
 	SelectionMode              *string
 	SyncThreshold              *uint32
@@ -44,6 +45,10 @@ func (c *MultiNodeConfig) PollFailureThreshold() uint32 {
 	return *c.MultiNode.PollFailureThreshold
 }
 
+func (c *MultiNodeConfig) PollSuccessThreshold() uint32 {
+	return *c.MultiNode.PollSuccessThreshold
+}
+
 func (c *MultiNodeConfig) PollInterval() time.Duration {
 	return c.MultiNode.PollInterval.Duration()
 }
@@ -103,6 +108,9 @@ func (c *MultiNodeConfig) SetFrom(f *MultiNodeConfig) {
 	if f.MultiNode.PollFailureThreshold != nil {
 		c.MultiNode.PollFailureThreshold = f.MultiNode.PollFailureThreshold
 	}
+	if f.MultiNode.PollSuccessThreshold != nil {
+		c.MultiNode.PollSuccessThreshold = f.MultiNode.PollSuccessThreshold
+	}
 	if f.MultiNode.PollInterval != nil {
 		c.MultiNode.PollInterval = f.MultiNode.PollInterval
 	}

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

@@ -527,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()
+	if threshold == 0 {
+		return true
+	}
+	pollInterval := n.nodePoolCfg.PollInterval()
+	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()
@@ -572,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:

multinode/node_lifecycle_test.go

@@ -1495,6 +1495,88 @@ func TestUnit_NodeLifecycle_unreachableLoop(t *testing.T) {
 			return node.State() == nodeStateAlive
 		})
 	})
+	t.Run("with PollSuccessThreshold set, without isSyncing, node becomes alive once all probe polls succeed", func(t *testing.T) {
+		t.Parallel()
+		rpc := newMockRPCClient[ID, Head](t)
+		nodeChainID := RandomID()
+		const pollSuccessThreshold = 2
+		node := newAliveNode(t, testNodeOpts{
+			rpc:     rpc,
+			chainID: nodeChainID,
+			config: testNodeConfig{
+				pollSuccessThreshold: pollSuccessThreshold,
+				pollInterval:         tests.TestInterval,
+			},
+		})
+		defer func() { assert.NoError(t, node.close()) }()
+
+		rpc.On("Dial", mock.Anything).Return(nil).Once()
+		rpc.On("ChainID", mock.Anything).Return(nodeChainID, nil).Once()
+		rpc.On("ClientVersion", mock.Anything).Return("", nil).Twice()
+		setupRPCForAliveLoop(t, rpc)
+
+		node.declareUnreachable()
+		tests.AssertEventually(t, func() bool {
+			return node.State() == nodeStateAlive
+		})
+	})
+	t.Run("with PollSuccessThreshold set, node becomes alive once all probe polls succeed", func(t *testing.T) {
+		t.Parallel()
+		rpc := newMockRPCClient[ID, Head](t)
+		nodeChainID := RandomID()
+		const pollSuccessThreshold = 2
+		node := newAliveNode(t, testNodeOpts{
+			rpc:     rpc,
+			chainID: nodeChainID,
+			config: testNodeConfig{
+				nodeIsSyncingEnabled: true,
+				pollSuccessThreshold: pollSuccessThreshold,
+				pollInterval:         tests.TestInterval,
+			},
+		})
+		defer func() { assert.NoError(t, node.close()) }()
+
+		rpc.On("Dial", mock.Anything).Return(nil).Once()
+		rpc.On("ChainID", mock.Anything).Return(nodeChainID, nil).Once()
+		rpc.On("IsSyncing", mock.Anything).Return(false, nil)
+		rpc.On("ClientVersion", mock.Anything).Return("", nil).Twice()
+		setupRPCForAliveLoop(t, rpc)
+
+		node.declareUnreachable()
+		tests.AssertEventually(t, func() bool {
+			return node.State() == nodeStateAlive
+		})
+	})
+	t.Run("with PollSuccessThreshold set, probe poll failure keeps node unreachable and restarts redial", func(t *testing.T) {
+		t.Parallel()
+		rpc := newMockRPCClient[ID, Head](t)
+		nodeChainID := RandomID()
+		lggr, observedLogs := logger.TestObserved(t, zap.WarnLevel)
+		const pollSuccessThreshold = 2
+		node := newAliveNode(t, testNodeOpts{
+			rpc:     rpc,
+			chainID: nodeChainID,
+			lggr:    lggr,
+			config: testNodeConfig{
+				pollSuccessThreshold: pollSuccessThreshold,
+				pollInterval:         tests.TestInterval,
+			},
+		})
+		defer func() { assert.NoError(t, node.close()) }()
+
+		rpc.On("Dial", mock.Anything).Return(nil).Once()
+		rpc.On("ChainID", mock.Anything).Return(nodeChainID, nil).Once()
+		rpc.On("ClientVersion", mock.Anything).Return("", nil).Once()
+		rpc.On("ClientVersion", mock.Anything).Return("", errors.New("probe poll failed")).Once()
+		// after the probe aborts, rpc.Close() is called and the redial backoff fires again; keep failing
+		rpc.On("Dial", mock.Anything).Return(errors.New("failed to dial"))
+		// guard: if current code (no probe) enters aliveLoop, fail the subscribe so the node returns to unreachable
+		rpc.On("SubscribeToHeads", mock.Anything).Return(nil, nil, errors.New("unexpected")).Maybe()
+
+		node.declareUnreachable()
+		tests.AssertLogEventually(t, observedLogs, "Recovery probe poll failed; restarting redial")
+		assert.Equal(t, nodeStateUnreachable, node.State())
+	})
 }
 
 func TestUnit_NodeLifecycle_invalidChainIDLoop(t *testing.T) {

multinode/node_test.go

@@ -16,6 +16,7 @@ import (
 
 type testNodeConfig struct {
 	pollFailureThreshold       uint32
+	pollSuccessThreshold       uint32
 	pollInterval               time.Duration
 	selectionMode              string
 	syncThreshold              uint32
@@ -34,6 +35,10 @@ func (n testNodeConfig) PollFailureThreshold() uint32 {
 	return n.pollFailureThreshold
 }
 
+func (n testNodeConfig) PollSuccessThreshold() uint32 {
+	return n.pollSuccessThreshold
+}
+
 func (n testNodeConfig) PollInterval() time.Duration {
 	return n.pollInterval
 }