moar tests for safe and finalized bans

proxyd/consensus_poller.go

@@ -406,11 +406,6 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 		}
 	}
 
-	// no block to propose (i.e. initializing consensus)
-	if lowestLatestBlock == 0 {
-		return
-	}
-
 	proposedBlock := lowestLatestBlock
 	proposedBlockHash := lowestLatestBlockHash
 	hasConsensus := false
@@ -424,7 +419,10 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 		log.Debug("validating consensus on block", "lowestLatestBlock", lowestLatestBlock)
 	}
 
-	broken := false
+	// if there is no block to propose, the consensus is automatically broken
+	broken := proposedBlock == 0 && currentConsensusBlockNumber > 0
+
+	if proposedBlock > 0 {
 		for !hasConsensus {
 			allAgreed := true
 			consensusBackends = consensusBackends[:0]
@@ -479,6 +477,7 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 				log.Debug("no consensus, now trying", "block:", proposedBlock)
 			}
 		}
+	}
 
 	if broken {
 		// propagate event to other interested parts, such as cache invalidator
@@ -521,6 +520,16 @@ func (cp *ConsensusPoller) Ban(be *Backend) {
 	defer bs.backendStateMux.Unlock()
 	bs.backendStateMux.Lock()
 	bs.bannedUntil = time.Now().Add(cp.banPeriod)
+	bs.safeBlockNumber = 0
+	bs.finalizedBlockNumber = 0
+}
+
+// Unban remove any bans from the backends
+func (cp *ConsensusPoller) Unban(be *Backend) {
+	bs := cp.backendState[be]
+	defer bs.backendStateMux.Unlock()
+	bs.backendStateMux.Lock()
+	bs.bannedUntil = time.Now().Add(-10 * time.Hour)
 }
 
 // Reset remove any bans from the backends and reset their states

proxyd/integration_tests/consensus_test.go

@@ -330,6 +330,82 @@ func TestConsensus(t *testing.T) {
 		require.Equal(t, 1, len(consensusGroup))
 	})
 
+	t.Run("recover after safe and finalized dropped", func(t *testing.T) {
+		reset()
+		useOnlyNode1()
+		overrideBlock("node1", "latest", "0xd1")
+		overrideBlock("node1", "safe", "0xb1")
+		overrideBlock("node1", "finalized", "0x91")
+		update()
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.NotContains(t, consensusGroup, nodes["node1"].backend)
+		require.True(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 0, len(consensusGroup))
+
+		// unban and see if it recovers
+		bg.Consensus.Unban(nodes["node1"].backend)
+		update()
+
+		consensusGroup = bg.Consensus.GetConsensusGroup()
+		require.Contains(t, consensusGroup, nodes["node1"].backend)
+		require.False(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 1, len(consensusGroup))
+
+		require.Equal(t, "0xd1", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0x91", bg.Consensus.GetFinalizedBlockNumber().String())
+		require.Equal(t, "0xb1", bg.Consensus.GetSafeBlockNumber().String())
+	})
+
+	t.Run("latest dropped below safe, then recovered", func(t *testing.T) {
+		reset()
+		useOnlyNode1()
+		overrideBlock("node1", "latest", "0xd1")
+		update()
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.NotContains(t, consensusGroup, nodes["node1"].backend)
+		require.True(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 0, len(consensusGroup))
+
+		// unban and see if it recovers
+		bg.Consensus.Unban(nodes["node1"].backend)
+		overrideBlock("node1", "safe", "0xb1")
+		overrideBlock("node1", "finalized", "0x91")
+		update()
+
+		consensusGroup = bg.Consensus.GetConsensusGroup()
+		require.Contains(t, consensusGroup, nodes["node1"].backend)
+		require.False(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 1, len(consensusGroup))
+
+		require.Equal(t, "0xd1", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0x91", bg.Consensus.GetFinalizedBlockNumber().String())
+		require.Equal(t, "0xb1", bg.Consensus.GetSafeBlockNumber().String())
+	})
+
+	t.Run("latest dropped below safe, and stayed inconsistent after ban", func(t *testing.T) {
+		reset()
+		useOnlyNode1()
+		overrideBlock("node1", "latest", "0xd1")
+		update()
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.NotContains(t, consensusGroup, nodes["node1"].backend)
+		require.True(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 0, len(consensusGroup))
+
+		// unban and see if it recovers - it should not since the blocks stays the same
+		bg.Consensus.Unban(nodes["node1"].backend)
+		update()
+
+		// should be banned again
+		consensusGroup = bg.Consensus.GetConsensusGroup()
+		require.NotContains(t, consensusGroup, nodes["node1"].backend)
+		require.True(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 0, len(consensusGroup))
+	})
+
 	t.Run("broken consensus", func(t *testing.T) {
 		reset()
 		listenerCalled := false

proxyd/integration_tests/testdata/consensus_responses.yml

@@ -107,6 +107,17 @@
         "number": "0x200"
       }
     }
+- method: eth_getBlockByNumber
+  block: 0x91
+  response: >
+    {
+      "jsonrpc": "2.0",
+      "id": 67,
+      "result": {
+        "hash": "hash_0x91",
+        "number": "0x91"
+      }
+    }
 - method: eth_getBlockByNumber
   block: safe
   response: >
@@ -151,3 +162,14 @@
         "number": "0xc1"
       }
     }
+- method: eth_getBlockByNumber
+  block: 0xd1
+  response: >
+    {
+      "jsonrpc": "2.0",
+      "id": 67,
+      "result": {
+        "hash": "hash_0xd1",
+        "number": "0xd1"
+      }
+    }