tweak(zkevm_api): impl concurrency limit on getBatchWitness (erigonte…

…ch#954)

closes erigontech#952

README.md

@@ -110,6 +110,9 @@ In order to enable the zkevm_ namespace, please add 'zkevm' to the http.api flag
 ### Supported (remote)
 - `zkevm_getBatchByNumber`
 
+### Configurable
+- `zkevm_getBatchWitness` - concurrency can be limited with `zkevm.rpc-get-batch-witness-concurrency-limit` flag which defaults to 1. Use 0 for no limit.
+
 ### Not yet supported
 - `zkevm_getNativeBlockHashesInRange`
 

cmd/rpcdaemon/commands/zkevm_api.go

@@ -65,6 +65,8 @@ type ZkEvmAPI interface {
 	GetExitRootTable(ctx context.Context) ([]l1InfoTreeData, error)
 }
 
+const getBatchWitness = "getBatchWitness"
+
 // APIImpl is implementation of the ZkEvmAPI interface based on remote Db access
 type ZkEvmAPIImpl struct {
 	ethApi *APIImpl
@@ -74,6 +76,17 @@ type ZkEvmAPIImpl struct {
 	config          *ethconfig.Config
 	l1Syncer        *syncer.L1Syncer
 	l2SequencerUrl  string
+	semaphores      map[string]chan struct{}
+}
+
+func (api *ZkEvmAPIImpl) initializeSemaphores(functionLimits map[string]int) {
+	api.semaphores = make(map[string]chan struct{})
+
+	for funcName, limit := range functionLimits {
+		if limit != 0 {
+			api.semaphores[funcName] = make(chan struct{}, limit)
+		}
+	}
 }
 
 // NewEthAPI returns ZkEvmAPIImpl instance
@@ -85,14 +98,21 @@ func NewZkEvmAPI(
 	l1Syncer *syncer.L1Syncer,
 	l2SequencerUrl string,
 ) *ZkEvmAPIImpl {
-	return &ZkEvmAPIImpl{
+
+	a := &ZkEvmAPIImpl{
 		ethApi:          base,
 		db:              db,
 		ReturnDataLimit: returnDataLimit,
 		config:          zkConfig,
 		l1Syncer:        l1Syncer,
 		l2SequencerUrl:  l2SequencerUrl,
 	}
+
+	a.initializeSemaphores(map[string]int{
+		getBatchWitness: zkConfig.Zk.RpcGetBatchWitnessConcurrencyLimit,
+	})
+
+	return a
 }
 
 // ConsolidatedBlockNumber returns the latest consolidated block number
@@ -836,6 +856,18 @@ func (api *ZkEvmAPIImpl) GetBlockRangeWitness(ctx context.Context, startBlockNrO
 }
 
 func (api *ZkEvmAPIImpl) getBatchWitness(ctx context.Context, tx kv.Tx, batchNum uint64, debug bool, mode WitnessMode) (hexutility.Bytes, error) {
+
+	// limit in-flight requests by name
+	semaphore := api.semaphores[getBatchWitness]
+	if semaphore != nil {
+		select {
+		case semaphore <- struct{}{}:
+			defer func() { <-semaphore }()
+		default:
+			return nil, fmt.Errorf("busy")
+		}
+	}
+
 	if api.ethApi.historyV3(tx) {
 		return nil, fmt.Errorf("not supported by Erigon3")
 	}

cmd/utils/flags.go

@@ -527,6 +527,11 @@ var (
 		Usage: "RPC rate limit in requests per second.",
 		Value: 0,
 	}
+	RpcGetBatchWitnessConcurrencyLimitFlag = cli.IntFlag{
+		Name:  "zkevm.rpc-get-batch-witness-concurrency-limit",
+		Usage: "The maximum number of concurrent requests to the executor for getBatchWitness.",
+		Value: 1,
+	}
 	DatastreamVersionFlag = cli.IntFlag{
 		Name:  "zkevm.datastream-version",
 		Usage: "Stream version indicator 1: PreBigEndian, 2: BigEndian.",

eth/ethconfig/config_zkevm.go

@@ -30,6 +30,7 @@ type Zk struct {
 	L1CacheEnabled                         bool
 	L1CachePort                            uint
 	RpcRateLimits                          int
+	RpcGetBatchWitnessConcurrencyLimit     int
 	DatastreamVersion                      int
 	SequencerBlockSealTime                 time.Duration
 	SequencerBatchSealTime                 time.Duration

turbo/cli/default_flags.go

@@ -187,6 +187,7 @@ var DefaultFlags = []cli.Flag{
 	&utils.L1MaticContractAddressFlag,
 	&utils.L1FirstBlockFlag,
 	&utils.RpcRateLimitsFlag,
+	&utils.RpcGetBatchWitnessConcurrencyLimitFlag,
 	&utils.DatastreamVersionFlag,
 	&utils.RebuildTreeAfterFlag,
 	&utils.IncrementTreeAlways,

turbo/cli/flags_zkevm.go

@@ -124,6 +124,7 @@ func ApplyFlagsForZkConfig(ctx *cli.Context, cfg *ethconfig.Config) {
 		L1MaticContractAddress:                 libcommon.HexToAddress(ctx.String(utils.L1MaticContractAddressFlag.Name)),
 		L1FirstBlock:                           ctx.Uint64(utils.L1FirstBlockFlag.Name),
 		RpcRateLimits:                          ctx.Int(utils.RpcRateLimitsFlag.Name),
+		RpcGetBatchWitnessConcurrencyLimit:     ctx.Int(utils.RpcGetBatchWitnessConcurrencyLimitFlag.Name),
 		DatastreamVersion:                      ctx.Int(utils.DatastreamVersionFlag.Name),
 		RebuildTreeAfter:                       ctx.Uint64(utils.RebuildTreeAfterFlag.Name),
 		IncrementTreeAlways:                    ctx.Bool(utils.IncrementTreeAlways.Name),
@@ -209,6 +210,7 @@ func ApplyFlagsForZkConfig(ctx *cli.Context, cfg *ethconfig.Config) {
 	checkFlag(utils.L1MaticContractAddressFlag.Name, cfg.L1MaticContractAddress.Hex())
 	checkFlag(utils.L1FirstBlockFlag.Name, cfg.L1FirstBlock)
 	checkFlag(utils.RpcRateLimitsFlag.Name, cfg.RpcRateLimits)
+	checkFlag(utils.RpcGetBatchWitnessConcurrencyLimitFlag.Name, cfg.RpcGetBatchWitnessConcurrencyLimit)
 	checkFlag(utils.RebuildTreeAfterFlag.Name, cfg.RebuildTreeAfter)
 	checkFlag(utils.L1BlockRangeFlag.Name, cfg.L1BlockRange)
 	checkFlag(utils.L1QueryDelayFlag.Name, cfg.L1QueryDelay)