Unify RPC method signatures and parameter decoding (#8397)

Pass all parameters from JSON-RPC requests to their corresponding handlers
using struct types instead of positional parameters. This allows us to control
encoding of arguments using only the standard library, and to eliminate the
remaining special-purpose JSON encoding hooks in the server.

To support existing use, the server still allows arguments to be encoded in
JSON as either an array or an object.

Related changes:

- Rework the RPCFunc constructor to reduce reflection during RPC call service.
- Add request parameter wrappers for each RPC service method.
- Update the RPC Environment methods to use these types.
- Update the interfaces and shims derived from Environment to the new
  signatures.
- Update and extend test cases.
This commit is contained in:
M. J. Fromberger
2022-04-27 15:57:30 -04:00
committed by tycho garen
parent 0dda3ff88b
commit 251e41895f
30 changed files with 983 additions and 779 deletions
+10 -10
View File
@@ -38,16 +38,16 @@ func Routes(cfg config.RPCConfig, s state.Store, bs state.BlockStore, es []index
Logger: logger,
}
return core.RoutesMap{
"blockchain": server.NewRPCFunc(env.BlockchainInfo, "minHeight", "maxHeight"),
"consensus_params": server.NewRPCFunc(env.ConsensusParams, "height"),
"block": server.NewRPCFunc(env.Block, "height"),
"block_by_hash": server.NewRPCFunc(env.BlockByHash, "hash"),
"block_results": server.NewRPCFunc(env.BlockResults, "height"),
"commit": server.NewRPCFunc(env.Commit, "height"),
"validators": server.NewRPCFunc(env.Validators, "height", "page", "per_page"),
"tx": server.NewRPCFunc(env.Tx, "hash", "prove"),
"tx_search": server.NewRPCFunc(env.TxSearch, "query", "prove", "page", "per_page", "order_by"),
"block_search": server.NewRPCFunc(env.BlockSearch, "query", "page", "per_page", "order_by"),
"blockchain": server.NewRPCFunc(env.BlockchainInfo),
"consensus_params": server.NewRPCFunc(env.ConsensusParams),
"block": server.NewRPCFunc(env.Block),
"block_by_hash": server.NewRPCFunc(env.BlockByHash),
"block_results": server.NewRPCFunc(env.BlockResults),
"commit": server.NewRPCFunc(env.Commit),
"validators": server.NewRPCFunc(env.Validators),
"tx": server.NewRPCFunc(env.Tx),
"tx_search": server.NewRPCFunc(env.TxSearch),
"block_search": server.NewRPCFunc(env.BlockSearch),
}
}
+5 -12
View File
@@ -5,24 +5,17 @@ import (
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/internal/proxy"
"github.com/tendermint/tendermint/libs/bytes"
"github.com/tendermint/tendermint/rpc/coretypes"
)
// ABCIQuery queries the application for some information.
// More: https://docs.tendermint.com/master/rpc/#/ABCI/abci_query
func (env *Environment) ABCIQuery(
ctx context.Context,
path string,
data bytes.HexBytes,
height int64,
prove bool,
) (*coretypes.ResultABCIQuery, error) {
func (env *Environment) ABCIQuery(ctx context.Context, req *coretypes.RequestABCIQuery) (*coretypes.ResultABCIQuery, error) {
resQuery, err := env.ProxyApp.Query(ctx, &abci.RequestQuery{
Path: path,
Data: data,
Height: height,
Prove: prove,
Path: req.Path,
Data: req.Data,
Height: int64(req.Height),
Prove: req.Prove,
})
if err != nil {
return nil, err
+25 -43
View File
@@ -7,7 +7,6 @@ import (
tmquery "github.com/tendermint/tendermint/internal/pubsub/query"
"github.com/tendermint/tendermint/internal/state/indexer"
"github.com/tendermint/tendermint/libs/bytes"
tmmath "github.com/tendermint/tendermint/libs/math"
"github.com/tendermint/tendermint/rpc/coretypes"
"github.com/tendermint/tendermint/types"
@@ -23,17 +22,15 @@ import (
// order (highest first).
//
// More: https://docs.tendermint.com/master/rpc/#/Info/blockchain
func (env *Environment) BlockchainInfo(ctx context.Context, minHeight, maxHeight int64) (*coretypes.ResultBlockchainInfo, error) {
const limit int64 = 20
var err error
minHeight, maxHeight, err = filterMinMax(
func (env *Environment) BlockchainInfo(ctx context.Context, req *coretypes.RequestBlockchainInfo) (*coretypes.ResultBlockchainInfo, error) {
const limit = 20
minHeight, maxHeight, err := filterMinMax(
env.BlockStore.Base(),
env.BlockStore.Height(),
minHeight,
maxHeight,
limit)
int64(req.MinHeight),
int64(req.MaxHeight),
limit,
)
if err != nil {
return nil, err
}
@@ -90,8 +87,8 @@ func filterMinMax(base, height, min, max, limit int64) (int64, int64, error) {
// 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 context.Context, heightPtr *int64) (*coretypes.ResultBlock, error) {
height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
func (env *Environment) Block(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultBlock, error) {
height, err := env.getHeight(env.BlockStore.Height(), (*int64)(req.Height))
if err != nil {
return nil, err
}
@@ -107,12 +104,8 @@ func (env *Environment) Block(ctx context.Context, heightPtr *int64) (*coretypes
// BlockByHash gets block by hash.
// More: https://docs.tendermint.com/master/rpc/#/Info/block_by_hash
func (env *Environment) BlockByHash(ctx context.Context, hash bytes.HexBytes) (*coretypes.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)
func (env *Environment) BlockByHash(ctx context.Context, req *coretypes.RequestBlockByHash) (*coretypes.ResultBlock, error) {
block := env.BlockStore.LoadBlockByHash(req.Hash)
if block == nil {
return &coretypes.ResultBlock{BlockID: types.BlockID{}, Block: nil}, nil
}
@@ -124,8 +117,8 @@ func (env *Environment) BlockByHash(ctx context.Context, hash bytes.HexBytes) (*
// Header gets block header at a given height.
// If no height is provided, it will fetch the latest header.
// More: https://docs.tendermint.com/master/rpc/#/Info/header
func (env *Environment) Header(ctx context.Context, heightPtr *int64) (*coretypes.ResultHeader, error) {
height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
func (env *Environment) Header(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultHeader, error) {
height, err := env.getHeight(env.BlockStore.Height(), (*int64)(req.Height))
if err != nil {
return nil, err
}
@@ -140,12 +133,8 @@ func (env *Environment) Header(ctx context.Context, heightPtr *int64) (*coretype
// HeaderByHash gets header by hash.
// More: https://docs.tendermint.com/master/rpc/#/Info/header_by_hash
func (env *Environment) HeaderByHash(ctx context.Context, hash bytes.HexBytes) (*coretypes.ResultHeader, 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.
blockMeta := env.BlockStore.LoadBlockMetaByHash(hash)
func (env *Environment) HeaderByHash(ctx context.Context, req *coretypes.RequestBlockByHash) (*coretypes.ResultHeader, error) {
blockMeta := env.BlockStore.LoadBlockMetaByHash(req.Hash)
if blockMeta == nil {
return &coretypes.ResultHeader{}, nil
}
@@ -156,8 +145,8 @@ func (env *Environment) HeaderByHash(ctx context.Context, hash bytes.HexBytes) (
// 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 context.Context, heightPtr *int64) (*coretypes.ResultCommit, error) {
height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
func (env *Environment) Commit(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultCommit, error) {
height, err := env.getHeight(env.BlockStore.Height(), (*int64)(req.Height))
if err != nil {
return nil, err
}
@@ -192,8 +181,8 @@ func (env *Environment) Commit(ctx context.Context, heightPtr *int64) (*coretype
//
// Results are for the height of the block containing the txs.
// More: https://docs.tendermint.com/master/rpc/#/Info/block_results
func (env *Environment) BlockResults(ctx context.Context, heightPtr *int64) (*coretypes.ResultBlockResults, error) {
height, err := env.getHeight(env.BlockStore.Height(), heightPtr)
func (env *Environment) BlockResults(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultBlockResults, error) {
height, err := env.getHeight(env.BlockStore.Height(), (*int64)(req.Height))
if err != nil {
return nil, err
}
@@ -218,20 +207,13 @@ func (env *Environment) BlockResults(ctx context.Context, heightPtr *int64) (*co
}, nil
}
// BlockSearch searches for a paginated set of blocks matching the provided
// query.
func (env *Environment) BlockSearch(
ctx context.Context,
query string,
pagePtr, perPagePtr *int,
orderBy string,
) (*coretypes.ResultBlockSearch, error) {
// BlockSearch searches for a paginated set of blocks matching the provided query.
func (env *Environment) BlockSearch(ctx context.Context, req *coretypes.RequestBlockSearch) (*coretypes.ResultBlockSearch, error) {
if !indexer.KVSinkEnabled(env.EventSinks) {
return nil, fmt.Errorf("block searching is disabled due to no kvEventSink")
}
q, err := tmquery.New(query)
q, err := tmquery.New(req.Query)
if err != nil {
return nil, err
}
@@ -249,7 +231,7 @@ func (env *Environment) BlockSearch(
}
// sort results (must be done before pagination)
switch orderBy {
switch req.OrderBy {
case "desc", "":
sort.Slice(results, func(i, j int) bool { return results[i] > results[j] })
@@ -262,9 +244,9 @@ func (env *Environment) BlockSearch(
// paginate results
totalCount := len(results)
perPage := env.validatePerPage(perPagePtr)
perPage := env.validatePerPage(req.PerPage.IntPtr())
page, err := validatePage(pagePtr, perPage, totalCount)
page, err := validatePage(req.Page.IntPtr(), perPage, totalCount)
if err != nil {
return nil, err
}
+3 -1
View File
@@ -109,7 +109,9 @@ func TestBlockResults(t *testing.T) {
ctx := context.Background()
for _, tc := range testCases {
res, err := env.BlockResults(ctx, &tc.height)
res, err := env.BlockResults(ctx, &coretypes.RequestBlockInfo{
Height: (*coretypes.Int64)(&tc.height),
})
if tc.wantErr {
assert.Error(t, err)
} else {
+10 -11
View File
@@ -14,10 +14,9 @@ import (
// 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 context.Context, heightPtr *int64, pagePtr, perPagePtr *int) (*coretypes.ResultValidators, error) {
func (env *Environment) Validators(ctx context.Context, req *coretypes.RequestValidators) (*coretypes.ResultValidators, error) {
// The latest validator that we know is the NextValidator of the last block.
height, err := env.getHeight(env.latestUncommittedHeight(), heightPtr)
height, err := env.getHeight(env.latestUncommittedHeight(), (*int64)(req.Height))
if err != nil {
return nil, err
}
@@ -28,8 +27,8 @@ func (env *Environment) Validators(ctx context.Context, heightPtr *int64, pagePt
}
totalCount := len(validators.Validators)
perPage := env.validatePerPage(perPagePtr)
page, err := validatePage(pagePtr, perPage, totalCount)
perPage := env.validatePerPage(req.PerPage.IntPtr())
page, err := validatePage(req.Page.IntPtr(), perPage, totalCount)
if err != nil {
return nil, err
}
@@ -42,7 +41,8 @@ func (env *Environment) Validators(ctx context.Context, heightPtr *int64, pagePt
BlockHeight: height,
Validators: v,
Count: len(v),
Total: totalCount}, nil
Total: totalCount,
}, nil
}
// DumpConsensusState dumps consensus state.
@@ -99,11 +99,10 @@ func (env *Environment) GetConsensusState(ctx context.Context) (*coretypes.Resul
// 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 context.Context, heightPtr *int64) (*coretypes.ResultConsensusParams, error) {
// The latest consensus params that we know is the consensus params after the
// last block.
height, err := env.getHeight(env.latestUncommittedHeight(), heightPtr)
func (env *Environment) ConsensusParams(ctx context.Context, req *coretypes.RequestConsensusParams) (*coretypes.ResultConsensusParams, error) {
// The latest consensus params that we know is the consensus params after
// the last block.
height, err := env.getHeight(env.latestUncommittedHeight(), (*int64)(req.Height))
if err != nil {
return nil, err
}
+23 -17
View File
@@ -26,7 +26,7 @@ const (
// Subscribe for events via WebSocket.
// More: https://docs.tendermint.com/master/rpc/#/Websocket/subscribe
func (env *Environment) Subscribe(ctx context.Context, query string) (*coretypes.ResultSubscribe, error) {
func (env *Environment) Subscribe(ctx context.Context, req *coretypes.RequestSubscribe) (*coretypes.ResultSubscribe, error) {
callInfo := rpctypes.GetCallInfo(ctx)
addr := callInfo.RemoteAddr()
@@ -34,15 +34,15 @@ func (env *Environment) Subscribe(ctx context.Context, query string) (*coretypes
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)
} else if len(query) > maxQueryLength {
} else if len(req.Query) > maxQueryLength {
return nil, errors.New("maximum query length exceeded")
}
env.Logger.Info("WARNING: Websocket subscriptions are deprecated and will be removed " +
"in Tendermint v0.37. See https://tinyurl.com/adr075 for more information.")
env.Logger.Info("Subscribe to query", "remote", addr, "query", query)
env.Logger.Info("Subscribe to query", "remote", addr, "query", req.Query)
q, err := tmquery.New(query)
q, err := tmquery.New(req.Query)
if err != nil {
return nil, fmt.Errorf("failed to parse query: %w", err)
}
@@ -83,7 +83,7 @@ func (env *Environment) Subscribe(ctx context.Context, query string) (*coretypes
// We have a message to deliver to the client.
resp := callInfo.RPCRequest.MakeResponse(&coretypes.ResultEvent{
Query: query,
Query: req.Query,
Data: msg.Data(),
Events: msg.Events(),
})
@@ -102,15 +102,15 @@ func (env *Environment) Subscribe(ctx context.Context, query string) (*coretypes
// Unsubscribe from events via WebSocket.
// More: https://docs.tendermint.com/master/rpc/#/Websocket/unsubscribe
func (env *Environment) Unsubscribe(ctx context.Context, query string) (*coretypes.ResultUnsubscribe, error) {
func (env *Environment) Unsubscribe(ctx context.Context, req *coretypes.RequestUnsubscribe) (*coretypes.ResultUnsubscribe, error) {
args := tmpubsub.UnsubscribeArgs{Subscriber: rpctypes.GetCallInfo(ctx).RemoteAddr()}
env.Logger.Info("Unsubscribe from query", "remote", args.Subscriber, "subscription", query)
env.Logger.Info("Unsubscribe from query", "remote", args.Subscriber, "subscription", req.Query)
var err error
args.Query, err = tmquery.New(query)
args.Query, err = tmquery.New(req.Query)
if err != nil {
args.ID = query
args.ID = req.Query
}
err = env.EventBus.Unsubscribe(ctx, args)
@@ -148,17 +148,13 @@ func (env *Environment) UnsubscribeAll(ctx context.Context) (*coretypes.ResultUn
// If maxItems ≤ 0, a default positive number of events is chosen. The values
// of maxItems and waitTime may be capped to sensible internal maxima without
// reporting an error to the caller.
func (env *Environment) Events(ctx context.Context,
filter *coretypes.EventFilter,
maxItems int,
before, after cursor.Cursor,
waitTime time.Duration,
) (*coretypes.ResultEvents, error) {
func (env *Environment) Events(ctx context.Context, req *coretypes.RequestEvents) (*coretypes.ResultEvents, error) {
if env.EventLog == nil {
return nil, errors.New("the event log is not enabled")
}
// Parse and validate parameters.
maxItems := req.MaxItems
if maxItems <= 0 {
maxItems = 10
} else if maxItems > 100 {
@@ -167,6 +163,8 @@ func (env *Environment) Events(ctx context.Context,
const minWaitTime = 1 * time.Second
const maxWaitTime = 30 * time.Second
waitTime := req.WaitTime
if waitTime < minWaitTime {
waitTime = minWaitTime
} else if waitTime > maxWaitTime {
@@ -174,14 +172,22 @@ func (env *Environment) Events(ctx context.Context,
}
query := tmquery.All
if filter != nil && filter.Query != "" {
q, err := tmquery.New(filter.Query)
if req.Filter != nil && req.Filter.Query != "" {
q, err := tmquery.New(req.Filter.Query)
if err != nil {
return nil, fmt.Errorf("invalid filter query: %w", err)
}
query = q
}
var before, after cursor.Cursor
if err := before.UnmarshalText([]byte(req.Before)); err != nil {
return nil, fmt.Errorf("invalid cursor %q: %w", req.Before, err)
}
if err := after.UnmarshalText([]byte(req.After)); err != nil {
return nil, fmt.Errorf("invalid cursor %q: %w", req.After, err)
}
var info eventlog.Info
var items []*eventlog.Item
var err error
+5 -8
View File
@@ -9,18 +9,15 @@ import (
// BroadcastEvidence broadcasts evidence of the misbehavior.
// More: https://docs.tendermint.com/master/rpc/#/Evidence/broadcast_evidence
func (env *Environment) BroadcastEvidence(
ctx context.Context,
ev coretypes.Evidence,
) (*coretypes.ResultBroadcastEvidence, error) {
if ev.Value == nil {
func (env *Environment) BroadcastEvidence(ctx context.Context, req *coretypes.RequestBroadcastEvidence) (*coretypes.ResultBroadcastEvidence, error) {
if req.Evidence == nil {
return nil, fmt.Errorf("%w: no evidence was provided", coretypes.ErrInvalidRequest)
}
if err := ev.Value.ValidateBasic(); err != nil {
if err := req.Evidence.ValidateBasic(); err != nil {
return nil, fmt.Errorf("evidence.ValidateBasic failed: %w", err)
}
if err := env.EvidencePool.AddEvidence(ctx, ev.Value); err != nil {
if err := env.EvidencePool.AddEvidence(ctx, req.Evidence); err != nil {
return nil, fmt.Errorf("failed to add evidence: %w", err)
}
return &coretypes.ResultBroadcastEvidence{Hash: ev.Value.Hash()}, nil
return &coretypes.ResultBroadcastEvidence{Hash: req.Evidence.Hash()}, nil
}
+26 -24
View File
@@ -12,7 +12,6 @@ import (
"github.com/tendermint/tendermint/internal/state/indexer"
tmmath "github.com/tendermint/tendermint/libs/math"
"github.com/tendermint/tendermint/rpc/coretypes"
"github.com/tendermint/tendermint/types"
)
//-----------------------------------------------------------------------------
@@ -21,23 +20,23 @@ import (
// 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 context.Context, tx types.Tx) (*coretypes.ResultBroadcastTx, error) {
err := env.Mempool.CheckTx(ctx, tx, nil, mempool.TxInfo{})
func (env *Environment) BroadcastTxAsync(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error) {
err := env.Mempool.CheckTx(ctx, req.Tx, nil, mempool.TxInfo{})
if err != nil {
return nil, err
}
return &coretypes.ResultBroadcastTx{Hash: tx.Hash()}, nil
return &coretypes.ResultBroadcastTx{Hash: req.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 context.Context, tx types.Tx) (*coretypes.ResultBroadcastTx, error) {
func (env *Environment) BroadcastTxSync(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error) {
resCh := make(chan *abci.ResponseCheckTx, 1)
err := env.Mempool.CheckTx(
ctx,
tx,
req.Tx,
func(res *abci.ResponseCheckTx) {
select {
case <-ctx.Done():
@@ -60,19 +59,18 @@ func (env *Environment) BroadcastTxSync(ctx context.Context, tx types.Tx) (*core
Log: r.Log,
Codespace: r.Codespace,
MempoolError: r.MempoolError,
Hash: tx.Hash(),
Hash: req.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 context.Context, tx types.Tx) (*coretypes.ResultBroadcastTxCommit, error) {
func (env *Environment) BroadcastTxCommit(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTxCommit, error) {
resCh := make(chan *abci.ResponseCheckTx, 1)
err := env.Mempool.CheckTx(
ctx,
tx,
req.Tx,
func(res *abci.ResponseCheckTx) {
select {
case <-ctx.Done():
@@ -92,14 +90,14 @@ func (env *Environment) BroadcastTxCommit(ctx context.Context, tx types.Tx) (*co
if r.Code != abci.CodeTypeOK {
return &coretypes.ResultBroadcastTxCommit{
CheckTx: *r,
Hash: tx.Hash(),
Hash: req.Tx.Hash(),
}, fmt.Errorf("transaction encountered error (%s)", r.MempoolError)
}
if !indexer.KVSinkEnabled(env.EventSinks) {
return &coretypes.ResultBroadcastTxCommit{
CheckTx: *r,
Hash: tx.Hash(),
Hash: req.Tx.Hash(),
},
errors.New("cannot confirm transaction because kvEventSink is not enabled")
}
@@ -118,11 +116,14 @@ func (env *Environment) BroadcastTxCommit(ctx context.Context, tx types.Tx) (*co
"err", err)
return &coretypes.ResultBroadcastTxCommit{
CheckTx: *r,
Hash: tx.Hash(),
Hash: req.Tx.Hash(),
}, fmt.Errorf("timeout waiting for commit of tx %s (%s)",
tx.Hash(), time.Since(startAt))
req.Tx.Hash(), time.Since(startAt))
case <-timer.C:
txres, err := env.Tx(ctx, tx.Hash(), false)
txres, err := env.Tx(ctx, &coretypes.RequestTx{
Hash: req.Tx.Hash(),
Prove: false,
})
if err != nil {
jitter := 100*time.Millisecond + time.Duration(rand.Int63n(int64(time.Second))) // nolint: gosec
backoff := 100 * time.Duration(count) * time.Millisecond
@@ -133,7 +134,7 @@ func (env *Environment) BroadcastTxCommit(ctx context.Context, tx types.Tx) (*co
return &coretypes.ResultBroadcastTxCommit{
CheckTx: *r,
TxResult: txres.TxResult,
Hash: tx.Hash(),
Hash: req.Tx.Hash(),
Height: txres.Height,
}, nil
}
@@ -143,10 +144,10 @@ func (env *Environment) BroadcastTxCommit(ctx context.Context, tx types.Tx) (*co
// UnconfirmedTxs gets unconfirmed transactions from the mempool in order of priority
// More: https://docs.tendermint.com/master/rpc/#/Info/unconfirmed_txs
func (env *Environment) UnconfirmedTxs(ctx context.Context, pagePtr, perPagePtr *int) (*coretypes.ResultUnconfirmedTxs, error) {
func (env *Environment) UnconfirmedTxs(ctx context.Context, req *coretypes.RequestUnconfirmedTxs) (*coretypes.ResultUnconfirmedTxs, error) {
totalCount := env.Mempool.Size()
perPage := env.validatePerPage(perPagePtr)
page, err := validatePage(pagePtr, perPage, totalCount)
perPage := env.validatePerPage(req.PerPage.IntPtr())
page, err := validatePage(req.Page.IntPtr(), perPage, totalCount)
if err != nil {
return nil, err
}
@@ -160,7 +161,8 @@ func (env *Environment) UnconfirmedTxs(ctx context.Context, pagePtr, perPagePtr
Count: len(result),
Total: totalCount,
TotalBytes: env.Mempool.SizeBytes(),
Txs: result}, nil
Txs: result,
}, nil
}
// NumUnconfirmedTxs gets number of unconfirmed transactions.
@@ -175,14 +177,14 @@ func (env *Environment) NumUnconfirmedTxs(ctx context.Context) (*coretypes.Resul
// 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 context.Context, tx types.Tx) (*coretypes.ResultCheckTx, error) {
res, err := env.ProxyApp.CheckTx(ctx, &abci.RequestCheckTx{Tx: tx})
func (env *Environment) CheckTx(ctx context.Context, req *coretypes.RequestCheckTx) (*coretypes.ResultCheckTx, error) {
res, err := env.ProxyApp.CheckTx(ctx, &abci.RequestCheckTx{Tx: req.Tx})
if err != nil {
return nil, err
}
return &coretypes.ResultCheckTx{ResponseCheckTx: *res}, nil
}
func (env *Environment) RemoveTx(ctx context.Context, txkey types.TxKey) error {
return env.Mempool.RemoveTxByKey(txkey)
func (env *Environment) RemoveTx(ctx context.Context, req *coretypes.RequestRemoveTx) error {
return env.Mempool.RemoveTxByKey(req.TxKey)
}
+2 -2
View File
@@ -44,7 +44,7 @@ func (env *Environment) Genesis(ctx context.Context) (*coretypes.ResultGenesis,
return &coretypes.ResultGenesis{Genesis: env.GenDoc}, nil
}
func (env *Environment) GenesisChunked(ctx context.Context, chunk uint) (*coretypes.ResultGenesisChunk, error) {
func (env *Environment) GenesisChunked(ctx context.Context, req *coretypes.RequestGenesisChunked) (*coretypes.ResultGenesisChunk, error) {
if env.genChunks == nil {
return nil, fmt.Errorf("service configuration error, genesis chunks are not initialized")
}
@@ -53,7 +53,7 @@ func (env *Environment) GenesisChunked(ctx context.Context, chunk uint) (*corety
return nil, fmt.Errorf("service configuration error, there are no chunks")
}
id := int(chunk)
id := int(req.Chunk)
if id > len(env.genChunks)-1 {
return nil, fmt.Errorf("there are %d chunks, %d is invalid", len(env.genChunks)-1, id)
+48 -52
View File
@@ -2,13 +2,9 @@ package core
import (
"context"
"time"
"github.com/tendermint/tendermint/internal/eventlog/cursor"
"github.com/tendermint/tendermint/libs/bytes"
"github.com/tendermint/tendermint/rpc/coretypes"
rpc "github.com/tendermint/tendermint/rpc/jsonrpc/server"
"github.com/tendermint/tendermint/types"
)
// TODO: better system than "unsafe" prefix
@@ -32,47 +28,47 @@ func NewRoutesMap(svc RPCService, opts *RouteOptions) RoutesMap {
out := RoutesMap{
// Event subscription. Note that subscribe, unsubscribe, and
// unsubscribe_all are only available via the websocket endpoint.
"events": rpc.NewRPCFunc(svc.Events, "filter", "maxItems", "before", "after", "waitTime"),
"subscribe": rpc.NewWSRPCFunc(svc.Subscribe, "query"),
"unsubscribe": rpc.NewWSRPCFunc(svc.Unsubscribe, "query"),
"events": rpc.NewRPCFunc(svc.Events),
"subscribe": rpc.NewWSRPCFunc(svc.Subscribe),
"unsubscribe": rpc.NewWSRPCFunc(svc.Unsubscribe),
"unsubscribe_all": rpc.NewWSRPCFunc(svc.UnsubscribeAll),
// info API
"health": rpc.NewRPCFunc(svc.Health),
"status": rpc.NewRPCFunc(svc.Status),
"net_info": rpc.NewRPCFunc(svc.NetInfo),
"blockchain": rpc.NewRPCFunc(svc.BlockchainInfo, "minHeight", "maxHeight"),
"blockchain": rpc.NewRPCFunc(svc.BlockchainInfo),
"genesis": rpc.NewRPCFunc(svc.Genesis),
"genesis_chunked": rpc.NewRPCFunc(svc.GenesisChunked, "chunk"),
"header": rpc.NewRPCFunc(svc.Header, "height"),
"header_by_hash": rpc.NewRPCFunc(svc.HeaderByHash, "hash"),
"block": rpc.NewRPCFunc(svc.Block, "height"),
"block_by_hash": rpc.NewRPCFunc(svc.BlockByHash, "hash"),
"block_results": rpc.NewRPCFunc(svc.BlockResults, "height"),
"commit": rpc.NewRPCFunc(svc.Commit, "height"),
"check_tx": rpc.NewRPCFunc(svc.CheckTx, "tx"),
"remove_tx": rpc.NewRPCFunc(svc.RemoveTx, "txkey"),
"tx": rpc.NewRPCFunc(svc.Tx, "hash", "prove"),
"tx_search": rpc.NewRPCFunc(svc.TxSearch, "query", "prove", "page", "per_page", "order_by"),
"block_search": rpc.NewRPCFunc(svc.BlockSearch, "query", "page", "per_page", "order_by"),
"validators": rpc.NewRPCFunc(svc.Validators, "height", "page", "per_page"),
"genesis_chunked": rpc.NewRPCFunc(svc.GenesisChunked),
"header": rpc.NewRPCFunc(svc.Header),
"header_by_hash": rpc.NewRPCFunc(svc.HeaderByHash),
"block": rpc.NewRPCFunc(svc.Block),
"block_by_hash": rpc.NewRPCFunc(svc.BlockByHash),
"block_results": rpc.NewRPCFunc(svc.BlockResults),
"commit": rpc.NewRPCFunc(svc.Commit),
"check_tx": rpc.NewRPCFunc(svc.CheckTx),
"remove_tx": rpc.NewRPCFunc(svc.RemoveTx),
"tx": rpc.NewRPCFunc(svc.Tx),
"tx_search": rpc.NewRPCFunc(svc.TxSearch),
"block_search": rpc.NewRPCFunc(svc.BlockSearch),
"validators": rpc.NewRPCFunc(svc.Validators),
"dump_consensus_state": rpc.NewRPCFunc(svc.DumpConsensusState),
"consensus_state": rpc.NewRPCFunc(svc.GetConsensusState),
"consensus_params": rpc.NewRPCFunc(svc.ConsensusParams, "height"),
"unconfirmed_txs": rpc.NewRPCFunc(svc.UnconfirmedTxs, "page", "per_page"),
"consensus_params": rpc.NewRPCFunc(svc.ConsensusParams),
"unconfirmed_txs": rpc.NewRPCFunc(svc.UnconfirmedTxs),
"num_unconfirmed_txs": rpc.NewRPCFunc(svc.NumUnconfirmedTxs),
// tx broadcast API
"broadcast_tx_commit": rpc.NewRPCFunc(svc.BroadcastTxCommit, "tx"),
"broadcast_tx_sync": rpc.NewRPCFunc(svc.BroadcastTxSync, "tx"),
"broadcast_tx_async": rpc.NewRPCFunc(svc.BroadcastTxAsync, "tx"),
"broadcast_tx_commit": rpc.NewRPCFunc(svc.BroadcastTxCommit),
"broadcast_tx_sync": rpc.NewRPCFunc(svc.BroadcastTxSync),
"broadcast_tx_async": rpc.NewRPCFunc(svc.BroadcastTxAsync),
// abci API
"abci_query": rpc.NewRPCFunc(svc.ABCIQuery, "path", "data", "height", "prove"),
"abci_query": rpc.NewRPCFunc(svc.ABCIQuery),
"abci_info": rpc.NewRPCFunc(svc.ABCIInfo),
// evidence API
"broadcast_evidence": rpc.NewRPCFunc(svc.BroadcastEvidence, "evidence"),
"broadcast_evidence": rpc.NewRPCFunc(svc.BroadcastEvidence),
}
if u, ok := svc.(RPCUnsafe); ok && opts.Unsafe {
out["unsafe_flush_mempool"] = rpc.NewRPCFunc(u.UnsafeFlushMempool)
@@ -84,38 +80,38 @@ func NewRoutesMap(svc RPCService, opts *RouteOptions) RoutesMap {
// implementation, for use in constructing a routing table.
type RPCService interface {
ABCIInfo(ctx context.Context) (*coretypes.ResultABCIInfo, error)
ABCIQuery(ctx context.Context, path string, data bytes.HexBytes, height int64, prove bool) (*coretypes.ResultABCIQuery, error)
Block(ctx context.Context, heightPtr *int64) (*coretypes.ResultBlock, error)
BlockByHash(ctx context.Context, hash bytes.HexBytes) (*coretypes.ResultBlock, error)
BlockResults(ctx context.Context, heightPtr *int64) (*coretypes.ResultBlockResults, error)
BlockSearch(ctx context.Context, query string, pagePtr, perPagePtr *int, orderBy string) (*coretypes.ResultBlockSearch, error)
BlockchainInfo(ctx context.Context, minHeight, maxHeight int64) (*coretypes.ResultBlockchainInfo, error)
BroadcastEvidence(ctx context.Context, ev coretypes.Evidence) (*coretypes.ResultBroadcastEvidence, error)
BroadcastTxAsync(ctx context.Context, tx types.Tx) (*coretypes.ResultBroadcastTx, error)
BroadcastTxCommit(ctx context.Context, tx types.Tx) (*coretypes.ResultBroadcastTxCommit, error)
BroadcastTxSync(ctx context.Context, tx types.Tx) (*coretypes.ResultBroadcastTx, error)
CheckTx(ctx context.Context, tx types.Tx) (*coretypes.ResultCheckTx, error)
Commit(ctx context.Context, heightPtr *int64) (*coretypes.ResultCommit, error)
ConsensusParams(ctx context.Context, heightPtr *int64) (*coretypes.ResultConsensusParams, error)
ABCIQuery(ctx context.Context, req *coretypes.RequestABCIQuery) (*coretypes.ResultABCIQuery, error)
Block(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultBlock, error)
BlockByHash(ctx context.Context, req *coretypes.RequestBlockByHash) (*coretypes.ResultBlock, error)
BlockResults(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultBlockResults, error)
BlockSearch(ctx context.Context, req *coretypes.RequestBlockSearch) (*coretypes.ResultBlockSearch, error)
BlockchainInfo(ctx context.Context, req *coretypes.RequestBlockchainInfo) (*coretypes.ResultBlockchainInfo, error)
BroadcastEvidence(ctx context.Context, req *coretypes.RequestBroadcastEvidence) (*coretypes.ResultBroadcastEvidence, error)
BroadcastTxAsync(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error)
BroadcastTxCommit(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTxCommit, error)
BroadcastTxSync(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTx, error)
CheckTx(ctx context.Context, req *coretypes.RequestCheckTx) (*coretypes.ResultCheckTx, error)
Commit(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultCommit, error)
ConsensusParams(ctx context.Context, req *coretypes.RequestConsensusParams) (*coretypes.ResultConsensusParams, error)
DumpConsensusState(ctx context.Context) (*coretypes.ResultDumpConsensusState, error)
Events(ctx context.Context, filter *coretypes.EventFilter, maxItems int, before, after cursor.Cursor, waitTime time.Duration) (*coretypes.ResultEvents, error)
Events(ctx context.Context, req *coretypes.RequestEvents) (*coretypes.ResultEvents, error)
Genesis(ctx context.Context) (*coretypes.ResultGenesis, error)
GenesisChunked(ctx context.Context, chunk uint) (*coretypes.ResultGenesisChunk, error)
GenesisChunked(ctx context.Context, req *coretypes.RequestGenesisChunked) (*coretypes.ResultGenesisChunk, error)
GetConsensusState(ctx context.Context) (*coretypes.ResultConsensusState, error)
Header(ctx context.Context, heightPtr *int64) (*coretypes.ResultHeader, error)
HeaderByHash(ctx context.Context, hash bytes.HexBytes) (*coretypes.ResultHeader, error)
Header(ctx context.Context, req *coretypes.RequestBlockInfo) (*coretypes.ResultHeader, error)
HeaderByHash(ctx context.Context, req *coretypes.RequestBlockByHash) (*coretypes.ResultHeader, error)
Health(ctx context.Context) (*coretypes.ResultHealth, error)
NetInfo(ctx context.Context) (*coretypes.ResultNetInfo, error)
NumUnconfirmedTxs(ctx context.Context) (*coretypes.ResultUnconfirmedTxs, error)
RemoveTx(ctx context.Context, txkey types.TxKey) error
RemoveTx(ctx context.Context, req *coretypes.RequestRemoveTx) error
Status(ctx context.Context) (*coretypes.ResultStatus, error)
Subscribe(ctx context.Context, query string) (*coretypes.ResultSubscribe, error)
Tx(ctx context.Context, hash bytes.HexBytes, prove bool) (*coretypes.ResultTx, error)
TxSearch(ctx context.Context, query string, prove bool, pagePtr, perPagePtr *int, orderBy string) (*coretypes.ResultTxSearch, error)
UnconfirmedTxs(ctx context.Context, page, perPage *int) (*coretypes.ResultUnconfirmedTxs, error)
Unsubscribe(ctx context.Context, query string) (*coretypes.ResultUnsubscribe, error)
Subscribe(ctx context.Context, req *coretypes.RequestSubscribe) (*coretypes.ResultSubscribe, error)
Tx(ctx context.Context, req *coretypes.RequestTx) (*coretypes.ResultTx, error)
TxSearch(ctx context.Context, req *coretypes.RequestTxSearch) (*coretypes.ResultTxSearch, error)
UnconfirmedTxs(ctx context.Context, req *coretypes.RequestUnconfirmedTxs) (*coretypes.ResultUnconfirmedTxs, error)
Unsubscribe(ctx context.Context, req *coretypes.RequestUnsubscribe) (*coretypes.ResultUnsubscribe, error)
UnsubscribeAll(ctx context.Context) (*coretypes.ResultUnsubscribe, error)
Validators(ctx context.Context, heightPtr *int64, pagePtr, perPagePtr *int) (*coretypes.ResultValidators, error)
Validators(ctx context.Context, req *coretypes.RequestValidators) (*coretypes.ResultValidators, error)
}
// RPCUnsafe defines the set of "unsafe" methods that may optionally be
+12 -25
View File
@@ -8,7 +8,6 @@ import (
tmquery "github.com/tendermint/tendermint/internal/pubsub/query"
"github.com/tendermint/tendermint/internal/state/indexer"
"github.com/tendermint/tendermint/libs/bytes"
tmmath "github.com/tendermint/tendermint/libs/math"
"github.com/tendermint/tendermint/rpc/coretypes"
"github.com/tendermint/tendermint/types"
@@ -18,32 +17,27 @@ import (
// 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 context.Context, hash bytes.HexBytes, prove bool) (*coretypes.ResultTx, error) {
func (env *Environment) Tx(ctx context.Context, req *coretypes.RequestTx) (*coretypes.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)
r, err := sink.GetTxByHash(req.Hash)
if r == nil {
return nil, fmt.Errorf("tx (%X) not found, err: %w", hash, err)
return nil, fmt.Errorf("tx (%X) not found, err: %w", req.Hash, err)
}
var proof types.TxProof
if prove {
if req.Prove {
block := env.BlockStore.LoadBlock(r.Height)
proof = block.Data.Txs.Proof(int(r.Index))
}
return &coretypes.ResultTx{
Hash: hash,
Hash: req.Hash,
Height: r.Height,
Index: r.Index,
TxResult: r.Result,
@@ -59,21 +53,14 @@ func (env *Environment) Tx(ctx context.Context, hash bytes.HexBytes, prove bool)
// 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 context.Context,
query string,
prove bool,
pagePtr, perPagePtr *int,
orderBy string,
) (*coretypes.ResultTxSearch, error) {
func (env *Environment) TxSearch(ctx context.Context, req *coretypes.RequestTxSearch) (*coretypes.ResultTxSearch, error) {
if !indexer.KVSinkEnabled(env.EventSinks) {
return nil, fmt.Errorf("transaction searching is disabled due to no kvEventSink")
} else if len(query) > maxQueryLength {
} else if len(req.Query) > maxQueryLength {
return nil, errors.New("maximum query length exceeded")
}
q, err := tmquery.New(query)
q, err := tmquery.New(req.Query)
if err != nil {
return nil, err
}
@@ -86,7 +73,7 @@ func (env *Environment) TxSearch(
}
// sort results (must be done before pagination)
switch orderBy {
switch req.OrderBy {
case "desc", "":
sort.Slice(results, func(i, j int) bool {
if results[i].Height == results[j].Height {
@@ -107,9 +94,9 @@ func (env *Environment) TxSearch(
// paginate results
totalCount := len(results)
perPage := env.validatePerPage(perPagePtr)
perPage := env.validatePerPage(req.PerPage.IntPtr())
page, err := validatePage(pagePtr, perPage, totalCount)
page, err := validatePage(req.Page.IntPtr(), perPage, totalCount)
if err != nil {
return nil, err
}
@@ -122,7 +109,7 @@ func (env *Environment) TxSearch(
r := results[i]
var proof types.TxProof
if prove {
if req.Prove {
block := env.BlockStore.LoadBlock(r.Height)
proof = block.Data.Txs.Proof(int(r.Index))
}