--api-timeout flag (#10260)

* `--api-timeout` flag

* simplify code

* review feedback

* better error handling

* better docs

Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

api/gateway/apimiddleware/api_middleware.go

@@ -3,6 +3,7 @@ package apimiddleware
 import (
 	"net/http"
 	"reflect"
+	"time"
 
 	"github.com/gorilla/mux"
 )
@@ -14,6 +15,7 @@ import (
 type ApiProxyMiddleware struct {
 	GatewayAddress  string
 	EndpointCreator EndpointFactory
+	Timeout         time.Duration
 	router          *mux.Router
 }
 
@@ -120,7 +122,7 @@ func (m *ApiProxyMiddleware) WithMiddleware(path string) http.HandlerFunc {
 			WriteError(w, errJson, nil)
 			return
 		}
-		grpcResp, errJson := ProxyRequest(req)
+		grpcResp, errJson := m.ProxyRequest(req)
 		if errJson != nil {
 			WriteError(w, errJson, nil)
 			return

api/gateway/apimiddleware/process_request.go

@@ -5,10 +5,10 @@ import (
 	"encoding/json"
 	"io"
 	"io/ioutil"
+	"net"
 	"net/http"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/prysm/api/grpc"
@@ -75,11 +75,14 @@ func (m *ApiProxyMiddleware) PrepareRequestForProxying(endpoint Endpoint, req *h
 }
 
 // ProxyRequest proxies the request to grpc-gateway.
-func ProxyRequest(req *http.Request) (*http.Response, ErrorJson) {
+func (m *ApiProxyMiddleware) ProxyRequest(req *http.Request) (*http.Response, ErrorJson) {
 	// We do not use http.DefaultClient because it does not have any timeout.
-	netClient := &http.Client{Timeout: time.Minute * 2}
+	netClient := &http.Client{Timeout: m.Timeout}
 	grpcResp, err := netClient.Do(req)
 	if err != nil {
+		if err, ok := err.(net.Error); ok && err.Timeout() {
+			return nil, TimeoutError()
+		}
 		return nil, InternalServerErrorWithMessage(err, "could not proxy request")
 	}
 	if grpcResp == nil {
@@ -111,9 +114,14 @@ func HandleGrpcResponseError(errJson ErrorJson, resp *http.Response, respBody []
 				w.Header().Set(h, v)
 			}
 		}
-		// Set code to HTTP code because unmarshalled body contained gRPC code.
-		errJson.SetCode(resp.StatusCode)
-		WriteError(w, errJson, resp.Header)
+		// Handle gRPC timeout.
+		if resp.StatusCode == http.StatusGatewayTimeout {
+			WriteError(w, TimeoutError(), resp.Header)
+		} else {
+			// Set code to HTTP code because unmarshalled body contained gRPC code.
+			errJson.SetCode(resp.StatusCode)
+			WriteError(w, errJson, resp.Header)
+		}
 	}
 	return responseHasError, nil
 }

api/gateway/apimiddleware/structs.go

@@ -41,6 +41,13 @@ func InternalServerError(err error) *DefaultErrorJson {
 	}
 }
 
+func TimeoutError() *DefaultErrorJson {
+	return &DefaultErrorJson{
+		Message: "Request timeout",
+		Code:    http.StatusRequestTimeout,
+	}
+}
+
 // StatusCode returns the error's underlying error code.
 func (e *DefaultErrorJson) StatusCode() int {
 	return e.Code

api/gateway/gateway.go

@@ -52,6 +52,7 @@ type config struct {
 	muxHandler                   MuxHandler
 	pbHandlers                   []*PbMux
 	router                       *mux.Router
+	timeout                      time.Duration
 }
 
 // Gateway is the gRPC gateway to serve HTTP JSON traffic as a proxy and forward it to the gRPC server.
@@ -248,6 +249,7 @@ func (g *Gateway) registerApiMiddleware() {
 	g.proxy = &apimiddleware.ApiProxyMiddleware{
 		GatewayAddress:  g.cfg.gatewayAddr,
 		EndpointCreator: g.cfg.apiMiddlewareEndpointFactory,
+		Timeout:         g.cfg.timeout,
 	}
 	log.Info("Starting API middleware")
 	g.proxy.Run(g.cfg.router)

api/gateway/options.go

@@ -1,7 +1,10 @@
 package gateway
 
 import (
+	"time"
+
 	"github.com/gorilla/mux"
+	gwruntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/prysmaticlabs/prysm/api/gateway/apimiddleware"
 )
 
@@ -79,3 +82,12 @@ func WithApiMiddleware(endpointFactory apimiddleware.EndpointFactory) Option {
 		return nil
 	}
 }
+
+// WithTimeout allows changing the timeout value for API calls.
+func WithTimeout(seconds uint64) Option {
+	return func(g *Gateway) error {
+		g.cfg.timeout = time.Second * time.Duration(seconds)
+		gwruntime.DefaultContextTimeout = time.Second * time.Duration(seconds)
+		return nil
+	}
+}

beacon-chain/node/node.go

@@ -834,6 +834,7 @@ func (b *BeaconNode) registerGRPCGateway() error {
 	selfCert := b.cliCtx.String(flags.CertFlag.Name)
 	maxCallSize := b.cliCtx.Uint64(cmd.GrpcMaxCallRecvMsgSizeFlag.Name)
 	httpModules := b.cliCtx.String(flags.HTTPModules.Name)
+	timeout := b.cliCtx.Int(cmd.ApiTimeoutFlag.Name)
 	if enableDebugRPCEndpoints {
 		maxCallSize = uint64(math.Max(float64(maxCallSize), debugGrpcMaxMsgSize))
 	}
@@ -855,6 +856,7 @@ func (b *BeaconNode) registerGRPCGateway() error {
 		apigateway.WithRemoteCert(selfCert),
 		apigateway.WithMaxCallRecvMsgSize(maxCallSize),
 		apigateway.WithAllowedOrigins(allowedOrigins),
+		apigateway.WithTimeout(uint64(timeout)),
 	}
 	if flags.EnableHTTPEthAPI(httpModules) {
 		opts = append(opts, apigateway.WithApiMiddleware(&apimiddleware.BeaconEndpointFactory{}))

beacon-chain/rpc/apimiddleware/custom_handlers.go

@@ -74,7 +74,7 @@ func handleGetSSZ(
 		apimiddleware.WriteError(w, errJson, nil)
 		return true
 	}
-	grpcResponse, errJson := apimiddleware.ProxyRequest(req)
+	grpcResponse, errJson := m.ProxyRequest(req)
 	if errJson != nil {
 		apimiddleware.WriteError(w, errJson, nil)
 		return true

cmd/beacon-chain/main.go

@@ -117,6 +117,7 @@ var appFlags = []cli.Flag{
 	cmd.RestoreTargetDirFlag,
 	cmd.BoltMMapInitialSizeFlag,
 	cmd.ValidatorMonitorIndicesFlag,
+	cmd.ApiTimeoutFlag,
 }
 
 func init() {

cmd/beacon-chain/usage.go

@@ -74,6 +74,7 @@ var appHelpFlagGroups = []flagGroup{
 			cmd.RestoreTargetDirFlag,
 			cmd.BoltMMapInitialSizeFlag,
 			cmd.ValidatorMonitorIndicesFlag,
+			cmd.ApiTimeoutFlag,
 		},
 	},
 	{

cmd/flags.go

@@ -249,6 +249,12 @@ var (
 		Usage: "Specifies the size in bytes of bolt db's mmap syscall allocation",
 		Value: 536870912, // 512 Mb as a default value.
 	}
+	// ApiTimeoutFlag specifies the timeout value for API requests in seconds. A timeout of zero means no timeout.
+	ApiTimeoutFlag = &cli.IntFlag{
+		Name:  "api-timeout",
+		Usage: "Specifies the timeout value for API requests in seconds",
+		Value: 120,
+	}
 )
 
 // LoadFlagsFromConfig sets flags values from config file if ConfigFileFlag is set.

cmd/validator/main.go

@@ -96,6 +96,7 @@ var appFlags = []cli.Flag{
 	cmd.ChainConfigFileFlag,
 	cmd.GrpcMaxCallRecvMsgSizeFlag,
 	cmd.BoltMMapInitialSizeFlag,
+	cmd.ApiTimeoutFlag,
 	debug.PProfFlag,
 	debug.PProfAddrFlag,
 	debug.PProfPortFlag,

cmd/validator/usage.go

@@ -66,6 +66,7 @@ var appHelpFlagGroups = []flagGroup{
 			cmd.GrpcMaxCallRecvMsgSizeFlag,
 			cmd.AcceptTosFlag,
 			cmd.BoltMMapInitialSizeFlag,
+			cmd.ApiTimeoutFlag,
 		},
 	},
 	{

validator/node/node.go

@@ -515,6 +515,7 @@ func (c *ValidatorClient) registerRPCGatewayService(cliCtx *cli.Context) error {
 	rpcPort := cliCtx.Int(flags.RPCPort.Name)
 	rpcAddr := fmt.Sprintf("%s:%d", rpcHost, rpcPort)
 	gatewayAddress := fmt.Sprintf("%s:%d", gatewayHost, gatewayPort)
+	timeout := cliCtx.Int(cmd.ApiTimeoutFlag.Name)
 	var allowedOrigins []string
 	if cliCtx.IsSet(flags.GPRCGatewayCorsDomain.Name) {
 		allowedOrigins = strings.Split(cliCtx.String(flags.GPRCGatewayCorsDomain.Name), ",")
@@ -580,6 +581,7 @@ func (c *ValidatorClient) registerRPCGatewayService(cliCtx *cli.Context) error {
 		gateway.WithAllowedOrigins(allowedOrigins),
 		gateway.WithApiMiddleware(&validatorMiddleware.ValidatorEndpointFactory{}),
 		gateway.WithMuxHandler(muxHandler),
+		gateway.WithTimeout(uint64(timeout)),
 	}
 	gw, err := gateway.New(cliCtx.Context, opts...)
 	if err != nil {