EVM-779 Throttling with concurrency (#1821)

Co-authored-by: stana-ethernal <stana.miric@ethernal.tech>

command/server/config/config.go

@@ -35,7 +35,7 @@ type Config struct {
 	Relayer               bool   `json:"relayer" yaml:"relayer"`
 	NumBlockConfirmations uint64 `json:"num_block_confirmations" yaml:"num_block_confirmations"`
 
-	RequestsPerSecondDebug uint64 `json:"requests_per_second_debug" yaml:"requests_per_second_debug"`
+	ConcurrentRequestsDebug uint64 `json:"concurrent_requests_debug" yaml:"concurrent_requests_debug"`
 }
 
 // Telemetry holds the config details for metric services.
@@ -82,8 +82,8 @@ const (
 	// on ethereum epoch lasts for 32 blocks. more details: https://www.alchemy.com/overviews/ethereum-commitment-levels
 	DefaultNumBlockConfirmations uint64 = 64
 
-	// Maximum number of allowed requests per second for debug endpoints
-	DefaultRequestsPerSecondDebug uint64 = 5
+	// Maximum number of allowed concurrent requests for debug endpoints
+	DefaultConcurrentRequestsDebug uint64 = 32
 )
 
 // DefaultConfig returns the default server configuration
@@ -121,7 +121,7 @@ func DefaultConfig() *Config {
 		JSONRPCBlockRangeLimit:   DefaultJSONRPCBlockRangeLimit,
 		Relayer:                  false,
 		NumBlockConfirmations:    DefaultNumBlockConfirmations,
-		RequestsPerSecondDebug:   DefaultRequestsPerSecondDebug,
+		ConcurrentRequestsDebug:  DefaultConcurrentRequestsDebug,
 	}
 }
 

command/server/params.go

@@ -41,7 +41,7 @@ const (
 	relayerFlag               = "relayer"
 	numBlockConfirmationsFlag = "num-block-confirmations"
 
-	requestsPerSecondDebugFlag = "requests-per-second-debug"
+	concurrentRequestsDebugFlag = "concurrent-requests-debug"
 )
 
 // Flags that are deprecated, but need to be preserved for
@@ -154,7 +154,7 @@ func (p *serverParams) generateConfig() *server.Config {
 			AccessControlAllowOrigin: p.rawConfig.CorsAllowedOrigins,
 			BatchLengthLimit:         p.rawConfig.JSONRPCBatchRequestLimit,
 			BlockRangeLimit:          p.rawConfig.JSONRPCBlockRangeLimit,
-			RequestsPerSecondDebug:   p.rawConfig.RequestsPerSecondDebug,
+			ConcurrentRequestsDebug:  p.rawConfig.ConcurrentRequestsDebug,
 		},
 		GRPCAddr:   p.grpcAddress,
 		LibP2PAddr: p.libp2pAddress,

command/server/server.go

@@ -229,10 +229,10 @@ func setFlags(cmd *cobra.Command) {
 	)
 
 	cmd.Flags().Uint64Var(
-		&params.rawConfig.RequestsPerSecondDebug,
-		requestsPerSecondDebugFlag,
-		defaultConfig.RequestsPerSecondDebug,
-		"maximal number of requests per second for debug endpoints",
+		&params.rawConfig.ConcurrentRequestsDebug,
+		concurrentRequestsDebugFlag,
+		defaultConfig.ConcurrentRequestsDebug,
+		"maximal number of concurrent requests for debug endpoints",
 	)
 
 	setLegacyFlags(cmd)

jsonrpc/debug_endpoint.go

@@ -69,10 +69,10 @@ type Debug struct {
 	throttling *Throttling
 }
 
