proxyd: add limit to consensus block lag

proxyd/config.go

@@ -105,6 +105,7 @@ type BackendGroupConfig struct {
 
 	ConsensusBanPeriod          TOMLDuration `toml:"consensus_ban_period"`
 	ConsensusMaxUpdateThreshold TOMLDuration `toml:"consensus_max_update_threshold"`
+	ConsensusMaxBlockLag        uint64       `toml:"consensus_max_block_lag"`
 	ConsensusMinPeerCount       int          `toml:"consensus_min_peer_count"`
 }
 

proxyd/consensus_poller.go

@@ -35,6 +35,7 @@ type ConsensusPoller struct {
 
 	banPeriod          time.Duration
 	maxUpdateThreshold time.Duration
+	maxBlockLag        uint64
 }
 
 type backendState struct {
@@ -160,6 +161,12 @@ func WithMaxUpdateThreshold(maxUpdateThreshold time.Duration) ConsensusOpt {
 	}
 }
 
+func WithMaxBlockLag(maxBlockLag uint64) ConsensusOpt {
+	return func(cp *ConsensusPoller) {
+		cp.maxBlockLag = maxBlockLag
+	}
+}
+
 func WithMinPeerCount(minPeerCount uint64) ConsensusOpt {
 	return func(cp *ConsensusPoller) {
 		cp.minPeerCount = minPeerCount
@@ -181,6 +188,7 @@ func NewConsensusPoller(bg *BackendGroup, opts ...ConsensusOpt) *ConsensusPoller
 
 		banPeriod:          5 * time.Minute,
 		maxUpdateThreshold: 30 * time.Second,
+		maxBlockLag:        50,
 		minPeerCount:       3,
 	}
 
@@ -270,7 +278,12 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 			continue
 		}
 
-		if lowestBlock == 0 || backendLatestBlockNumber < lowestBlock {
+		// find the highest common ancestor, ignoring backends that are too far lagging behind
+		// when the backend is too far ahead from current lowest, the current lowest is ignored
+		// when the backend if too far behind, the backend itself is ignored
+		if lowestBlock == 0 ||
+			backendLatestBlockNumber > lowestBlock && uint64(backendLatestBlockNumber-lowestBlock) > cp.maxBlockLag ||
+			backendLatestBlockNumber < lowestBlock && uint64(lowestBlock-backendLatestBlockNumber) < cp.maxBlockLag {
 			lowestBlock = backendLatestBlockNumber
 			lowestBlockHash = backendLatestBlockHash
 		}
@@ -308,12 +321,14 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
 				- not banned
 				- with minimum peer count
 				- updated recently
+				- not lagging
 			*/
-			peerCount, _, _, lastUpdate, bannedUntil := cp.getBackendState(be)
+			peerCount, latestBlockNumber, _, lastUpdate, bannedUntil := cp.getBackendState(be)
 			notUpdated := lastUpdate.Add(cp.maxUpdateThreshold).Before(time.Now())
 			isBanned := time.Now().Before(bannedUntil)
 			notEnoughPeers := !be.skipPeerCountCheck && peerCount < cp.minPeerCount
-			if !be.IsHealthy() || be.IsRateLimited() || !be.Online() || notUpdated || isBanned || notEnoughPeers {
+			lagging := (latestBlockNumber < proposedBlock) && uint64(proposedBlock-latestBlockNumber) >= cp.maxBlockLag
+			if !be.IsHealthy() || be.IsRateLimited() || !be.Online() || notUpdated || isBanned || notEnoughPeers || lagging {
 				filteredBackendsNames = append(filteredBackendsNames, be.Name)
 				continue
 			}

proxyd/example.config.toml

@@ -93,6 +93,8 @@ backends = ["infura"]
 # consensus_ban_period = "1m"
 # Maximum delay for update the backend, default 30s
 # consensus_max_update_threshold = "20s"
+# Maximum block lag, default 50
+# consensus_max_block_lag = 10
 # Minimum peer count, default 3
 # consensus_min_peer_count = 4
 

proxyd/integration_tests/consensus_test.go

@@ -97,6 +97,44 @@ func TestConsensus(t *testing.T) {
 		require.Equal(t, 1, len(consensusGroup))
 	})
 
+	t.Run("prevent using a backend lagging behind", func(t *testing.T) {
+		h1.ResetOverrides()
+		h2.ResetOverrides()
+		bg.Consensus.Unban()
+
+		h1.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x1", "hash1"),
+		})
+
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x100", "hash0x100"),
+		})
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "0x100",
+			Response: buildGetBlockResponse("0x100", "hash0x100"),
+		})
+
+		for _, be := range bg.Backends {
+			bg.Consensus.UpdateBackend(ctx, be)
+		}
+		bg.Consensus.UpdateBackendGroupConsensus(ctx)
+
+		// since we ignored node1, the consensus should be at 0x100
+		require.Equal(t, "0x100", bg.Consensus.GetConsensusBlockNumber().String())
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+
+		be := backend(bg, "node1")
+		require.NotNil(t, be)
+		require.NotContains(t, consensusGroup, be)
+		require.Equal(t, 1, len(consensusGroup))
+	})
+
 	t.Run("prevent using a backend not in sync", func(t *testing.T) {
 		h1.ResetOverrides()
 		h2.ResetOverrides()

proxyd/integration_tests/testdata/consensus.toml

@@ -18,6 +18,7 @@ consensus_aware = true
 consensus_handler = "noop" # allow more control over the consensus poller for tests
 consensus_ban_period = "1m"
 consensus_max_update_threshold = "2m"
+consensus_max_block_lag = 50
 consensus_min_peer_count = 4
 
 [rpc_method_mappings]

proxyd/proxyd.go

@@ -328,6 +328,9 @@ func Start(config *Config) (*Server, func(), error) {
 			if bgcfg.ConsensusMaxUpdateThreshold > 0 {
 				copts = append(copts, WithMaxUpdateThreshold(time.Duration(bgcfg.ConsensusMaxUpdateThreshold)))
 			}
+			if bgcfg.ConsensusMaxBlockLag > 0 {
+				copts = append(copts, WithMaxBlockLag(bgcfg.ConsensusMaxBlockLag))
+			}
 			if bgcfg.ConsensusMinPeerCount > 0 {
 				copts = append(copts, WithMinPeerCount(uint64(bgcfg.ConsensusMinPeerCount)))
 			}