diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index 65ab5ee3b..d38caf50b 100644 --- a/CHANGELOG_PENDING.md +++ b/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 diff --git a/cmd/tendermint/commands/light.go b/cmd/tendermint/commands/light.go index 8e39d7900..2b812fe18 100644 --- a/cmd/tendermint/commands/light.go +++ b/cmd/tendermint/commands/light.go @@ -171,7 +171,8 @@ for applications built w/ Cosmos SDK). // If necessary adjust global WriteTimeout to ensure it's greater than // TimeoutBroadcastTxCommit. // See https://github.com/tendermint/tendermint/issues/3435 - if cfg.WriteTimeout <= conf.RPC.TimeoutBroadcastTxCommit { + // Note we don't need to adjust anything if the timeout is already unlimited. + if cfg.WriteTimeout > 0 && cfg.WriteTimeout <= conf.RPC.TimeoutBroadcastTxCommit { cfg.WriteTimeout = conf.RPC.TimeoutBroadcastTxCommit + 1*time.Second } diff --git a/config/config.go b/config/config.go index 500e3f7d6..c1fa4223a 100644 --- a/config/config.go +++ b/config/config.go @@ -523,7 +523,7 @@ func DefaultRPCConfig() *RPCConfig { MaxSubscriptionClients: 100, MaxSubscriptionsPerClient: 5, ExperimentalDisableWebsocket: false, // compatible with TM v0.35 and earlier - EventLogWindowSize: 0, // disables /events RPC by default + EventLogWindowSize: 30 * time.Second, EventLogMaxItems: 0, TimeoutBroadcastTxCommit: 10 * time.Second, diff --git a/go.mod b/go.mod index 2151f4b41..90142c5b1 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/gorilla/websocket v1.5.0 github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 - github.com/lib/pq v1.10.5 + github.com/lib/pq v1.10.6 github.com/libp2p/go-buffer-pool v0.0.2 github.com/mroth/weightedrand v0.4.1 github.com/oasisprotocol/curve25519-voi v0.0.0-20210609091139-0a56a4bca00b diff --git a/go.sum b/go.sum index 791fd4b73..c26a91a66 100644 --- a/go.sum +++ b/go.sum @@ -678,8 +678,8 @@ github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo= github.com/lib/pq v1.8.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/lib/pq v1.9.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/lib/pq v1.10.4/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= -github.com/lib/pq v1.10.5 h1:J+gdV2cUmX7ZqL2B0lFcW0m+egaHC2V3lpO8nWxyYiQ= -github.com/lib/pq v1.10.5/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/lib/pq v1.10.6 h1:jbk+ZieJ0D7EVGJYpL9QTz7/YW6UHbmdnZWYyK5cdBs= +github.com/lib/pq v1.10.6/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/libp2p/go-buffer-pool v0.0.2 h1:QNK2iAFa8gjAe1SPz6mHSMuCcjs+X1wlHzeOSqcmlfs= github.com/libp2p/go-buffer-pool v0.0.2/go.mod h1:MvaB6xw5vOrDl8rYZGLFdKAuk/hRoRZd1Vi32+RXyFM= github.com/lufeee/execinquery v1.0.0 h1:1XUTuLIVPDlFvUU3LXmmZwHDsolsxXnY67lzhpeqe0I= diff --git a/internal/inspect/rpc/rpc.go b/internal/inspect/rpc/rpc.go index 00c3e52ef..d70616834 100644 --- a/internal/inspect/rpc/rpc.go +++ b/internal/inspect/rpc/rpc.go @@ -125,7 +125,8 @@ func serverRPCConfig(r *config.RPCConfig) *server.Config { // If necessary adjust global WriteTimeout to ensure it's greater than // TimeoutBroadcastTxCommit. // See https://github.com/tendermint/tendermint/issues/3435 - if cfg.WriteTimeout <= r.TimeoutBroadcastTxCommit { + // Note we don't need to adjust anything if the timeout is already unlimited. + if cfg.WriteTimeout > 0 && cfg.WriteTimeout <= r.TimeoutBroadcastTxCommit { cfg.WriteTimeout = r.TimeoutBroadcastTxCommit + 1*time.Second } return cfg diff --git a/internal/mempool/reactor.go b/internal/mempool/reactor.go index 3c22988ee..28ee9e334 100644 --- a/internal/mempool/reactor.go +++ b/internal/mempool/reactor.go @@ -6,7 +6,6 @@ import ( "fmt" "runtime/debug" "sync" - "time" "github.com/tendermint/tendermint/config" "github.com/tendermint/tendermint/internal/libs/clist" @@ -22,13 +21,6 @@ var ( _ p2p.Wrapper = (*protomem.Message)(nil) ) -// PeerManager defines the interface contract required for getting necessary -// peer information. This should eventually be replaced with a message-oriented -// approach utilizing the p2p stack. -type PeerManager interface { - GetHeight(types.NodeID) int64 -} - // Reactor implements a service that contains mempool of txs that are broadcasted // amongst peers. It maintains a map from peer ID to counter, to prevent gossiping // txs to the peers you received it from. @@ -40,9 +32,8 @@ type Reactor struct { mempool *TxMempool ids *IDs - getPeerHeight func(types.NodeID) int64 - peerEvents p2p.PeerEventSubscriber - chCreator p2p.ChannelCreator + peerEvents p2p.PeerEventSubscriber + chCreator p2p.ChannelCreator // observePanic is a function for observing panics that were recovered in methods on // Reactor. observePanic is called with the recovered value. @@ -59,18 +50,16 @@ func NewReactor( txmp *TxMempool, chCreator p2p.ChannelCreator, peerEvents p2p.PeerEventSubscriber, - getPeerHeight func(types.NodeID) int64, ) *Reactor { r := &Reactor{ - logger: logger, - cfg: cfg, - mempool: txmp, - ids: NewMempoolIDs(), - chCreator: chCreator, - peerEvents: peerEvents, - getPeerHeight: getPeerHeight, - peerRoutines: make(map[types.NodeID]context.CancelFunc), - observePanic: defaultObservePanic, + logger: logger, + cfg: cfg, + mempool: txmp, + ids: NewMempoolIDs(), + chCreator: chCreator, + peerEvents: peerEvents, + peerRoutines: make(map[types.NodeID]context.CancelFunc), + observePanic: defaultObservePanic, } r.BaseService = *service.NewBaseService(logger, "Mempool", r) @@ -327,15 +316,6 @@ func (r *Reactor) broadcastTxRoutine(ctx context.Context, peerID types.NodeID, m memTx := nextGossipTx.Value.(*WrappedTx) - if r.getPeerHeight != nil { - height := r.getPeerHeight(peerID) - if height > 0 && height < memTx.height-1 { - // allow for a lag of one block - time.Sleep(PeerCatchupSleepIntervalMS * time.Millisecond) - continue - } - } - // NOTE: Transaction batching was disabled due to: // https://github.com/tendermint/tendermint/issues/5796 if ok := r.mempool.txStore.TxHasPeer(memTx.hash, peerMempoolID); !ok { diff --git a/internal/mempool/reactor_test.go b/internal/mempool/reactor_test.go index 8ceae2013..351315bae 100644 --- a/internal/mempool/reactor_test.go +++ b/internal/mempool/reactor_test.go @@ -85,7 +85,6 @@ func setupReactors(ctx context.Context, t *testing.T, logger log.Logger, numNode mempool, chCreator, func(ctx context.Context) *p2p.PeerUpdates { return rts.peerUpdates[nodeID] }, - rts.network.Nodes[nodeID].PeerManager.GetHeight, ) rts.nodes = append(rts.nodes, nodeID) diff --git a/internal/p2p/peermanager.go b/internal/p2p/peermanager.go index 756551a49..165b00e61 100644 --- a/internal/p2p/peermanager.go +++ b/internal/p2p/peermanager.go @@ -1027,37 +1027,6 @@ func (m *PeerManager) retryDelay(failures uint32, persistent bool) time.Duration return delay } -// GetHeight returns a peer's height, as reported via SetHeight, or 0 if the -// peer or height is unknown. -// -// FIXME: This is a temporary workaround to share state between the consensus -// and mempool reactors, carried over from the legacy P2P stack. Reactors should -// not have dependencies on each other, instead tracking this themselves. -func (m *PeerManager) GetHeight(peerID types.NodeID) int64 { - m.mtx.Lock() - defer m.mtx.Unlock() - - peer, _ := m.store.Get(peerID) - return peer.Height -} - -// SetHeight stores a peer's height, making it available via GetHeight. -// -// FIXME: This is a temporary workaround to share state between the consensus -// and mempool reactors, carried over from the legacy P2P stack. Reactors should -// not have dependencies on each other, instead tracking this themselves. -func (m *PeerManager) SetHeight(peerID types.NodeID, height int64) error { - m.mtx.Lock() - defer m.mtx.Unlock() - - peer, ok := m.store.Get(peerID) - if !ok { - peer = m.newPeerInfo(peerID) - } - peer.Height = height - return m.store.Set(peer) -} - // peerStore stores information about peers. It is not thread-safe, assuming it // is only used by PeerManager which handles concurrency control. This allows // the manager to execute multiple operations atomically via its own mutex. diff --git a/internal/p2p/peermanager_test.go b/internal/p2p/peermanager_test.go index 82d1e2693..47e8462a4 100644 --- a/internal/p2p/peermanager_test.go +++ b/internal/p2p/peermanager_test.go @@ -1868,38 +1868,3 @@ func TestPeerManager_Advertise_Self(t *testing.T) { self, }, peerManager.Advertise(dID, 100)) } - -func TestPeerManager_SetHeight_GetHeight(t *testing.T) { - a := p2p.NodeAddress{Protocol: "memory", NodeID: types.NodeID(strings.Repeat("a", 40))} - b := p2p.NodeAddress{Protocol: "memory", NodeID: types.NodeID(strings.Repeat("b", 40))} - - db := dbm.NewMemDB() - peerManager, err := p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{}) - require.NoError(t, err) - - // Getting a height should default to 0, for unknown peers and - // for known peers without height. - added, err := peerManager.Add(a) - require.NoError(t, err) - require.True(t, added) - require.EqualValues(t, 0, peerManager.GetHeight(a.NodeID)) - require.EqualValues(t, 0, peerManager.GetHeight(b.NodeID)) - - // Setting a height should work for a known node. - require.NoError(t, peerManager.SetHeight(a.NodeID, 3)) - require.EqualValues(t, 3, peerManager.GetHeight(a.NodeID)) - - // Setting a height should add an unknown node. - require.Equal(t, []types.NodeID{a.NodeID}, peerManager.Peers()) - require.NoError(t, peerManager.SetHeight(b.NodeID, 7)) - require.EqualValues(t, 7, peerManager.GetHeight(b.NodeID)) - require.ElementsMatch(t, []types.NodeID{a.NodeID, b.NodeID}, peerManager.Peers()) - - // The heights should not be persisted. - peerManager, err = p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{}) - require.NoError(t, err) - - require.ElementsMatch(t, []types.NodeID{a.NodeID, b.NodeID}, peerManager.Peers()) - require.Zero(t, peerManager.GetHeight(a.NodeID)) - require.Zero(t, peerManager.GetHeight(b.NodeID)) -} diff --git a/internal/rpc/core/env.go b/internal/rpc/core/env.go index 24f43a4a7..124525f26 100644 --- a/internal/rpc/core/env.go +++ b/internal/rpc/core/env.go @@ -236,7 +236,8 @@ func (env *Environment) StartService(ctx context.Context, conf *config.Config) ( // If necessary adjust global WriteTimeout to ensure it's greater than // TimeoutBroadcastTxCommit. // See https://github.com/tendermint/tendermint/issues/3435 - if cfg.WriteTimeout <= conf.RPC.TimeoutBroadcastTxCommit { + // Note we don't need to adjust anything if the timeout is already unlimited. + if cfg.WriteTimeout > 0 && cfg.WriteTimeout <= conf.RPC.TimeoutBroadcastTxCommit { cfg.WriteTimeout = conf.RPC.TimeoutBroadcastTxCommit + 1*time.Second } diff --git a/internal/rpc/core/routes.go b/internal/rpc/core/routes.go index 4bc1ca414..cafb92094 100644 --- a/internal/rpc/core/routes.go +++ b/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), diff --git a/node/node.go b/node/node.go index 56379d2e2..1bda1f0f7 100644 --- a/node/node.go +++ b/node/node.go @@ -266,7 +266,7 @@ func makeNode( node.evPool = evPool mpReactor, mp := createMempoolReactor(logger, cfg, proxyApp, stateStore, nodeMetrics.mempool, - peerManager.Subscribe, node.router.OpenChannel, peerManager.GetHeight) + peerManager.Subscribe, node.router.OpenChannel) node.rpcEnv.Mempool = mp node.services = append(node.services, mpReactor) diff --git a/node/setup.go b/node/setup.go index d6966800a..8089ea466 100644 --- a/node/setup.go +++ b/node/setup.go @@ -147,7 +147,6 @@ func createMempoolReactor( memplMetrics *mempool.Metrics, peerEvents p2p.PeerEventSubscriber, chCreator p2p.ChannelCreator, - peerHeight func(types.NodeID) int64, ) (service.Service, mempool.Mempool) { logger = logger.With("module", "mempool") @@ -166,7 +165,6 @@ func createMempoolReactor( mp, chCreator, peerEvents, - peerHeight, ) if cfg.Consensus.WaitForTxs() { diff --git a/rpc/jsonrpc/server/http_server.go b/rpc/jsonrpc/server/http_server.go index 0b715835d..50a37158e 100644 --- a/rpc/jsonrpc/server/http_server.go +++ b/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 } } diff --git a/rpc/jsonrpc/server/rpc_func.go b/rpc/jsonrpc/server/rpc_func.go index 8eba28728..1fff323d7 100644 --- a/rpc/jsonrpc/server/rpc_func.go +++ b/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 } diff --git a/test/e2e/node/main.go b/test/e2e/node/main.go index 2cbb9e4b0..94c1af1ab 100644 --- a/test/e2e/node/main.go +++ b/test/e2e/node/main.go @@ -210,7 +210,8 @@ func startLightNode(ctx context.Context, logger log.Logger, cfg *Config) error { // If necessary adjust global WriteTimeout to ensure it's greater than // TimeoutBroadcastTxCommit. // See https://github.com/tendermint/tendermint/issues/3435 - if rpccfg.WriteTimeout <= tmcfg.RPC.TimeoutBroadcastTxCommit { + // Note we don't need to adjust anything if the timeout is already unlimited. + if rpccfg.WriteTimeout > 0 && rpccfg.WriteTimeout <= tmcfg.RPC.TimeoutBroadcastTxCommit { rpccfg.WriteTimeout = tmcfg.RPC.TimeoutBroadcastTxCommit + 1*time.Second }