state: move package to internal (#6964)

internal/rpc/core/CONTRIBUTING.md

@@ -0,0 +1,4 @@
+# Openapi docs
+
+Do not forget to update ../openapi/openapi.yaml if making changes to any
+endpoint.

internal/rpc/core/README.md

@@ -0,0 +1,18 @@
+# Tendermint RPC
+
+## Pagination
+
+Requests that return multiple items will be paginated to 30 items by default.
+You can specify further pages with the ?page parameter. You can also set a
+custom page size up to 100 with the ?per_page parameter.
+
+## Subscribing to events
+
+The user can subscribe to events emitted by Tendermint, using `/subscribe`. If
+the maximum number of clients is reached or the client has too many
+subscriptions, an error will be returned. The subscription timeout is 5 sec.
+Each subscription has a buffer to accommodate short bursts of events or some
+slowness in clients. If the buffer gets full, the subscription will be canceled
+("client is not pulling messages fast enough"). If Tendermint exits, all
+subscriptions are canceled ("Tendermint exited"). The user can unsubscribe
+using either `/unsubscribe` or `/unsubscribe_all`.

internal/rpc/core/abci.go

@@ -0,0 +1,42 @@
+package core
+
+import (
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/internal/proxy"
+	"github.com/tendermint/tendermint/libs/bytes"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+)
+
+// ABCIQuery queries the application for some information.
+// More: https://docs.tendermint.com/master/rpc/#/ABCI/abci_query
+func (env *Environment) ABCIQuery(
+	ctx *rpctypes.Context,
+	path string,
+	data bytes.HexBytes,
+	height int64,
+	prove bool,
+) (*ctypes.ResultABCIQuery, error) {
+	resQuery, err := env.ProxyAppQuery.QuerySync(ctx.Context(), abci.RequestQuery{
+		Path:   path,
+		Data:   data,
+		Height: height,
+		Prove:  prove,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &ctypes.ResultABCIQuery{Response: *resQuery}, nil
+}
+
+// ABCIInfo gets some info about the application.
+// More: https://docs.tendermint.com/master/rpc/#/ABCI/abci_info
+func (env *Environment) ABCIInfo(ctx *rpctypes.Context) (*ctypes.ResultABCIInfo, error) {
+	resInfo, err := env.ProxyAppQuery.InfoSync(ctx.Context(), proxy.RequestInfo)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ctypes.ResultABCIInfo{Response: *resInfo}, nil
+}

internal/rpc/core/blocks.go

@@ -0,0 +1,262 @@
+package core
+
+import (
+	"fmt"
+	"sort"
+
+	"github.com/tendermint/tendermint/internal/state/indexer"
+	"github.com/tendermint/tendermint/libs/bytes"
+	tmmath "github.com/tendermint/tendermint/libs/math"
+	tmquery "github.com/tendermint/tendermint/libs/pubsub/query"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+// BlockchainInfo gets block headers for minHeight <= height <= maxHeight.
+//
+// If maxHeight does not yet exist, blocks up to the current height will be
+// returned. If minHeight does not exist (due to pruning), earliest existing
+// height will be used.
+//
+// At most 20 items will be returned. Block headers are returned in descending
+// order (highest first).
+//
+// More: https://docs.tendermint.com/master/rpc/#/Info/blockchain
+func (env *Environment) BlockchainInfo(
+	ctx *rpctypes.Context,
+	minHeight, maxHeight int64) (*ctypes.ResultBlockchainInfo, error) {
+
+	const limit int64 = 20
+
+	var err error
+	minHeight, maxHeight, err = filterMinMax(
+		env.BlockStore.Base(),
+		env.BlockStore.Height(),
+		minHeight,
+		maxHeight,
+		limit)
+	if err != nil {
+		return nil, err
+	}
+	env.Logger.Debug("BlockchainInfo", "maxHeight", maxHeight, "minHeight", minHeight)
+
+	blockMetas := make([]*types.BlockMeta, 0, maxHeight-minHeight+1)
+	for height := maxHeight; height >= minHeight; height-- {
+		blockMeta := env.BlockStore.LoadBlockMeta(height)
+		if blockMeta != nil {
+			blockMetas = append(blockMetas, blockMeta)
+		}
+	}
+
+	return &ctypes.ResultBlockchainInfo{
+		LastHeight: env.BlockStore.Height(),
+		BlockMetas: blockMetas}, nil
+}
+
+// error if either min or max are negative or min > max
+// if 0, use blockstore base for min, latest block height for max
+// enforce limit.
+func filterMinMax(base, height, min, max, limit int64) (int64, int64, error) {
+	// filter negatives
+	if min < 0 || max < 0 {
+		return min, max, ctypes.ErrZeroOrNegativeHeight
+	}
+
+	// adjust for default values
+	if min == 0 {
+		min = 1
+	}
+	if max == 0 {
+		max = height
+	}
+
+	// limit max to the height
+	max = tmmath.MinInt64(height, max)
+
+	// limit min to the base
+	min = tmmath.MaxInt64(base, min)
+
+	// limit min to within `limit` of max
+	// so the total number of blocks returned will be `limit`
+	min = tmmath.MaxInt64(min, max-limit+1)
+
+	if min > max {
+		return min, max, fmt.Errorf("%w: min height %d can't be greater than max height %d",
+			ctypes.ErrInvalidRequest, min, max)
+	}
+	return min, max, nil
+}
+
+// Block gets block at a given height.
+// If no height is provided, it will fetch the latest block.
+// More: https://docs.tendermint.com/master/rpc/#/Info/block
+func (env *Environment) Block(ctx *rpctypes.Context, heightPtr *int64) (*ctypes.ResultBlock, error) {
+	height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
+	if err != nil {
+		return nil, err
+	}
+
+	blockMeta := env.BlockStore.LoadBlockMeta(height)
+	if blockMeta == nil {
+		return &ctypes.ResultBlock{BlockID: types.BlockID{}, Block: nil}, nil
+	}
+
+	block := env.BlockStore.LoadBlock(height)
+	return &ctypes.ResultBlock{BlockID: blockMeta.BlockID, Block: block}, nil
+}
+
+// BlockByHash gets block by hash.
+// More: https://docs.tendermint.com/master/rpc/#/Info/block_by_hash
+func (env *Environment) BlockByHash(ctx *rpctypes.Context, hash bytes.HexBytes) (*ctypes.ResultBlock, error) {
+	// N.B. The hash parameter is HexBytes so that the reflective parameter
+	// decoding logic in the HTTP service will correctly translate from JSON.
+	// See https://github.com/tendermint/tendermint/issues/6802 for context.
+
+	block := env.BlockStore.LoadBlockByHash(hash)
+	if block == nil {
+		return &ctypes.ResultBlock{BlockID: types.BlockID{}, Block: nil}, nil
+	}
+	// If block is not nil, then blockMeta can't be nil.
+	blockMeta := env.BlockStore.LoadBlockMeta(block.Height)
+	return &ctypes.ResultBlock{BlockID: blockMeta.BlockID, Block: block}, nil
+}
+
+// Commit gets block commit at a given height.
+// If no height is provided, it will fetch the commit for the latest block.
+// More: https://docs.tendermint.com/master/rpc/#/Info/commit
+func (env *Environment) Commit(ctx *rpctypes.Context, heightPtr *int64) (*ctypes.ResultCommit, error) {
+	height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
+	if err != nil {
+		return nil, err
+	}
+
+	blockMeta := env.BlockStore.LoadBlockMeta(height)
+	if blockMeta == nil {
+		return nil, nil
+	}
+	header := blockMeta.Header
+
+	// If the next block has not been committed yet,
+	// use a non-canonical commit
+	if height == env.BlockStore.Height() {
+		commit := env.BlockStore.LoadSeenCommit()
+		// NOTE: we can't yet ensure atomicity of operations in asserting
+		// whether this is the latest height and retrieving the seen commit
+		if commit != nil && commit.Height == height {
+			return ctypes.NewResultCommit(&header, commit, false), nil
+		}
+	}
+
+	// Return the canonical commit (comes from the block at height+1)
+	commit := env.BlockStore.LoadBlockCommit(height)
+	if commit == nil {
+		return nil, nil
+	}
+	return ctypes.NewResultCommit(&header, commit, true), nil
+}
+
+// BlockResults gets ABCIResults at a given height.
+// If no height is provided, it will fetch results for the latest block.
+//
+// Results are for the height of the block containing the txs.
+// Thus response.results.deliver_tx[5] is the results of executing
+// getBlock(h).Txs[5]
+// More: https://docs.tendermint.com/master/rpc/#/Info/block_results
+func (env *Environment) BlockResults(ctx *rpctypes.Context, heightPtr *int64) (*ctypes.ResultBlockResults, error) {
+	height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
+	if err != nil {
+		return nil, err
+	}
+
+	results, err := env.StateStore.LoadABCIResponses(height)
+	if err != nil {
+		return nil, err
+	}
+
+	var totalGasUsed int64
+	for _, tx := range results.GetDeliverTxs() {
+		totalGasUsed += tx.GetGasUsed()
+	}
+
+	return &ctypes.ResultBlockResults{
+		Height:                height,
+		TxsResults:            results.DeliverTxs,
+		TotalGasUsed:          totalGasUsed,
+		BeginBlockEvents:      results.BeginBlock.Events,
+		EndBlockEvents:        results.EndBlock.Events,
+		ValidatorUpdates:      results.EndBlock.ValidatorUpdates,
+		ConsensusParamUpdates: results.EndBlock.ConsensusParamUpdates,
+	}, nil
+}
+
+// BlockSearch searches for a paginated set of blocks matching BeginBlock and
+// EndBlock event search criteria.
+func (env *Environment) BlockSearch(
+	ctx *rpctypes.Context,
+	query string,
+	pagePtr, perPagePtr *int,
+	orderBy string,
+) (*ctypes.ResultBlockSearch, error) {
+
+	if !indexer.KVSinkEnabled(env.EventSinks) {
+		return nil, fmt.Errorf("block searching is disabled due to no kvEventSink")
+	}
+
+	q, err := tmquery.New(query)
+	if err != nil {
+		return nil, err
+	}
+
+	var kvsink indexer.EventSink
+	for _, sink := range env.EventSinks {
+		if sink.Type() == indexer.KV {
+			kvsink = sink
+		}
+	}
+
+	results, err := kvsink.SearchBlockEvents(ctx.Context(), q)
+	if err != nil {
+		return nil, err
+	}
+
+	// sort results (must be done before pagination)
+	switch orderBy {
+	case "desc", "":
+		sort.Slice(results, func(i, j int) bool { return results[i] > results[j] })
+
+	case "asc":
+		sort.Slice(results, func(i, j int) bool { return results[i] < results[j] })
+
+	default:
+		return nil, fmt.Errorf("expected order_by to be either `asc` or `desc` or empty: %w", ctypes.ErrInvalidRequest)
+	}
+
+	// paginate results
+	totalCount := len(results)
+	perPage := env.validatePerPage(perPagePtr)
+
+	page, err := validatePage(pagePtr, perPage, totalCount)
+	if err != nil {
+		return nil, err
+	}
+
+	skipCount := validateSkipCount(page, perPage)
+	pageSize := tmmath.MinInt(perPage, totalCount-skipCount)
+
+	apiResults := make([]*ctypes.ResultBlock, 0, pageSize)
+	for i := skipCount; i < skipCount+pageSize; i++ {
+		block := env.BlockStore.LoadBlock(results[i])
+		if block != nil {
+			blockMeta := env.BlockStore.LoadBlockMeta(block.Height)
+			if blockMeta != nil {
+				apiResults = append(apiResults, &ctypes.ResultBlock{
+					Block:   block,
+					BlockID: blockMeta.BlockID,
+				})
+			}
+		}
+	}
+
+	return &ctypes.ResultBlockSearch{Blocks: apiResults, TotalCount: totalCount}, nil
+}

internal/rpc/core/blocks_test.go

@@ -0,0 +1,135 @@
+package core
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	dbm "github.com/tendermint/tm-db"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+	sm "github.com/tendermint/tendermint/internal/state"
+	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+func TestBlockchainInfo(t *testing.T) {
+	cases := []struct {
+		min, max     int64
+		base, height int64
+		limit        int64
+		resultLength int64
+		wantErr      bool
+	}{
+
+		// min > max
+		{0, 0, 0, 0, 10, 0, true},  // min set to 1
+		{0, 1, 0, 0, 10, 0, true},  // max set to height (0)
+		{0, 0, 0, 1, 10, 1, false}, // max set to height (1)
+		{2, 0, 0, 1, 10, 0, true},  // max set to height (1)
+		{2, 1, 0, 5, 10, 0, true},
+
+		// negative
+		{1, 10, 0, 14, 10, 10, false}, // control
+		{-1, 10, 0, 14, 10, 0, true},
+		{1, -10, 0, 14, 10, 0, true},
+		{-9223372036854775808, -9223372036854775788, 0, 100, 20, 0, true},
+
+		// check base
+		{1, 1, 1, 1, 1, 1, false},
+		{2, 5, 3, 5, 5, 3, false},
+
+		// check limit and height
+		{1, 1, 0, 1, 10, 1, false},
+		{1, 1, 0, 5, 10, 1, false},
+		{2, 2, 0, 5, 10, 1, false},
+		{1, 2, 0, 5, 10, 2, false},
+		{1, 5, 0, 1, 10, 1, false},
+		{1, 5, 0, 10, 10, 5, false},
+		{1, 15, 0, 10, 10, 10, false},
+		{1, 15, 0, 15, 10, 10, false},
+		{1, 15, 0, 15, 20, 15, false},
+		{1, 20, 0, 15, 20, 15, false},
+		{1, 20, 0, 20, 20, 20, false},
+	}
+
+	for i, c := range cases {
+		caseString := fmt.Sprintf("test %d failed", i)
+		min, max, err := filterMinMax(c.base, c.height, c.min, c.max, c.limit)
+		if c.wantErr {
+			require.Error(t, err, caseString)
+		} else {
+			require.NoError(t, err, caseString)
+			require.Equal(t, 1+max-min, c.resultLength, caseString)
+		}
+	}
+}
+
+func TestBlockResults(t *testing.T) {
+	results := &tmstate.ABCIResponses{
+		DeliverTxs: []*abci.ResponseDeliverTx{
+			{Code: 0, Data: []byte{0x01}, Log: "ok", GasUsed: 10},
+			{Code: 0, Data: []byte{0x02}, Log: "ok", GasUsed: 5},
+			{Code: 1, Log: "not ok", GasUsed: 0},
+		},
+		EndBlock:   &abci.ResponseEndBlock{},
+		BeginBlock: &abci.ResponseBeginBlock{},
+	}
+
+	env := &Environment{}
+	env.StateStore = sm.NewStore(dbm.NewMemDB())
+	err := env.StateStore.SaveABCIResponses(100, results)
+	require.NoError(t, err)
+	env.BlockStore = mockBlockStore{height: 100}
+
+	testCases := []struct {
+		height  int64
+		wantErr bool
+		wantRes *ctypes.ResultBlockResults
+	}{
+		{-1, true, nil},
+		{0, true, nil},
+		{101, true, nil},
+		{100, false, &ctypes.ResultBlockResults{
+			Height:                100,
+			TxsResults:            results.DeliverTxs,
+			TotalGasUsed:          15,
+			BeginBlockEvents:      results.BeginBlock.Events,
+			EndBlockEvents:        results.EndBlock.Events,
+			ValidatorUpdates:      results.EndBlock.ValidatorUpdates,
+			ConsensusParamUpdates: results.EndBlock.ConsensusParamUpdates,
+		}},
+	}
+
+	for _, tc := range testCases {
+		res, err := env.BlockResults(&rpctypes.Context{}, &tc.height)
+		if tc.wantErr {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+			assert.Equal(t, tc.wantRes, res)
+		}
+	}
+}
+
+type mockBlockStore struct {
+	height int64
+}
+
+func (mockBlockStore) Base() int64                                       { return 1 }
+func (store mockBlockStore) Height() int64                               { return store.height }
+func (store mockBlockStore) Size() int64                                 { return store.height }
+func (mockBlockStore) LoadBaseMeta() *types.BlockMeta                    { return nil }
+func (mockBlockStore) LoadBlockMeta(height int64) *types.BlockMeta       { return nil }
+func (mockBlockStore) LoadBlock(height int64) *types.Block               { return nil }
+func (mockBlockStore) LoadBlockByHash(hash []byte) *types.Block          { return nil }
+func (mockBlockStore) LoadBlockPart(height int64, index int) *types.Part { return nil }
+func (mockBlockStore) LoadBlockCommit(height int64) *types.Commit        { return nil }
+func (mockBlockStore) LoadSeenCommit() *types.Commit                     { return nil }
+func (mockBlockStore) PruneBlocks(height int64) (uint64, error)          { return 0, nil }
+func (mockBlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) {
+}

internal/rpc/core/consensus.go

@@ -0,0 +1,150 @@
+package core
+
+import (
+	"errors"
+
+	cm "github.com/tendermint/tendermint/internal/consensus"
+	tmmath "github.com/tendermint/tendermint/libs/math"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+// Validators gets the validator set at the given block height.
+//
+// If no height is provided, it will fetch the latest validator set. Note the
+// validators are sorted by their voting power - this is the canonical order
+// for the validators in the set as used in computing their Merkle root.
+//
+// More: https://docs.tendermint.com/master/rpc/#/Info/validators
+func (env *Environment) Validators(
+	ctx *rpctypes.Context,
+	heightPtr *int64,
+	pagePtr, perPagePtr *int) (*ctypes.ResultValidators, error) {
+
+	// The latest validator that we know is the NextValidator of the last block.
+	height, err := env.getHeight(env.latestUncommittedHeight(), heightPtr)
+	if err != nil {
+		return nil, err
+	}
+
+	validators, err := env.StateStore.LoadValidators(height)
+	if err != nil {
+		return nil, err
+	}
+
+	totalCount := len(validators.Validators)
+	perPage := env.validatePerPage(perPagePtr)
+	page, err := validatePage(pagePtr, perPage, totalCount)
+	if err != nil {
+		return nil, err
+	}
+
+	skipCount := validateSkipCount(page, perPage)
+
+	v := validators.Validators[skipCount : skipCount+tmmath.MinInt(perPage, totalCount-skipCount)]
+
+	return &ctypes.ResultValidators{
+		BlockHeight: height,
+		Validators:  v,
+		Count:       len(v),
+		Total:       totalCount}, nil
+}
+
+// DumpConsensusState dumps consensus state.
+// UNSTABLE
+// More: https://docs.tendermint.com/master/rpc/#/Info/dump_consensus_state
+func (env *Environment) DumpConsensusState(ctx *rpctypes.Context) (*ctypes.ResultDumpConsensusState, error) {
+	// Get Peer consensus states.
+
+	var peerStates []ctypes.PeerStateInfo
+	switch {
+	case env.P2PPeers != nil:
+		peers := env.P2PPeers.Peers().List()
+		peerStates = make([]ctypes.PeerStateInfo, 0, len(peers))
+		for _, peer := range peers {
+			peerState, ok := peer.Get(types.PeerStateKey).(*cm.PeerState)
+			if !ok { // peer does not have a state yet
+				continue
+			}
+			peerStateJSON, err := peerState.ToJSON()
+			if err != nil {
+				return nil, err
+			}
+			peerStates = append(peerStates, ctypes.PeerStateInfo{
+				// Peer basic info.
+				NodeAddress: peer.SocketAddr().String(),
+				// Peer consensus state.
+				PeerState: peerStateJSON,
+			})
+		}
+	case env.PeerManager != nil:
+		peers := env.PeerManager.Peers()
+		peerStates = make([]ctypes.PeerStateInfo, 0, len(peers))
+		for _, pid := range peers {
+			peerState, ok := env.ConsensusReactor.GetPeerState(pid)
+			if !ok {
+				continue
+			}
+
+			peerStateJSON, err := peerState.ToJSON()
+			if err != nil {
+				return nil, err
+			}
+
+			addr := env.PeerManager.Addresses(pid)
+			if len(addr) >= 1 {
+				peerStates = append(peerStates, ctypes.PeerStateInfo{
+					// Peer basic info.
+					NodeAddress: addr[0].String(),
+					// Peer consensus state.
+					PeerState: peerStateJSON,
+				})
+			}
+		}
+	default:
+		return nil, errors.New("no peer system configured")
+	}
+
+	// Get self round state.
+	roundState, err := env.ConsensusState.GetRoundStateJSON()
+	if err != nil {
+		return nil, err
+	}
+	return &ctypes.ResultDumpConsensusState{
+		RoundState: roundState,
+		Peers:      peerStates}, nil
+}
+
+// ConsensusState returns a concise summary of the consensus state.
+// UNSTABLE
+// More: https://docs.tendermint.com/master/rpc/#/Info/consensus_state
+func (env *Environment) GetConsensusState(ctx *rpctypes.Context) (*ctypes.ResultConsensusState, error) {
+	// Get self round state.
+	bz, err := env.ConsensusState.GetRoundStateSimpleJSON()
+	return &ctypes.ResultConsensusState{RoundState: bz}, err
+}
+
+// ConsensusParams gets the consensus parameters at the given block height.
+// If no height is provided, it will fetch the latest consensus params.
+// More: https://docs.tendermint.com/master/rpc/#/Info/consensus_params
+func (env *Environment) ConsensusParams(
+	ctx *rpctypes.Context,
+	heightPtr *int64) (*ctypes.ResultConsensusParams, error) {
+
+	// The latest consensus params that we know is the consensus params after the
+	// last block.
+	height, err := env.getHeight(env.latestUncommittedHeight(), heightPtr)
+	if err != nil {
+		return nil, err
+	}
+
+	consensusParams, err := env.StateStore.LoadConsensusParams(height)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ctypes.ResultConsensusParams{
+		BlockHeight:     height,
+		ConsensusParams: consensusParams}, nil
+}

internal/rpc/core/dev.go

@@ -0,0 +1,12 @@
+package core
+
+import (
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+)
+
+// UnsafeFlushMempool removes all transactions from the mempool.
+func (env *Environment) UnsafeFlushMempool(ctx *rpctypes.Context) (*ctypes.ResultUnsafeFlushMempool, error) {
+	env.Mempool.Flush()
+	return &ctypes.ResultUnsafeFlushMempool{}, nil
+}

internal/rpc/core/doc.go

@@ -0,0 +1,45 @@
+/*
+Package core defines the Tendermint RPC endpoints.
+
+Tendermint ships with its own JSONRPC library -
+https://github.com/tendermint/tendermint/tree/master/rpc/jsonrpc.
+
+## Get the list
+
+An HTTP Get request to the root RPC endpoint shows a list of available endpoints.
+
+```bash
+curl 'localhost:26657'
+```
+
+> Response:
+
+```plain
+Available endpoints:
+/abci_info
+/dump_consensus_state
+/genesis
+/net_info
+/num_unconfirmed_txs
+/status
+/health
+/unconfirmed_txs
+/unsafe_flush_mempool
+/validators
+
+Endpoints that require arguments:
+/abci_query?path=_&data=_&prove=_
+/block?height=_
+/blockchain?minHeight=_&maxHeight=_
+/broadcast_tx_async?tx=_
+/broadcast_tx_commit?tx=_
+/broadcast_tx_sync?tx=_
+/commit?height=_
+/dial_seeds?seeds=_
+/dial_persistent_peers?persistent_peers=_
+/subscribe?event=_
+/tx?hash=_&prove=_
+/unsubscribe?event=_
+```
+*/
+package core

internal/rpc/core/doc_template.txt

@@ -0,0 +1,8 @@
+{{with .PDoc}}
+{{comment_md .Doc}}
+{{example_html $ ""}}
+
+{{range .Funcs}}{{$name_html := html .Name}}## [{{$name_html}}]({{posLink_url $ .Decl}})
+{{comment_md .Doc}}{{end}}
+{{end}}
+---

internal/rpc/core/env.go

@@ -0,0 +1,215 @@
+package core
+
+import (
+	"encoding/base64"
+	"fmt"
+	"time"
+
+	cfg "github.com/tendermint/tendermint/config"
+	"github.com/tendermint/tendermint/crypto"
+	"github.com/tendermint/tendermint/internal/consensus"
+	mempl "github.com/tendermint/tendermint/internal/mempool"
+	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/proxy"
+	sm "github.com/tendermint/tendermint/internal/state"
+	"github.com/tendermint/tendermint/internal/state/indexer"
+	"github.com/tendermint/tendermint/internal/statesync"
+	tmjson "github.com/tendermint/tendermint/libs/json"
+	"github.com/tendermint/tendermint/libs/log"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	"github.com/tendermint/tendermint/types"
+)
+
+const (
+	// see README
+	defaultPerPage = 30
+	maxPerPage     = 100
+
+	// SubscribeTimeout is the maximum time we wait to subscribe for an event.
+	// must be less than the server's write timeout (see rpcserver.DefaultConfig)
+	SubscribeTimeout = 5 * time.Second
+
+	// genesisChunkSize is the maximum size, in bytes, of each
+	// chunk in the genesis structure for the chunked API
+	genesisChunkSize = 16 * 1024 * 1024 // 16
+)
+
+//----------------------------------------------
+// These interfaces are used by RPC and must be thread safe
+
+type consensusState interface {
+	GetState() sm.State
+	GetValidators() (int64, []*types.Validator)
+	GetLastHeight() int64
+	GetRoundStateJSON() ([]byte, error)
+	GetRoundStateSimpleJSON() ([]byte, error)
+}
+
+type transport interface {
+	Listeners() []string
+	IsListening() bool
+	NodeInfo() types.NodeInfo
+}
+
+type peers interface {
+	AddPersistentPeers([]string) error
+	AddUnconditionalPeerIDs([]string) error
+	AddPrivatePeerIDs([]string) error
+	DialPeersAsync([]string) error
+	Peers() p2p.IPeerSet
+}
+
+type consensusReactor interface {
+	WaitSync() bool
+	GetPeerState(peerID types.NodeID) (*consensus.PeerState, bool)
+}
+
+type peerManager interface {
+	Peers() []types.NodeID
+	Addresses(types.NodeID) []p2p.NodeAddress
+}
+
+//----------------------------------------------
+// Environment contains objects and interfaces used by the RPC. It is expected
+// to be setup once during startup.
+type Environment struct {
+	// external, thread safe interfaces
+	ProxyAppQuery   proxy.AppConnQuery
+	ProxyAppMempool proxy.AppConnMempool
+
+	// interfaces defined in types and above
+	StateStore       sm.Store
+	BlockStore       sm.BlockStore
+	EvidencePool     sm.EvidencePool
+	ConsensusState   consensusState
+	ConsensusReactor consensusReactor
+	P2PPeers         peers
+
+	// Legacy p2p stack
+	P2PTransport transport
+
+	// interfaces for new p2p interfaces
+	PeerManager peerManager
+
+	// objects
+	PubKey            crypto.PubKey
+	GenDoc            *types.GenesisDoc // cache the genesis structure
+	EventSinks        []indexer.EventSink
+	EventBus          *types.EventBus // thread safe
+	Mempool           mempl.Mempool
+	BlockSyncReactor  consensus.BlockSyncReactor
+	StateSyncMetricer statesync.Metricer
+
+	Logger log.Logger
+
+	Config cfg.RPCConfig
+
+	// cache of chunked genesis data.
+	genChunks []string
+}
+
+//----------------------------------------------
+
+func validatePage(pagePtr *int, perPage, totalCount int) (int, error) {
+	// this can only happen if we haven't first run validatePerPage
+	if perPage < 1 {
+		panic(fmt.Errorf("%w (%d)", ctypes.ErrZeroOrNegativePerPage, perPage))
+	}
+
+	if pagePtr == nil { // no page parameter
+		return 1, nil
+	}
+
+	pages := ((totalCount - 1) / perPage) + 1
+	if pages == 0 {
+		pages = 1 // one page (even if it's empty)
+	}
+	page := *pagePtr
+	if page <= 0 || page > pages {
+		return 1, fmt.Errorf("%w expected range: [1, %d], given %d", ctypes.ErrPageOutOfRange, pages, page)
+	}
+
+	return page, nil
+}
+
+func (env *Environment) validatePerPage(perPagePtr *int) int {
+	if perPagePtr == nil { // no per_page parameter
+		return defaultPerPage
+	}
+
+	perPage := *perPagePtr
+	if perPage < 1 {
+		return defaultPerPage
+		// in unsafe mode there is no max on the page size but in safe mode
+		// we cap it to maxPerPage
+	} else if perPage > maxPerPage && !env.Config.Unsafe {
+		return maxPerPage
+	}
+	return perPage
+}
+
+// InitGenesisChunks configures the environment and should be called on service
+// startup.
+func (env *Environment) InitGenesisChunks() error {
+	if env.genChunks != nil {
+		return nil
+	}
+
+	if env.GenDoc == nil {
+		return nil
+	}
+
+	data, err := tmjson.Marshal(env.GenDoc)
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < len(data); i += genesisChunkSize {
+		end := i + genesisChunkSize
+
+		if end > len(data) {
+			end = len(data)
+		}
+
+		env.genChunks = append(env.genChunks, base64.StdEncoding.EncodeToString(data[i:end]))
+	}
+
+	return nil
+}
+
+func validateSkipCount(page, perPage int) int {
+	skipCount := (page - 1) * perPage
+	if skipCount < 0 {
+		return 0
+	}
+
+	return skipCount
+}
+
+// latestHeight can be either latest committed or uncommitted (+1) height.
+func (env *Environment) getHeight(latestHeight int64, heightPtr *int64) (int64, error) {
+	if heightPtr != nil {
+		height := *heightPtr
+		if height <= 0 {
+			return 0, fmt.Errorf("%w (requested height: %d)", ctypes.ErrZeroOrNegativeHeight, height)
+		}
+		if height > latestHeight {
+			return 0, fmt.Errorf("%w (requested height: %d, blockchain height: %d)",
+				ctypes.ErrHeightExceedsChainHead, height, latestHeight)
+		}
+		base := env.BlockStore.Base()
+		if height < base {
+			return 0, fmt.Errorf("%w (requested height: %d, base height: %d)", ctypes.ErrHeightNotAvailable, height, base)
+		}
+		return height, nil
+	}
+	return latestHeight, nil
+}
+
+func (env *Environment) latestUncommittedHeight() int64 {
+	nodeIsSyncing := env.ConsensusReactor.WaitSync()
+	if nodeIsSyncing {
+		return env.BlockStore.Height()
+	}
+	return env.BlockStore.Height() + 1
+}

internal/rpc/core/env_test.go

@@ -0,0 +1,90 @@
+package core
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPaginationPage(t *testing.T) {
+	cases := []struct {
+		totalCount int
+		perPage    int
+		page       int
+		newPage    int
+		expErr     bool
+	}{
+		{0, 10, 1, 1, false},
+
+		{0, 10, 0, 1, false},
+		{0, 10, 1, 1, false},
+		{0, 10, 2, 0, true},
+
+		{5, 10, -1, 0, true},
+		{5, 10, 0, 1, false},
+		{5, 10, 1, 1, false},
+		{5, 10, 2, 0, true},
+		{5, 10, 2, 0, true},
+
+		{5, 5, 1, 1, false},
+		{5, 5, 2, 0, true},
+		{5, 5, 3, 0, true},
+
+		{5, 3, 2, 2, false},
+		{5, 3, 3, 0, true},
+
+		{5, 2, 2, 2, false},
+		{5, 2, 3, 3, false},
+		{5, 2, 4, 0, true},
+	}
+
+	for _, c := range cases {
+		p, err := validatePage(&c.page, c.perPage, c.totalCount)
+		if c.expErr {
+			assert.Error(t, err)
+			continue
+		}
+
+		assert.Equal(t, c.newPage, p, fmt.Sprintf("%v", c))
+	}
+
+	// nil case
+	p, err := validatePage(nil, 1, 1)
+	if assert.NoError(t, err) {
+		assert.Equal(t, 1, p)
+	}
+}
+
+func TestPaginationPerPage(t *testing.T) {
+	cases := []struct {
+		perPage    int
+		newPerPage int
+	}{
+		{0, defaultPerPage},
+		{1, 1},
+		{2, 2},
+		{defaultPerPage, defaultPerPage},
+		{maxPerPage - 1, maxPerPage - 1},
+		{maxPerPage, maxPerPage},
+		{maxPerPage + 1, maxPerPage},
+	}
+
+	env := &Environment{}
+
+	for _, c := range cases {
+		p := env.validatePerPage(&c.perPage)
+		assert.Equal(t, c.newPerPage, p, fmt.Sprintf("%v", c))
+	}
+
+	// nil case
+	p := env.validatePerPage(nil)
+	assert.Equal(t, defaultPerPage, p)
+
+	// test in unsafe mode
+	env.Config.Unsafe = true
+	perPage := 1000
+	p = env.validatePerPage(&perPage)
+	assert.Equal(t, perPage, p)
+	env.Config.Unsafe = false
+}

internal/rpc/core/events.go

@@ -0,0 +1,116 @@
+package core
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
+	tmquery "github.com/tendermint/tendermint/libs/pubsub/query"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+)
+
+const (
+	// Buffer on the Tendermint (server) side to allow some slowness in clients.
+	subBufferSize = 100
+)
+
+// Subscribe for events via WebSocket.
+// More: https://docs.tendermint.com/master/rpc/#/Websocket/subscribe
+func (env *Environment) Subscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultSubscribe, error) {
+	addr := ctx.RemoteAddr()
+
+	if env.EventBus.NumClients() >= env.Config.MaxSubscriptionClients {
+		return nil, fmt.Errorf("max_subscription_clients %d reached", env.Config.MaxSubscriptionClients)
+	} else if env.EventBus.NumClientSubscriptions(addr) >= env.Config.MaxSubscriptionsPerClient {
+		return nil, fmt.Errorf("max_subscriptions_per_client %d reached", env.Config.MaxSubscriptionsPerClient)
+	}
+
+	env.Logger.Info("Subscribe to query", "remote", addr, "query", query)
+
+	q, err := tmquery.New(query)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse query: %w", err)
+	}
+
+	subCtx, cancel := context.WithTimeout(ctx.Context(), SubscribeTimeout)
+	defer cancel()
+
+	sub, err := env.EventBus.Subscribe(subCtx, addr, q, subBufferSize)
+	if err != nil {
+		return nil, err
+	}
+
+	// Capture the current ID, since it can change in the future.
+	subscriptionID := ctx.JSONReq.ID
+	go func() {
+		for {
+			select {
+			case msg := <-sub.Out():
+				var (
+					resultEvent = &ctypes.ResultEvent{Query: query, Data: msg.Data(), Events: msg.Events()}
+					resp        = rpctypes.NewRPCSuccessResponse(subscriptionID, resultEvent)
+				)
+				writeCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+				defer cancel()
+				if err := ctx.WSConn.WriteRPCResponse(writeCtx, resp); err != nil {
+					env.Logger.Info("Can't write response (slow client)",
+						"to", addr, "subscriptionID", subscriptionID, "err", err)
+				}
+			case <-sub.Canceled():
+				if sub.Err() != tmpubsub.ErrUnsubscribed {
+					var reason string
+					if sub.Err() == nil {
+						reason = "Tendermint exited"
+					} else {
+						reason = sub.Err().Error()
+					}
+					var (
+						err  = fmt.Errorf("subscription was canceled (reason: %s)", reason)
+						resp = rpctypes.RPCServerError(subscriptionID, err)
+					)
+					if ok := ctx.WSConn.TryWriteRPCResponse(resp); !ok {
+						env.Logger.Info("Can't write response (slow client)",
+							"to", addr, "subscriptionID", subscriptionID, "err", err)
+					}
+				}
+				return
+			}
+		}
+	}()
+
+	return &ctypes.ResultSubscribe{}, nil
+}
+
+// Unsubscribe from events via WebSocket.
+// More: https://docs.tendermint.com/master/rpc/#/Websocket/unsubscribe
+func (env *Environment) Unsubscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultUnsubscribe, error) {
+	args := tmpubsub.UnsubscribeArgs{Subscriber: ctx.RemoteAddr()}
+	env.Logger.Info("Unsubscribe from query", "remote", args.Subscriber, "subscription", query)
+
+	var err error
+	args.Query, err = tmquery.New(query)
+
+	if err != nil {
+		args.ID = query
+	}
+
+	err = env.EventBus.Unsubscribe(ctx.Context(), args)
+	if err != nil {
+		return nil, err
+	}
+	return &ctypes.ResultUnsubscribe{}, nil
+}
+
+// UnsubscribeAll from all events via WebSocket.
+// More: https://docs.tendermint.com/master/rpc/#/Websocket/unsubscribe_all
+func (env *Environment) UnsubscribeAll(ctx *rpctypes.Context) (*ctypes.ResultUnsubscribe, error) {
+	addr := ctx.RemoteAddr()
+	env.Logger.Info("Unsubscribe from all", "remote", addr)
+	err := env.EventBus.UnsubscribeAll(ctx.Context(), addr)
+	if err != nil {
+		return nil, err
+	}
+	return &ctypes.ResultUnsubscribe{}, nil
+}

internal/rpc/core/evidence.go

@@ -0,0 +1,29 @@
+package core
+
+import (
+	"fmt"
+
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+// BroadcastEvidence broadcasts evidence of the misbehavior.
+// More: https://docs.tendermint.com/master/rpc/#/Evidence/broadcast_evidence
+func (env *Environment) BroadcastEvidence(
+	ctx *rpctypes.Context,
+	ev types.Evidence) (*ctypes.ResultBroadcastEvidence, error) {
+
+	if ev == nil {
+		return nil, fmt.Errorf("%w: no evidence was provided", ctypes.ErrInvalidRequest)
+	}
+
+	if err := ev.ValidateBasic(); err != nil {
+		return nil, fmt.Errorf("evidence.ValidateBasic failed: %w", err)
+	}
+
+	if err := env.EvidencePool.AddEvidence(ev); err != nil {
+		return nil, fmt.Errorf("failed to add evidence: %w", err)
+	}
+	return &ctypes.ResultBroadcastEvidence{Hash: ev.Hash()}, nil
+}

internal/rpc/core/health.go

@@ -0,0 +1,13 @@
+package core
+
+import (
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+)
+
+// Health gets node health. Returns empty result (200 OK) on success, no
+// response - in case of an error.
+// More: https://docs.tendermint.com/master/rpc/#/Info/health
+func (env *Environment) Health(ctx *rpctypes.Context) (*ctypes.ResultHealth, error) {
+	return &ctypes.ResultHealth{}, nil
+}

internal/rpc/core/mempool.go

@@ -0,0 +1,180 @@
+package core
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+	mempl "github.com/tendermint/tendermint/internal/mempool"
+	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+//-----------------------------------------------------------------------------
+// NOTE: tx should be signed, but this is only checked at the app level (not by Tendermint!)
+
+// BroadcastTxAsync returns right away, with no response. Does not wait for
+// CheckTx nor DeliverTx results.
+// More: https://docs.tendermint.com/master/rpc/#/Tx/broadcast_tx_async
+func (env *Environment) BroadcastTxAsync(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTx, error) {
+	err := env.Mempool.CheckTx(ctx.Context(), tx, nil, mempl.TxInfo{})
+	if err != nil {
+		return nil, err
+	}
+
+	return &ctypes.ResultBroadcastTx{Hash: tx.Hash()}, nil
+}
+
+// BroadcastTxSync returns with the response from CheckTx. Does not wait for
+// DeliverTx result.
+// More: https://docs.tendermint.com/master/rpc/#/Tx/broadcast_tx_sync
+func (env *Environment) BroadcastTxSync(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTx, error) {
+	resCh := make(chan *abci.Response, 1)
+	err := env.Mempool.CheckTx(
+		ctx.Context(),
+		tx,
+		func(res *abci.Response) { resCh <- res },
+		mempl.TxInfo{},
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	res := <-resCh
+	r := res.GetCheckTx()
+
+	return &ctypes.ResultBroadcastTx{
+		Code:         r.Code,
+		Data:         r.Data,
+		Log:          r.Log,
+		Codespace:    r.Codespace,
+		MempoolError: r.MempoolError,
+		Hash:         tx.Hash(),
+	}, nil
+}
+
+// BroadcastTxCommit returns with the responses from CheckTx and DeliverTx.
+// More: https://docs.tendermint.com/master/rpc/#/Tx/broadcast_tx_commit
+func (env *Environment) BroadcastTxCommit(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTxCommit, error) {
+	subscriber := ctx.RemoteAddr()
+
+	if env.EventBus.NumClients() >= env.Config.MaxSubscriptionClients {
+		return nil, fmt.Errorf("max_subscription_clients %d reached", env.Config.MaxSubscriptionClients)
+	} else if env.EventBus.NumClientSubscriptions(subscriber) >= env.Config.MaxSubscriptionsPerClient {
+		return nil, fmt.Errorf("max_subscriptions_per_client %d reached", env.Config.MaxSubscriptionsPerClient)
+	}
+
+	// Subscribe to tx being committed in block.
+	subCtx, cancel := context.WithTimeout(ctx.Context(), SubscribeTimeout)
+	defer cancel()
+	q := types.EventQueryTxFor(tx)
+	deliverTxSub, err := env.EventBus.Subscribe(subCtx, subscriber, q)
+	if err != nil {
+		err = fmt.Errorf("failed to subscribe to tx: %w", err)
+		env.Logger.Error("Error on broadcast_tx_commit", "err", err)
+		return nil, err
+	}
+	defer func() {
+		args := tmpubsub.UnsubscribeArgs{Subscriber: subscriber, Query: q}
+		if err := env.EventBus.Unsubscribe(context.Background(), args); err != nil {
+			env.Logger.Error("Error unsubscribing from eventBus", "err", err)
+		}
+	}()
+
+	// Broadcast tx and wait for CheckTx result
+	checkTxResCh := make(chan *abci.Response, 1)
+	err = env.Mempool.CheckTx(
+		ctx.Context(),
+		tx,
+		func(res *abci.Response) { checkTxResCh <- res },
+		mempl.TxInfo{},
+	)
+	if err != nil {
+		env.Logger.Error("Error on broadcastTxCommit", "err", err)
+		return nil, fmt.Errorf("error on broadcastTxCommit: %v", err)
+	}
+
+	checkTxResMsg := <-checkTxResCh
+	checkTxRes := checkTxResMsg.GetCheckTx()
+
+	if checkTxRes.Code != abci.CodeTypeOK {
+		return &ctypes.ResultBroadcastTxCommit{
+			CheckTx:   *checkTxRes,
+			DeliverTx: abci.ResponseDeliverTx{},
+			Hash:      tx.Hash(),
+		}, nil
+	}
+
+	// Wait for the tx to be included in a block or timeout.
+	select {
+	case msg := <-deliverTxSub.Out(): // The tx was included in a block.
+		deliverTxRes := msg.Data().(types.EventDataTx)
+		return &ctypes.ResultBroadcastTxCommit{
+			CheckTx:   *checkTxRes,
+			DeliverTx: deliverTxRes.Result,
+			Hash:      tx.Hash(),
+			Height:    deliverTxRes.Height,
+		}, nil
+	case <-deliverTxSub.Canceled():
+		var reason string
+		if deliverTxSub.Err() == nil {
+			reason = "Tendermint exited"
+		} else {
+			reason = deliverTxSub.Err().Error()
+		}
+		err = fmt.Errorf("deliverTxSub was canceled (reason: %s)", reason)
+		env.Logger.Error("Error on broadcastTxCommit", "err", err)
+		return &ctypes.ResultBroadcastTxCommit{
+			CheckTx:   *checkTxRes,
+			DeliverTx: abci.ResponseDeliverTx{},
+			Hash:      tx.Hash(),
+		}, err
+	case <-time.After(env.Config.TimeoutBroadcastTxCommit):
+		err = errors.New("timed out waiting for tx to be included in a block")
+		env.Logger.Error("Error on broadcastTxCommit", "err", err)
+		return &ctypes.ResultBroadcastTxCommit{
+			CheckTx:   *checkTxRes,
+			DeliverTx: abci.ResponseDeliverTx{},
+			Hash:      tx.Hash(),
+		}, err
+	}
+}
+
+// UnconfirmedTxs gets unconfirmed transactions (maximum ?limit entries)
+// including their number.
+// More: https://docs.tendermint.com/master/rpc/#/Info/unconfirmed_txs
+func (env *Environment) UnconfirmedTxs(ctx *rpctypes.Context, limitPtr *int) (*ctypes.ResultUnconfirmedTxs, error) {
+	// reuse per_page validator
+	limit := env.validatePerPage(limitPtr)
+
+	txs := env.Mempool.ReapMaxTxs(limit)
+	return &ctypes.ResultUnconfirmedTxs{
+		Count:      len(txs),
+		Total:      env.Mempool.Size(),
+		TotalBytes: env.Mempool.SizeBytes(),
+		Txs:        txs}, nil
+}
+
+// NumUnconfirmedTxs gets number of unconfirmed transactions.
+// More: https://docs.tendermint.com/master/rpc/#/Info/num_unconfirmed_txs
+func (env *Environment) NumUnconfirmedTxs(ctx *rpctypes.Context) (*ctypes.ResultUnconfirmedTxs, error) {
+	return &ctypes.ResultUnconfirmedTxs{
+		Count:      env.Mempool.Size(),
+		Total:      env.Mempool.Size(),
+		TotalBytes: env.Mempool.SizeBytes()}, nil
+}
+
+// CheckTx checks the transaction without executing it. The transaction won't
+// be added to the mempool either.
+// More: https://docs.tendermint.com/master/rpc/#/Tx/check_tx
+func (env *Environment) CheckTx(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultCheckTx, error) {
+	res, err := env.ProxyAppMempool.CheckTxSync(ctx.Context(), abci.RequestCheckTx{Tx: tx})
+	if err != nil {
+		return nil, err
+	}
+	return &ctypes.ResultCheckTx{ResponseCheckTx: *res}, nil
+}

internal/rpc/core/net.go

@@ -0,0 +1,162 @@
+package core
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/tendermint/tendermint/internal/p2p"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+)
+
+// NetInfo returns network info.
+// More: https://docs.tendermint.com/master/rpc/#/Info/net_info
+func (env *Environment) NetInfo(ctx *rpctypes.Context) (*ctypes.ResultNetInfo, error) {
+	var peers []ctypes.Peer
+
+	switch {
+	case env.P2PPeers != nil:
+		peersList := env.P2PPeers.Peers().List()
+		peers = make([]ctypes.Peer, 0, len(peersList))
+		for _, peer := range peersList {
+			peers = append(peers, ctypes.Peer{
+				ID:  peer.ID(),
+				URL: peer.SocketAddr().String(),
+			})
+		}
+	case env.PeerManager != nil:
+		peerList := env.PeerManager.Peers()
+		for _, peer := range peerList {
+			addrs := env.PeerManager.Addresses(peer)
+			if len(addrs) == 0 {
+				continue
+			}
+
+			peers = append(peers, ctypes.Peer{
+				ID:  peer,
+				URL: addrs[0].String(),
+			})
+		}
+	default:
+		return nil, errors.New("peer management system does not support NetInfo responses")
+	}
+
+	return &ctypes.ResultNetInfo{
+		Listening: env.P2PTransport.IsListening(),
+		Listeners: env.P2PTransport.Listeners(),
+		NPeers:    len(peers),
+		Peers:     peers,
+	}, nil
+}
+
+// UnsafeDialSeeds dials the given seeds (comma-separated id@IP:PORT).
+func (env *Environment) UnsafeDialSeeds(ctx *rpctypes.Context, seeds []string) (*ctypes.ResultDialSeeds, error) {
+	if env.P2PPeers == nil {
+		return nil, errors.New("peer management system does not support this operation")
+	}
+
+	if len(seeds) == 0 {
+		return &ctypes.ResultDialSeeds{}, fmt.Errorf("%w: no seeds provided", ctypes.ErrInvalidRequest)
+	}
+	env.Logger.Info("DialSeeds", "seeds", seeds)
+	if err := env.P2PPeers.DialPeersAsync(seeds); err != nil {
+		return &ctypes.ResultDialSeeds{}, err
+	}
+	return &ctypes.ResultDialSeeds{Log: "Dialing seeds in progress. See /net_info for details"}, nil
+}
+
+// UnsafeDialPeers dials the given peers (comma-separated id@IP:PORT),
+// optionally making them persistent.
+func (env *Environment) UnsafeDialPeers(
+	ctx *rpctypes.Context,
+	peers []string,
+	persistent, unconditional, private bool) (*ctypes.ResultDialPeers, error) {
+
+	if env.P2PPeers == nil {
+		return nil, errors.New("peer management system does not support this operation")
+	}
+
+	if len(peers) == 0 {
+		return &ctypes.ResultDialPeers{}, fmt.Errorf("%w: no peers provided", ctypes.ErrInvalidRequest)
+	}
+
+	ids, err := getIDs(peers)
+	if err != nil {
+		return &ctypes.ResultDialPeers{}, err
+	}
+
+	env.Logger.Info("DialPeers", "peers", peers, "persistent",
+		persistent, "unconditional", unconditional, "private", private)
+
+	if persistent {
+		if err := env.P2PPeers.AddPersistentPeers(peers); err != nil {
+			return &ctypes.ResultDialPeers{}, err
+		}
+	}
+
+	if private {
+		if err := env.P2PPeers.AddPrivatePeerIDs(ids); err != nil {
+			return &ctypes.ResultDialPeers{}, err
+		}
+	}
+
+	if unconditional {
+		if err := env.P2PPeers.AddUnconditionalPeerIDs(ids); err != nil {
+			return &ctypes.ResultDialPeers{}, err
+		}
+	}
+
+	if err := env.P2PPeers.DialPeersAsync(peers); err != nil {
+		return &ctypes.ResultDialPeers{}, err
+	}
+
+	return &ctypes.ResultDialPeers{Log: "Dialing peers in progress. See /net_info for details"}, nil
+}
+
+// Genesis returns genesis file.
+// More: https://docs.tendermint.com/master/rpc/#/Info/genesis
+func (env *Environment) Genesis(ctx *rpctypes.Context) (*ctypes.ResultGenesis, error) {
+	if len(env.genChunks) > 1 {
+		return nil, errors.New("genesis response is large, please use the genesis_chunked API instead")
+	}
+
+	return &ctypes.ResultGenesis{Genesis: env.GenDoc}, nil
+}
+
+func (env *Environment) GenesisChunked(ctx *rpctypes.Context, chunk uint) (*ctypes.ResultGenesisChunk, error) {
+	if env.genChunks == nil {
+		return nil, fmt.Errorf("service configuration error, genesis chunks are not initialized")
+	}
+
+	if len(env.genChunks) == 0 {
+		return nil, fmt.Errorf("service configuration error, there are no chunks")
+	}
+
+	id := int(chunk)
+
+	if id > len(env.genChunks)-1 {
+		return nil, fmt.Errorf("there are %d chunks, %d is invalid", len(env.genChunks)-1, id)
+	}
+
+	return &ctypes.ResultGenesisChunk{
+		TotalChunks: len(env.genChunks),
+		ChunkNumber: id,
+		Data:        env.genChunks[id],
+	}, nil
+}
+
+func getIDs(peers []string) ([]string, error) {
+	ids := make([]string, 0, len(peers))
+
+	for _, peer := range peers {
+
+		spl := strings.Split(peer, "@")
+		if len(spl) != 2 {
+			return nil, p2p.ErrNetAddressNoID{Addr: peer}
+		}
+		ids = append(ids, spl[0])
+
+	}
+	return ids, nil
+}

internal/rpc/core/net_test.go

@@ -0,0 +1,89 @@
+package core
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	cfg "github.com/tendermint/tendermint/config"
+	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/libs/log"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+)
+
+func TestUnsafeDialSeeds(t *testing.T) {
+	sw := p2p.MakeSwitch(cfg.DefaultP2PConfig(), 1, "testing", "123.123.123",
+		func(n int, sw *p2p.Switch) *p2p.Switch { return sw }, log.TestingLogger())
+	err := sw.Start()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := sw.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+
+	env := &Environment{}
+	env.Logger = log.TestingLogger()
+	env.P2PPeers = sw
+
+	testCases := []struct {
+		seeds []string
+		isErr bool
+	}{
+		{[]string{}, true},
+		{[]string{"d51fb70907db1c6c2d5237e78379b25cf1a37ab4@127.0.0.1:41198"}, false},
+		{[]string{"127.0.0.1:41198"}, true},
+	}
+
+	for _, tc := range testCases {
+		res, err := env.UnsafeDialSeeds(&rpctypes.Context{}, tc.seeds)
+		if tc.isErr {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+			assert.NotNil(t, res)
+		}
+	}
+}
+
+func TestUnsafeDialPeers(t *testing.T) {
+	sw := p2p.MakeSwitch(cfg.DefaultP2PConfig(), 1, "testing", "123.123.123",
+		func(n int, sw *p2p.Switch) *p2p.Switch { return sw }, log.TestingLogger())
+	sw.SetAddrBook(&p2p.AddrBookMock{
+		Addrs:        make(map[string]struct{}),
+		OurAddrs:     make(map[string]struct{}),
+		PrivateAddrs: make(map[string]struct{}),
+	})
+	err := sw.Start()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := sw.Stop(); err != nil {
+			t.Error(err)
+		}
+	})
+
+	env := &Environment{}
+	env.Logger = log.TestingLogger()
+	env.P2PPeers = sw
+
+	testCases := []struct {
+		peers                               []string
+		persistence, unconditional, private bool
+		isErr                               bool
+	}{
+		{[]string{}, false, false, false, true},
+		{[]string{"d51fb70907db1c6c2d5237e78379b25cf1a37ab4@127.0.0.1:41198"}, true, true, true, false},
+		{[]string{"127.0.0.1:41198"}, true, true, false, true},
+	}
+
+	for _, tc := range testCases {
+		res, err := env.UnsafeDialPeers(&rpctypes.Context{}, tc.peers, tc.persistence, tc.unconditional, tc.private)
+		if tc.isErr {
+			assert.Error(t, err)
+		} else {
+			assert.NoError(t, err)
+			assert.NotNil(t, res)
+		}
+	}
+}

internal/rpc/core/routes.go

@@ -0,0 +1,61 @@
+package core
+
+import (
+	rpc "github.com/tendermint/tendermint/rpc/jsonrpc/server"
+)
+
+// TODO: better system than "unsafe" prefix
+
+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, ""),
+
+		// info API
+		"health":               rpc.NewRPCFunc(env.Health, "", false),
+		"status":               rpc.NewRPCFunc(env.Status, "", false),
+		"net_info":             rpc.NewRPCFunc(env.NetInfo, "", false),
+		"blockchain":           rpc.NewRPCFunc(env.BlockchainInfo, "minHeight,maxHeight", true),
+		"genesis":              rpc.NewRPCFunc(env.Genesis, "", true),
+		"genesis_chunked":      rpc.NewRPCFunc(env.GenesisChunked, "chunk", true),
+		"block":                rpc.NewRPCFunc(env.Block, "height", true),
+		"block_by_hash":        rpc.NewRPCFunc(env.BlockByHash, "hash", true),
+		"block_results":        rpc.NewRPCFunc(env.BlockResults, "height", true),
+		"commit":               rpc.NewRPCFunc(env.Commit, "height", true),
+		"check_tx":             rpc.NewRPCFunc(env.CheckTx, "tx", true),
+		"tx":                   rpc.NewRPCFunc(env.Tx, "hash,prove", true),
+		"tx_search":            rpc.NewRPCFunc(env.TxSearch, "query,prove,page,per_page,order_by", false),
+		"block_search":         rpc.NewRPCFunc(env.BlockSearch, "query,page,per_page,order_by", false),
+		"validators":           rpc.NewRPCFunc(env.Validators, "height,page,per_page", true),
+		"dump_consensus_state": rpc.NewRPCFunc(env.DumpConsensusState, "", false),
+		"consensus_state":      rpc.NewRPCFunc(env.GetConsensusState, "", false),
+		"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
+		"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
+		"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)
+}

