diff --git a/.changeset/rude-mayflies-shout.md b/.changeset/rude-mayflies-shout.md new file mode 100644 index 00000000000..dcfcf8e35ce --- /dev/null +++ b/.changeset/rude-mayflies-shout.md @@ -0,0 +1,5 @@ +--- +'@eth-optimism/proxyd': patch +--- + +proxyd: Limit the number of concurrent RPCs to backends diff --git a/go/proxyd/backend.go b/go/proxyd/backend.go index a2b1200c6d5..17a7e754b0c 100644 --- a/go/proxyd/backend.go +++ b/go/proxyd/backend.go @@ -19,6 +19,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/gorilla/websocket" "github.com/prometheus/client_golang/prometheus" + "golang.org/x/sync/semaphore" ) const ( @@ -88,7 +89,7 @@ type Backend struct { authUsername string authPassword string rateLimiter RateLimiter - client *http.Client + client *LimitedHTTPClient dialer *websocket.Dialer maxRetries int maxResponseSize int64 @@ -170,6 +171,7 @@ func NewBackend( rpcURL string, wsURL string, rateLimiter RateLimiter, + rpcSemaphore *semaphore.Weighted, opts ...BackendOpt, ) *Backend { backend := &Backend{ @@ -178,8 +180,10 @@ func NewBackend( wsURL: wsURL, rateLimiter: rateLimiter, maxResponseSize: math.MaxInt64, - client: &http.Client{ - Timeout: 5 * time.Second, + client: &LimitedHTTPClient{ + Client: http.Client{Timeout: 5 * time.Second}, + sem: rpcSemaphore, + backendName: name, }, dialer: &websocket.Dialer{}, } @@ -358,7 +362,7 @@ func (b *Backend) doForward(ctx context.Context, rpcReq *RPCReq) (*RPCRes, error httpReq.Header.Set("content-type", "application/json") httpReq.Header.Set("X-Forwarded-For", xForwardedFor) - httpRes, err := b.client.Do(httpReq) + httpRes, err := b.client.DoLimited(httpReq) if err != nil { return nil, wrapErr(err, "error in backend request") } @@ -693,3 +697,18 @@ func sleepContext(ctx context.Context, duration time.Duration) { case <-time.After(duration): } } + +type LimitedHTTPClient struct { + http.Client + sem *semaphore.Weighted + backendName string +} + +func (c *LimitedHTTPClient) DoLimited(req *http.Request) (*http.Response, error) { + if err := c.sem.Acquire(req.Context(), 1); err != nil { + tooManyRequestErrorsTotal.WithLabelValues(c.backendName).Inc() + return nil, wrapErr(err, "too many requests") + } + defer c.sem.Release(1) + return c.Do(req) +} diff --git a/go/proxyd/config.go b/go/proxyd/config.go index ac83114cdba..332df4c3987 100644 --- a/go/proxyd/config.go +++ b/go/proxyd/config.go @@ -7,11 +7,12 @@ import ( ) type ServerConfig struct { - RPCHost string `toml:"rpc_host"` - RPCPort int `toml:"rpc_port"` - WSHost string `toml:"ws_host"` - WSPort int `toml:"ws_port"` - MaxBodySizeBytes int64 `toml:"max_body_size_bytes"` + RPCHost string `toml:"rpc_host"` + RPCPort int `toml:"rpc_port"` + WSHost string `toml:"ws_host"` + WSPort int `toml:"ws_port"` + MaxBodySizeBytes int64 `toml:"max_body_size_bytes"` + MaxConcurrentRPCs int64 `toml:"max_concurrent_rpcs"` // TimeoutSeconds specifies the maximum time spent serving an HTTP request. Note that isn't used for websocket connections TimeoutSeconds int `toml:"timeout_seconds"` diff --git a/go/proxyd/example.config.toml b/go/proxyd/example.config.toml index c0318d4f24b..25739106b81 100644 --- a/go/proxyd/example.config.toml +++ b/go/proxyd/example.config.toml @@ -18,6 +18,7 @@ ws_host = "0.0.0.0" ws_port = 8085 # Maximum client body size, in bytes, that the server will accept. max_body_size_bytes = 10485760 +max_concurrent_rpcs = 1000 [redis] # URL to a Redis instance. diff --git a/go/proxyd/go.mod b/go/proxyd/go.mod index 44659c3199c..e9a28a93184 100644 --- a/go/proxyd/go.mod +++ b/go/proxyd/go.mod @@ -17,5 +17,6 @@ require ( github.com/rs/cors v1.8.0 github.com/stretchr/testify v1.7.0 github.com/yuin/gopher-lua v0.0.0-20210529063254-f4c35e4016d9 // indirect + golang.org/x/sync v0.0.0-20210220032951-036812b2e83c gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect ) diff --git a/go/proxyd/integration_tests/max_rpc_conns_test.go b/go/proxyd/integration_tests/max_rpc_conns_test.go new file mode 100644 index 00000000000..155fa841660 --- /dev/null +++ b/go/proxyd/integration_tests/max_rpc_conns_test.go @@ -0,0 +1,79 @@ +package integration_tests + +import ( + "net/http" + "net/http/httptest" + "os" + "sync" + "testing" + "time" + + "github.com/ethereum-optimism/optimism/go/proxyd" + "github.com/stretchr/testify/require" +) + +func TestMaxConcurrentRPCs(t *testing.T) { + var ( + mu sync.Mutex + concurrentRPCs int + maxConcurrentRPCs int + ) + handler := func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + concurrentRPCs++ + if maxConcurrentRPCs < concurrentRPCs { + maxConcurrentRPCs = concurrentRPCs + } + mu.Unlock() + + time.Sleep(time.Second * 2) + SingleResponseHandler(200, goodResponse)(w, r) + + mu.Lock() + concurrentRPCs-- + mu.Unlock() + } + // We don't use the MockBackend because it serializes requests to the handler + slowBackend := httptest.NewServer(http.HandlerFunc(handler)) + defer slowBackend.Close() + + require.NoError(t, os.Setenv("GOOD_BACKEND_RPC_URL", slowBackend.URL)) + + config := ReadConfig("max_rpc_conns") + client := NewProxydClient("http://127.0.0.1:8545") + shutdown, err := proxyd.Start(config) + require.NoError(t, err) + defer shutdown() + + type resWithCodeErr struct { + res []byte + code int + err error + } + resCh := make(chan *resWithCodeErr) + for i := 0; i < 3; i++ { + go func() { + res, code, err := client.SendRPC("eth_chainId", nil) + resCh <- &resWithCodeErr{ + res: res, + code: code, + err: err, + } + }() + } + res1 := <-resCh + res2 := <-resCh + res3 := <-resCh + + require.NoError(t, res1.err) + require.NoError(t, res2.err) + require.NoError(t, res3.err) + require.Equal(t, 200, res1.code) + require.Equal(t, 200, res2.code) + require.Equal(t, 200, res3.code) + RequireEqualJSON(t, []byte(goodResponse), res1.res) + RequireEqualJSON(t, []byte(goodResponse), res2.res) + RequireEqualJSON(t, []byte(goodResponse), res3.res) + + require.EqualValues(t, 2, maxConcurrentRPCs) +} diff --git a/go/proxyd/integration_tests/testdata/max_rpc_conns.toml b/go/proxyd/integration_tests/testdata/max_rpc_conns.toml new file mode 100644 index 00000000000..68d7c19c72e --- /dev/null +++ b/go/proxyd/integration_tests/testdata/max_rpc_conns.toml @@ -0,0 +1,19 @@ +[server] +rpc_port = 8545 +max_concurrent_rpcs = 2 + +[backend] +# this should cover blocked requests due to max_concurrent_rpcs +response_timeout_seconds = 12 + +[backends] +[backends.good] +rpc_url = "$GOOD_BACKEND_RPC_URL" +ws_url = "$GOOD_BACKEND_RPC_URL" + +[backend_groups] +[backend_groups.main] +backends = ["good"] + +[rpc_method_mappings] +eth_chainId = "main" diff --git a/go/proxyd/metrics.go b/go/proxyd/metrics.go index 7463b7acc06..faf2534c5b3 100644 --- a/go/proxyd/metrics.go +++ b/go/proxyd/metrics.go @@ -212,6 +212,14 @@ var ( Help: "Histogram of Redis command durations, in milliseconds.", Buckets: MillisecondDurationBuckets, }, []string{"command"}) + + tooManyRequestErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: MetricsNamespace, + Name: "too_many_request_errors_total", + Help: "Count of request timeouts due to too many concurrent RPCs.", + }, []string{ + "backend_name", + }) ) func RecordRedisError(source string) { diff --git a/go/proxyd/proxyd.go b/go/proxyd/proxyd.go index e71e14c2222..a51e043d7c1 100644 --- a/go/proxyd/proxyd.go +++ b/go/proxyd/proxyd.go @@ -10,9 +10,11 @@ import ( "strconv" "time" + "github.com/ethereum/go-ethereum/common/math" "github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/log" "github.com/prometheus/client_golang/prometheus/promhttp" + "golang.org/x/sync/semaphore" ) func Start(config *Config) (func(), error) { @@ -53,6 +55,12 @@ func Start(config *Config) (func(), error) { } } + maxConcurrentRPCs := config.Server.MaxConcurrentRPCs + if maxConcurrentRPCs == 0 { + maxConcurrentRPCs = math.MaxInt64 + } + rpcRequestSemaphore := semaphore.NewWeighted(maxConcurrentRPCs) + backendNames := make([]string, 0) backendsByName := make(map[string]*Backend) for name, cfg := range config.Backends { @@ -111,7 +119,7 @@ func Start(config *Config) (func(), error) { opts = append(opts, WithStrippedTrailingXFF()) } opts = append(opts, WithProxydIP(os.Getenv("PROXYD_IP"))) - back := NewBackend(name, rpcURL, wsURL, lim, opts...) + back := NewBackend(name, rpcURL, wsURL, lim, rpcRequestSemaphore, opts...) backendNames = append(backendNames, name) backendsByName[name] = back log.Info("configured backend", "name", name, "rpc_url", rpcURL, "ws_url", wsURL)