You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

133 lines
3.4 KiB

7 years ago
limit number of /subscribe clients and queries per client (#3269) * limit number of /subscribe clients and queries per client Add the following config variables (under [rpc] section): * max_subscription_clients * max_subscriptions_per_client * timeout_broadcast_tx_commit Fixes #2826 new HTTPClient interface for subscriptions finalize HTTPClient events interface remove EventSubscriber fix data race ``` WARNING: DATA RACE Read at 0x00c000a36060 by goroutine 129: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe.func1() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:168 +0x1f0 Previous write at 0x00c000a36060 by goroutine 132: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:191 +0x4e0 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 129 (running) created at: github.com/tendermint/tendermint/rpc/client.(*Local).Subscribe() /go/src/github.com/tendermint/tendermint/rpc/client/localclient.go:164 +0x4b7 github.com/tendermint/tendermint/rpc/client.WaitForOneEvent() /go/src/github.com/tendermint/tendermint/rpc/client/helpers.go:64 +0x178 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync.func1() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:139 +0x298 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Goroutine 132 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:878 +0x659 github.com/tendermint/tendermint/rpc/client_test.TestTxEventsSentWithBroadcastTxSync() /go/src/github.com/tendermint/tendermint/rpc/client/event_test.go:119 +0x186 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 ================== ``` lite client works (tested manually) godoc comments httpclient: do not close the out channel use TimeoutBroadcastTxCommit no timeout for unsubscribe but 1s Local (5s HTTP) timeout for resubscribe format code change Subscribe#out cap to 1 and replace config vars with RPCConfig TimeoutBroadcastTxCommit can't be greater than rpcserver.WriteTimeout rpc: Context as first parameter to all functions reformat code fixes after my own review fixes after Ethan's review add test stubs fix config.toml * fixes after manual testing - rpc: do not recommend to use BroadcastTxCommit because it's slow and wastes Tendermint resources (pubsub) - rpc: better error in Subscribe and BroadcastTxCommit - HTTPClient: do not resubscribe if err = ErrAlreadySubscribed * fixes after Ismail's review * Update rpc/grpc/grpc_test.go Co-Authored-By: melekes <anton.kalyaev@gmail.com>
6 years ago
  1. package core
  2. import (
  3. "errors"
  4. "fmt"
  5. "sort"
  6. tmmath "github.com/tendermint/tendermint/libs/math"
  7. tmquery "github.com/tendermint/tendermint/libs/pubsub/query"
  8. ctypes "github.com/tendermint/tendermint/rpc/core/types"
  9. rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
  10. "github.com/tendermint/tendermint/state/indexer/tx/null"
  11. "github.com/tendermint/tendermint/types"
  12. )
  13. // Tx allows you to query the transaction results. `nil` could mean the
  14. // transaction is in the mempool, invalidated, or was not sent in the first
  15. // place.
  16. // More: https://docs.tendermint.com/master/rpc/#/Info/tx
  17. func Tx(ctx *rpctypes.Context, hash []byte, prove bool) (*ctypes.ResultTx, error) {
  18. // if index is disabled, return error
  19. if _, ok := env.TxIndexer.(*null.TxIndex); ok {
  20. return nil, fmt.Errorf("transaction indexing is disabled")
  21. }
  22. r, err := env.TxIndexer.Get(hash)
  23. if err != nil {
  24. return nil, err
  25. }
  26. if r == nil {
  27. return nil, fmt.Errorf("tx (%X) not found", hash)
  28. }
  29. height := r.Height
  30. index := r.Index
  31. var proof types.TxProof
  32. if prove {
  33. block := env.BlockStore.LoadBlock(height)
  34. proof = block.Data.Txs.Proof(int(index)) // XXX: overflow on 32-bit machines
  35. }
  36. return &ctypes.ResultTx{
  37. Hash: hash,
  38. Height: height,
  39. Index: index,
  40. TxResult: r.Result,
  41. Tx: r.Tx,
  42. Proof: proof,
  43. }, nil
  44. }
  45. // TxSearch allows you to query for multiple transactions results. It returns a
  46. // list of transactions (maximum ?per_page entries) and the total count.
  47. // More: https://docs.tendermint.com/master/rpc/#/Info/tx_search
  48. func TxSearch(
  49. ctx *rpctypes.Context,
  50. query string,
  51. prove bool,
  52. pagePtr, perPagePtr *int,
  53. orderBy string,
  54. ) (*ctypes.ResultTxSearch, error) {
  55. // if index is disabled, return error
  56. if _, ok := env.TxIndexer.(*null.TxIndex); ok {
  57. return nil, errors.New("transaction indexing is disabled")
  58. }
  59. q, err := tmquery.New(query)
  60. if err != nil {
  61. return nil, err
  62. }
  63. results, err := env.TxIndexer.Search(ctx.Context(), q)
  64. if err != nil {
  65. return nil, err
  66. }
  67. // sort results (must be done before pagination)
  68. switch orderBy {
  69. case "desc", "":
  70. sort.Slice(results, func(i, j int) bool {
  71. if results[i].Height == results[j].Height {
  72. return results[i].Index > results[j].Index
  73. }
  74. return results[i].Height > results[j].Height
  75. })
  76. case "asc":
  77. sort.Slice(results, func(i, j int) bool {
  78. if results[i].Height == results[j].Height {
  79. return results[i].Index < results[j].Index
  80. }
  81. return results[i].Height < results[j].Height
  82. })
  83. default:
  84. return nil, fmt.Errorf("%w: expected order_by to be either `asc` or `desc` or empty", ctypes.ErrInvalidRequest)
  85. }
  86. // paginate results
  87. totalCount := len(results)
  88. perPage := validatePerPage(perPagePtr)
  89. page, err := validatePage(pagePtr, perPage, totalCount)
  90. if err != nil {
  91. return nil, err
  92. }
  93. skipCount := validateSkipCount(page, perPage)
  94. pageSize := tmmath.MinInt(perPage, totalCount-skipCount)
  95. apiResults := make([]*ctypes.ResultTx, 0, pageSize)
  96. for i := skipCount; i < skipCount+pageSize; i++ {
  97. r := results[i]
  98. var proof types.TxProof
  99. if prove {
  100. block := env.BlockStore.LoadBlock(r.Height)
  101. proof = block.Data.Txs.Proof(int(r.Index)) // XXX: overflow on 32-bit machines
  102. }
  103. apiResults = append(apiResults, &ctypes.ResultTx{
  104. Hash: types.Tx(r.Tx).Hash(),
  105. Height: r.Height,
  106. Index: r.Index,
  107. TxResult: r.Result,
  108. Tx: r.Tx,
  109. Proof: proof,
  110. })
  111. }
  112. return &ctypes.ResultTxSearch{Txs: apiResults, TotalCount: totalCount}, nil
  113. }