diff --git a/multinode/config/config.go b/multinode/config/config.go index be7540e..5ae615e 100644 --- a/multinode/config/config.go +++ b/multinode/config/config.go @@ -48,6 +48,10 @@ func (c *MultiNodeConfig) PollSuccessThreshold() uint32 { return 0 // retaining source compat for -solana; -evm sets via NodePoolConfig } +func (c *MultiNodeConfig) RedialBackoffMin() time.Duration { + return time.Second // retaining source compat for -solana; -evm sets via NodePoolConfig +} + func (c *MultiNodeConfig) PollInterval() time.Duration { return c.MultiNode.PollInterval.Duration() } diff --git a/multinode/node.go b/multinode/node.go index 1db52b9..be30917 100644 --- a/multinode/node.go +++ b/multinode/node.go @@ -19,6 +19,7 @@ var errInvalidChainID = errors.New("invalid chain id") type NodeConfig interface { PollFailureThreshold() uint32 PollSuccessThreshold() uint32 + RedialBackoffMin() time.Duration PollInterval() time.Duration SelectionMode() string SyncThreshold() uint32 diff --git a/multinode/node_lifecycle.go b/multinode/node_lifecycle.go index 46b09a7..9c23515 100644 --- a/multinode/node_lifecycle.go +++ b/multinode/node_lifecycle.go @@ -631,12 +631,20 @@ func (n *node[CHAIN_ID, HEAD, RPC]) unreachableLoop() { lggr.Debugw("Trying to revive unreachable RPC node", "nodeState", n.getCachedState()) dialRetryBackoff := NewRedialBackoff() + dialRetryBackoff.Min = n.nodePoolCfg.RedialBackoffMin() + // The first redial waits RedialBackoffMin, so 0 redials immediately. Later attempts + // back off from there (backoff treats a zero Min as 100ms) up to the usual max. + var redialDelay time.Duration + if dialRetryBackoff.Min > 0 { + redialDelay = dialRetryBackoff.Duration() + } for { select { case <-ctx.Done(): return - case <-time.After(dialRetryBackoff.Duration()): + case <-time.After(redialDelay): + redialDelay = dialRetryBackoff.Duration() lggr.Tracew("Trying to re-dial RPC node", "nodeState", n.getCachedState()) err := n.rpc.Dial(ctx) diff --git a/multinode/node_lifecycle_test.go b/multinode/node_lifecycle_test.go index 38c1deb..068028b 100644 --- a/multinode/node_lifecycle_test.go +++ b/multinode/node_lifecycle_test.go @@ -242,6 +242,7 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { const pollFailureThreshold = 3 node := newSubscribedNode(t, testNodeOpts{ config: testNodeConfig{ + redialBackoffMin: time.Second, // no Dial mock: keep the node unreachable while asserting pollFailureThreshold: pollFailureThreshold, pollInterval: tests.TestInterval, }, @@ -271,9 +272,10 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { const syncThreshold = 10 node := newSubscribedNode(t, testNodeOpts{ config: testNodeConfig{ - pollInterval: tests.TestInterval, - syncThreshold: syncThreshold, - selectionMode: NodeSelectionModeRoundRobin, + redialBackoffMin: time.Second, // no Dial mock: keep the node unreachable while asserting + pollInterval: tests.TestInterval, + syncThreshold: syncThreshold, + selectionMode: NodeSelectionModeRoundRobin, }, rpc: rpc, lggr: lggr, @@ -333,9 +335,10 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { const syncThreshold = 10 node := newSubscribedNode(t, testNodeOpts{ config: testNodeConfig{ - pollInterval: tests.TestInterval, - syncThreshold: syncThreshold, - selectionMode: NodeSelectionModeRoundRobin, + redialBackoffMin: time.Second, // no Dial mock: keep the node unreachable while asserting + pollInterval: tests.TestInterval, + syncThreshold: syncThreshold, + selectionMode: NodeSelectionModeRoundRobin, }, rpc: rpc, lggr: lggr, @@ -525,6 +528,7 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) { lggr := logger.Test(t) node := newDialedNode(t, testNodeOpts{ config: testNodeConfig{ + redialBackoffMin: time.Second, // no Dial mock: keep the node unreachable while asserting finalizedBlockPollInterval: tests.TestInterval, }, chainConfig: clientMocks.ChainConfig{ @@ -1495,6 +1499,52 @@ func TestUnit_NodeLifecycle_unreachableLoop(t *testing.T) { return node.State() == nodeStateAlive }) }) + t.Run("with RedialBackoffMin zero, redials immediately", func(t *testing.T) { + t.Parallel() + rpc := newMockRPCClient[ID, Head](t) + nodeChainID := RandomID() + node := newAliveNode(t, testNodeOpts{ + rpc: rpc, + chainID: nodeChainID, + }) + defer func() { assert.NoError(t, node.close()) }() + + rpc.On("Dial", mock.Anything).Return(nil) + rpc.On("ChainID", mock.Anything).Return(nodeChainID, nil) + setupRPCForAliveLoop(t, rpc) + + start := time.Now() + node.declareUnreachable() + tests.AssertEventually(t, func() bool { + return node.State() == nodeStateAlive + }) + assert.Less(t, time.Since(start), 500*time.Millisecond) + }) + t.Run("with RedialBackoffMin set, waits it before the first redial", func(t *testing.T) { + t.Parallel() + rpc := newMockRPCClient[ID, Head](t) + nodeChainID := RandomID() + node := newAliveNode(t, testNodeOpts{ + rpc: rpc, + chainID: nodeChainID, + config: testNodeConfig{redialBackoffMin: time.Second}, + }) + defer func() { assert.NoError(t, node.close()) }() + + var dialedAt atomic.Int64 + rpc.On("Dial", mock.Anything).Run(func(_ mock.Arguments) { + dialedAt.CompareAndSwap(0, time.Now().UnixNano()) + }).Return(nil) + rpc.On("ChainID", mock.Anything).Return(nodeChainID, nil) + setupRPCForAliveLoop(t, rpc) + + start := time.Now() + node.declareUnreachable() + tests.AssertEventually(t, func() bool { + return node.State() == nodeStateAlive + }) + assert.GreaterOrEqual(t, time.Duration(dialedAt.Load()-start.UnixNano()), time.Second) + }) 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) diff --git a/multinode/node_test.go b/multinode/node_test.go index 7e7444d..78f9077 100644 --- a/multinode/node_test.go +++ b/multinode/node_test.go @@ -17,6 +17,7 @@ import ( type testNodeConfig struct { pollFailureThreshold uint32 pollSuccessThreshold uint32 + redialBackoffMin time.Duration pollInterval time.Duration selectionMode string syncThreshold uint32 @@ -40,6 +41,10 @@ func (n testNodeConfig) PollSuccessThreshold() uint32 { return n.pollSuccessThreshold } +func (n testNodeConfig) RedialBackoffMin() time.Duration { + return n.redialBackoffMin +} + func (n testNodeConfig) PollInterval() time.Duration { return n.pollInterval }