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.

125 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>
5 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/txindex/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(ctx *rpctypes.Context, query string, prove bool, pagePtr, perPagePtr *int, orderBy string) (
  49. *ctypes.ResultTxSearch, error) {
  50. // if index is disabled, return error
  51. if _, ok := env.TxIndexer.(*null.TxIndex); ok {
  52. return nil, errors.New("transaction indexing is disabled")
  53. }
  54. q, err := tmquery.New(query)
  55. if err != nil {
  56. return nil, err
  57. }
  58. results, err := env.TxIndexer.Search(ctx.Context(), q)
  59. if err != nil {
  60. return nil, err
  61. }
  62. // sort results (must be done before pagination)
  63. switch orderBy {
  64. case "desc":
  65. sort.Slice(results, func(i, j int) bool {
  66. if results[i].Height == results[j].Height {
  67. return results[i].Index > results[j].Index
  68. }
  69. return results[i].Height > results[j].Height
  70. })
  71. case "asc", "":
  72. sort.Slice(results, func(i, j int) bool {
  73. if results[i].Height == results[j].Height {
  74. return results[i].Index < results[j].Index
  75. }
  76. return results[i].Height < results[j].Height
  77. })
  78. default:
  79. return nil, errors.New("expected order_by to be either `asc` or `desc` or empty")
  80. }
  81. // paginate results
  82. totalCount := len(results)
  83. perPage := validatePerPage(perPagePtr)
  84. page, err := validatePage(pagePtr, perPage, totalCount)
  85. if err != nil {
  86. return nil, err
  87. }
  88. skipCount := validateSkipCount(page, perPage)
  89. pageSize := tmmath.MinInt(perPage, totalCount-skipCount)
  90. apiResults := make([]*ctypes.ResultTx, 0, pageSize)
  91. for i := skipCount; i < skipCount+pageSize; i++ {
  92. r := results[i]
  93. var proof types.TxProof
  94. if prove {
  95. block := env.BlockStore.LoadBlock(r.Height)
  96. proof = block.Data.Txs.Proof(int(r.Index)) // XXX: overflow on 32-bit machines
  97. }
  98. apiResults = append(apiResults, &ctypes.ResultTx{
  99. Hash: types.Tx(r.Tx).Hash(),
  100. Height: r.Height,
  101. Index: r.Index,
  102. TxResult: r.Result,
  103. Tx: r.Tx,
  104. Proof: proof,
  105. })
  106. }
  107. return &ctypes.ResultTxSearch{Txs: apiResults, TotalCount: totalCount}, nil
  108. }