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
4.4 KiB

7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
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 client_test
  2. import (
  3. "context"
  4. "fmt"
  5. "reflect"
  6. "testing"
  7. "time"
  8. "github.com/stretchr/testify/assert"
  9. "github.com/stretchr/testify/require"
  10. abci "github.com/tendermint/tendermint/abci/types"
  11. tmrand "github.com/tendermint/tendermint/libs/rand"
  12. "github.com/tendermint/tendermint/rpc/client"
  13. ctypes "github.com/tendermint/tendermint/rpc/core/types"
  14. "github.com/tendermint/tendermint/types"
  15. )
  16. var waitForEventTimeout = 8 * time.Second
  17. // MakeTxKV returns a text transaction, allong with expected key, value pair
  18. func MakeTxKV() ([]byte, []byte, []byte) {
  19. k := []byte(tmrand.Str(8))
  20. v := []byte(tmrand.Str(8))
  21. return k, v, append(k, append([]byte("="), v...)...)
  22. }
  23. func TestHeaderEvents(t *testing.T) {
  24. for i, c := range GetClients() {
  25. i, c := i, c // capture params
  26. t.Run(reflect.TypeOf(c).String(), func(t *testing.T) {
  27. // start for this test it if it wasn't already running
  28. if !c.IsRunning() {
  29. // if so, then we start it, listen, and stop it.
  30. err := c.Start()
  31. require.Nil(t, err, "%d: %+v", i, err)
  32. defer c.Stop()
  33. }
  34. evtTyp := types.EventNewBlockHeader
  35. evt, err := client.WaitForOneEvent(c, evtTyp, waitForEventTimeout)
  36. require.Nil(t, err, "%d: %+v", i, err)
  37. _, ok := evt.(types.EventDataNewBlockHeader)
  38. require.True(t, ok, "%d: %#v", i, evt)
  39. // TODO: more checks...
  40. })
  41. }
  42. }
  43. func TestBlockEvents(t *testing.T) {
  44. for i, c := range GetClients() {
  45. i, c := i, c // capture params
  46. t.Run(reflect.TypeOf(c).String(), func(t *testing.T) {
  47. // start for this test it if it wasn't already running
  48. if !c.IsRunning() {
  49. // if so, then we start it, listen, and stop it.
  50. err := c.Start()
  51. require.Nil(t, err, "%d: %+v", i, err)
  52. defer c.Stop()
  53. }
  54. // listen for a new block; ensure height increases by 1
  55. var firstBlockHeight int64
  56. for j := 0; j < 3; j++ {
  57. evtTyp := types.EventNewBlock
  58. evt, err := client.WaitForOneEvent(c, evtTyp, waitForEventTimeout)
  59. require.Nil(t, err, "%d: %+v", j, err)
  60. blockEvent, ok := evt.(types.EventDataNewBlock)
  61. require.True(t, ok, "%d: %#v", j, evt)
  62. block := blockEvent.Block
  63. if j == 0 {
  64. firstBlockHeight = block.Header.Height
  65. continue
  66. }
  67. require.Equal(t, block.Header.Height, firstBlockHeight+int64(j))
  68. }
  69. })
  70. }
  71. }
  72. func TestTxEventsSentWithBroadcastTxAsync(t *testing.T) { testTxEventsSent(t, "async") }
  73. func TestTxEventsSentWithBroadcastTxSync(t *testing.T) { testTxEventsSent(t, "sync") }
  74. func testTxEventsSent(t *testing.T, broadcastMethod string) {
  75. for i, c := range GetClients() {
  76. i, c := i, c // capture params
  77. t.Run(reflect.TypeOf(c).String(), func(t *testing.T) {
  78. // start for this test it if it wasn't already running
  79. if !c.IsRunning() {
  80. // if so, then we start it, listen, and stop it.
  81. err := c.Start()
  82. require.Nil(t, err, "%d: %+v", i, err)
  83. defer c.Stop()
  84. }
  85. // wait for the client subscription to get set up
  86. time.Sleep(100 * time.Millisecond)
  87. // make the tx
  88. _, _, tx := MakeTxKV()
  89. evtTyp := types.EventTx
  90. // send
  91. var (
  92. txres *ctypes.ResultBroadcastTx
  93. err error
  94. )
  95. switch broadcastMethod {
  96. case "async":
  97. txres, err = c.BroadcastTxAsync(tx)
  98. case "sync":
  99. txres, err = c.BroadcastTxSync(tx)
  100. default:
  101. panic(fmt.Sprintf("Unknown broadcastMethod %s", broadcastMethod))
  102. }
  103. require.NoError(t, err)
  104. require.Equal(t, txres.Code, abci.CodeTypeOK)
  105. // and wait for confirmation
  106. evt, err := client.WaitForOneEvent(c, evtTyp, waitForEventTimeout)
  107. require.Nil(t, err, "%d: %+v", i, err)
  108. // and make sure it has the proper info
  109. txe, ok := evt.(types.EventDataTx)
  110. require.True(t, ok, "%d: %#v", i, evt)
  111. // make sure this is the proper tx
  112. require.EqualValues(t, tx, txe.Tx)
  113. require.True(t, txe.Result.IsOK())
  114. })
  115. }
  116. }
  117. // Test HTTPClient resubscribes upon disconnect && subscription error.
  118. // Test Local client resubscribes upon subscription error.
  119. func TestClientsResubscribe(t *testing.T) {
  120. // TODO(melekes)
  121. }
  122. func TestHTTPReturnsErrorIfClientIsNotRunning(t *testing.T) {
  123. c := getHTTPClient()
  124. // on Subscribe
  125. _, err := c.Subscribe(context.Background(), "TestHeaderEvents",
  126. types.QueryForEvent(types.EventNewBlockHeader).String())
  127. assert.Error(t, err)
  128. // on Unsubscribe
  129. err = c.Unsubscribe(context.Background(), "TestHeaderEvents",
  130. types.QueryForEvent(types.EventNewBlockHeader).String())
  131. assert.Error(t, err)
  132. // on UnsubscribeAll
  133. err = c.UnsubscribeAll(context.Background(), "TestHeaderEvents")
  134. assert.Error(t, err)
  135. }