-func NewDebug(store debugStore, requestsPerSecond int) *Debug {
+func NewDebug(store debugStore, requestsPerSecond uint64) *Debug {
 	return &Debug{
 		store:      store,
-		throttling: NewThrottling(requestsPerSecond),
+		throttling: NewThrottling(requestsPerSecond, time.Second),
 	}
 }
 
@@ -88,119 +88,124 @@ func (d *Debug) TraceBlockByNumber(
 	blockNumber BlockNumber,
 	config *TraceConfig,
 ) (interface{}, error) {
-	if err := d.throttling.AttemptRequest(); err != nil {
-		return nil, err
-	}
-
-	num, err := GetNumericBlockNumber(blockNumber, d.store)
-	if err != nil {
-		return nil, err
-	}
-
-	block, ok := d.store.GetBlockByNumber(num, true)
-	if !ok {
-		return nil, fmt.Errorf("block %d not found", num)
-	}
-
-	return d.traceBlock(block, config)
+	return d.throttling.AttemptRequest(
+		context.Background(),
+		func() (interface{}, error) {
+			num, err := GetNumericBlockNumber(blockNumber, d.store)
+			if err != nil {
+				return nil, err
+			}
+
+			block, ok := d.store.GetBlockByNumber(num, true)
+			if !ok {
+				return nil, fmt.Errorf("block %d not found", num)
+			}
+
+			return d.traceBlock(block, config)
+		},
+	)
 }
 
 func (d *Debug) TraceBlockByHash(
 	blockHash types.Hash,
 	config *TraceConfig,
 ) (interface{}, error) {
-	if err := d.throttling.AttemptRequest(); err != nil {
-		return nil, err
-	}
-
-	block, ok := d.store.GetBlockByHash(blockHash, true)
-	if !ok {
-		return nil, fmt.Errorf("block %s not found", blockHash)
-	}
-
-	return d.traceBlock(block, config)
+	return d.throttling.AttemptRequest(
+		context.Background(),
+		func() (interface{}, error) {
+			block, ok := d.store.GetBlockByHash(blockHash, true)
+			if !ok {
+				return nil, fmt.Errorf("block %s not found", blockHash)
+			}
+
+			return d.traceBlock(block, config)
+		},
+	)
 }
 
 func (d *Debug) TraceBlock(
 	input string,
 	config *TraceConfig,
 ) (interface{}, error) {
-	if err := d.throttling.AttemptRequest(); err != nil {
-		return nil, err
-	}
-
-	blockByte, decodeErr := hex.DecodeHex(input)
-	if decodeErr != nil {
-		return nil, fmt.Errorf("unable to decode block, %w", decodeErr)
-	}
-
-	block := &types.Block{}
-	if err := block.UnmarshalRLP(blockByte); err != nil {
-		return nil, err
-	}
-
-	return d.traceBlock(block, config)
+	return d.throttling.AttemptRequest(
+		context.Background(),
+		func() (interface{}, error) {
+			blockByte, decodeErr := hex.DecodeHex(input)
+			if decodeErr != nil {
+				return nil, fmt.Errorf("unable to decode block, %w", decodeErr)
+			}
+
+			block := &types.Block{}
+			if err := block.UnmarshalRLP(blockByte); err != nil {
+				return nil, err
+			}
+
+			return d.traceBlock(block, config)
+		},
+	)
 }
 
 func (d *Debug) TraceTransaction(
 	txHash types.Hash,
 	config *TraceConfig,
 ) (interface{}, error) {
-	if err := d.throttling.AttemptRequest(); err != nil {
-		return nil, err
-	}
-
-	tx, block := GetTxAndBlockByTxHash(txHash, d.store)
-	if tx == nil {
-		return nil, fmt.Errorf("tx %s not found", txHash.String())
-	}
-
-	if block.Number() == 0 {
-		return nil, ErrTraceGenesisBlock
-	}
-
-	tracer, cancel, err := newTracer(config)
-	if err != nil {
-		return nil, err
-	}
-
-	defer cancel()
-
-	return d.store.TraceTxn(block, tx.Hash, tracer)
+	return d.throttling.AttemptRequest(
+		context.Background(),
+		func() (interface{}, error) {
+			tx, block := GetTxAndBlockByTxHash(txHash, d.store)
+			if tx == nil {
+				return nil, fmt.Errorf("tx %s not found", txHash.String())
+			}
+
+			if block.Number() == 0 {
+				return nil, ErrTraceGenesisBlock
+			}
+
+			tracer, cancel, err := newTracer(config)
+			if err != nil {
+				return nil, err
+			}
+
+			defer cancel()
+
+			return d.store.TraceTxn(block, tx.Hash, tracer)
+		},
+	)
 }
 
 func (d *Debug) TraceCall(
 	arg *txnArgs,
 	filter BlockNumberOrHash,
 	config *TraceConfig,
 ) (interface{}, error) {
-	if err := d.throttling.AttemptRequest(); err != nil {
-		return nil, err
-	}
-
-	header, err := GetHeaderFromBlockNumberOrHash(filter, d.store)
-	if err != nil {
-		return nil, ErrHeaderNotFound
-	}
-
-	tx, err := DecodeTxn(arg, header.Number, d.store, true)
-	if err != nil {
-		return nil, err
-	}
-
-	// If the caller didn't supply the gas limit in the message, then we set it to maximum possible => block gas limit
-	if tx.Gas == 0 {
-		tx.Gas = header.GasLimit
-	}
-
-	tracer, cancel, err := newTracer(config)
-	defer cancel()
-
-	if err != nil {
-		return nil, err
-	}
-
-	return d.store.TraceCall(tx, header, tracer)
+	return d.throttling.AttemptRequest(
+		context.Background(),
+		func() (interface{}, error) {
+			header, err := GetHeaderFromBlockNumberOrHash(filter, d.store)
+			if err != nil {
+				return nil, ErrHeaderNotFound
+			}
+
+			tx, err := DecodeTxn(arg, header.Number, d.store, true)
+			if err != nil {
+				return nil, err
+			}
+
+			// If the caller didn't supply the gas limit in the message, then we set it to maximum possible => block gas limit
+			if tx.Gas == 0 {
+				tx.Gas = header.GasLimit
+			}
+
+			tracer, cancel, err := newTracer(config)
+			defer cancel()
+
+			if err != nil {
+				return nil, err
+			}
+
+			return d.store.TraceCall(tx, header, tracer)
+		},
+	)
 }
 
 func (d *Debug) traceBlock(

jsonrpc/dispatcher.go

@@ -60,7 +60,7 @@ type dispatcherParams struct {
 	jsonRPCBatchLengthLimit uint64
 	blockRangeLimit         uint64
 
-	requestsPerSecondDebug uint64
+	concurrentRequestsDebug uint64
 }
 
 func (dp dispatcherParams) isExceedingBatchLengthLimit(value uint64) bool {
@@ -111,7 +111,7 @@ func (d *Dispatcher) registerEndpoints(store JSONRPCStore) error {
 	d.endpoints.Bridge = &Bridge{
 		store,
 	}
-	d.endpoints.Debug = NewDebug(store, int(d.params.requestsPerSecondDebug))
+	d.endpoints.Debug = NewDebug(store, d.params.concurrentRequestsDebug)
 
 	var err error
 

jsonrpc/jsonrpc.go

@@ -69,7 +69,7 @@ type Config struct {
 	BatchLengthLimit         uint64
 	BlockRangeLimit          uint64
 
-	RequestsPerSecondDebug uint64
+	ConcurrentRequestsDebug uint64
 }
 
 // NewJSONRPC returns the JSONRPC http server
@@ -83,7 +83,7 @@ func NewJSONRPC(logger hclog.Logger, config *Config) (*JSONRPC, error) {
 			priceLimit:              config.PriceLimit,
 			jsonRPCBatchLengthLimit: config.BatchLengthLimit,
 			blockRangeLimit:         config.BlockRangeLimit,
-			requestsPerSecondDebug:  config.RequestsPerSecondDebug,
+			concurrentRequestsDebug: config.ConcurrentRequestsDebug,
 		},
 	)