local rate limiting

Author: dvush

cmd/receiver-proxy/main.go

@@ -80,6 +80,12 @@ var flags []cli.Flag = []cli.Flag{
 		Usage:   "Number of parallel connections for each peer and archival RPC",
 		EnvVars: []string{"CONN_PER_PEER"},
 	},
+	&cli.IntFlag{
+		Name:    "max-local-requests-per-second",
+		Value:   100,
+		Usage:   "Maximum number of unique local requests per second",
+		EnvVars: []string{"MAX_LOCAL_RPS"},
+	},
 
 	// certificate config
 	&cli.DurationFlag{
@@ -198,6 +204,7 @@ func main() {
 			flashbotsSignerAddress := eth.HexToAddress(flashbotsSignerStr)
 			maxRequestBodySizeBytes := cCtx.Int64("max-request-body-size-bytes")
 			connectionsPerPeer := cCtx.Int("connections-per-peer")
+			maxLocalRPS := cCtx.Int("max-local-requests-per-second")
 
 			proxyConfig := &proxy.ReceiverProxyConfig{
 				ReceiverProxyConstantConfig: proxy.ReceiverProxyConstantConfig{Log: log, FlashbotsSignerAddress: flashbotsSignerAddress},
@@ -210,6 +217,7 @@ func main() {
 				EthRPC:                      rpcEndpoint,
 				MaxRequestBodySizeBytes:     maxRequestBodySizeBytes,
 				ConnectionsPerPeer:          connectionsPerPeer,
+				MaxLocalRPS:                 maxLocalRPS,
 			}
 
 			instance, err := proxy.NewReceiverProxy(*proxyConfig)

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/stretchr/testify v1.9.0
 	github.com/urfave/cli/v2 v2.27.2
+	golang.org/x/time v0.5.0
 )
 
 require (

go.sum

@@ -151,6 +151,8 @@ golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
 golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
 golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
+golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
 google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

proxy/api.go renamed to proxy/receiver_api.go

@@ -28,6 +28,7 @@ var (
 	errUnknownPeer          = errors.New("unknown peers can't send to the public address")
 	errSubsidyWrongEndpoint = errors.New("subsidy can only be called on public method")
 	errSubsidyWrongCaller   = errors.New("subsidy can only be called by Flashbots")
+	errRateLimiting         = errors.New("requests to local API are rate limited")
 
 	errUUIDParse = errors.New("failed to parse UUID")
 
@@ -315,6 +316,12 @@ func (prx *ReceiverProxy) HandleParsedRequest(ctx context.Context, parsedRequest
 		}
 		prx.requestUniqueKeysRLU.Add(*parsedRequest.requestArgUniqueKey, struct{}{})
 	}
+	if !parsedRequest.publicEndpoint {
+		err := prx.localAPIRateLimiter.Wait(ctx)
+		if err != nil {
+			return errors.Join(errRateLimiting, err)
+		}
+	}
 	select {
 	case <-ctx.Done():
 	case prx.shareQueue <- &parsedRequest:

proxy/receiver_proxy.go

@@ -13,6 +13,7 @@ import (
 	"github.com/flashbots/go-utils/signature"
 	"github.com/google/uuid"
 	"github.com/hashicorp/golang-lru/v2/expirable"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -61,6 +62,8 @@ type ReceiverProxy struct {
 	replacementNonceRLU *expirable.LRU[replacementNonceKey, int]
 
 	peerUpdaterClose chan struct{}
+
+	localAPIRateLimiter *rate.Limiter
 }
 
 type ReceiverProxyConstantConfig struct {
@@ -86,6 +89,7 @@ type ReceiverProxyConfig struct {
 	MaxRequestBodySizeBytes int64
 
 	ConnectionsPerPeer int
+	MaxLocalRPS        int
 }
 
 func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
@@ -105,6 +109,11 @@ func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
 
 	localBuilder := rpcclient.NewClient(config.LocalBuilderEndpoint)
 
+	limit := rate.Limit(config.MaxLocalRPS)
+	if config.MaxLocalRPS == 0 {
+		limit = rate.Inf
+	}
+	localAPIRateLimiter := rate.NewLimiter(limit, config.MaxLocalRPS)
 	prx := &ReceiverProxy{
 		ReceiverProxyConstantConfig: config.ReceiverProxyConstantConfig,
 		ConfigHub:                   NewBuilderConfigHub(config.Log, config.BuilderConfigHubEndpoint),
@@ -114,6 +123,7 @@ func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
 		localBuilder:                localBuilder,
 		requestUniqueKeysRLU:        expirable.NewLRU[uuid.UUID, struct{}](requestsRLUSize, nil, requestsRLUTTL),
 		replacementNonceRLU:         expirable.NewLRU[replacementNonceKey, int](replacementNonceSize, nil, replacementNonceTTL),
+		localAPIRateLimiter:         localAPIRateLimiter,
 	}
 	maxRequestBodySizeBytes := DefaultMaxRequestBodySizeBytes
 	if config.MaxRequestBodySizeBytes != 0 {

proxy/receiver_proxy_test.go

@@ -189,6 +189,7 @@ func createProxy(localBuilder, name string) *ReceiverProxy {
 		ArchiveEndpoint:          archiveServer.URL,
 		LocalBuilderEndpoint:     localBuilder,
 		EthRPC:                   "eth-rpc-not-set",
+		MaxLocalRPS:              10,
 	})
 	if err != nil {
 		panic(err)