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.

160 lines
3.8 KiB

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
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. "fmt"
  4. "time"
  5. dbm "github.com/tendermint/tm-db"
  6. cfg "github.com/tendermint/tendermint/config"
  7. "github.com/tendermint/tendermint/consensus"
  8. "github.com/tendermint/tendermint/crypto"
  9. "github.com/tendermint/tendermint/libs/log"
  10. mempl "github.com/tendermint/tendermint/mempool"
  11. "github.com/tendermint/tendermint/p2p"
  12. "github.com/tendermint/tendermint/proxy"
  13. sm "github.com/tendermint/tendermint/state"
  14. "github.com/tendermint/tendermint/state/txindex"
  15. "github.com/tendermint/tendermint/types"
  16. )
  17. const (
  18. // see README
  19. defaultPerPage = 30
  20. maxPerPage = 100
  21. // SubscribeTimeout is the maximum time we wait to subscribe for an event.
  22. // must be less than the server's write timeout (see rpcserver.DefaultConfig)
  23. SubscribeTimeout = 5 * time.Second
  24. )
  25. var (
  26. // set by Node
  27. env *Environment
  28. )
  29. // SetEnvironment sets up the given Environment.
  30. // It will race if multiple Node call SetEnvironment.
  31. func SetEnvironment(e *Environment) {
  32. env = e
  33. }
  34. //----------------------------------------------
  35. // These interfaces are used by RPC and must be thread safe
  36. type Consensus interface {
  37. GetState() sm.State
  38. GetValidators() (int64, []*types.Validator)
  39. GetLastHeight() int64
  40. GetRoundStateJSON() ([]byte, error)
  41. GetRoundStateSimpleJSON() ([]byte, error)
  42. }
  43. type transport interface {
  44. Listeners() []string
  45. IsListening() bool
  46. NodeInfo() p2p.NodeInfo
  47. }
  48. type peers interface {
  49. AddPersistentPeers([]string) error
  50. DialPeersAsync([]string) error
  51. Peers() p2p.IPeerSet
  52. }
  53. //----------------------------------------------
  54. // Environment contains objects and interfaces used by the RPC. It is expected
  55. // to be setup once during startup.
  56. type Environment struct {
  57. // external, thread safe interfaces
  58. ProxyAppQuery proxy.AppConnQuery
  59. // interfaces defined in types and above
  60. StateDB dbm.DB
  61. BlockStore sm.BlockStore
  62. EvidencePool sm.EvidencePool
  63. ConsensusState Consensus
  64. P2PPeers peers
  65. P2PTransport transport
  66. // objects
  67. PubKey crypto.PubKey
  68. GenDoc *types.GenesisDoc // cache the genesis structure
  69. TxIndexer txindex.TxIndexer
  70. ConsensusReactor *consensus.Reactor
  71. EventBus *types.EventBus // thread safe
  72. Mempool mempl.Mempool
  73. Logger log.Logger
  74. Config cfg.RPCConfig
  75. }
  76. //----------------------------------------------
  77. func validatePage(page, perPage, totalCount int) (int, error) {
  78. if perPage < 1 {
  79. panic(fmt.Sprintf("zero or negative perPage: %d", perPage))
  80. }
  81. if page == 0 {
  82. return 1, nil // default
  83. }
  84. pages := ((totalCount - 1) / perPage) + 1
  85. if pages == 0 {
  86. pages = 1 // one page (even if it's empty)
  87. }
  88. if page < 0 || page > pages {
  89. return 1, fmt.Errorf("page should be within [0, %d] range, given %d", pages, page)
  90. }
  91. return page, nil
  92. }
  93. func validatePerPage(perPage int) int {
  94. if perPage < 1 {
  95. return defaultPerPage
  96. } else if perPage > maxPerPage {
  97. return maxPerPage
  98. }
  99. return perPage
  100. }
  101. func validateSkipCount(page, perPage int) int {
  102. skipCount := (page - 1) * perPage
  103. if skipCount < 0 {
  104. return 0
  105. }
  106. return skipCount
  107. }
  108. // latestHeight can be either latest committed or uncommitted (+1) height.
  109. func getHeight(latestHeight int64, heightPtr *int64) (int64, error) {
  110. if heightPtr != nil {
  111. height := *heightPtr
  112. if height <= 0 {
  113. return 0, fmt.Errorf("height must be greater than 0, but got %d", height)
  114. }
  115. if height > latestHeight {
  116. return 0, fmt.Errorf("height %d must be less than or equal to the current blockchain height %d",
  117. height, latestHeight)
  118. }
  119. base := env.BlockStore.Base()
  120. if height < base {
  121. return 0, fmt.Errorf("height %v is not available, blocks pruned at height %v",
  122. height, base)
  123. }
  124. return height, nil
  125. }
  126. return latestHeight, nil
  127. }
  128. func latestUncommittedHeight() int64 {
  129. nodeIsSyncing := env.ConsensusReactor.WaitSync()
  130. if nodeIsSyncing {
  131. return env.BlockStore.Height()
  132. }
  133. return env.BlockStore.Height() + 1
  134. }