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.

167 lines
4.0 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>
5 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. "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. ProxyAppMempool proxy.AppConnMempool
  60. // interfaces defined in types and above
  61. StateDB dbm.DB
  62. BlockStore sm.BlockStore
  63. EvidencePool sm.EvidencePool
  64. ConsensusState Consensus
  65. P2PPeers peers
  66. P2PTransport transport
  67. // objects
  68. PubKey crypto.PubKey
  69. GenDoc *types.GenesisDoc // cache the genesis structure
  70. TxIndexer txindex.TxIndexer
  71. ConsensusReactor *consensus.Reactor
  72. EventBus *types.EventBus // thread safe
  73. Mempool mempl.Mempool
  74. Logger log.Logger
  75. Config cfg.RPCConfig
  76. }
  77. //----------------------------------------------
  78. func validatePage(pagePtr *int, perPage, totalCount int) (int, error) {
  79. if perPage < 1 {
  80. panic(fmt.Sprintf("zero or negative perPage: %d", perPage))
  81. }
  82. if pagePtr == nil { // no page parameter
  83. return 1, nil
  84. }
  85. pages := ((totalCount - 1) / perPage) + 1
  86. if pages == 0 {
  87. pages = 1 // one page (even if it's empty)
  88. }
  89. page := *pagePtr
  90. if page <= 0 || page > pages {
  91. return 1, fmt.Errorf("page should be within [1, %d] range, given %d", pages, page)
  92. }
  93. return page, nil
  94. }
  95. func validatePerPage(perPagePtr *int) int {
  96. if perPagePtr == nil { // no per_page parameter
  97. return defaultPerPage
  98. }
  99. perPage := *perPagePtr
  100. if perPage < 1 {
  101. return defaultPerPage
  102. } else if perPage > maxPerPage {
  103. return maxPerPage
  104. }
  105. return perPage
  106. }
  107. func validateSkipCount(page, perPage int) int {
  108. skipCount := (page - 1) * perPage
  109. if skipCount < 0 {
  110. return 0
  111. }
  112. return skipCount
  113. }
  114. // latestHeight can be either latest committed or uncommitted (+1) height.
  115. func getHeight(latestHeight int64, heightPtr *int64) (int64, error) {
  116. if heightPtr != nil {
  117. height := *heightPtr
  118. if height <= 0 {
  119. return 0, fmt.Errorf("height must be greater than 0, but got %d", height)
  120. }
  121. if height > latestHeight {
  122. return 0, fmt.Errorf("height %d must be less than or equal to the current blockchain height %d",
  123. height, latestHeight)
  124. }
  125. base := env.BlockStore.Base()
  126. if height < base {
  127. return 0, fmt.Errorf("height %v is not available, lowest height is %v",
  128. height, base)
  129. }
  130. return height, nil
  131. }
  132. return latestHeight, nil
  133. }
  134. func latestUncommittedHeight() int64 {
  135. nodeIsSyncing := env.ConsensusReactor.WaitSync()
  136. if nodeIsSyncing {
  137. return env.BlockStore.Height()
  138. }
  139. return env.BlockStore.Height() + 1
  140. }