Merge branch 'develop' into aj/change-check

op-batcher/flags/flags.go

@@ -110,6 +110,7 @@ var optionalFlags = []cli.Flag{
 	TargetNumFramesFlag,
 	ApproxComprRatioFlag,
 	StoppedFlag,
+	SequencerHDPathFlag,
 }
 
 func init() {

op-program/host/cmd/main_test.go

@@ -78,6 +78,11 @@ func TestNetwork(t *testing.T) {
 		name := name
 		expected := cfg
 		t.Run("Network_"+name, func(t *testing.T) {
+			// TODO(CLI-3936) Re-enable test for other networks once bedrock migration is complete
+			if name != "goerli" {
+				t.Skipf("Not requiring chain config for network %s", name)
+				return
+			}
 			args := replaceRequiredArg("--network", name)
 			cfg := configForArgs(t, args)
 			require.Equal(t, expected, *cfg.Rollup)

op-proposer/flags/flags.go

@@ -59,11 +59,11 @@ var requiredFlags = []cli.Flag{
 var optionalFlags = []cli.Flag{
 	PollIntervalFlag,
 	AllowNonFinalizedFlag,
+	L2OutputHDPathFlag,
 }
 
 func init() {
-	requiredFlags = append(requiredFlags, oprpc.CLIFlags(envVarPrefix)...)
-
+	optionalFlags = append(optionalFlags, oprpc.CLIFlags(envVarPrefix)...)
 	optionalFlags = append(optionalFlags, oplog.CLIFlags(envVarPrefix)...)
 	optionalFlags = append(optionalFlags, opmetrics.CLIFlags(envVarPrefix)...)
 	optionalFlags = append(optionalFlags, oppprof.CLIFlags(envVarPrefix)...)

op-service/txmgr/cli.go

@@ -60,10 +60,8 @@ func CLIFlags(envPrefix string) []cli.Flag {
 			Usage:  "The HD path used to derive the sequencer wallet from the mnemonic. The mnemonic flag must also be set.",
 			EnvVar: opservice.PrefixEnvVar(envPrefix, "HD_PATH"),
 		},
-		SequencerHDPathFlag,
-		L2OutputHDPathFlag,
 		cli.StringFlag{
-			Name:   "private-key",
+			Name:   PrivateKeyFlagName,
 			Usage:  "The private key to use with the service. Must not be used with mnemonic.",
 			EnvVar: opservice.PrefixEnvVar(envPrefix, "PRIVATE_KEY"),
 		},

packages/contracts-bedrock/scripts/universal/SafeBuilder.sol

@@ -34,14 +34,10 @@ abstract contract SafeBuilder is EnhancedScript, GlobalConstants {
     address[] internal approvals;
 
     /**
-     * @notice The entrypoint to this script.
+     * -----------------------------------------------------------
+     * Virtual Functions
+     * -----------------------------------------------------------
      */
-    function run(address _safe, address _proxyAdmin) public returns (bool) {
-        vm.startBroadcast();
-        bool success = _run(_safe, _proxyAdmin);
-        if (success) _postCheck();
-        return success;
-    }
 
     /**
      * @notice Follow up assertions to ensure that the script ran to completion.
@@ -56,7 +52,30 @@ abstract contract SafeBuilder is EnhancedScript, GlobalConstants {
     /**
      * @notice Internal helper function to compute the safe transaction hash.
      */
-    function _getTransactionHash(address _safe, address _proxyAdmin) internal returns (bytes32) {
+    function computeSafeTransactionHash(address _safe, address _proxyAdmin) public virtual returns (bytes32) {
+        return _getTransactionHash(_safe, _proxyAdmin);
+    }
+
+    /**
+     * -----------------------------------------------------------
+     * Implemented Functions
+     * -----------------------------------------------------------
+     */
+
+    /**
+     * @notice The entrypoint to this script.
+     */
+    function run(address _safe, address _proxyAdmin) public returns (bool) {
+        vm.startBroadcast();
+        bool success = _run(_safe, _proxyAdmin);
+        if (success) _postCheck();
+        return success;
+    }
+
+    /**
+     * @notice Computes the safe transaction hash for the provided safe and proxy admin.
+     */
+    function _getTransactionHash(address _safe, address _proxyAdmin) internal view returns (bytes32) {
         // Ensure that the required contracts exist
         require(address(multicall).code.length > 0, "multicall3 not deployed");
         require(_safe.code.length > 0, "no code at safe address");
@@ -84,7 +103,6 @@ abstract contract SafeBuilder is EnhancedScript, GlobalConstants {
         return hash;
     }
 
-
     /**
      * @notice The implementation of the upgrade. Split into its own function
      *         to allow for testability. This is subject to a race condition if
@@ -189,6 +207,5 @@ abstract contract SafeBuilder is EnhancedScript, GlobalConstants {
         }
         return signatures;
     }
-
 }
 

packages/contracts-bedrock/scripts/upgrades/PostSherlock.s.sol

@@ -115,24 +115,24 @@ contract PostSherlockL1 is SafeBuilder {
      *         could be added.
      */
     function test_script_succeeds() skipWhenNotForking external {
-        address safe;
-        address proxyAdmin;
+        address _safe;
+        address _proxyAdmin;
 
         if (block.chainid == GOERLI) {
-            safe = 0xBc1233d0C3e6B5d53Ab455cF65A6623F6dCd7e4f;
-            proxyAdmin = 0x01d3670863c3F4b24D7b107900f0b75d4BbC6e0d;
+            _safe = 0xBc1233d0C3e6B5d53Ab455cF65A6623F6dCd7e4f;
+            _proxyAdmin = 0x01d3670863c3F4b24D7b107900f0b75d4BbC6e0d;
             // Set the proxy admin for the `_postCheck` function
-            PROXY_ADMIN = ProxyAdmin(proxyAdmin);
+            PROXY_ADMIN = ProxyAdmin(_proxyAdmin);
         }
 
-        require(safe != address(0) && proxyAdmin != address(0));
+        require(_safe != address(0) && _proxyAdmin != address(0));
 
-        address[] memory owners = IGnosisSafe(payable(safe)).getOwners();
+        address[] memory owners = IGnosisSafe(payable(_safe)).getOwners();
 
         for (uint256 i; i < owners.length; i++) {
             address owner = owners[i];
             vm.startBroadcast(owner);
-            bool success = _run(safe, proxyAdmin);
+            bool success = _run(_safe, _proxyAdmin);
             vm.stopBroadcast();
 
             if (success) {

packages/contracts-bedrock/scripts/upgrades/PostSherlockL2.s.sol

@@ -3,7 +3,7 @@ pragma solidity 0.8.15;
 
 import { console } from "forge-std/console.sol";
 import { SafeBuilder } from "../universal/SafeBuilder.sol";
-import { IGnosisSafe, Enum } from "../libraries/IGnosisSafe.sol";
+import { IGnosisSafe, Enum } from "../interfaces/IGnosisSafe.sol";
 import { IMulticall3 } from "forge-std/interfaces/IMulticall3.sol";
 import { Predeploys } from "../../contracts/libraries/Predeploys.sol";
 import { ProxyAdmin } from "../../contracts/universal/ProxyAdmin.sol";
@@ -131,22 +131,22 @@ contract PostSherlockL2 is SafeBuilder {
      *         could be added.
      */
     function test_script_succeeds() skipWhenNotForking external {
-        address safe;
-        address proxyAdmin;
+        address _safe;
+        address _proxyAdmin;
 
         if (block.chainid == OP_GOERLI) {
-            safe = 0xE534ccA2753aCFbcDBCeB2291F596fc60495257e;
-            proxyAdmin = 0x4200000000000000000000000000000000000018;
+            _safe = 0xE534ccA2753aCFbcDBCeB2291F596fc60495257e;
+            _proxyAdmin = 0x4200000000000000000000000000000000000018;
         }
 
-        require(safe != address(0) && proxyAdmin != address(0));
+        require(_safe != address(0) && _proxyAdmin != address(0));
 
-        address[] memory owners = IGnosisSafe(payable(safe)).getOwners();
+        address[] memory owners = IGnosisSafe(payable(_safe)).getOwners();
 
         for (uint256 i; i < owners.length; i++) {
             address owner = owners[i];
             vm.startBroadcast(owner);
-            bool success = _run(safe, proxyAdmin);
+            bool success = _run(_safe, _proxyAdmin);
             vm.stopBroadcast();
 
             if (success) {

packages/sdk/tasks/deposit-eth.ts

@@ -27,12 +27,6 @@ task('deposit-eth', 'Deposits ether to L2.')
     'http://localhost:9545',
     types.string
   )
-  .addParam(
-    'opNodeProviderUrl',
-    'op-node provider URL',
-    'http://localhost:7545',
-    types.string
-  )
   .addOptionalParam('to', 'Recipient of the ether', '', types.string)
   .addOptionalParam(
     'amount',

proxyd/backend.go

@@ -90,6 +90,11 @@ var (
 		Message:       "backend is currently not healthy to serve traffic",
 		HTTPErrorCode: 503,
 	}
+	ErrBlockOutOfRange = &RPCErr{
+		Code:          JSONRPCErrorInternal - 19,
+		Message:       "block is out of range",
+		HTTPErrorCode: 400,
+	}
 
 	ErrBackendUnexpectedJSONRPC = errors.New("backend returned an unexpected JSON-RPC response")
 )
@@ -202,6 +207,12 @@ func WithProxydIP(ip string) BackendOpt {
 	}
 }
 
+func WithMaxDegradedLatencyThreshold(maxDegradedLatencyThreshold time.Duration) BackendOpt {
+	return func(b *Backend) {
+		b.maxDegradedLatencyThreshold = maxDegradedLatencyThreshold
+	}
+}
+
 func WithMaxLatencyThreshold(maxLatencyThreshold time.Duration) BackendOpt {
 	return func(b *Backend) {
 		b.maxLatencyThreshold = maxLatencyThreshold
@@ -214,6 +225,12 @@ func WithMaxErrorRateThreshold(maxErrorRateThreshold float64) BackendOpt {
 	}
 }
 
+type indexedReqRes struct {
+	index int
+	req   *RPCReq
+	res   *RPCRes
+}
+
 func NewBackend(
 	name string,
 	rpcURL string,
@@ -540,7 +557,11 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
 
 // IsHealthy checks if the backend is able to serve traffic, based on dynamic parameters
 func (b *Backend) IsHealthy() bool {
-	errorRate := b.networkErrorsSlidingWindow.Sum() / b.networkRequestsSlidingWindow.Sum()
+	errorRate := float64(0)
+	// avoid division-by-zero when the window is empty
+	if b.networkRequestsSlidingWindow.Sum() >= 10 {
+		errorRate = b.networkErrorsSlidingWindow.Sum() / b.networkRequestsSlidingWindow.Sum()
+	}
 	avgLatency := time.Duration(b.latencySlidingWindow.Avg())
 	if errorRate >= b.maxErrorRateThreshold {
 		return false
@@ -593,47 +614,96 @@ func (b *BackendGroup) Forward(ctx context.Context, rpcReqs []*RPCReq, isBatch b
 
 	backends := b.Backends
 
-	// When `consensus_aware` is set to `true`, the backend group acts as a load balancer
-	// serving traffic from any backend that agrees in the consensus group
+	overriddenResponses := make([]*indexedReqRes, 0)
+	rewrittenReqs := make([]*RPCReq, 0, len(rpcReqs))
+
 	if b.Consensus != nil {
+		// When `consensus_aware` is set to `true`, the backend group acts as a load balancer
+		// serving traffic from any backend that agrees in the consensus group
 		backends = b.loadBalancedConsensusGroup()
+
+		// We also rewrite block tags to enforce compliance with consensus
+		rctx := RewriteContext{latest: b.Consensus.GetConsensusBlockNumber()}
+
+		for i, req := range rpcReqs {
+			res := RPCRes{JSONRPC: JSONRPCVersion, ID: req.ID}
+			result, err := RewriteTags(rctx, req, &res)
+			switch result {
+			case RewriteOverrideError:
+				overriddenResponses = append(overriddenResponses, &indexedReqRes{
+					index: i,
+					req:   req,
+					res:   &res,
+				})
+				if errors.Is(err, ErrRewriteBlockOutOfRange) {
+					res.Error = ErrBlockOutOfRange
+				} else {
+					res.Error = ErrParseErr
+				}
+			case RewriteOverrideResponse:
+				overriddenResponses = append(overriddenResponses, &indexedReqRes{
+					index: i,
+					req:   req,
+					res:   &res,
+				})
+			case RewriteOverrideRequest, RewriteNone:
+				rewrittenReqs = append(rewrittenReqs, req)
+			}
+		}
+		rpcReqs = rewrittenReqs
 	}
 
 	rpcRequestsTotal.Inc()
 
 	for _, back := range backends {
-		res, err := back.Forward(ctx, rpcReqs, isBatch)
-		if errors.Is(err, ErrMethodNotWhitelisted) {
-			return nil, err
-		}
-		if errors.Is(err, ErrBackendOffline) {
-			log.Warn(
-				"skipping offline backend",
-				"name", back.Name,
-				"auth", GetAuthCtx(ctx),
-				"req_id", GetReqID(ctx),
-			)
-			continue
-		}
-		if errors.Is(err, ErrBackendOverCapacity) {
-			log.Warn(
-				"skipping over-capacity backend",
-				"name", back.Name,
-				"auth", GetAuthCtx(ctx),
-				"req_id", GetReqID(ctx),
-			)
-			continue
+		res := make([]*RPCRes, 0)
+		var err error
+
+		if len(rpcReqs) > 0 {
+			res, err = back.Forward(ctx, rpcReqs, isBatch)
+			if errors.Is(err, ErrMethodNotWhitelisted) {
+				return nil, err
+			}
+			if errors.Is(err, ErrBackendOffline) {
+				log.Warn(
+					"skipping offline backend",
+					"name", back.Name,
+					"auth", GetAuthCtx(ctx),
+					"req_id", GetReqID(ctx),
+				)
+				continue
+			}
+			if errors.Is(err, ErrBackendOverCapacity) {
+				log.Warn(
+					"skipping over-capacity backend",
+					"name", back.Name,
+					"auth", GetAuthCtx(ctx),
+					"req_id", GetReqID(ctx),
+				)
+				continue
+			}
+			if err != nil {
+				log.Error(
+					"error forwarding request to backend",
+					"name", back.Name,
+					"req_id", GetReqID(ctx),
+					"auth", GetAuthCtx(ctx),
+					"err", err,
+				)
+				continue
+			}
 		}
-		if err != nil {
-			log.Error(
-				"error forwarding request to backend",
-				"name", back.Name,
-				"req_id", GetReqID(ctx),
-				"auth", GetAuthCtx(ctx),
-				"err", err,
-			)
-			continue
+
+		// re-apply overridden responses
+		for _, ov := range overriddenResponses {
+			if len(res) > 0 {
+				// insert ov.res at position ov.index
+				res = append(res[:ov.index], append([]*RPCRes{ov.res}, res[ov.index:]...)...)
+			} else {
+				res = append(res, ov.res)
+			}
 		}
+
 		return res, nil
 	}
 

proxyd/config.go

@@ -71,10 +71,13 @@ func (t *TOMLDuration) UnmarshalText(b []byte) error {
 }
 
 type BackendOptions struct {
-	ResponseTimeoutSeconds int   `toml:"response_timeout_seconds"`
-	MaxResponseSizeBytes   int64 `toml:"max_response_size_bytes"`
-	MaxRetries             int   `toml:"max_retries"`
-	OutOfServiceSeconds    int   `toml:"out_of_service_seconds"`
+	ResponseTimeoutSeconds      int          `toml:"response_timeout_seconds"`
+	MaxResponseSizeBytes        int64        `toml:"max_response_size_bytes"`
+	MaxRetries                  int          `toml:"max_retries"`
+	OutOfServiceSeconds         int          `toml:"out_of_service_seconds"`
+	MaxDegradedLatencyThreshold TOMLDuration `toml:"max_degraded_latency_threshold"`
+	MaxLatencyThreshold         TOMLDuration `toml:"max_latency_threshold"`
+	MaxErrorRateThreshold       float64      `toml:"max_error_rate_threshold"`
 }
 
 type BackendConfig struct {
@@ -94,9 +97,14 @@ type BackendConfig struct {
 type BackendsConfig map[string]*BackendConfig
 
 type BackendGroupConfig struct {
-	Backends              []string `toml:"backends"`
-	ConsensusAware        bool     `toml:"consensus_aware"`
-	ConsensusAsyncHandler string   `toml:"consensus_handler"`
+	Backends []string `toml:"backends"`
+
+	ConsensusAware        bool   `toml:"consensus_aware"`
+	ConsensusAsyncHandler string `toml:"consensus_handler"`
+
+	ConsensusBanPeriod          TOMLDuration `toml:"consensus_ban_period"`
+	ConsensusMaxUpdateThreshold TOMLDuration `toml:"consensus_max_update_threshold"`
+	ConsensusMinPeerCount       int          `toml:"consensus_min_peer_count"`
 }
 
 type BackendGroupsConfig map[string]*BackendGroupConfig

proxyd/example.config.toml

@@ -44,6 +44,12 @@ max_response_size_bytes = 5242880
 max_retries = 3
 # Number of seconds to wait before trying an unhealthy backend again.
 out_of_service_seconds = 600
+# Maximum latency accepted to serve requests, default 10s
+max_latency_threshold = "30s"
+# Maximum latency accepted to serve requests before degraded, default 5s
+max_degraded_latency_threshold = "10s"
+# Maximum error rate accepted to serve requests, default 0.5 (i.e. 50%)
+max_error_rate_threshold = 0.3
 
 [backends]
 # A map of backends by name.
@@ -78,6 +84,14 @@ max_ws_conns = 1
 [backend_groups]
 [backend_groups.main]
 backends = ["infura"]
+# Enable consensus awareness for backend group, making it act as a load balancer, default false
+# consensus_aware = true
+# Period in which the backend wont serve requests if banned, default 5m
+# consensus_ban_period = "1m"
+# Maximum delay for update the backend, default 30s
+# consensus_max_update_threshold = "20s"
+# Minimum peer count, default 3
+# consensus_min_peer_count = 4
 
 [backend_groups.alchemy]
 backends = ["alchemy"]

proxyd/integration_tests/consensus_test.go

@@ -433,6 +433,184 @@ func TestConsensus(t *testing.T) {
 		require.Equal(t, len(node1.Requests()), 0, msg)
 		require.Equal(t, len(node2.Requests()), 10, msg)
 	})
+
+	t.Run("rewrite response of eth_blockNumber", func(t *testing.T) {
+		h1.ResetOverrides()
+		h2.ResetOverrides()
+		node1.Reset()
+		node2.Reset()
+		bg.Consensus.Unban()
+
+		// establish the consensus
+
+		h1.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x2", "hash2"),
+		})
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x2", "hash2"),
+		})
+
+		for _, be := range bg.Backends {
+			bg.Consensus.UpdateBackend(ctx, be)
+		}
+		bg.Consensus.UpdateBackendGroupConsensus(ctx)
+
+		totalRequests := len(node1.Requests()) + len(node2.Requests())
+
+		require.Equal(t, 2, len(bg.Consensus.GetConsensusGroup()))
+
+		// pretend backends advanced in consensus, but we are still serving the latest value of the consensus
+		// until it gets updated again
+
+		h1.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x3", "hash3"),
+		})
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x3", "hash3"),
+		})
+
+		resRaw, statusCode, err := client.SendRPC("eth_blockNumber", nil)
+		require.NoError(t, err)
+		require.Equal(t, 200, statusCode)
+
+		var jsonMap map[string]interface{}
+		err = json.Unmarshal(resRaw, &jsonMap)
+		require.NoError(t, err)
+		require.Equal(t, "0x2", jsonMap["result"])
+
+		// no extra request hit the backends
+		require.Equal(t, totalRequests, len(node1.Requests())+len(node2.Requests()))
+	})
+
+	t.Run("rewrite request of eth_getBlockByNumber", func(t *testing.T) {
+		h1.ResetOverrides()
+		h2.ResetOverrides()
+		bg.Consensus.Unban()
+
+		// establish the consensus and ban node2 for now
+		h1.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x2", "hash2"),
+		})
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "net_peerCount",
+			Block:    "",
+			Response: buildPeerCountResponse(1),
+		})
+
+		for _, be := range bg.Backends {
+			bg.Consensus.UpdateBackend(ctx, be)
+		}
+		bg.Consensus.UpdateBackendGroupConsensus(ctx)
+
+		require.Equal(t, 1, len(bg.Consensus.GetConsensusGroup()))
+
+		node1.Reset()
+
+		_, statusCode, err := client.SendRPC("eth_getBlockByNumber", []interface{}{"latest"})
+		require.NoError(t, err)
+		require.Equal(t, 200, statusCode)
+
+		var jsonMap map[string]interface{}
+		err = json.Unmarshal(node1.Requests()[0].Body, &jsonMap)
+		require.NoError(t, err)
+		require.Equal(t, "0x2", jsonMap["params"].([]interface{})[0])
+	})
+
+	t.Run("rewrite request of eth_getBlockByNumber - out of range", func(t *testing.T) {
+		h1.ResetOverrides()
+		h2.ResetOverrides()
+		bg.Consensus.Unban()
+
+		// establish the consensus and ban node2 for now
+		h1.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x2", "hash2"),
+		})
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "net_peerCount",
+			Block:    "",
+			Response: buildPeerCountResponse(1),
+		})
+
+		for _, be := range bg.Backends {
+			bg.Consensus.UpdateBackend(ctx, be)
+		}
+		bg.Consensus.UpdateBackendGroupConsensus(ctx)
+
+		require.Equal(t, 1, len(bg.Consensus.GetConsensusGroup()))
+
+		node1.Reset()
+
+		resRaw, statusCode, err := client.SendRPC("eth_getBlockByNumber", []interface{}{"0x10"})
+		require.NoError(t, err)
+		require.Equal(t, 400, statusCode)
+
+		var jsonMap map[string]interface{}
+		err = json.Unmarshal(resRaw, &jsonMap)
+		require.NoError(t, err)
+		require.Equal(t, -32019, int(jsonMap["error"].(map[string]interface{})["code"].(float64)))
+		require.Equal(t, "block is out of range", jsonMap["error"].(map[string]interface{})["message"])
+	})
+
+	t.Run("batched rewrite", func(t *testing.T) {
+		h1.ResetOverrides()
+		h2.ResetOverrides()
+		bg.Consensus.Unban()
+
+		// establish the consensus and ban node2 for now
+		h1.AddOverride(&ms.MethodTemplate{
+			Method:   "eth_getBlockByNumber",
+			Block:    "latest",
+			Response: buildGetBlockResponse("0x2", "hash2"),
+		})
+		h2.AddOverride(&ms.MethodTemplate{
+			Method:   "net_peerCount",
+			Block:    "",
+			Response: buildPeerCountResponse(1),
+		})
+
+		for _, be := range bg.Backends {
+			bg.Consensus.UpdateBackend(ctx, be)
+		}
+		bg.Consensus.UpdateBackendGroupConsensus(ctx)
+
+		require.Equal(t, 1, len(bg.Consensus.GetConsensusGroup()))
+
+		node1.Reset()
+
+		resRaw, statusCode, err := client.SendBatchRPC(
+			NewRPCReq("1", "eth_getBlockByNumber", []interface{}{"latest"}),
+			NewRPCReq("2", "eth_getBlockByNumber", []interface{}{"0x10"}),
+			NewRPCReq("3", "eth_getBlockByNumber", []interface{}{"0x1"}))
+		require.NoError(t, err)
+		require.Equal(t, 200, statusCode)
+
+		var jsonMap []map[string]interface{}
+		err = json.Unmarshal(resRaw, &jsonMap)
+		require.NoError(t, err)
+		require.Equal(t, 3, len(jsonMap))
+
+		// rewrite latest to 0x2
+		require.Equal(t, "0x2", jsonMap[0]["result"].(map[string]interface{})["number"])
+
+		// out of bounds for block 0x10
+		require.Equal(t, -32019, int(jsonMap[1]["error"].(map[string]interface{})["code"].(float64)))
+		require.Equal(t, "block is out of range", jsonMap[1]["error"].(map[string]interface{})["message"])
+
+		// dont rewrite for 0x1
+		require.Equal(t, "0x1", jsonMap[2]["result"].(map[string]interface{})["number"])
+	})
 }
 
 func backend(bg *proxyd.BackendGroup, name string) *proxyd.Backend {

proxyd/integration_tests/testdata/consensus.toml

@@ -16,6 +16,9 @@ rpc_url = "$NODE2_URL"
 backends = ["node1", "node2"]
 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_min_peer_count = 4
 
 [rpc_method_mappings]
 eth_call = "node"

proxyd/proxyd.go

@@ -123,6 +123,15 @@ func Start(config *Config) (*Server, func(), error) {
 		if config.BackendOptions.OutOfServiceSeconds != 0 {
 			opts = append(opts, WithOutOfServiceDuration(secondsToDuration(config.BackendOptions.OutOfServiceSeconds)))
 		}
+		if config.BackendOptions.MaxDegradedLatencyThreshold > 0 {
+			opts = append(opts, WithMaxDegradedLatencyThreshold(time.Duration(config.BackendOptions.MaxDegradedLatencyThreshold)))
+		}
+		if config.BackendOptions.MaxLatencyThreshold > 0 {
+			opts = append(opts, WithMaxLatencyThreshold(time.Duration(config.BackendOptions.MaxLatencyThreshold)))
+		}
+		if config.BackendOptions.MaxErrorRateThreshold > 0 {
+			opts = append(opts, WithMaxErrorRateThreshold(config.BackendOptions.MaxErrorRateThreshold))
+		}
 		if cfg.MaxRPS != 0 {
 			opts = append(opts, WithMaxRPS(cfg.MaxRPS))
 		}
@@ -148,6 +157,7 @@ func Start(config *Config) (*Server, func(), error) {
 			opts = append(opts, WithStrippedTrailingXFF())
 		}
 		opts = append(opts, WithProxydIP(os.Getenv("PROXYD_IP")))
+
 		back := NewBackend(name, rpcURL, wsURL, lim, rpcRequestSemaphore, opts...)
 		backendNames = append(backendNames, name)
 		backendsByName[name] = back
@@ -302,14 +312,25 @@ func Start(config *Config) (*Server, func(), error) {
 	}
 
 	for bgName, bg := range backendGroups {
-		if config.BackendGroups[bgName].ConsensusAware {
+		bgcfg := config.BackendGroups[bgName]
+		if bgcfg.ConsensusAware {
 			log.Info("creating poller for consensus aware backend_group", "name", bgName)
 
 			copts := make([]ConsensusOpt, 0)
 
-			if config.BackendGroups[bgName].ConsensusAsyncHandler == "noop" {
+			if bgcfg.ConsensusAsyncHandler == "noop" {
 				copts = append(copts, WithAsyncHandler(NewNoopAsyncHandler()))
 			}
+			if bgcfg.ConsensusBanPeriod > 0 {
+				copts = append(copts, WithBanPeriod(time.Duration(bgcfg.ConsensusBanPeriod)))
+			}
+			if bgcfg.ConsensusMaxUpdateThreshold > 0 {
+				copts = append(copts, WithMaxUpdateThreshold(time.Duration(bgcfg.ConsensusMaxUpdateThreshold)))
+			}
+			if bgcfg.ConsensusMinPeerCount > 0 {
+				copts = append(copts, WithMinPeerCount(uint64(bgcfg.ConsensusMinPeerCount)))
+			}
+
 			cp := NewConsensusPoller(bg, copts...)
 			bg.Consensus = cp
 		}

proxyd/rewriter.go

+package proxyd
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/common/hexutil"
+)
+
+type RewriteContext struct {
+	latest hexutil.Uint64
+}
+
+type RewriteResult uint8
+
+const (
+	// RewriteNone means request should be forwarded as-is
+	RewriteNone RewriteResult = iota
+
+	// RewriteOverrideError means there was an error attempting to rewrite
+	RewriteOverrideError
+
+	// RewriteOverrideRequest means the modified request should be forwarded to the backend
+	RewriteOverrideRequest
+
+	// RewriteOverrideResponse means to skip calling the backend and serve the overridden response
+	RewriteOverrideResponse
+)
+
+var (
+	ErrRewriteBlockOutOfRange = errors.New("block is out of range")
+)
+
+// RewriteTags modifies the request and the response based on block tags
+func RewriteTags(rctx RewriteContext, req *RPCReq, res *RPCRes) (RewriteResult, error) {
+	rw, err := RewriteResponse(rctx, req, res)
+	if rw == RewriteOverrideResponse {
+		return rw, err
+	}
+	return RewriteRequest(rctx, req, res)
+}
+
+// RewriteResponse modifies the response object to comply with the rewrite context
+// after the method has been called at the backend
+// RewriteResult informs the decision of the rewrite
+func RewriteResponse(rctx RewriteContext, req *RPCReq, res *RPCRes) (RewriteResult, error) {
+	switch req.Method {
+	case "eth_blockNumber":
+		res.Result = rctx.latest
+		return RewriteOverrideResponse, nil
+	}
+	return RewriteNone, nil
+}
+
+// RewriteRequest modifies the request object to comply with the rewrite context
+// before the method has been called at the backend
+// it returns false if nothing was changed
+func RewriteRequest(rctx RewriteContext, req *RPCReq, res *RPCRes) (RewriteResult, error) {
+	switch req.Method {
+	case "eth_getLogs",
+		"eth_newFilter":
+		return rewriteRange(rctx, req, res, 0)
+	case "eth_getBalance",
+		"eth_getCode",
+		"eth_getTransactionCount",
+		"eth_call":
+		return rewriteParam(rctx, req, res, 1)
+	case "eth_getStorageAt":
+		return rewriteParam(rctx, req, res, 2)
+	case "eth_getBlockTransactionCountByNumber",
+		"eth_getUncleCountByBlockNumber",
+		"eth_getBlockByNumber",
+		"eth_getTransactionByBlockNumberAndIndex",
+		"eth_getUncleByBlockNumberAndIndex":
+		return rewriteParam(rctx, req, res, 0)
+	}
+	return RewriteNone, nil
+}
+
+func rewriteParam(rctx RewriteContext, req *RPCReq, res *RPCRes, pos int) (RewriteResult, error) {
+	var p []interface{}
+	err := json.Unmarshal(req.Params, &p)
+	if err != nil {
+		return RewriteOverrideError, err
+	}
+
+	if len(p) <= pos {
+		p = append(p, "latest")
+	}
+
+	val, rw, err := rewriteTag(rctx, p[pos].(string))
+	if err != nil {
+		return RewriteOverrideError, err
+	}
+
+	if rw {
+		p[pos] = val
+		paramRaw, err := json.Marshal(p)
+		if err != nil {
+			return RewriteOverrideError, err
+		}
+		req.Params = paramRaw
+		return RewriteOverrideRequest, nil
+	}
+	return RewriteNone, nil
+}
+
+func rewriteRange(rctx RewriteContext, req *RPCReq, res *RPCRes, pos int) (RewriteResult, error) {
+	var p []map[string]interface{}
+	err := json.Unmarshal(req.Params, &p)
+	if err != nil {
+		return RewriteOverrideError, err
+	}
+
+	modifiedFrom, err := rewriteTagMap(rctx, p[pos], "fromBlock")
+	if err != nil {
+		return RewriteOverrideError, err
+	}
+
+	modifiedTo, err := rewriteTagMap(rctx, p[pos], "toBlock")
+	if err != nil {
+		return RewriteOverrideError, err
+	}
+
+	// if any of the fields the request have been changed, re-marshal the params
+	if modifiedFrom || modifiedTo {
+		paramsRaw, err := json.Marshal(p)
+		req.Params = paramsRaw
+		if err != nil {
+			return RewriteOverrideError, err
+		}
+		return RewriteOverrideRequest, nil
+	}
+
+	return RewriteNone, nil
+}
+
+func rewriteTagMap(rctx RewriteContext, m map[string]interface{}, key string) (bool, error) {
+	if m[key] == nil || m[key] == "" {
+		return false, nil
+	}
+
+	current, ok := m[key].(string)
+	if !ok {
+		return false, errors.New("expected string")
+	}
+
+	val, rw, err := rewriteTag(rctx, current)
+	if err != nil {
+		return false, err
+	}
+	if rw {
+		m[key] = val
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func rewriteTag(rctx RewriteContext, current string) (string, bool, error) {
+	if current == "latest" {
+		return rctx.latest.String(), true, nil
+	} else if strings.HasPrefix(current, "0x") {
+		decode, err := hexutil.DecodeUint64(current)
+		if err != nil {
+			return current, false, err
+		}
+		b := hexutil.Uint64(decode)
+		if b > rctx.latest {
+			return "", false, ErrRewriteBlockOutOfRange
+		}
+	}
+	return current, false, nil
+}

proxyd/tools/mockserver/handler/handler.go

@@ -6,6 +6,9 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"strings"
+
+	"github.com/ethereum-optimism/optimism/proxyd"
 
 	"github.com/gorilla/mux"
 	"github.com/pkg/errors"
@@ -46,12 +49,6 @@ func (mh *MockedHandler) Handler(w http.ResponseWriter, req *http.Request) {
 		fmt.Printf("error reading request: %v\n", err)
 	}
 
-	var j map[string]interface{}
-	err = json.Unmarshal(body, &j)
-	if err != nil {
-		fmt.Printf("error reading request: %v\n", err)
-	}
-
 	var template []*MethodTemplate
 	if mh.Autoload {
 		template = append(template, mh.LoadFromFile(mh.AutoloadFile)...)
@@ -60,23 +57,51 @@ func (mh *MockedHandler) Handler(w http.ResponseWriter, req *http.Request) {
 		template = append(template, mh.Overrides...)
 	}
 
-	method := j["method"]
-	block := ""
-	if method == "eth_getBlockByNumber" {
-		block = (j["params"].([]interface{})[0]).(string)
+	batched := proxyd.IsBatch(body)
+	var requests []map[string]interface{}
+	if batched {
+		err = json.Unmarshal(body, &requests)
+		if err != nil {
+			fmt.Printf("error reading request: %v\n", err)
+		}
+	} else {
+		var j map[string]interface{}
+		err = json.Unmarshal(body, &j)
+		if err != nil {
+			fmt.Printf("error reading request: %v\n", err)
+		}
+		requests = append(requests, j)
 	}
 
-	var selectedResponse *string
-	for _, r := range template {
-		if r.Method == method && r.Block == block {
-			selectedResponse = &r.Response
+	var responses []string
+	for _, r := range requests {
+		method := r["method"]
+		block := ""
+		if method == "eth_getBlockByNumber" {
+			block = (r["params"].([]interface{})[0]).(string)
 		}
-	}
-	if selectedResponse != nil {
-		_, err := fmt.Fprintf(w, *selectedResponse)
-		if err != nil {
-			fmt.Printf("error writing response: %v\n", err)
+
+		var selectedResponse string
+		for _, r := range template {
+			if r.Method == method && r.Block == block {
+				selectedResponse = r.Response
+			}
 		}
+		if selectedResponse != "" {
+			responses = append(responses, selectedResponse)
+		}
+	}
+
+	resBody := ""
+	if batched {
+		resBody = "[" + strings.Join(responses, ",") + "]"
+	} else {
+		resBody = responses[0]
+	}
+
+	_, err = fmt.Fprint(w, resBody)
+	if err != nil {
+		fmt.Printf("error writing response: %v\n", err)
 	}
 }