node: sketch of debug node logic

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

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
 	}

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

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

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 {

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

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

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
 }
 

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 }

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
 }

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

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

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

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

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 {

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