internal/rpc/core/status.go

@@ -0,0 +1,115 @@
+package core
+
+import (
+	"bytes"
+	"time"
+
+	tmbytes "github.com/tendermint/tendermint/libs/bytes"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+// Status returns Tendermint status including node info, pubkey, latest block
+// hash, app hash, block height, current max peer block height, and time.
+// More: https://docs.tendermint.com/master/rpc/#/Info/status
+func (env *Environment) Status(ctx *rpctypes.Context) (*ctypes.ResultStatus, error) {
+	var (
+		earliestBlockHeight   int64
+		earliestBlockHash     tmbytes.HexBytes
+		earliestAppHash       tmbytes.HexBytes
+		earliestBlockTimeNano int64
+	)
+
+	if earliestBlockMeta := env.BlockStore.LoadBaseMeta(); earliestBlockMeta != nil {
+		earliestBlockHeight = earliestBlockMeta.Header.Height
+		earliestAppHash = earliestBlockMeta.Header.AppHash
+		earliestBlockHash = earliestBlockMeta.BlockID.Hash
+		earliestBlockTimeNano = earliestBlockMeta.Header.Time.UnixNano()
+	}
+
+	var (
+		latestBlockHash     tmbytes.HexBytes
+		latestAppHash       tmbytes.HexBytes
+		latestBlockTimeNano int64
+
+		latestHeight = env.BlockStore.Height()
+	)
+
+	if latestHeight != 0 {
+		if latestBlockMeta := env.BlockStore.LoadBlockMeta(latestHeight); latestBlockMeta != nil {
+			latestBlockHash = latestBlockMeta.BlockID.Hash
+			latestAppHash = latestBlockMeta.Header.AppHash
+			latestBlockTimeNano = latestBlockMeta.Header.Time.UnixNano()
+		}
+	}
+
+	// Return the very last voting power, not the voting power of this validator
+	// during the last block.
+	var votingPower int64
+	if val := env.validatorAtHeight(env.latestUncommittedHeight()); val != nil {
+		votingPower = val.VotingPower
+	}
+	validatorInfo := ctypes.ValidatorInfo{}
+	if env.PubKey != nil {
+		validatorInfo = ctypes.ValidatorInfo{
+			Address:     env.PubKey.Address(),
+			PubKey:      env.PubKey,
+			VotingPower: votingPower,
+		}
+	}
+	result := &ctypes.ResultStatus{
+		NodeInfo: env.P2PTransport.NodeInfo(),
+		SyncInfo: ctypes.SyncInfo{
+			LatestBlockHash:     latestBlockHash,
+			LatestAppHash:       latestAppHash,
+			LatestBlockHeight:   latestHeight,
+			LatestBlockTime:     time.Unix(0, latestBlockTimeNano),
+			EarliestBlockHash:   earliestBlockHash,
+			EarliestAppHash:     earliestAppHash,
+			EarliestBlockHeight: earliestBlockHeight,
+			EarliestBlockTime:   time.Unix(0, earliestBlockTimeNano),
+			MaxPeerBlockHeight:  env.BlockSyncReactor.GetMaxPeerBlockHeight(),
+			CatchingUp:          env.ConsensusReactor.WaitSync(),
+			TotalSyncedTime:     env.BlockSyncReactor.GetTotalSyncedTime(),
+			RemainingTime:       env.BlockSyncReactor.GetRemainingSyncTime(),
+		},
+		ValidatorInfo: validatorInfo,
+	}
+
+	if env.StateSyncMetricer != nil {
+		result.SyncInfo.TotalSnapshots = env.StateSyncMetricer.TotalSnapshots()
+		result.SyncInfo.ChunkProcessAvgTime = env.StateSyncMetricer.ChunkProcessAvgTime()
+		result.SyncInfo.SnapshotHeight = env.StateSyncMetricer.SnapshotHeight()
+		result.SyncInfo.SnapshotChunksCount = env.StateSyncMetricer.SnapshotChunksCount()
+		result.SyncInfo.SnapshotChunksTotal = env.StateSyncMetricer.SnapshotChunksTotal()
+		result.SyncInfo.BackFilledBlocks = env.StateSyncMetricer.BackFilledBlocks()
+		result.SyncInfo.BackFillBlocksTotal = env.StateSyncMetricer.BackFillBlocksTotal()
+	}
+
+	return result, nil
+}
+
+func (env *Environment) validatorAtHeight(h int64) *types.Validator {
+	valsWithH, err := env.StateStore.LoadValidators(h)
+	if err != nil {
+		return nil
+	}
+	if env.PubKey == nil {
+		return nil
+	}
+	privValAddress := env.PubKey.Address()
+
+	// If we're still at height h, search in the current validator set.
+	lastBlockHeight, vals := env.ConsensusState.GetValidators()
+	if lastBlockHeight == h {
+		for _, val := range vals {
+			if bytes.Equal(val.Address, privValAddress) {
+				return val
+			}
+		}
+	}
+
+	_, val := valsWithH.GetByAddress(privValAddress)
+	return val
+}

internal/rpc/core/tx.go

@@ -0,0 +1,146 @@
+package core
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/tendermint/tendermint/internal/state/indexer"
+	"github.com/tendermint/tendermint/libs/bytes"
+	tmmath "github.com/tendermint/tendermint/libs/math"
+	tmquery "github.com/tendermint/tendermint/libs/pubsub/query"
+	ctypes "github.com/tendermint/tendermint/rpc/coretypes"
+	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	"github.com/tendermint/tendermint/types"
+)
+
+// Tx allows you to query the transaction results. `nil` could mean the
+// transaction is in the mempool, invalidated, or was not sent in the first
+// place.
+// More: https://docs.tendermint.com/master/rpc/#/Info/tx
+func (env *Environment) Tx(ctx *rpctypes.Context, hash bytes.HexBytes, prove bool) (*ctypes.ResultTx, error) {
+	// if index is disabled, return error
+
+	// N.B. The hash parameter is HexBytes so that the reflective parameter
+	// decoding logic in the HTTP service will correctly translate from JSON.
+	// See https://github.com/tendermint/tendermint/issues/6802 for context.
+
+	if !indexer.KVSinkEnabled(env.EventSinks) {
+		return nil, errors.New("transaction querying is disabled due to no kvEventSink")
+	}
+
+	for _, sink := range env.EventSinks {
+		if sink.Type() == indexer.KV {
+			r, err := sink.GetTxByHash(hash)
+			if r == nil {
+				return nil, fmt.Errorf("tx (%X) not found, err: %w", hash, err)
+			}
+
+			height := r.Height
+			index := r.Index
+
+			var proof types.TxProof
+			if prove {
+				block := env.BlockStore.LoadBlock(height)
+				proof = block.Data.Txs.Proof(int(index)) // XXX: overflow on 32-bit machines
+			}
+
+			return &ctypes.ResultTx{
+				Hash:     hash,
+				Height:   height,
+				Index:    index,
+				TxResult: r.Result,
+				Tx:       r.Tx,
+				Proof:    proof,
+			}, nil
+		}
+	}
+
+	return nil, fmt.Errorf("transaction querying is disabled on this node due to the KV event sink being disabled")
+}
+
+// TxSearch allows you to query for multiple transactions results. It returns a
+// list of transactions (maximum ?per_page entries) and the total count.
+// More: https://docs.tendermint.com/master/rpc/#/Info/tx_search
+func (env *Environment) TxSearch(
+	ctx *rpctypes.Context,
+	query string,
+	prove bool,
+	pagePtr, perPagePtr *int,
+	orderBy string,
+) (*ctypes.ResultTxSearch, error) {
+
+	if !indexer.KVSinkEnabled(env.EventSinks) {
+		return nil, fmt.Errorf("transaction searching is disabled due to no kvEventSink")
+	}
+
+	q, err := tmquery.New(query)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, sink := range env.EventSinks {
+		if sink.Type() == indexer.KV {
+			results, err := sink.SearchTxEvents(ctx.Context(), q)
+			if err != nil {
+				return nil, err
+			}
+
+			// sort results (must be done before pagination)
+			switch orderBy {
+			case "desc", "":
+				sort.Slice(results, func(i, j int) bool {
+					if results[i].Height == results[j].Height {
+						return results[i].Index > results[j].Index
+					}
+					return results[i].Height > results[j].Height
+				})
+			case "asc":
+				sort.Slice(results, func(i, j int) bool {
+					if results[i].Height == results[j].Height {
+						return results[i].Index < results[j].Index
+					}
+					return results[i].Height < results[j].Height
+				})
+			default:
+				return nil, fmt.Errorf("expected order_by to be either `asc` or `desc` or empty: %w", ctypes.ErrInvalidRequest)
+			}
+
+			// paginate results
+			totalCount := len(results)
+			perPage := env.validatePerPage(perPagePtr)
+
+			page, err := validatePage(pagePtr, perPage, totalCount)
+			if err != nil {
+				return nil, err
+			}
+
+			skipCount := validateSkipCount(page, perPage)
+			pageSize := tmmath.MinInt(perPage, totalCount-skipCount)
+
+			apiResults := make([]*ctypes.ResultTx, 0, pageSize)
+			for i := skipCount; i < skipCount+pageSize; i++ {
+				r := results[i]
+
+				var proof types.TxProof
+				if prove {
+					block := env.BlockStore.LoadBlock(r.Height)
+					proof = block.Data.Txs.Proof(int(r.Index)) // XXX: overflow on 32-bit machines
+				}
+
+				apiResults = append(apiResults, &ctypes.ResultTx{
+					Hash:     types.Tx(r.Tx).Hash(),
+					Height:   r.Height,
+					Index:    r.Index,
+					TxResult: r.Result,
+					Tx:       r.Tx,
+					Proof:    proof,
+				})
+			}
+
+			return &ctypes.ResultTxSearch{Txs: apiResults, TotalCount: totalCount}, nil
+		}
+	}
+
+	return nil, fmt.Errorf("transaction searching is disabled on this node due to the KV event sink being disabled")
+}