rpc: rework timeouts to be per-method instead of global (#8570)

* rpc: rework timeouts to be per-method instead of global

Prior to this change, we set a 10-second global timeout for all RPC methods
using the net/http Server type's WriteTimeout. This meant that any request
whose handler did not return within that period would simply drop the
connection to the client.

This timeout is too short for a default, as evidenced by issues like [1] and
[2].  In addition, the mode of failure on the client side is confusing; it
shows up as a dropped connection (EOF) rather than a meaningful error from the
service. More importantly, various methods have diffent constraints: Some
should be able to return quickly, others may need to adjust based on the
application workload.

This is a first step toward supporting configurable timeouts. This change:

- Removes the server-wide default global timeout, and instead:
- Wires up a default context timeout for all RPC handlers.
- Increases the default timeout from 10s to 60s.
- Adds a hook to override this per-method as needed.

This does NOT expose the timeouts in the configuration file (yet).

[1] https://github.com/osmosis-labs/osmosis/issues/1391
[2] https://github.com/tendermint/tendermint/issues/8465

CHANGELOG_PENDING.md

@@ -21,6 +21,7 @@ Special thanks to external contributors on this release:
   - [rpc] \#7982 Add new Events interface and deprecate Subscribe. (@creachadair)
   - [cli] \#8081 make the reset command safe to use by intoducing `reset-state` command. Fixed by \#8259. (@marbar3778, @cmwaters)
   - [config] \#8222 default indexer configuration to null. (@creachadair)
+  - [rpc] \#8570 rework timeouts to be per-method instead of global. (@creachadair)
 
 - Apps
 

internal/rpc/core/routes.go

@@ -28,7 +28,7 @@ func NewRoutesMap(svc RPCService, opts *RouteOptions) RoutesMap {
 	out := RoutesMap{
 		// Event subscription. Note that subscribe, unsubscribe, and
 		// unsubscribe_all are only available via the websocket endpoint.
-		"events":          rpc.NewRPCFunc(svc.Events),
+		"events":          rpc.NewRPCFunc(svc.Events).Timeout(0),
 		"subscribe":       rpc.NewWSRPCFunc(svc.Subscribe),
 		"unsubscribe":     rpc.NewWSRPCFunc(svc.Unsubscribe),
 		"unsubscribe_all": rpc.NewWSRPCFunc(svc.UnsubscribeAll),

rpc/jsonrpc/server/http_server.go

@@ -20,16 +20,27 @@ import (
 
 // Config is a RPC server configuration.
 type Config struct {
-	// see netutil.LimitListener
+	// The maximum number of connections that will be accepted by the listener.
+	// See https://godoc.org/golang.org/x/net/netutil#LimitListener
 	MaxOpenConnections int
-	// mirrors http.Server#ReadTimeout
+
+	// Used to set the HTTP server's per-request read timeout.
+	// See https://godoc.org/net/http#Server.ReadTimeout
 	ReadTimeout time.Duration
-	// mirrors http.Server#WriteTimeout
+
+	// Used to set the HTTP server's per-request write timeout.  Note that this
+	// affects ALL methods on the server, so it should not be set too low. This
+	// should be used as a safety valve, not a resource-control timeout.
+	//
+	// See https://godoc.org/net/http#Server.WriteTimeout
 	WriteTimeout time.Duration
-	// MaxBodyBytes controls the maximum number of bytes the
-	// server will read parsing the request body.
+
+	// Controls the maximum number of bytes the server will read parsing the
+	// request body.
 	MaxBodyBytes int64
-	// mirrors http.Server#MaxHeaderBytes
+
+	// Controls the maximum size of a request header.
+	// See https://godoc.org/net/http#Server.MaxHeaderBytes
 	MaxHeaderBytes int
 }
 
@@ -38,9 +49,9 @@ func DefaultConfig() *Config {
 	return &Config{
 		MaxOpenConnections: 0, // unlimited
 		ReadTimeout:        10 * time.Second,
-		WriteTimeout:       10 * time.Second,
-		MaxBodyBytes:       int64(1000000), // 1MB
-		MaxHeaderBytes:     1 << 20,        // same as the net/http default
+		WriteTimeout:       0,       // no default timeout
+		MaxBodyBytes:       1000000, // 1MB
+		MaxHeaderBytes:     1 << 20, // same as the net/http default
 	}
 }
 

rpc/jsonrpc/server/rpc_func.go

@@ -9,11 +9,16 @@ import (
 	"net/http"
 	"reflect"
 	"strings"
+	"time"
 
 	"github.com/tendermint/tendermint/libs/log"
 	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
 )
 
+// DefaultRPCTimeout is the default context timeout for calls to any RPC method
+// that does not override it with a more specific timeout.
+const DefaultRPCTimeout = 60 * time.Second
+
 // RegisterRPCFuncs adds a route to mux for each non-websocket function in the
 // funcMap, and also a root JSON-RPC POST handler.
 func RegisterRPCFuncs(mux *http.ServeMux, funcMap map[string]*RPCFunc, logger log.Logger) {
@@ -32,11 +37,12 @@ func RegisterRPCFuncs(mux *http.ServeMux, funcMap map[string]*RPCFunc, logger lo
 
 // RPCFunc contains the introspected type information for a function.
 type RPCFunc struct {
-	f      reflect.Value // underlying rpc function
-	param  reflect.Type  // the parameter struct, or nil
-	result reflect.Type  // the non-error result type, or nil
-	args   []argInfo     // names and type information (for URL decoding)
-	ws     bool          // websocket only
+	f       reflect.Value // underlying rpc function
+	param   reflect.Type  // the parameter struct, or nil
+	result  reflect.Type  // the non-error result type, or nil
+	args    []argInfo     // names and type information (for URL decoding)
+	timeout time.Duration // default request timeout, 0 means none
+	ws      bool          // websocket only
 }
 
 // argInfo records the name of a field, along with a bit to tell whether the
@@ -52,6 +58,12 @@ type argInfo struct {
 // with the resulting argument value. It reports an error if parameter parsing
 // fails, otherwise it returns the result from the wrapped function.
 func (rf *RPCFunc) Call(ctx context.Context, params json.RawMessage) (interface{}, error) {
+	// If ctx has its own deadline we will respect it; otherwise use rf.timeout.
+	if _, ok := ctx.Deadline(); !ok && rf.timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, rf.timeout)
+		defer cancel()
+	}
 	args, err := rf.parseParams(ctx, params)
 	if err != nil {
 		return nil, err
@@ -74,6 +86,11 @@ func (rf *RPCFunc) Call(ctx context.Context, params json.RawMessage) (interface{
 	return returns[0].Interface(), nil
 }
 
+// Timeout updates rf to include a default timeout for calls to rf. This
+// timeout is used if one is not already provided on the request context.
+// Setting d == 0 means there will be no timeout. Returns rf to allow chaining.
+func (rf *RPCFunc) Timeout(d time.Duration) *RPCFunc { rf.timeout = d; return rf }
+
 // parseParams parses the parameters of a JSON-RPC request and returns the
 // corresponding argument values. On success, the first argument value will be
 // the value of ctx.
@@ -129,7 +146,9 @@ func (rf *RPCFunc) adjustParams(data []byte) (json.RawMessage, error) {
 //     func(context.Context, *T) (R, error)
 //
 // for an arbitrary struct type T and type R. NewRPCFunc will panic if f does
-// not have one of these forms.
+// not have one of these forms.  A newly-constructed RPCFunc has a default
+// timeout of DefaultRPCTimeout; use the Timeout method to adjust this as
+// needed.
 func NewRPCFunc(f interface{}) *RPCFunc {
 	rf, err := newRPCFunc(f)
 	if err != nil {
@@ -215,10 +234,11 @@ func newRPCFunc(f interface{}) (*RPCFunc, error) {
 	}
 
 	return &RPCFunc{
-		f:      fv,
-		param:  ptype,
-		result: rtype,
-		args:   args,
+		f:       fv,
+		param:   ptype,
+		result:  rtype,
+		args:    args,
+		timeout: DefaultRPCTimeout, // until overridden
 	}, nil
 }