EVM-779 Debug Transaction endpoint - throttling (#1818)

command/server/config/config.go

@@ -34,6 +34,8 @@ 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"`
 }
 
 // Telemetry holds the config details for metric services.
@@ -79,6 +81,9 @@ const (
 	// DefaultNumBlockConfirmations minimal number of child blocks required for the parent block to be considered final
 	// 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
 )
 
 // DefaultConfig returns the default server configuration
@@ -116,6 +121,7 @@ func DefaultConfig() *Config {
 		JSONRPCBlockRangeLimit:   DefaultJSONRPCBlockRangeLimit,
 		Relayer:                  false,
 		NumBlockConfirmations:    DefaultNumBlockConfirmations,
+		RequestsPerSecondDebug:   DefaultRequestsPerSecondDebug,
 	}
 }
 

command/server/params.go

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

command/server/server.go

@@ -228,6 +228,13 @@ func setFlags(cmd *cobra.Command) {
 		"minimal number of child blocks required for the parent block to be considered final",
 	)
 
+	cmd.Flags().Uint64Var(
+		&params.rawConfig.RequestsPerSecondDebug,
+		requestsPerSecondDebugFlag,
+		defaultConfig.RequestsPerSecondDebug,
+		"maximal number of requests per second for debug endpoints",
+	)
+
 	setLegacyFlags(cmd)
 
 	setDevFlags(cmd)

jsonrpc/debug_endpoint.go

@@ -65,7 +65,15 @@ type debugStore interface {
 
 // Debug is the debug jsonrpc endpoint
 type Debug struct {
-	store debugStore
+	store      debugStore
+	throttling *Throttling
+}
+
+func NewDebug(store debugStore, requestsPerSecond int) *Debug {
+	return &Debug{
+		store:      store,
+		throttling: NewThrottling(requestsPerSecond),
+	}
 }
 
 type TraceConfig struct {
@@ -80,6 +88,10 @@ 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
@@ -97,6 +109,10 @@ 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)
@@ -109,6 +125,10 @@ 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)
@@ -126,6 +146,10 @@ 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())
@@ -150,6 +174,10 @@ func (d *Debug) TraceCall(
 	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

jsonrpc/debug_endpoint_test.go

@@ -269,7 +269,7 @@ func TestTraceBlockByNumber(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 
-			endpoint := &Debug{test.store}
+			endpoint := NewDebug(test.store, 100000)
 
 			res, err := endpoint.TraceBlockByNumber(test.blockNumber, test.config)
 
@@ -338,7 +338,7 @@ func TestTraceBlockByHash(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 
-			endpoint := &Debug{test.store}
+			endpoint := NewDebug(test.store, 100000)
 
 			res, err := endpoint.TraceBlockByHash(test.blockHash, test.config)
 
@@ -397,7 +397,7 @@ func TestTraceBlock(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 
-			endpoint := &Debug{test.store}
+			endpoint := NewDebug(test.store, 100000)
 
 			res, err := endpoint.TraceBlock(test.input, test.config)
 
@@ -543,7 +543,7 @@ func TestTraceTransaction(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 
-			endpoint := &Debug{test.store}
+			endpoint := NewDebug(test.store, 100000)
 
 			res, err := endpoint.TraceTransaction(test.txHash, test.config)
 
@@ -679,7 +679,7 @@ func TestTraceCall(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 
-			endpoint := &Debug{test.store}
+			endpoint := NewDebug(test.store, 100000)
 
 			res, err := endpoint.TraceCall(test.arg, test.filter, test.config)
 

jsonrpc/dispatcher.go

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

jsonrpc/jsonrpc.go

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

jsonrpc/throttling.go

@@ -0,0 +1,70 @@
+package jsonrpc
+
+import (
+	"container/heap"
+	"errors"
+	"sync"
+	"time"
+)
+
+var errRequestLimitExceeded = errors.New("request limit exceeded")
+
+type Throttling struct {
+	requestsPerSeconds int
+	requests           timeQueue
+	lock               sync.Mutex
+}
+
+func NewThrottling(requestsPerSeconds int) *Throttling {
+	return &Throttling{
+		requestsPerSeconds: requestsPerSeconds,
+	}
+}
+
+func (t *Throttling) AttemptRequest() error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	currTime := time.Now().UTC()
+
+	// remove all old requests
+	for t.requests.Len() > 0 && currTime.Sub(t.requests[0]) > time.Second {
+		heap.Pop(&t.requests)
+	}
+
+	// if too many requests in one second return error
+	if t.requests.Len() == t.requestsPerSeconds {
+		return errRequestLimitExceeded
+	}
+
+	heap.Push(&t.requests, currTime)
+
+	return nil
+}
+
+type timeQueue []time.Time
+
+func (t *timeQueue) Len() int {
+	return len(*t)
+}
+
+func (t *timeQueue) Swap(i, j int) {
+	(*t)[i], (*t)[j] = (*t)[j], (*t)[i]
+}
+
+func (t *timeQueue) Push(x interface{}) {
+	if time, ok := x.(time.Time); ok {
+		*t = append(*t, time)
+	}
+}
+
+func (t *timeQueue) Pop() interface{} {
+	x := (*t)[len(*t)-1]
+	*t = (*t)[0 : len(*t)-1]
+
+	return x
+}
+
+func (t *timeQueue) Less(i, j int) bool {
+	return (*t)[i].Compare((*t)[j]) < 0
+}

jsonrpc/throttling_test.go

@@ -0,0 +1,41 @@
+package jsonrpc
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestThrottling(t *testing.T) {
+	t.Parallel()
+
+	th := NewThrottling(5)
+
+	assert.NoError(t, th.AttemptRequest())
+	assert.NoError(t, th.AttemptRequest())
+
+	time.Sleep(time.Millisecond * 100)
+
+	assert.NoError(t, th.AttemptRequest())
+	assert.NoError(t, th.AttemptRequest())
+
+	time.Sleep(time.Millisecond * 100)
+
+	assert.NoError(t, th.AttemptRequest())
+	assert.ErrorIs(t, th.AttemptRequest(), errRequestLimitExceeded)
+
+	time.Sleep(time.Millisecond * 100)
+
+	assert.ErrorIs(t, th.AttemptRequest(), errRequestLimitExceeded)
+
+	time.Sleep(time.Millisecond * 701)
+
+	assert.NoError(t, th.AttemptRequest())
+	assert.Equal(t, 4, th.requests.Len())
+
+	time.Sleep(time.Millisecond * 1002)
+
+	assert.NoError(t, th.AttemptRequest())
+	assert.Equal(t, 1, th.requests.Len())
+}

server/config.go

@@ -57,4 +57,5 @@ type JSONRPC struct {
 	AccessControlAllowOrigin []string
 	BatchLengthLimit         uint64
 	BlockRangeLimit          uint64
+	RequestsPerSecondDebug   uint64
 }

server/server.go

@@ -908,6 +908,7 @@ func (s *Server) setupJSONRPC() error {
 		PriceLimit:               s.config.PriceLimit,
 		BatchLengthLimit:         s.config.JSONRPC.BatchLengthLimit,
 		BlockRangeLimit:          s.config.JSONRPC.BlockRangeLimit,
+		RequestsPerSecondDebug:   s.config.JSONRPC.RequestsPerSecondDebug,
 	}
 
 	srv, err := jsonrpc.NewJSONRPC(s.logger, conf)