From df669c7bed69d12c5fdf7c830991278a9f936c8b Mon Sep 17 00:00:00 2001 From: William Banfield Date: Thu, 29 Jul 2021 19:14:26 -0400 Subject: [PATCH] node: sketch of debug node logic --- internal/consensus/replay_test.go | 3 + light/proxy/proxy.go | 2 +- node/debug.go | 78 +++++++++++++ node/debug_test.go | 73 ++++++++++++ node/node.go | 147 ++++-------------------- node/node_test.go | 6 +- node/rpc.go | 150 +++++++++++++++++++++++++ node/setup.go | 9 +- rpc/core/blocks_test.go | 1 + rpc/core/routes.go | 72 +++++++++--- rpc/jsonrpc/jsonrpc_test.go | 4 +- rpc/jsonrpc/server/http_server.go | 18 +-- rpc/jsonrpc/server/http_server_test.go | 2 +- rpc/jsonrpc/test/main.go | 2 +- store/store.go | 4 + types/event_bus.go | 14 ++- 16 files changed, 418 insertions(+), 167 deletions(-) create mode 100644 node/debug.go create mode 100644 node/debug_test.go create mode 100644 node/rpc.go diff --git a/internal/consensus/replay_test.go b/internal/consensus/replay_test.go index 4d1c9c6b2..81ff39ff2 100644 --- a/internal/consensus/replay_test.go +++ b/internal/consensus/replay_test.go @@ -1217,6 +1217,9 @@ func (bs *mockBlockStore) PruneBlocks(height int64) (uint64, error) { bs.base = height return pruned, nil } +func (bs *mockBlockStore) Close() error { + return nil +} //--------------------------------------- // Test handshake/init chain diff --git a/light/proxy/proxy.go b/light/proxy/proxy.go index 8f1e7bf87..6f2622588 100644 --- a/light/proxy/proxy.go +++ b/light/proxy/proxy.go @@ -113,7 +113,7 @@ func (p *Proxy) listen() (net.Listener, *http.ServeMux, error) { } // 4) Start listening for new connections. - listener, err := rpcserver.Listen(p.Addr, p.Config) + listener, err := rpcserver.Listen(p.Addr, p.Config.MaxOpenConnections) if err != nil { return nil, mux, err } diff --git a/node/debug.go b/node/debug.go new file mode 100644 index 000000000..d73dbc61c --- /dev/null +++ b/node/debug.go @@ -0,0 +1,78 @@ +package node + +import ( + "net" + + "github.com/tendermint/tendermint/config" + cfg "github.com/tendermint/tendermint/config" + "github.com/tendermint/tendermint/libs/log" + "github.com/tendermint/tendermint/libs/service" + rpccore "github.com/tendermint/tendermint/rpc/core" + sm "github.com/tendermint/tendermint/state" + "github.com/tendermint/tendermint/store" + "github.com/tendermint/tendermint/types" +) + +// Debug is a type useful for debugging tendermint problems. +// Tendermint nodes will shutdown if a divergent hash is detected. Once in this +// state, they will not start up again. Debug runs just an RPC server on the +// tendermint data stores without running any other components. This way a user +// can query the RPC server to diagnose the issue that caused a crash to begin with. +type Debug struct { + service.BaseService + + blockStore sm.BlockStore + stateStore sm.Store + + rpcConfig *cfg.RPCConfig + listeners []net.Listener +} + +func NewDebugFromConfig(cfg *config.Config) (*Debug, error) { + blockStoreDB, err := config.DefaultDBProvider(&config.DBContext{ID: _blockStoreID, Config: cfg}) + if err != nil { + return nil, err + } + blockStore := store.NewBlockStore(blockStoreDB) + stateDB, err := config.DefaultDBProvider(&config.DBContext{ID: _stateStoreID, Config: cfg}) + stateStore := sm.NewStore(stateDB) + + return NewDebug(cfg.RPC, blockStore, stateStore), nil +} + +func NewDebug(rpcConfig *cfg.RPCConfig, blockStore sm.BlockStore, stateStore sm.Store) *Debug { + return &Debug{ + blockStore: blockStore, + stateStore: stateStore, + rpcConfig: rpcConfig, + } +} + +func NewDefaultDebug() (*Debug, error) { + cfg := config.Config{ + BaseConfig: config.DefaultBaseConfig(), + RPC: config.DefaultRPCConfig(), + } + return NewDebugFromConfig(&cfg) +} + +func (debug *Debug) OnStart() error { + rpcCoreEnv := rpccore.Environment{ + StateStore: debug.stateStore, + BlockStore: debug.blockStore, + } + routes := rpcCoreEnv.InfoRoutes() + l := log.MustNewDefaultLogger(log.LogFormatPlain, log.LogLevelInfo, false) + listeners, err := startHTTPRPCServer(debug.rpcConfig, l, routes, types.NopEventBus{}) + if err != nil { + return err + } + debug.listeners = listeners + return nil +} + +func (debug *Debug) OnStop() { + for _, listener := range debug.listeners { + listener.Close() + } +} diff --git a/node/debug_test.go b/node/debug_test.go new file mode 100644 index 000000000..11a810fc7 --- /dev/null +++ b/node/debug_test.go @@ -0,0 +1,73 @@ +package node_test + +import ( + "context" + "testing" + + "github.com/fortytw2/leaktest" + "github.com/stretchr/testify/require" + "github.com/tendermint/tendermint/config" + "github.com/tendermint/tendermint/node" + http_client "github.com/tendermint/tendermint/rpc/client/http" + state_mocks "github.com/tendermint/tendermint/state/mocks" + "github.com/tendermint/tendermint/types" +) + +func TestDebugConstructor(t *testing.T) { + t.Cleanup(leaktest.Check(t)) + t.Run("from config", func(t *testing.T) { + d, err := node.NewDebugFromConfig(&config.Config{ + BaseConfig: config.TestBaseConfig(), + RPC: config.TestRPCConfig(), + }) + require.NoError(t, err) + require.NotNil(t, d) + + d.OnStop() + }) + +} + +func TestDebugRun(t *testing.T) { + t.Cleanup(leaktest.Check(t)) + t.Run("from config", func(t *testing.T) { + d, err := node.NewDebugFromConfig(&config.Config{ + BaseConfig: config.TestBaseConfig(), + RPC: config.TestRPCConfig(), + }) + require.NoError(t, err) + err = d.OnStart() + require.NoError(t, err) + d.OnStop() + }) + +} + +func TestDebugServeInfoRPC(t *testing.T) { + testHeight := int64(1) + testBlock := new(types.Block) + testBlock.Header.Height = testHeight + testBlock.Header.LastCommitHash = []byte("test hash") + stateStoreMock := &state_mocks.Store{} + + blockStoreMock := &state_mocks.BlockStore{} + blockStoreMock.On("Height").Return(testHeight) + blockStoreMock.On("Base").Return(int64(0)) + blockStoreMock.On("LoadBlockMeta", testHeight).Return(&types.BlockMeta{}) + blockStoreMock.On("LoadBlock", testHeight).Return(testBlock) + + rpcConfig := config.TestRPCConfig() + d := node.NewDebug(rpcConfig, blockStoreMock, stateStoreMock) + require.NoError(t, d.OnStart()) + cli, err := http_client.New(rpcConfig.ListenAddress) + require.NoError(t, err) + resultBlock, err := cli.Block(context.Background(), &testHeight) + require.NoError(t, err) + require.Equal(t, testBlock.Height, resultBlock.Block.Height) + require.Equal(t, testBlock.LastCommitHash, resultBlock.Block.LastCommitHash) + + d.OnStop() + + blockStoreMock.AssertExpectations(t) + stateStoreMock.AssertExpectations(t) +} diff --git a/node/node.go b/node/node.go index ced8af729..e50d2166b 100644 --- a/node/node.go +++ b/node/node.go @@ -13,7 +13,6 @@ import ( _ "github.com/lib/pq" // provide the psql db driver "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" - "github.com/rs/cors" abci "github.com/tendermint/tendermint/abci/types" cfg "github.com/tendermint/tendermint/config" "github.com/tendermint/tendermint/crypto" @@ -25,7 +24,6 @@ import ( "github.com/tendermint/tendermint/internal/statesync" "github.com/tendermint/tendermint/libs/log" tmnet "github.com/tendermint/tendermint/libs/net" - tmpubsub "github.com/tendermint/tendermint/libs/pubsub" "github.com/tendermint/tendermint/libs/service" "github.com/tendermint/tendermint/libs/strings" tmtime "github.com/tendermint/tendermint/libs/time" @@ -34,8 +32,6 @@ import ( tmgrpc "github.com/tendermint/tendermint/privval/grpc" "github.com/tendermint/tendermint/proxy" rpccore "github.com/tendermint/tendermint/rpc/core" - grpccore "github.com/tendermint/tendermint/rpc/grpc" - rpcserver "github.com/tendermint/tendermint/rpc/jsonrpc/server" sm "github.com/tendermint/tendermint/state" "github.com/tendermint/tendermint/state/indexer" "github.com/tendermint/tendermint/store" @@ -573,13 +569,35 @@ func (n *nodeImpl) OnStart() error { // Start the RPC server before the P2P server // so we can eg. receive txs for the first block if n.config.RPC.ListenAddress != "" && n.config.Mode != cfg.ModeSeed { - listeners, err := n.startRPC() + env, err := n.ConfigureRPC() + if err != nil { + return err + } + + routes := env.GetRoutes() + if n.config.RPC.Unsafe { + routes = rpccore.CombineRoutes(routes, env.UnsafeRoutes()) + } + + listeners, err := startHTTPRPCServer(n.config.RPC, env.Logger, routes, n.eventBus) if err != nil { return err } n.rpcListeners = listeners } + if n.config.RPC.GRPCListenAddress != "" && n.config.Mode != cfg.ModeSeed { + env, err := n.ConfigureRPC() + if err != nil { + return err + } + listener, err := startGRPCServer(n.config.RPC, env, env.Logger) + if err != nil { + return err + } + n.rpcListeners = append(n.rpcListeners, listener) + } + if n.config.Instrumentation.Prometheus && n.config.Instrumentation.PrometheusListenAddr != "" { n.prometheusSrv = n.startPrometheusServer(n.config.Instrumentation.PrometheusListenAddr) @@ -805,125 +823,6 @@ func (n *nodeImpl) ConfigureRPC() (*rpccore.Environment, error) { return &rpcCoreEnv, nil } -func (n *nodeImpl) startRPC() ([]net.Listener, error) { - env, err := n.ConfigureRPC() - if err != nil { - return nil, err - } - - listenAddrs := strings.SplitAndTrimEmpty(n.config.RPC.ListenAddress, ",", " ") - routes := env.GetRoutes() - - if n.config.RPC.Unsafe { - env.AddUnsafe(routes) - } - - config := rpcserver.DefaultConfig() - config.MaxBodyBytes = n.config.RPC.MaxBodyBytes - config.MaxHeaderBytes = n.config.RPC.MaxHeaderBytes - config.MaxOpenConnections = n.config.RPC.MaxOpenConnections - // If necessary adjust global WriteTimeout to ensure it's greater than - // TimeoutBroadcastTxCommit. - // See https://github.com/tendermint/tendermint/issues/3435 - if config.WriteTimeout <= n.config.RPC.TimeoutBroadcastTxCommit { - config.WriteTimeout = n.config.RPC.TimeoutBroadcastTxCommit + 1*time.Second - } - - // we may expose the rpc over both a unix and tcp socket - listeners := make([]net.Listener, len(listenAddrs)) - for i, listenAddr := range listenAddrs { - mux := http.NewServeMux() - rpcLogger := n.Logger.With("module", "rpc-server") - wmLogger := rpcLogger.With("protocol", "websocket") - wm := rpcserver.NewWebsocketManager(routes, - rpcserver.OnDisconnect(func(remoteAddr string) { - err := n.eventBus.UnsubscribeAll(context.Background(), remoteAddr) - if err != nil && err != tmpubsub.ErrSubscriptionNotFound { - wmLogger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err) - } - }), - rpcserver.ReadLimit(config.MaxBodyBytes), - ) - wm.SetLogger(wmLogger) - mux.HandleFunc("/websocket", wm.WebsocketHandler) - rpcserver.RegisterRPCFuncs(mux, routes, rpcLogger) - listener, err := rpcserver.Listen( - listenAddr, - config, - ) - if err != nil { - return nil, err - } - - var rootHandler http.Handler = mux - if n.config.RPC.IsCorsEnabled() { - corsMiddleware := cors.New(cors.Options{ - AllowedOrigins: n.config.RPC.CORSAllowedOrigins, - AllowedMethods: n.config.RPC.CORSAllowedMethods, - AllowedHeaders: n.config.RPC.CORSAllowedHeaders, - }) - rootHandler = corsMiddleware.Handler(mux) - } - if n.config.RPC.IsTLSEnabled() { - go func() { - if err := rpcserver.ServeTLS( - listener, - rootHandler, - n.config.RPC.CertFile(), - n.config.RPC.KeyFile(), - rpcLogger, - config, - ); err != nil { - n.Logger.Error("Error serving server with TLS", "err", err) - } - }() - } else { - go func() { - if err := rpcserver.Serve( - listener, - rootHandler, - rpcLogger, - config, - ); err != nil { - n.Logger.Error("Error serving server", "err", err) - } - }() - } - - listeners[i] = listener - } - - // we expose a simplified api over grpc for convenience to app devs - grpcListenAddr := n.config.RPC.GRPCListenAddress - if grpcListenAddr != "" { - config := rpcserver.DefaultConfig() - config.MaxBodyBytes = n.config.RPC.MaxBodyBytes - config.MaxHeaderBytes = n.config.RPC.MaxHeaderBytes - // NOTE: GRPCMaxOpenConnections is used, not MaxOpenConnections - config.MaxOpenConnections = n.config.RPC.GRPCMaxOpenConnections - // If necessary adjust global WriteTimeout to ensure it's greater than - // TimeoutBroadcastTxCommit. - // See https://github.com/tendermint/tendermint/issues/3435 - if config.WriteTimeout <= n.config.RPC.TimeoutBroadcastTxCommit { - config.WriteTimeout = n.config.RPC.TimeoutBroadcastTxCommit + 1*time.Second - } - listener, err := rpcserver.Listen(grpcListenAddr, config) - if err != nil { - return nil, err - } - go func() { - if err := grpccore.StartGRPCServer(env, listener); err != nil { - n.Logger.Error("Error starting gRPC server", "err", err) - } - }() - listeners = append(listeners, listener) - - } - - return listeners, nil - -} - // startPrometheusServer starts a Prometheus HTTP server, listening for metrics // collectors on addr. func (n *nodeImpl) startPrometheusServer(addr string) *http.Server { diff --git a/node/node_test.go b/node/node_test.go index 16edb4210..d80eacae1 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -81,8 +81,10 @@ func TestNodeStartStop(t *testing.T) { panic(err) } err = p.Signal(syscall.SIGABRT) - fmt.Println(err) - t.Fatal("timed out waiting for shutdown") + if err != nil { + t.Logf("err: %s", err) + } + t.Fatalf("timed out waiting for shutdown") } } diff --git a/node/rpc.go b/node/rpc.go new file mode 100644 index 000000000..d4a7cbd12 --- /dev/null +++ b/node/rpc.go @@ -0,0 +1,150 @@ +package node + +import ( + "context" + "errors" + "net" + "net/http" + "time" + + "github.com/rs/cors" + cfg "github.com/tendermint/tendermint/config" + "github.com/tendermint/tendermint/libs/log" + tmpubsub "github.com/tendermint/tendermint/libs/pubsub" + "github.com/tendermint/tendermint/libs/strings" + rpccore "github.com/tendermint/tendermint/rpc/core" + grpccore "github.com/tendermint/tendermint/rpc/grpc" + rpcserver "github.com/tendermint/tendermint/rpc/jsonrpc/server" + "github.com/tendermint/tendermint/types" +) + +func startGRPCServer(rpcConfig *cfg.RPCConfig, + env *rpccore.Environment, + logger log.Logger) (net.Listener, error) { + // we expose a simplified api over grpc for convenience to app devs + listener, err := rpcserver.Listen(rpcConfig.GRPCListenAddress, rpcConfig.GRPCMaxOpenConnections) + if err != nil { + return nil, err + } + go func() { + if err := grpccore.StartGRPCServer(env, listener); err != nil { + logger.Error("Error starting gRPC server", "err", err) + } + }() + return listener, nil +} + +func startHTTPRPCServer(rpcConfig *cfg.RPCConfig, + logger log.Logger, + routes rpccore.RoutesMap, + eventBus types.EventBusSubscriber) ([]net.Listener, error) { + + config := rpcserver.DefaultConfig() + config.MaxBodyBytes = rpcConfig.MaxBodyBytes + config.MaxHeaderBytes = rpcConfig.MaxHeaderBytes + // If necessary adjust global WriteTimeout to ensure it's greater than + // TimeoutBroadcastTxCommit. + // See https://github.com/tendermint/tendermint/issues/3435 + if config.WriteTimeout <= rpcConfig.TimeoutBroadcastTxCommit { + config.WriteTimeout = rpcConfig.TimeoutBroadcastTxCommit + 1*time.Second + } + + // we may expose the rpc over both a unix and tcp socket + listeners, err := listenersFromRPCConfig(rpcConfig) + if err != nil { + return nil, err + } + for _, listener := range listeners { + mux := http.NewServeMux() + registerWebsocketHandler(rpcConfig, mux, routes, logger, eventBus) + rpcserver.RegisterRPCFuncs(mux, routes, logger) + + var rootHandler http.Handler = mux + if rpcConfig.IsCorsEnabled() { + rootHandler = addCORSHandler(rpcConfig, mux) + } + if rpcConfig.IsTLSEnabled() { + go func() { + listenerAddr := listener.Addr().String() + keyFile := rpcConfig.KeyFile() + certFile := rpcConfig.CertFile() + logger.Info("RPC HTTPS server starting", "address", listenerAddr, + "certfile", certFile, "keyfile", keyFile) + + err := rpcserver.ServeTLS(listener, rootHandler, keyFile, certFile, logger, config) + if !errors.Is(err, net.ErrClosed) { + logger.Error("RPC HTTPS server stopped with error", "address", listener, "err", err) + return + } + logger.Info("RPC HTTPS server stopped", "address", listenerAddr) + }() + } else { + go func() { + listenerAddr := listener.Addr().String() + logger.Info("RPC HTTPS server starting", "address", listenerAddr) + + err := rpcserver.Serve(listener, rootHandler, logger, config) + if !errors.Is(err, net.ErrClosed) { + logger.Error("RPC HTTP server stopped with error", "address", listener, "err", err) + return + } + logger.Info("RPC HTTP server stopped", "address", listenerAddr) + }() + } + } + + return listeners, nil +} + +func listenersFromRPCConfig(rpcConfig *cfg.RPCConfig) ([]net.Listener, error) { + listenAddrs := strings.SplitAndTrimEmpty(rpcConfig.ListenAddress, ",", " ") + listeners := make([]net.Listener, len(listenAddrs)) + for i, listenAddr := range listenAddrs { + listener, err := rpcserver.Listen(listenAddr, rpcConfig.MaxOpenConnections) + if err != nil { + closeOpenListeners(listeners) + return nil, err + } + listeners[i] = listener + } + return listeners, nil +} + +func registerWebsocketHandler(rpcConfig *cfg.RPCConfig, + mux *http.ServeMux, + routes rpccore.RoutesMap, + logger log.Logger, + eventBus types.EventBusSubscriber) { + wmLogger := logger.With("protocol", "websocket") + + websocketDisconnectFn := func(remoteAddr string) { + err := eventBus.UnsubscribeAll(context.Background(), remoteAddr) + if err != nil && err != tmpubsub.ErrSubscriptionNotFound { + wmLogger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err) + } + } + + wm := rpcserver.NewWebsocketManager(routes, + rpcserver.OnDisconnect(websocketDisconnectFn), + rpcserver.ReadLimit(rpcConfig.MaxBodyBytes)) + wm.SetLogger(wmLogger) + mux.HandleFunc("/websocket", wm.WebsocketHandler) +} + +func addCORSHandler(rpcConfig *cfg.RPCConfig, h http.Handler) http.Handler { + if rpcConfig.IsCorsEnabled() { + corsMiddleware := cors.New(cors.Options{ + AllowedOrigins: rpcConfig.CORSAllowedOrigins, + AllowedMethods: rpcConfig.CORSAllowedMethods, + AllowedHeaders: rpcConfig.CORSAllowedHeaders, + }) + h = corsMiddleware.Handler(h) + } + return h +} + +func closeOpenListeners(listeners []net.Listener) { + for _, listener := range listeners { + listener.Close() + } +} diff --git a/node/setup.go b/node/setup.go index af48fb382..0739282c6 100644 --- a/node/setup.go +++ b/node/setup.go @@ -41,15 +41,20 @@ import ( "github.com/tendermint/tendermint/version" ) +const ( + _blockStoreID = "blockstore" + _stateStoreID = "state" +) + func initDBs(config *cfg.Config, dbProvider cfg.DBProvider) (blockStore *store.BlockStore, stateDB dbm.DB, err error) { var blockStoreDB dbm.DB - blockStoreDB, err = dbProvider(&cfg.DBContext{ID: "blockstore", Config: config}) + blockStoreDB, err = dbProvider(&cfg.DBContext{ID: _blockStoreID, Config: config}) if err != nil { return } blockStore = store.NewBlockStore(blockStoreDB) - stateDB, err = dbProvider(&cfg.DBContext{ID: "state", Config: config}) + stateDB, err = dbProvider(&cfg.DBContext{ID: _stateStoreID, Config: config}) return } diff --git a/rpc/core/blocks_test.go b/rpc/core/blocks_test.go index 29db2f094..cfa970cda 100644 --- a/rpc/core/blocks_test.go +++ b/rpc/core/blocks_test.go @@ -133,3 +133,4 @@ func (mockBlockStore) LoadSeenCommit() *types.Commit { retur func (mockBlockStore) PruneBlocks(height int64) (uint64, error) { return 0, nil } func (mockBlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) { } +func (mockBlockStore) Close() error { return nil } diff --git a/rpc/core/routes.go b/rpc/core/routes.go index 1eb50fe4e..d150b949f 100644 --- a/rpc/core/routes.go +++ b/rpc/core/routes.go @@ -10,13 +10,16 @@ type RoutesMap map[string]*rpc.RPCFunc // Routes is a map of available routes. func (env *Environment) GetRoutes() RoutesMap { - return RoutesMap{ - // subscribe/unsubscribe are reserved for websocket events. - "subscribe": rpc.NewWSRPCFunc(env.Subscribe, "query"), - "unsubscribe": rpc.NewWSRPCFunc(env.Unsubscribe, "query"), - "unsubscribe_all": rpc.NewWSRPCFunc(env.UnsubscribeAll, ""), + return CombineRoutes(env.InfoRoutes(), + env.SubscribeRoutes(), + env.BroadcastTxRoutes(), + env.ABCIQueryRoutes(), + env.EvidenceRoutes(), + ) +} - // info API +func (env *Environment) InfoRoutes() RoutesMap { + return RoutesMap{ "health": rpc.NewRPCFunc(env.Health, "", false), "status": rpc.NewRPCFunc(env.Status, "", false), "net_info": rpc.NewRPCFunc(env.NetInfo, "", false), @@ -37,25 +40,58 @@ func (env *Environment) GetRoutes() RoutesMap { "consensus_params": rpc.NewRPCFunc(env.ConsensusParams, "height", true), "unconfirmed_txs": rpc.NewRPCFunc(env.UnconfirmedTxs, "limit", false), "num_unconfirmed_txs": rpc.NewRPCFunc(env.NumUnconfirmedTxs, "", false), + } +} - // tx broadcast API +func (env *Environment) SubscribeRoutes() RoutesMap { + return RoutesMap{ + // subscribe/unsubscribe are reserved for websocket events. + "subscribe": rpc.NewWSRPCFunc(env.Subscribe, "query"), + "unsubscribe": rpc.NewWSRPCFunc(env.Unsubscribe, "query"), + "unsubscribe_all": rpc.NewWSRPCFunc(env.UnsubscribeAll, ""), + } +} + +func (env *Environment) BroadcastTxRoutes() RoutesMap { + // tx broadcast API + return RoutesMap{ "broadcast_tx_commit": rpc.NewRPCFunc(env.BroadcastTxCommit, "tx", false), "broadcast_tx_sync": rpc.NewRPCFunc(env.BroadcastTxSync, "tx", false), "broadcast_tx_async": rpc.NewRPCFunc(env.BroadcastTxAsync, "tx", false), + } +} - // abci API - "abci_query": rpc.NewRPCFunc(env.ABCIQuery, "path,data,height,prove", false), - "abci_info": rpc.NewRPCFunc(env.ABCIInfo, "", true), - - // evidence API +func (env *Environment) EvidenceRoutes() RoutesMap { + // evidence API + return RoutesMap{ "broadcast_evidence": rpc.NewRPCFunc(env.BroadcastEvidence, "evidence", false), } } -// AddUnsafeRoutes adds unsafe routes. -func (env *Environment) AddUnsafe(routes RoutesMap) { - // control API - routes["dial_seeds"] = rpc.NewRPCFunc(env.UnsafeDialSeeds, "seeds", false) - routes["dial_peers"] = rpc.NewRPCFunc(env.UnsafeDialPeers, "peers,persistent,unconditional,private", false) - routes["unsafe_flush_mempool"] = rpc.NewRPCFunc(env.UnsafeFlushMempool, "", false) +func (env *Environment) ABCIQueryRoutes() RoutesMap { + // abci API + return RoutesMap{ + "abci_query": rpc.NewRPCFunc(env.ABCIQuery, "path,data,height,prove", false), + "abci_info": rpc.NewRPCFunc(env.ABCIInfo, "", true), + } +} + +// AddUnsafeRoutes adds unsafe routes. +func (env *Environment) UnsafeRoutes() RoutesMap { + // control API + return RoutesMap{ + "dial_seeds": rpc.NewRPCFunc(env.UnsafeDialSeeds, "seeds", false), + "dial_peers": rpc.NewRPCFunc(env.UnsafeDialPeers, "peers,persistent,unconditional,private", false), + "unsafe_flush_mempool": rpc.NewRPCFunc(env.UnsafeFlushMempool, "", false), + } +} + +func CombineRoutes(routesMaps ...RoutesMap) RoutesMap { + res := RoutesMap{} + for _, routesMap := range routesMaps { + for path, rpcFunc := range routesMap { + res[path] = rpcFunc + } + } + return res } diff --git a/rpc/jsonrpc/jsonrpc_test.go b/rpc/jsonrpc/jsonrpc_test.go index 6e0c03f00..b5e422280 100644 --- a/rpc/jsonrpc/jsonrpc_test.go +++ b/rpc/jsonrpc/jsonrpc_test.go @@ -110,7 +110,7 @@ func setup() { wm.SetLogger(tcpLogger) mux.HandleFunc(websocketEndpoint, wm.WebsocketHandler) config := server.DefaultConfig() - listener1, err := server.Listen(tcpAddr, config) + listener1, err := server.Listen(tcpAddr, config.MaxOpenConnections) if err != nil { panic(err) } @@ -126,7 +126,7 @@ func setup() { wm = server.NewWebsocketManager(Routes) wm.SetLogger(unixLogger) mux2.HandleFunc(websocketEndpoint, wm.WebsocketHandler) - listener2, err := server.Listen(unixAddr, config) + listener2, err := server.Listen(unixAddr, config.MaxOpenConnections) if err != nil { panic(err) } diff --git a/rpc/jsonrpc/server/http_server.go b/rpc/jsonrpc/server/http_server.go index c21c71c49..04ec23474 100644 --- a/rpc/jsonrpc/server/http_server.go +++ b/rpc/jsonrpc/server/http_server.go @@ -51,16 +51,13 @@ func DefaultConfig() *Config { // // NOTE: This function blocks - you may want to call it in a go-routine. func Serve(listener net.Listener, handler http.Handler, logger log.Logger, config *Config) error { - logger.Info(fmt.Sprintf("Starting RPC HTTP server on %s", listener.Addr())) s := &http.Server{ Handler: RecoverAndLogHandler(maxBytesHandler{h: handler, n: config.MaxBodyBytes}, logger), ReadTimeout: config.ReadTimeout, WriteTimeout: config.WriteTimeout, MaxHeaderBytes: config.MaxHeaderBytes, } - err := s.Serve(listener) - logger.Info("RPC HTTP server stopped", "err", err) - return err + return s.Serve(listener) } // Serve creates a http.Server and calls ServeTLS with the given listener, @@ -75,18 +72,13 @@ func ServeTLS( logger log.Logger, config *Config, ) error { - logger.Info(fmt.Sprintf("Starting RPC HTTPS server on %s (cert: %q, key: %q)", - listener.Addr(), certFile, keyFile)) s := &http.Server{ Handler: RecoverAndLogHandler(maxBytesHandler{h: handler, n: config.MaxBodyBytes}, logger), ReadTimeout: config.ReadTimeout, WriteTimeout: config.WriteTimeout, MaxHeaderBytes: config.MaxHeaderBytes, } - err := s.ServeTLS(listener, certFile, keyFile) - - logger.Error("RPC HTTPS server stopped", "err", err) - return err + return s.ServeTLS(listener, certFile, keyFile) } // WriteRPCResponseHTTPError marshals res as JSON (with indent) and writes it @@ -261,7 +253,7 @@ func (h maxBytesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Listen starts a new net.Listener on the given address. // It returns an error if the address is invalid or the call to Listen() fails. -func Listen(addr string, config *Config) (listener net.Listener, err error) { +func Listen(addr string, maxOpenConnections int) (listener net.Listener, err error) { parts := strings.SplitN(addr, "://", 2) if len(parts) != 2 { return nil, fmt.Errorf( @@ -274,8 +266,8 @@ func Listen(addr string, config *Config) (listener net.Listener, err error) { if err != nil { return nil, fmt.Errorf("failed to listen on %v: %v", addr, err) } - if config.MaxOpenConnections > 0 { - listener = netutil.LimitListener(listener, config.MaxOpenConnections) + if maxOpenConnections > 0 { + listener = netutil.LimitListener(listener, maxOpenConnections) } return listener, nil diff --git a/rpc/jsonrpc/server/http_server_test.go b/rpc/jsonrpc/server/http_server_test.go index e7c517cde..a8e46d279 100644 --- a/rpc/jsonrpc/server/http_server_test.go +++ b/rpc/jsonrpc/server/http_server_test.go @@ -40,7 +40,7 @@ func TestMaxOpenConnections(t *testing.T) { }) config := DefaultConfig() config.MaxOpenConnections = max - l, err := Listen("tcp://127.0.0.1:0", config) + l, err := Listen("tcp://127.0.0.1:0", config.MaxOpenConnections) require.NoError(t, err) defer l.Close() go Serve(l, mux, log.TestingLogger(), config) //nolint:errcheck // ignore for tests diff --git a/rpc/jsonrpc/test/main.go b/rpc/jsonrpc/test/main.go index 1c949571f..d348e1639 100644 --- a/rpc/jsonrpc/test/main.go +++ b/rpc/jsonrpc/test/main.go @@ -33,7 +33,7 @@ func main() { rpcserver.RegisterRPCFuncs(mux, routes, logger) config := rpcserver.DefaultConfig() - listener, err := rpcserver.Listen("tcp://127.0.0.1:8008", config) + listener, err := rpcserver.Listen("tcp://127.0.0.1:8008", config.MaxOpenConnections) if err != nil { tmos.Exit(err.Error()) } diff --git a/store/store.go b/store/store.go index 8848b76d9..04d21de92 100644 --- a/store/store.go +++ b/store/store.go @@ -283,6 +283,10 @@ func (bs *BlockStore) LoadSeenCommit() *types.Commit { return commit } +func (bs *BlockStore) Close() error { + return bs.db.Close() +} + // PruneBlocks removes block up to (but not including) a height. It returns the number of blocks pruned. func (bs *BlockStore) PruneBlocks(height int64) (uint64, error) { if height <= 0 { diff --git a/types/event_bus.go b/types/event_bus.go index dfe3a0664..75265a584 100644 --- a/types/event_bus.go +++ b/types/event_bus.go @@ -244,9 +244,9 @@ func (NopEventBus) Subscribe( ctx context.Context, subscriber string, query tmpubsub.Query, - out chan<- interface{}, -) error { - return nil + outCapacity ...int, +) (Subscription, error) { + return nil, nil } func (NopEventBus) Unsubscribe(ctx context.Context, args tmpubsub.UnsubscribeArgs) error { @@ -324,3 +324,11 @@ func (NopEventBus) PublishEventBlockSyncStatus(data EventDataBlockSyncStatus) er func (NopEventBus) PublishEventStateSyncStatus(data EventDataStateSyncStatus) error { return nil } + +func (NopEventBus) NumClientSubscriptions(clientID string) int { + return 0 +} + +func (NopEventBus) NumClients() int { + return 0 +}