Merge remote-tracking branch 'origin/wb/abci-metrics' into wb/abci-metrics

2026-09-21 07:24:36 +00:00 · 2021-10-13 15:27:30 -04:00
parent e852dcd554 0961b53f0a
commit 24513eec28
44 changed files with 1116 additions and 1098 deletions
@@ -23,6 +23,7 @@ import (
 	"github.com/tendermint/tendermint/internal/proxy"
 	rpccore "github.com/tendermint/tendermint/internal/rpc/core"
 	sm "github.com/tendermint/tendermint/internal/state"
+	"github.com/tendermint/tendermint/internal/state/indexer"
 	"github.com/tendermint/tendermint/internal/statesync"
 	"github.com/tendermint/tendermint/internal/store"
 	"github.com/tendermint/tendermint/libs/log"
@@ -62,6 +63,7 @@ type nodeImpl struct {

 	// services
 	eventBus         *types.EventBus // pub/sub for services
+	eventSinks       []indexer.EventSink
 	stateStore       sm.Store
 	blockStore       *store.BlockStore // store the blockchain to disk
 	bcReactor        service.Service   // for block-syncing
@@ -73,6 +75,7 @@ type nodeImpl struct {
 	pexReactor       service.Service    // for exchanging peer addresses
 	evidenceReactor  service.Service
 	rpcListeners     []net.Listener // rpc servers
+	shutdownOps      closer
 	indexerService   service.Service
 	rpcEnv           *rpccore.Environment
 	prometheusSrv    *http.Server
@@ -106,6 +109,7 @@ func newDefaultNode(cfg *config.Config, logger log.Logger) (service.Service, err
 	}

 	appClient, _ := proxy.DefaultClientCreator(cfg.ProxyApp, cfg.ABCI, cfg.DBDir())
+
 	return makeNode(cfg,
 		pval,
 		nodeKey,
@@ -123,27 +127,34 @@ func makeNode(cfg *config.Config,
 	clientCreator abciclient.Creator,
 	genesisDocProvider genesisDocProvider,
 	dbProvider config.DBProvider,
-	logger log.Logger) (service.Service, error) {
+	logger log.Logger,
+) (service.Service, error) {
+	closers := []closer{}

-	blockStore, stateDB, err := initDBs(cfg, dbProvider)
+	blockStore, stateDB, dbCloser, err := initDBs(cfg, dbProvider)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, dbCloser)
 	}
+	closers = append(closers, dbCloser)
+
 	stateStore := sm.NewStore(stateDB)

 	genDoc, err := genesisDocProvider()
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
 	}

 	err = genDoc.ValidateAndComplete()
 	if err != nil {
-		return nil, fmt.Errorf("error in genesis doc: %w", err)
+		return nil, combineCloseError(
+			fmt.Errorf("error in genesis doc: %w", err),
+			makeCloser(closers))
 	}

 	state, err := loadStateFromDBOrGenesisDocProvider(stateStore, genDoc)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	nodeMetrics := defaultMetricsProvider(cfg.Instrumentation)(genDoc.ChainID)
@@ -151,7 +162,8 @@ func makeNode(cfg *config.Config,
 	// Create the proxyApp and establish connections to the ABCI app (consensus, mempool, query).
 	proxyApp, err := createAndStartProxyAppConns(clientCreator, logger, nodeMetrics.proxy)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	// EventBus and IndexerService must be started before the handshake because
@@ -160,12 +172,13 @@ func makeNode(cfg *config.Config,
 	// but before it indexed the txs, or, endblocker panicked)
 	eventBus, err := createAndStartEventBus(logger)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	indexerService, eventSinks, err := createAndStartIndexerService(cfg, dbProvider, eventBus, logger, genDoc.ChainID)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
 	}

 	// If an address is provided, listen on the socket for a connection from an
@@ -177,12 +190,16 @@ func makeNode(cfg *config.Config,
 		case "grpc":
 			privValidator, err = createAndStartPrivValidatorGRPCClient(cfg, genDoc.ChainID, logger)
 			if err != nil {
-				return nil, fmt.Errorf("error with private validator grpc client: %w", err)
+				return nil, combineCloseError(
+					fmt.Errorf("error with private validator grpc client: %w", err),
+					makeCloser(closers))
 			}
 		default:
 			privValidator, err = createAndStartPrivValidatorSocketClient(cfg.PrivValidator.ListenAddr, genDoc.ChainID, logger)
 			if err != nil {
-				return nil, fmt.Errorf("error with private validator socket client: %w", err)
+				return nil, combineCloseError(
+					fmt.Errorf("error with private validator socket client: %w", err),
+					makeCloser(closers))
 			}
 		}
 	}
@@ -190,10 +207,14 @@ func makeNode(cfg *config.Config,
 	if cfg.Mode == config.ModeValidator {
 		pubKey, err = privValidator.GetPubKey(context.TODO())
 		if err != nil {
-			return nil, fmt.Errorf("can't get pubkey: %w", err)
+			return nil, combineCloseError(fmt.Errorf("can't get pubkey: %w", err),
+				makeCloser(closers))
+
 		}
 		if pubKey == nil {
-			return nil, errors.New("could not retrieve public key from private validator")
+			return nil, combineCloseError(
+				errors.New("could not retrieve public key from private validator"),
+				makeCloser(closers))
 		}
 	}

@@ -209,7 +230,8 @@ func makeNode(cfg *config.Config,
 	consensusLogger := logger.With("module", "consensus")
 	if !stateSync {
 		if err := doHandshake(stateStore, state, blockStore, genDoc, eventBus, proxyApp, consensusLogger); err != nil {
-			return nil, err
+			return nil, combineCloseError(err, makeCloser(closers))
+
 		}

 		// Reload the state. It will have the Version.Consensus.App set by the
@@ -217,7 +239,9 @@ func makeNode(cfg *config.Config,
 		// what happened during block replay).
 		state, err = stateStore.Load()
 		if err != nil {
-			return nil, fmt.Errorf("cannot load state: %w", err)
+			return nil, combineCloseError(
+				fmt.Errorf("cannot load state: %w", err),
+				makeCloser(closers))
 		}
 	}

@@ -231,35 +255,43 @@ func makeNode(cfg *config.Config,
 	// TODO: Use a persistent peer database.
 	nodeInfo, err := makeNodeInfo(cfg, nodeKey, eventSinks, genDoc, state)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	p2pLogger := logger.With("module", "p2p")
 	transport := createTransport(p2pLogger, cfg)

-	peerManager, err := createPeerManager(cfg, dbProvider, nodeKey.ID)
+	peerManager, peerCloser, err := createPeerManager(cfg, dbProvider, nodeKey.ID)
+	closers = append(closers, peerCloser)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create peer manager: %w", err)
+		return nil, combineCloseError(
+			fmt.Errorf("failed to create peer manager: %w", err),
+			makeCloser(closers))
 	}
-
+  
 	router, err := createRouter(p2pLogger, nodeMetrics.p2p, nodeInfo, nodeKey.PrivKey,
 		peerManager, transport, getRouterConfig(cfg, proxyApp))
 	if err != nil {
-		return nil, fmt.Errorf("failed to create router: %w", err)
+		return nil, combineCloseError(
+			fmt.Errorf("failed to create router: %w", err),
+			makeCloser(closers))
 	}

 	mpReactorShim, mpReactor, mp, err := createMempoolReactor(
 		cfg, proxyApp, state, nodeMetrics.mempool, peerManager, router, logger,
 	)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	evReactorShim, evReactor, evPool, err := createEvidenceReactor(
 		cfg, dbProvider, stateDB, blockStore, peerManager, router, logger,
 	)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	// make block executor for consensus and blockchain reactors to execute blocks
@@ -286,7 +318,9 @@ func makeNode(cfg *config.Config,
 		peerManager, router, blockSync && !stateSync, nodeMetrics.consensus,
 	)
 	if err != nil {
-		return nil, fmt.Errorf("could not create blockchain reactor: %w", err)
+		return nil, combineCloseError(
+			fmt.Errorf("could not create blockchain reactor: %w", err),
+			makeCloser(closers))
 	}

 	// Make ConsensusReactor. Don't enable fully if doing a state sync and/or block sync first.
@@ -304,6 +338,7 @@ func makeNode(cfg *config.Config,
 	ssLogger := logger.With("module", "statesync")
 	ssReactorShim := p2p.NewReactorShim(ssLogger, "StateSyncShim", statesync.ChannelShims)
 	channels := makeChannelsFromShims(router, statesync.ChannelShims)
+
 	peerUpdates := peerManager.Subscribe()
 	stateSyncReactor := statesync.NewReactor(
 		genDoc.ChainID,
@@ -353,7 +388,8 @@ func makeNode(cfg *config.Config,

 	pexReactor, err = createPEXReactor(logger, peerManager, router)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, makeCloser(closers))
+
 	}

 	if cfg.RPC.PprofListenAddress != "" {
@@ -386,6 +422,9 @@ func makeNode(cfg *config.Config,
 		evidenceReactor:  evReactor,
 		indexerService:   indexerService,
 		eventBus:         eventBus,
+		eventSinks:       eventSinks,
+
+		shutdownOps: makeCloser(closers),

 		rpcEnv: &rpccore.Environment{
 			ProxyAppQuery:   proxyApp.Query(),
@@ -433,6 +472,7 @@ func makeSeedNode(cfg *config.Config,
 	state, err := sm.MakeGenesisState(genDoc)
 	if err != nil {
 		return nil, err
+
 	}

 	nodeInfo, err := makeSeedNodeInfo(cfg, nodeKey, genDoc, state)
@@ -445,15 +485,19 @@ func makeSeedNode(cfg *config.Config,
 	p2pLogger := logger.With("module", "p2p")
 	transport := createTransport(p2pLogger, cfg)

-	peerManager, err := createPeerManager(cfg, dbProvider, nodeKey.ID)
+	peerManager, closer, err := createPeerManager(cfg, dbProvider, nodeKey.ID)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create peer manager: %w", err)
+		return nil, combineCloseError(
+			fmt.Errorf("failed to create peer manager: %w", err),
+			closer)
 	}

 	router, err := createRouter(p2pLogger, p2pMetrics, nodeInfo, nodeKey.PrivKey,
 		peerManager, transport, getRouterConfig(cfg, nil))
 	if err != nil {
-		return nil, fmt.Errorf("failed to create router: %w", err)
+		return nil, combineCloseError(
+			fmt.Errorf("failed to create router: %w", err),
+			closer)
 	}

 	var pexReactor service.Service
@@ -467,7 +511,8 @@ func makeSeedNode(cfg *config.Config,

 	pexReactor, err = createPEXReactor(logger, peerManager, router)
 	if err != nil {
-		return nil, err
+		return nil, combineCloseError(err, closer)
+
 	}

 	if cfg.RPC.PprofListenAddress != "" {
@@ -487,6 +532,8 @@ func makeSeedNode(cfg *config.Config,
 		peerManager: peerManager,
 		router:      router,

+		shutdownOps: closer,
+
 		pexReactor: pexReactor,
 	}
 	node.BaseService = *service.NewBaseService(logger, "SeedNode", node)
@@ -640,12 +687,22 @@ func (n *nodeImpl) OnStop() {

 	n.Logger.Info("Stopping Node")

-	// first stop the non-reactor services
-	if err := n.eventBus.Stop(); err != nil {
-		n.Logger.Error("Error closing eventBus", "err", err)
+	if n.eventBus != nil {
+		// first stop the non-reactor services
+		if err := n.eventBus.Stop(); err != nil {
+			n.Logger.Error("Error closing eventBus", "err", err)
+		}
 	}
-	if err := n.indexerService.Stop(); err != nil {
-		n.Logger.Error("Error closing indexerService", "err", err)
+	if n.indexerService != nil {
+		if err := n.indexerService.Stop(); err != nil {
+			n.Logger.Error("Error closing indexerService", "err", err)
+		}
+	}
+
+	for _, es := range n.eventSinks {
+		if err := es.Stop(); err != nil {
+			n.Logger.Error("failed to stop event sink", "err", err)
+		}
 	}

 	if n.config.Mode != config.ModeSeed {
@@ -711,6 +768,10 @@ func (n *nodeImpl) OnStop() {
 			// Error from closing listeners, or context timeout:
 			n.Logger.Error("Prometheus HTTP server Shutdown", "err", err)
 		}
+
+	}
+	if err := n.shutdownOps(); err != nil {
+		n.Logger.Error("problem shutting down additional services", "err", err)
 	}
 }

@@ -31,6 +31,7 @@ import (
 	"github.com/tendermint/tendermint/internal/test/factory"
 	"github.com/tendermint/tendermint/libs/log"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
+	"github.com/tendermint/tendermint/libs/service"
 	tmtime "github.com/tendermint/tendermint/libs/time"
 	"github.com/tendermint/tendermint/privval"
 	"github.com/tendermint/tendermint/types"
@@ -46,11 +47,21 @@ func TestNodeStartStop(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, ns.Start())

+	t.Cleanup(func() {
+		if ns.IsRunning() {
+			assert.NoError(t, ns.Stop())
+			ns.Wait()
+		}
+	})
+
 	n, ok := ns.(*nodeImpl)
 	require.True(t, ok)

+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	// wait for the node to produce a block
-	blocksSub, err := n.EventBus().Subscribe(context.Background(), "node_test", types.EventQueryNewBlock)
+	blocksSub, err := n.EventBus().Subscribe(ctx, "node_test", types.EventQueryNewBlock)
 	require.NoError(t, err)
 	select {
 	case <-blocksSub.Out():
@@ -87,6 +98,14 @@ func getTestNode(t *testing.T, conf *config.Config, logger log.Logger) *nodeImpl

 	n, ok := ns.(*nodeImpl)
 	require.True(t, ok)
+
+	t.Cleanup(func() {
+		if ns.IsRunning() {
+			assert.NoError(t, ns.Stop())
+			ns.Wait()
+		}
+	})
+
 	return n
 }

@@ -100,7 +119,6 @@ func TestNodeDelayedStart(t *testing.T) {
 	n.GenesisDoc().GenesisTime = now.Add(2 * time.Second)

 	require.NoError(t, n.Start())
-	defer n.Stop() //nolint:errcheck // ignore for tests

 	startTime := tmtime.Now()
 	assert.Equal(t, true, startTime.After(n.GenesisDoc().GenesisTime))
@@ -165,8 +183,13 @@ func TestPrivValidatorListenAddrNoProtocol(t *testing.T) {
 	defer os.RemoveAll(cfg.RootDir)
 	cfg.PrivValidator.ListenAddr = addrNoPrefix

-	_, err := newDefaultNode(cfg, log.TestingLogger())
+	n, err := newDefaultNode(cfg, log.TestingLogger())
 	assert.Error(t, err)
+
+	if n != nil && n.IsRunning() {
+		assert.NoError(t, n.Stop())
+		n.Wait()
+	}
 }

 func TestNodeSetPrivValIPC(t *testing.T) {
@@ -211,6 +234,9 @@ func testFreeAddr(t *testing.T) string {
 // create a proposal block using real and full
 // mempool and evidence pool and validate it.
 func TestCreateProposalBlock(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	cfg := config.ResetTestRoot("node_create_proposal")
 	defer os.RemoveAll(cfg.RootDir)
 	cc := abciclient.NewLocalCreator(kvstore.NewApplication())
@@ -222,7 +248,7 @@ func TestCreateProposalBlock(t *testing.T) {
 	logger := log.TestingLogger()

 	const height int64 = 1
-	state, stateDB, privVals := state(1, height)
+	state, stateDB, privVals := state(t, 1, height)
 	stateStore := sm.NewStore(stateDB)
 	maxBytes := 16384
 	const partSize uint32 = 256
@@ -249,7 +275,7 @@ func TestCreateProposalBlock(t *testing.T) {

 	// fill the evidence pool with more evidence
 	// than can fit in a block
-	var currentBytes int64 = 0
+	var currentBytes int64
 	for currentBytes <= maxEvidenceBytes {
 		ev := types.NewMockDuplicateVoteEvidenceWithValidator(height, time.Now(), privVals[0], "test-chain")
 		currentBytes += int64(len(ev.Bytes()))
@@ -266,7 +292,7 @@ func TestCreateProposalBlock(t *testing.T) {
 	txLength := 100
 	for i := 0; i <= maxBytes/txLength; i++ {
 		tx := tmrand.Bytes(txLength)
-		err := mp.CheckTx(context.Background(), tx, nil, mempool.TxInfo{})
+		err := mp.CheckTx(ctx, tx, nil, mempool.TxInfo{})
 		assert.NoError(t, err)
 	}

@@ -303,6 +329,9 @@ func TestCreateProposalBlock(t *testing.T) {
 }

 func TestMaxTxsProposalBlockSize(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	cfg := config.ResetTestRoot("node_create_proposal")
 	defer os.RemoveAll(cfg.RootDir)
 	cc := abciclient.NewLocalCreator(kvstore.NewApplication())
@@ -314,7 +343,7 @@ func TestMaxTxsProposalBlockSize(t *testing.T) {
 	logger := log.TestingLogger()

 	const height int64 = 1
-	state, stateDB, _ := state(1, height)
+	state, stateDB, _ := state(t, 1, height)
 	stateStore := sm.NewStore(stateDB)
 	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	const maxBytes int64 = 16384
@@ -336,7 +365,7 @@ func TestMaxTxsProposalBlockSize(t *testing.T) {
 	// fill the mempool with one txs just below the maximum size
 	txLength := int(types.MaxDataBytesNoEvidence(maxBytes, 1))
 	tx := tmrand.Bytes(txLength - 4) // to account for the varint
-	err = mp.CheckTx(context.Background(), tx, nil, mempool.TxInfo{})
+	err = mp.CheckTx(ctx, tx, nil, mempool.TxInfo{})
 	assert.NoError(t, err)

 	blockExec := sm.NewBlockExecutor(
@@ -365,6 +394,9 @@ func TestMaxTxsProposalBlockSize(t *testing.T) {
 }

 func TestMaxProposalBlockSize(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	cfg := config.ResetTestRoot("node_create_proposal")
 	defer os.RemoveAll(cfg.RootDir)
 	cc := abciclient.NewLocalCreator(kvstore.NewApplication())
@@ -375,7 +407,7 @@ func TestMaxProposalBlockSize(t *testing.T) {

 	logger := log.TestingLogger()

-	state, stateDB, _ := state(types.MaxVotesCount, int64(1))
+	state, stateDB, _ := state(t, types.MaxVotesCount, int64(1))
 	stateStore := sm.NewStore(stateDB)
 	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	const maxBytes int64 = 1024 * 1024 * 2
@@ -402,7 +434,7 @@ func TestMaxProposalBlockSize(t *testing.T) {
 	// At the end of the test, only the single big tx should be added
 	for i := 0; i < 10; i++ {
 		tx := tmrand.Bytes(10)
-		err = mp.CheckTx(context.Background(), tx, nil, mempool.TxInfo{})
+		err = mp.CheckTx(ctx, tx, nil, mempool.TxInfo{})
 		assert.NoError(t, err)
 	}

@@ -493,14 +525,17 @@ func TestNodeNewSeedNode(t *testing.T) {
 		defaultGenesisDocProviderFunc(cfg),
 		log.TestingLogger(),
 	)
+
 	require.NoError(t, err)
 	n, ok := ns.(*nodeImpl)
 	require.True(t, ok)

 	err = n.Start()
 	require.NoError(t, err)
-
 	assert.True(t, n.pexReactor.IsRunning())
+
+	require.NoError(t, n.Stop())
+
 }

 func TestNodeSetEventSink(t *testing.T) {
@@ -511,7 +546,7 @@ func TestNodeSetEventSink(t *testing.T) {
 	setupTest := func(t *testing.T, conf *config.Config) []indexer.EventSink {
 		eventBus, err := createAndStartEventBus(logger)
 		require.NoError(t, err)
-
+		t.Cleanup(func() { require.NoError(t, eventBus.Stop()) })
 		genDoc, err := types.GenesisDocFromFile(cfg.GenesisFile())
 		require.NoError(t, err)

@@ -521,6 +556,22 @@ func TestNodeSetEventSink(t *testing.T) {
 		t.Cleanup(func() { require.NoError(t, indexService.Stop()) })
 		return eventSinks
 	}
+	cleanup := func(ns service.Service) func() {
+		return func() {
+			n, ok := ns.(*nodeImpl)
+			if !ok {
+				return
+			}
+			if n == nil {
+				return
+			}
+			if !n.IsRunning() {
+				return
+			}
+			assert.NoError(t, n.Stop())
+			n.Wait()
+		}
+	}

 	eventSinks := setupTest(t, cfg)
 	assert.Equal(t, 1, len(eventSinks))
@@ -541,7 +592,8 @@ func TestNodeSetEventSink(t *testing.T) {
 	cfg.TxIndex.Indexer = []string{"kvv"}
 	ns, err := newDefaultNode(cfg, logger)
 	assert.Nil(t, ns)
-	assert.Equal(t, errors.New("unsupported event sink type"), err)
+	assert.Contains(t, err.Error(), "unsupported event sink type")
+	t.Cleanup(cleanup(ns))

 	cfg.TxIndex.Indexer = []string{}
 	eventSinks = setupTest(t, cfg)
@@ -552,7 +604,8 @@ func TestNodeSetEventSink(t *testing.T) {
 	cfg.TxIndex.Indexer = []string{"psql"}
 	ns, err = newDefaultNode(cfg, logger)
 	assert.Nil(t, ns)
-	assert.Equal(t, errors.New("the psql connection settings cannot be empty"), err)
+	assert.Contains(t, err.Error(), "the psql connection settings cannot be empty")
+	t.Cleanup(cleanup(ns))

 	var psqlConn = "test"

@@ -591,18 +644,21 @@ func TestNodeSetEventSink(t *testing.T) {
 	var e = errors.New("found duplicated sinks, please check the tx-index section in the config.toml")
 	cfg.TxIndex.Indexer = []string{"psql", "kv", "Kv"}
 	cfg.TxIndex.PsqlConn = psqlConn
-	_, err = newDefaultNode(cfg, logger)
+	ns, err = newDefaultNode(cfg, logger)
 	require.Error(t, err)
-	assert.Equal(t, e, err)
+	assert.Contains(t, err.Error(), e.Error())
+	t.Cleanup(cleanup(ns))

 	cfg.TxIndex.Indexer = []string{"Psql", "kV", "kv", "pSql"}
 	cfg.TxIndex.PsqlConn = psqlConn
-	_, err = newDefaultNode(cfg, logger)
+	ns, err = newDefaultNode(cfg, logger)
 	require.Error(t, err)
-	assert.Equal(t, e, err)
+	assert.Contains(t, err.Error(), e.Error())
+	t.Cleanup(cleanup(ns))
 }

-func state(nVals int, height int64) (sm.State, dbm.DB, []types.PrivValidator) {
+func state(t *testing.T, nVals int, height int64) (sm.State, dbm.DB, []types.PrivValidator) {
+	t.Helper()
 	privVals := make([]types.PrivValidator, nVals)
 	vals := make([]types.GenesisValidator, nVals)
 	for i := 0; i < nVals; i++ {
@@ -623,17 +679,15 @@ func state(nVals int, height int64) (sm.State, dbm.DB, []types.PrivValidator) {

 	// save validators to db for 2 heights
 	stateDB := dbm.NewMemDB()
+	t.Cleanup(func() { require.NoError(t, stateDB.Close()) })
+
 	stateStore := sm.NewStore(stateDB)
-	if err := stateStore.Save(s); err != nil {
-		panic(err)
-	}
+	require.NoError(t, stateStore.Save(s))

 	for i := 1; i < int(height); i++ {
 		s.LastBlockHeight++
 		s.LastValidators = s.Validators.Copy()
-		if err := stateStore.Save(s); err != nil {
-			panic(err)
-		}
+		require.NoError(t, stateStore.Save(s))
 	}
 	return s, stateDB, privVals
 }
@@ -2,7 +2,9 @@ package node

 import (
 	"bytes"
+	"errors"
 	"fmt"
+	"strings"
 	"time"

 	dbm "github.com/tendermint/tm-db"
@@ -35,16 +37,57 @@ import (
 	_ "net/http/pprof" // nolint: gosec // securely exposed on separate, optional port
 )

-func initDBs(cfg *config.Config, dbProvider config.DBProvider) (blockStore *store.BlockStore, stateDB dbm.DB, err error) { //nolint:lll
-	var blockStoreDB dbm.DB
-	blockStoreDB, err = dbProvider(&config.DBContext{ID: "blockstore", Config: cfg})
-	if err != nil {
-		return
-	}
-	blockStore = store.NewBlockStore(blockStoreDB)
+type closer func() error

-	stateDB, err = dbProvider(&config.DBContext{ID: "state", Config: cfg})
-	return
+func makeCloser(cs []closer) closer {
+	return func() error {
+		errs := make([]string, 0, len(cs))
+		for _, cl := range cs {
+			if err := cl(); err != nil {
+				errs = append(errs, err.Error())
+			}
+		}
+		if len(errs) >= 0 {
+			return errors.New(strings.Join(errs, "; "))
+		}
+		return nil
+	}
+}
+
+func combineCloseError(err error, cl closer) error {
+	if err == nil {
+		return cl()
+	}
+
+	clerr := cl()
+	if clerr == nil {
+		return err
+	}
+
+	return fmt.Errorf("error=%q closerError=%q", err.Error(), clerr.Error())
+}
+
+func initDBs(
+	cfg *config.Config,
+	dbProvider config.DBProvider,
+) (*store.BlockStore, dbm.DB, closer, error) {
+
+	blockStoreDB, err := dbProvider(&config.DBContext{ID: "blockstore", Config: cfg})
+	if err != nil {
+		return nil, nil, func() error { return nil }, err
+	}
+	closers := []closer{}
+	blockStore := store.NewBlockStore(blockStoreDB)
+	closers = append(closers, blockStoreDB.Close)
+
+	stateDB, err := dbProvider(&config.DBContext{ID: "state", Config: cfg})
+	if err != nil {
+		return nil, nil, makeCloser(closers), err
+	}
+
+	closers = append(closers, stateDB.Close)
+
+	return blockStore, stateDB, makeCloser(closers), nil
 }

 // nolint:lll
@@ -355,7 +398,7 @@ func createPeerManager(
 	cfg *config.Config,
 	dbProvider config.DBProvider,
 	nodeID types.NodeID,
-) (*p2p.PeerManager, error) {
+) (*p2p.PeerManager, closer, error) {

 	var maxConns uint16

@@ -386,7 +429,7 @@ func createPeerManager(
 	for _, p := range tmstrings.SplitAndTrimEmpty(cfg.P2P.PersistentPeers, ",", " ") {
 		address, err := p2p.ParseNodeAddress(p)
 		if err != nil {
-			return nil, fmt.Errorf("invalid peer address %q: %w", p, err)
+			return nil, func() error { return nil }, fmt.Errorf("invalid peer address %q: %w", p, err)
 		}

 		peers = append(peers, address)
@@ -396,28 +439,28 @@ func createPeerManager(
 	for _, p := range tmstrings.SplitAndTrimEmpty(cfg.P2P.BootstrapPeers, ",", " ") {
 		address, err := p2p.ParseNodeAddress(p)
 		if err != nil {
-			return nil, fmt.Errorf("invalid peer address %q: %w", p, err)
+			return nil, func() error { return nil }, fmt.Errorf("invalid peer address %q: %w", p, err)
 		}
 		peers = append(peers, address)
 	}

 	peerDB, err := dbProvider(&config.DBContext{ID: "peerstore", Config: cfg})
 	if err != nil {
-		return nil, err
+		return nil, func() error { return nil }, err
 	}

 	peerManager, err := p2p.NewPeerManager(nodeID, peerDB, options)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create peer manager: %w", err)
+		return nil, peerDB.Close, fmt.Errorf("failed to create peer manager: %w", err)
 	}

 	for _, peer := range peers {
 		if _, err := peerManager.Add(peer); err != nil {
-			return nil, fmt.Errorf("failed to add peer %q: %w", peer, err)
+			return nil, peerDB.Close, fmt.Errorf("failed to add peer %q: %w", peer, err)
 		}
 	}

-	return peerManager, nil
+	return peerManager, peerDB.Close, nil
 }

 func createRouter(