proxyd: Limit the number of concurrent RPCs to backends (#2464)

* proxyd: Limit the number of concurrent RPCs to backends

We add a new config `max_concurrent_rpcs` under the server section to
prevent too many RPC requests, (and, more loosely, the number of connections),
from being sent upstream. Requests will block, in a FIFO fashion, until the
number of in-flight RPCs is under the limit.

* fix typo
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

.changeset/rude-mayflies-shout.md

+---
+'@eth-optimism/proxyd': patch
+---
+
+proxyd: Limit the number of concurrent RPCs to backends

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)
+}

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"`

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.

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
 )

go/proxyd/integration_tests/max_rpc_conns_test.go

+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)
+}

go/proxyd/integration_tests/testdata/max_rpc_conns.toml

+[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"

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) {

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)