feat(proxyd): add threshold configs to handle safe and finalized block drift (#229)

* feat(proxyd): add threshold configs to handle safe and finalized block drift

Author: hmvien

proxyd/backend.go

@@ -155,9 +155,11 @@ type Backend struct {
 	skipPeerCountCheck bool
 	forcedCandidate    bool
 
-	maxDegradedLatencyThreshold time.Duration
-	maxLatencyThreshold         time.Duration
-	maxErrorRateThreshold       float64
+	safeBlockDriftThreshold      uint64
+	finalizedBlockDriftThreshold uint64
+	maxDegradedLatencyThreshold  time.Duration
+	maxLatencyThreshold          time.Duration
+	maxErrorRateThreshold        float64
 
 	latencySlidingWindow            *sw.AvgSlidingWindow
 	networkRequestsSlidingWindow    *sw.AvgSlidingWindow
@@ -244,6 +246,18 @@ func WithSkipIsSyncingCheck(skipIsSyncingCheck bool) BackendOpt {
 	}
 }
 
+func WithSafeBlockDriftThreshold(safeBlockDriftThreshold uint64) BackendOpt {
+	return func(b *Backend) {
+		b.safeBlockDriftThreshold = safeBlockDriftThreshold
+	}
+}
+
+func WithFinalizedBlockDriftThreshold(finalizedBlockDriftThreshold uint64) BackendOpt {
+	return func(b *Backend) {
+		b.finalizedBlockDriftThreshold = finalizedBlockDriftThreshold
+	}
+}
+
 func WithConsensusSkipPeerCountCheck(skipPeerCountCheck bool) BackendOpt {
 	return func(b *Backend) {
 		b.skipPeerCountCheck = skipPeerCountCheck

proxyd/config.go

@@ -112,6 +112,9 @@ type BackendConfig struct {
 
 	SkipIsSyncingCheck bool `toml:"skip_is_syncing_check"`
 
+	SafeBlockDriftThreshold      uint64 `toml:"safe_block_drift_threshold"`
+	FinalizedBlockDriftThreshold uint64 `toml:"finalized_block_drift_threshold"`
+
 	ConsensusSkipPeerCountCheck bool   `toml:"consensus_skip_peer_count"`
 	ConsensusForcedCandidate    bool   `toml:"consensus_forced_candidate"`
 	ConsensusReceiptsTarget     string `toml:"consensus_receipts_target"`

proxyd/consensus_poller.go

@@ -400,6 +400,8 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 
 	// sanity check for latest, safe and finalized block tags
 	expectedBlockTags := cp.checkExpectedBlockTags(
+		be.safeBlockDriftThreshold,
+		be.finalizedBlockDriftThreshold,
 		latestBlockNumber,
 		bs.safeBlockNumber, safeBlockNumber,
 		bs.finalizedBlockNumber, finalizedBlockNumber)
@@ -420,15 +422,27 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
 }
 
 // checkExpectedBlockTags for unexpected conditions on block tags
-// - finalized block number should never decrease
-// - safe block number should never decrease
+// - finalized block number should never decrease by more than finalizedBlockDriftThreshold
+// - safe block number should never decrease by more than safeBlockDriftThreshold
 // - finalized block should be <= safe block <= latest block
 func (cp *ConsensusPoller) checkExpectedBlockTags(
+	safeBlockDriftThreshold uint64,
+	finalizedBlockDriftThreshold uint64,
 	currentLatest hexutil.Uint64,
 	oldSafe hexutil.Uint64, currentSafe hexutil.Uint64,
 	oldFinalized hexutil.Uint64, currentFinalized hexutil.Uint64) bool {
-	return currentFinalized >= oldFinalized &&
-		currentSafe >= oldSafe &&
+
+	minSafeBlockAllowance := oldSafe
+	minFinalizedBlockAllowance := oldFinalized
+	if minSafeBlockAllowance > hexutil.Uint64(safeBlockDriftThreshold) {
+		minSafeBlockAllowance -= hexutil.Uint64(safeBlockDriftThreshold)
+	}
+	if minFinalizedBlockAllowance > hexutil.Uint64(finalizedBlockDriftThreshold) {
+		minFinalizedBlockAllowance -= hexutil.Uint64(finalizedBlockDriftThreshold)
+	}
+
+	return currentFinalized >= minFinalizedBlockAllowance &&
+		currentSafe >= minSafeBlockAllowance &&
 		currentFinalized <= currentSafe &&
 		currentSafe <= currentLatest
 }

proxyd/example.config.toml

@@ -80,6 +80,12 @@ consensus_receipts_target = "eth_getBlockReceipts"
 # Allow backends to skip eth_syncing checks, default false
 # skip_is_syncing_check = false
 
+# Allow backends safe and finalized block to drift backward up to the threshold.
+# Default is 0, meaning no drift is allowed e.g. newSafe >= oldSafe and newFinalized >= oldFinalized
+# Drift allows newSafe >= (oldSafe - safe_block_drift_threshold) and newFinalized >= (oldFinalized - finalized_block_drift_threshold)
+# safe_block_drift_threshold = 0
+# finalized_block_drift_threshold = 0
+
 [backends.alchemy]
 rpc_url = ""
 ws_url = ""

proxyd/integration_tests/consensus_custom_config_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/ethereum-optimism/infra/proxyd"
 	ms "github.com/ethereum-optimism/infra/proxyd/tools/mockserver/handler"
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -129,3 +130,165 @@ func TestConsensusSkipSyncTest(t *testing.T) {
 		require.Equal(t, 2, len(consensusGroup))
 	})
 }
+
+func TestConsensusBlockDriftThreshold(t *testing.T) {
+	nodes, bg, _, shutdown := setupCustomConfig(t, "consensus_block_drift_threshold")
+	defer nodes["node1"].mockBackend.Close()
+	defer nodes["node2"].mockBackend.Close()
+	defer shutdown()
+
+	ctx := context.Background()
+
+	// poll for updated consensus
+	update := func() {
+		for _, be := range bg.Backends {
+			bg.Consensus.UpdateBackend(ctx, be)
+		}
+		bg.Consensus.UpdateBackendGroupConsensus(ctx)
+	}
+
+	// convenient methods to manipulate state and mock responses
+	reset := func() {
+		for _, node := range nodes {
+			node.handler.ResetOverrides()
+			node.mockBackend.Reset()
+			node.backend.ClearSlidingWindows()
+		}
+		bg.Consensus.ClearListeners()
+		bg.Consensus.Reset()
+
+	}
+
+	initSetupAndAssetions := func() {
+		reset()
+		update()
+
+		require.Equal(t, "0x101", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0xe1", bg.Consensus.GetSafeBlockNumber().String())
+		require.Equal(t, "0xc1", bg.Consensus.GetFinalizedBlockNumber().String())
+	}
+
+	override := func(node string, method string, block string, response string) {
+		if _, ok := nodes[node]; !ok {
+			t.Fatalf("node %s does not exist in the nodes map", node)
+		}
+		nodes[node].handler.AddOverride(&ms.MethodTemplate{
+			Method:   method,
+			Block:    block,
+			Response: response,
+		})
+	}
+
+	overrideBlock := func(node string, blockRequest string, blockResponse string) {
+		override(node,
+			"eth_getBlockByNumber",
+			blockRequest,
+			buildResponse(map[string]string{
+				"number": blockResponse,
+				"hash":   "hash_" + blockResponse,
+			}))
+	}
+
+	overridePeerCount := func(node string, count int) {
+		override(node, "net_peerCount", "", buildResponse(hexutil.Uint64(count).String()))
+	}
+
+	// force ban node2 and make sure node1 is the only one in consensus
+	useOnlyNode1 := func() {
+		overridePeerCount("node2", 0)
+		update()
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.Equal(t, 1, len(consensusGroup))
+		require.Contains(t, consensusGroup, nodes["node1"].backend)
+		nodes["node1"].mockBackend.Reset()
+	}
+
+	t.Run("allow backend if tags are messed if below tolerance - safe dropped", func(t *testing.T) {
+		initSetupAndAssetions()
+
+		overrideBlock("node1", "safe", "0xe0") // 1 blocks behind is ok
+		update()
+
+		require.Equal(t, "0x101", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0xe0", bg.Consensus.GetSafeBlockNumber().String())
+		require.Equal(t, "0xc1", bg.Consensus.GetFinalizedBlockNumber().String())
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.Contains(t, consensusGroup, nodes["node1"].backend)
+		require.False(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 2, len(consensusGroup))
+	})
+
+	t.Run("ban backend if tags are messed above tolerance - safe dropped", func(t *testing.T) {
+		initSetupAndAssetions()
+		overrideBlock("node1", "safe", "0xdf") // 2 blocks behind is not ok
+		update()
+
+		require.Equal(t, "0x101", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0xe1", bg.Consensus.GetSafeBlockNumber().String())
+		require.Equal(t, "0xc1", bg.Consensus.GetFinalizedBlockNumber().String())
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.NotContains(t, consensusGroup, nodes["node1"].backend)
+		require.True(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 1, len(consensusGroup))
+	})
+
+	t.Run("allow backend if tags are messed if below tolerance - finalized dropped", func(t *testing.T) {
+		initSetupAndAssetions()
+		overrideBlock("node1", "finalized", "0xbf") // finalized 2 blocks behind is ok
+		update()
+
+		require.Equal(t, "0x101", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0xe1", bg.Consensus.GetSafeBlockNumber().String())
+		require.Equal(t, "0xbf", bg.Consensus.GetFinalizedBlockNumber().String())
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.Contains(t, consensusGroup, nodes["node1"].backend)
+		require.False(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		require.Equal(t, 2, len(consensusGroup))
+	})
+
+	t.Run("ban backend if tags are messed - finalized dropped", func(t *testing.T) {
+		initSetupAndAssetions()
+		overrideBlock("node1", "finalized", "0xbe") // finalized 3 blocks behind is not ok
+		update()
+
+		require.Equal(t, "0x101", bg.Consensus.GetLatestBlockNumber().String())
+		require.Equal(t, "0xe1", bg.Consensus.GetSafeBlockNumber().String())
+		require.Equal(t, "0xc1", bg.Consensus.GetFinalizedBlockNumber().String())
+
+		consensusGroup := bg.Consensus.GetConsensusGroup()
+		require.NotContains(t, consensusGroup, nodes["node1"].backend)
+		require.True(t, bg.Consensus.IsBanned(nodes["node1"].backend))
+		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, "0xb1", bg.Consensus.GetSafeBlockNumber().String())
+		require.Equal(t, "0x91", bg.Consensus.GetFinalizedBlockNumber().String())
+	})
+}

proxyd/integration_tests/testdata/consensus_block_drift_threshold.toml

@@ -0,0 +1,33 @@
+[server]
+rpc_port = 8545
+
+[backend]
+response_timeout_seconds = 1
+max_degraded_latency_threshold = "30ms"
+
+[backends]
+[backends.node1]
+rpc_url = "$NODE1_URL"
+safe_block_drift_threshold = 1
+finalized_block_drift_threshold = 2
+
+[backends.node2]
+rpc_url = "$NODE2_URL"
+
+[backend_groups]
+[backend_groups.node]
+backends = ["node1", "node2"]
+routing_strategy = "consensus_aware"
+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 = 8
+consensus_min_peer_count = 4
+
+
+[rpc_method_mappings]
+eth_call = "node"
+eth_chainId = "node"
+eth_blockNumber = "node"
+eth_getBlockByNumber = "node"
+consensus_getReceipts = "node"

proxyd/proxyd.go

@@ -194,6 +194,8 @@ func Start(config *Config) (*Server, func(), error) {
 		}
 		opts = append(opts, WithProxydIP(os.Getenv("PROXYD_IP")))
 		opts = append(opts, WithSkipIsSyncingCheck(cfg.SkipIsSyncingCheck))
+		opts = append(opts, WithSafeBlockDriftThreshold(cfg.SafeBlockDriftThreshold))
+		opts = append(opts, WithFinalizedBlockDriftThreshold(cfg.FinalizedBlockDriftThreshold))
 		opts = append(opts, WithConsensusSkipPeerCountCheck(cfg.ConsensusSkipPeerCountCheck))
 		opts = append(opts, WithConsensusForcedCandidate(cfg.ConsensusForcedCandidate))
 		opts = append(opts, WithWeight(cfg.Weight))