@@ -202,6 +202,33 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
202202 tests .AssertLogEventually (t , observedLogs , fmt .Sprintf ("RPC endpoint failed to respond to %d consecutive polls" , pollFailureThreshold ))
203203 assert .Equal (t , nodeStateAlive , node .State ())
204204 })
205+ t .Run ("with threshold poll failures, we are the last node alive, but is a proxy, transitions to unreachable" , func (t * testing.T ) {
206+ t .Parallel ()
207+ rpc := newMockRPCClient [ID , Head ](t )
208+ lggr , observedLogs := logger .TestObserved (t , zap .DebugLevel )
209+ const pollFailureThreshold = 3
210+ node := newSubscribedNode (t , testNodeOpts {
211+ config : testNodeConfig {
212+ pollFailureThreshold : pollFailureThreshold ,
213+ pollInterval : tests .TestInterval ,
214+ },
215+ rpc : rpc ,
216+ lggr : lggr ,
217+ isLoadBalancedRPC : true ,
218+ })
219+ defer func () { assert .NoError (t , node .close ()) }()
220+ poolInfo := newMockPoolChainInfoProvider (t )
221+ poolInfo .On ("LatestChainInfo" ).Return (1 , ChainInfo {
222+ BlockNumber : 20 ,
223+ }).Once ()
224+ node .SetPoolChainInfoProvider (poolInfo )
225+ rpc .On ("GetInterceptedChainInfo" ).Return (ChainInfo {BlockNumber : 20 }, ChainInfo {BlockNumber : 20 })
226+ pollError := errors .New ("failed to get ClientVersion" )
227+ rpc .On ("ClientVersion" , mock .Anything ).Return ("" , pollError )
228+ node .declareAlive ()
229+ tests .AssertLogEventually (t , observedLogs , fmt .Sprintf ("RPC endpoint failed to respond to %d consecutive polls" , pollFailureThreshold ))
230+ assert .Equal (t , nodeStateUnreachable , node .State ())
231+ })
205232 t .Run ("when behind more than SyncThreshold, transitions to out of sync" , func (t * testing.T ) {
206233 t .Parallel ()
207234 rpc := newMockRPCClient [ID , Head ](t )
@@ -264,6 +291,42 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
264291 node .declareAlive ()
265292 tests .AssertLogEventually (t , observedLogs , fmt .Sprintf ("RPC endpoint has fallen behind; %s %s" , msgCannotDisable , msgDegradedState ))
266293 })
294+ 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 ) {
295+ t .Parallel ()
296+ rpc := newMockRPCClient [ID , Head ](t )
297+ lggr , observedLogs := logger .TestObserved (t , zap .DebugLevel )
298+ const syncThreshold = 10
299+ node := newSubscribedNode (t , testNodeOpts {
300+ config : testNodeConfig {
301+ pollInterval : tests .TestInterval ,
302+ syncThreshold : syncThreshold ,
303+ selectionMode : NodeSelectionModeRoundRobin ,
304+ },
305+ rpc : rpc ,
306+ lggr : lggr ,
307+ isLoadBalancedRPC : true ,
308+ })
309+ defer func () { assert .NoError (t , node .close ()) }()
310+ rpc .On ("ClientVersion" , mock .Anything ).Return ("" , nil )
311+ const mostRecentBlock = 20
312+ rpc .On ("GetInterceptedChainInfo" ).Return (ChainInfo {BlockNumber : mostRecentBlock }, ChainInfo {BlockNumber : 30 }).Twice ()
313+ poolInfo := newMockPoolChainInfoProvider (t )
314+ poolInfo .On ("LatestChainInfo" ).Return (1 , ChainInfo {
315+ BlockNumber : syncThreshold + mostRecentBlock + 1 ,
316+ TotalDifficulty : big .NewInt (10 ),
317+ })
318+ node .SetPoolChainInfoProvider (poolInfo )
319+ // tries to redial in outOfSync
320+ rpc .On ("Dial" , mock .Anything ).Return (errors .New ("failed to dial" )).Run (func (_ mock.Arguments ) {
321+ assert .Equal (t , nodeStateOutOfSync , node .State ())
322+ }).Once ()
323+ rpc .On ("Dial" , mock .Anything ).Run (func (_ mock.Arguments ) {
324+ require .Equal (t , nodeStateOutOfSync , node .State ())
325+ }).Return (errors .New ("failed to dial" )).Maybe ()
326+ node .declareAlive ()
327+ tests .AssertLogEventually (t , observedLogs , "Dial failed: Node is unreachable" )
328+ assert .Equal (t , nodeStateUnreachable , node .State ())
329+ })
267330 t .Run ("when behind but SyncThreshold=0, stay alive" , func (t * testing.T ) {
268331 t .Parallel ()
269332 rpc := newMockRPCClient [ID , Head ](t )
@@ -333,7 +396,36 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
333396 tests .AssertLogEventually (t , observedLogs , fmt .Sprintf ("RPC endpoint detected out of sync; %s %s" , msgCannotDisable , msgDegradedState ))
334397 assert .Equal (t , nodeStateAlive , node .State ())
335398 })
336-
399+ 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 ) {
400+ t .Parallel ()
401+ rpc := newMockRPCClient [ID , Head ](t )
402+ rpc .On ("GetInterceptedChainInfo" ).Return (ChainInfo {}, ChainInfo {})
403+ lggr , observedLogs := logger .TestObserved (t , zap .DebugLevel )
404+ node := newSubscribedNode (t , testNodeOpts {
405+ config : testNodeConfig {},
406+ lggr : lggr ,
407+ chainConfig : clientMocks.ChainConfig {
408+ NoNewHeadsThresholdVal : tests .TestInterval ,
409+ },
410+ rpc : rpc ,
411+ isLoadBalancedRPC : true ,
412+ })
413+ defer func () { assert .NoError (t , node .close ()) }()
414+ poolInfo := newMockPoolChainInfoProvider (t )
415+ poolInfo .On ("LatestChainInfo" ).Return (1 , ChainInfo {
416+ BlockNumber : 20 ,
417+ TotalDifficulty : big .NewInt (10 ),
418+ }).Once ()
419+ node .SetPoolChainInfoProvider (poolInfo )
420+ // tries to redial in outOfSync
421+ rpc .On ("Dial" , mock .Anything ).Return (errors .New ("failed to dial" )).Run (func (_ mock.Arguments ) {
422+ assert .Equal (t , nodeStateOutOfSync , node .State ())
423+ }).Once ()
424+ rpc .On ("Dial" , mock .Anything ).Return (errors .New ("failed to dial" )).Maybe ()
425+ node .declareAlive ()
426+ tests .AssertLogEventually (t , observedLogs , "Dial failed: Node is unreachable" )
427+ assert .Equal (t , nodeStateUnreachable , node .State ())
428+ })
337429 t .Run ("rpc closed head channel" , func (t * testing.T ) {
338430 t .Parallel ()
339431 rpc := newMockRPCClient [ID , Head ](t )
@@ -555,6 +647,40 @@ func TestUnit_NodeLifecycle_aliveLoop(t *testing.T) {
555647 tests .AssertLogEventually (t , observed , fmt .Sprintf ("RPC's finalized state is out of sync; %s %s" , msgCannotDisable , msgDegradedState ))
556648 assert .Equal (t , nodeStateAlive , node .State ())
557649 })
650+ 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 ) {
651+ t .Parallel ()
652+ rpc := newMockRPCClient [ID , Head ](t )
653+ rpc .On ("GetInterceptedChainInfo" ).Return (ChainInfo {}, ChainInfo {}).Once ()
654+ rpc .On ("SubscribeToFinalizedHeads" , mock .Anything ).Return (make (<- chan Head ), newSub (t ), nil ).Once ()
655+ lggr , observedLogs := logger .TestObserved (t , zap .DebugLevel )
656+ noNewFinalizedHeadsThreshold := tests .TestInterval
657+ node := newSubscribedNode (t , testNodeOpts {
658+ config : testNodeConfig {},
659+ chainConfig : clientMocks.ChainConfig {
660+ NoNewFinalizedHeadsThresholdVal : noNewFinalizedHeadsThreshold ,
661+ IsFinalityTagEnabled : true ,
662+ },
663+ rpc : rpc ,
664+ lggr : lggr ,
665+ isLoadBalancedRPC : true ,
666+ })
667+ defer func () { assert .NoError (t , node .close ()) }()
668+ poolInfo := newMockPoolChainInfoProvider (t )
669+ poolInfo .On ("LatestChainInfo" ).Return (1 , ChainInfo {
670+ BlockNumber : 20 ,
671+ TotalDifficulty : big .NewInt (10 ),
672+ }).Once ()
673+ node .SetPoolChainInfoProvider (poolInfo )
674+ // tries to redial in outOfSync
675+ // tries to redial in outOfSync
676+ rpc .On ("Dial" , mock .Anything ).Return (errors .New ("failed to dial" )).Run (func (_ mock.Arguments ) {
677+ assert .Equal (t , nodeStateOutOfSync , node .State ())
678+ }).Once ()
679+ rpc .On ("Dial" , mock .Anything ).Return (errors .New ("failed to dial" )).Maybe ()
680+ node .declareAlive ()
681+ tests .AssertLogEventually (t , observedLogs , "Dial failed: Node is unreachable" )
682+ assert .Equal (t , nodeStateUnreachable , node .State ())
683+ })
558684 t .Run ("If finalized subscription returns an error, transitions to unreachable" , func (t * testing.T ) {
559685 t .Parallel ()
560686 rpc := newMockRPCClient [ID , Head ](t )
@@ -937,6 +1063,42 @@ func TestUnit_NodeLifecycle_outOfSyncLoop(t *testing.T) {
9371063 return node .State () == nodeStateAlive
9381064 })
9391065 })
1066+ t .Run ("becomes alive if there is no other nodes, unless proxy" , func (t * testing.T ) {
1067+ t .Parallel ()
1068+ rpc := newMockRPCClient [ID , Head ](t )
1069+ nodeChainID := RandomID ()
1070+ lggr , _ := logger .TestObserved (t , zap .DebugLevel )
1071+ node := newAliveNode (t , testNodeOpts {
1072+ chainConfig : clientMocks.ChainConfig {
1073+ NoNewHeadsThresholdVal : tests .TestInterval ,
1074+ },
1075+ rpc : rpc ,
1076+ chainID : nodeChainID ,
1077+ lggr : lggr ,
1078+ isLoadBalancedRPC : true ,
1079+ })
1080+ defer func () { assert .NoError (t , node .close ()) }()
1081+ poolInfo := newMockPoolChainInfoProvider (t )
1082+ poolInfo .On ("LatestChainInfo" ).Return (0 , ChainInfo {
1083+ BlockNumber : 100 ,
1084+ TotalDifficulty : big .NewInt (200 ),
1085+ })
1086+ node .SetPoolChainInfoProvider (poolInfo )
1087+ rpc .On ("GetInterceptedChainInfo" ).Return (ChainInfo {}, ChainInfo {})
1088+
1089+ rpc .On ("Dial" , mock .Anything ).Return (nil ).Once ()
1090+ rpc .On ("ChainID" , mock .Anything ).Return (nodeChainID , nil ).Once ()
1091+
1092+ outOfSyncSubscription := newMockSubscription (t )
1093+ outOfSyncSubscription .On ("Err" ).Return ((<- chan error )(nil ))
1094+ outOfSyncSubscription .On ("Unsubscribe" ).Once ()
1095+ rpc .On ("SubscribeToHeads" , mock .Anything ).Return (make (<- chan Head ), outOfSyncSubscription , nil ).Once ()
1096+ rpc .On ("Dial" , mock .Anything ).Return (errors .New ("failed to redial" )).Maybe ()
1097+ node .declareOutOfSync (syncStatusNoNewHead )
1098+ tests .AssertEventually (t , func () bool {
1099+ return node .State () == nodeStateUnreachable
1100+ })
1101+ })
9401102 t .Run ("Stays out-of-sync if received new head, but lags behind pool" , func (t * testing.T ) {
9411103 t .Parallel ()
9421104 rpc := newMockRPCClient [ID , Head ](t )
0 commit comments