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.

285 lines
9.5 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
rpc: add support for batched requests/responses (#3534) Continues from #3280 in building support for batched requests/responses in the JSON RPC (as per issue #3213). * Add JSON RPC batching for client and server As per #3213, this adds support for [JSON RPC batch requests and responses](https://www.jsonrpc.org/specification#batch). * Add additional checks to ensure client responses are the same as results * Fix case where a notification is sent and no response is expected * Add test to check that JSON RPC notifications in a batch are left out in responses * Update CHANGELOG_PENDING.md * Update PR number now that PR has been created * Make errors start with lowercase letter * Refactor batch functionality to be standalone This refactors the batching functionality to rather act in a standalone way. In light of supporting concurrent goroutines making use of the same client, it would make sense to have batching functionality where one could create a batch of requests per goroutine and send that batch without interfering with a batch from another goroutine. * Add examples for simple and batch HTTP client usage * Check errors from writer and remove nolinter directives * Make error strings start with lowercase letter * Refactor examples to make them testable * Use safer deferred shutdown for example Tendermint test node * Recompose rpcClient interface from pre-existing interface components * Rename WaitGroup for brevity * Replace empty ID string with request ID * Remove extraneous test case * Convert first letter of errors.Wrap() messages to lowercase * Remove extraneous function parameter * Make variable declaration terse * Reorder WaitGroup.Done call to help prevent race conditions in the face of failure * Swap mutex to value representation and remove initialization * Restore empty JSONRPC string ID in response to prevent nil * Make JSONRPCBufferedRequest private * Revert PR hard link in CHANGELOG_PENDING * Add client ID for JSONRPCClient This adds code to automatically generate a randomized client ID for the JSONRPCClient, and adds a check of the IDs in the responses (if one was set in the requests). * Extract response ID validation into separate function * Remove extraneous comments * Reorder fields to indicate clearly which are protected by the mutex * Refactor for loop to remove indexing * Restructure and combine loop * Flatten conditional block for better readability * Make multi-variable declaration slightly more readable * Change for loop style * Compress error check statements * Make function description more generic to show that we support different protocols * Preallocate memory for request and result objects
6 years ago
rpc: add support for batched requests/responses (#3534) Continues from #3280 in building support for batched requests/responses in the JSON RPC (as per issue #3213). * Add JSON RPC batching for client and server As per #3213, this adds support for [JSON RPC batch requests and responses](https://www.jsonrpc.org/specification#batch). * Add additional checks to ensure client responses are the same as results * Fix case where a notification is sent and no response is expected * Add test to check that JSON RPC notifications in a batch are left out in responses * Update CHANGELOG_PENDING.md * Update PR number now that PR has been created * Make errors start with lowercase letter * Refactor batch functionality to be standalone This refactors the batching functionality to rather act in a standalone way. In light of supporting concurrent goroutines making use of the same client, it would make sense to have batching functionality where one could create a batch of requests per goroutine and send that batch without interfering with a batch from another goroutine. * Add examples for simple and batch HTTP client usage * Check errors from writer and remove nolinter directives * Make error strings start with lowercase letter * Refactor examples to make them testable * Use safer deferred shutdown for example Tendermint test node * Recompose rpcClient interface from pre-existing interface components * Rename WaitGroup for brevity * Replace empty ID string with request ID * Remove extraneous test case * Convert first letter of errors.Wrap() messages to lowercase * Remove extraneous function parameter * Make variable declaration terse * Reorder WaitGroup.Done call to help prevent race conditions in the face of failure * Swap mutex to value representation and remove initialization * Restore empty JSONRPC string ID in response to prevent nil * Make JSONRPCBufferedRequest private * Revert PR hard link in CHANGELOG_PENDING * Add client ID for JSONRPCClient This adds code to automatically generate a randomized client ID for the JSONRPCClient, and adds a check of the IDs in the responses (if one was set in the requests). * Extract response ID validation into separate function * Remove extraneous comments * Reorder fields to indicate clearly which are protected by the mutex * Refactor for loop to remove indexing * Restructure and combine loop * Flatten conditional block for better readability * Make multi-variable declaration slightly more readable * Change for loop style * Compress error check statements * Make function description more generic to show that we support different protocols * Preallocate memory for request and result objects
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 rpcserver_test
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "io/ioutil"
  6. "net/http"
  7. "net/http/httptest"
  8. "strings"
  9. "testing"
  10. "github.com/gorilla/websocket"
  11. "github.com/stretchr/testify/assert"
  12. "github.com/stretchr/testify/require"
  13. amino "github.com/tendermint/go-amino"
  14. "github.com/tendermint/tendermint/libs/log"
  15. rs "github.com/tendermint/tendermint/rpc/lib/server"
  16. types "github.com/tendermint/tendermint/rpc/lib/types"
  17. )
  18. //////////////////////////////////////////////////////////////////////////////
  19. // HTTP REST API
  20. // TODO
  21. //////////////////////////////////////////////////////////////////////////////
  22. // JSON-RPC over HTTP
  23. func testMux() *http.ServeMux {
  24. funcMap := map[string]*rs.RPCFunc{
  25. "c": rs.NewRPCFunc(func(ctx *types.Context, s string, i int) (string, error) { return "foo", nil }, "s,i"),
  26. }
  27. cdc := amino.NewCodec()
  28. mux := http.NewServeMux()
  29. buf := new(bytes.Buffer)
  30. logger := log.NewTMLogger(buf)
  31. rs.RegisterRPCFuncs(mux, funcMap, cdc, logger)
  32. return mux
  33. }
  34. func statusOK(code int) bool { return code >= 200 && code <= 299 }
  35. // Ensure that nefarious/unintended inputs to `params`
  36. // do not crash our RPC handlers.
  37. // See Issue https://github.com/tendermint/tendermint/issues/708.
  38. func TestRPCParams(t *testing.T) {
  39. mux := testMux()
  40. tests := []struct {
  41. payload string
  42. wantErr string
  43. expectedId interface{}
  44. }{
  45. // bad
  46. {`{"jsonrpc": "2.0", "id": "0"}`, "Method not found", types.JSONRPCStringID("0")},
  47. {`{"jsonrpc": "2.0", "method": "y", "id": "0"}`, "Method not found", types.JSONRPCStringID("0")},
  48. // id not captured in JSON parsing failures
  49. {`{"method": "c", "id": "0", "params": a}`, "invalid character", types.JSONRPCStringID("")},
  50. {`{"method": "c", "id": "0", "params": ["a"]}`, "got 1", types.JSONRPCStringID("0")},
  51. {`{"method": "c", "id": "0", "params": ["a", "b"]}`, "invalid character", types.JSONRPCStringID("0")},
  52. {`{"method": "c", "id": "0", "params": [1, 1]}`, "of type string", types.JSONRPCStringID("0")},
  53. // good
  54. {`{"jsonrpc": "2.0", "method": "c", "id": "0", "params": null}`, "", types.JSONRPCStringID("0")},
  55. {`{"method": "c", "id": "0", "params": {}}`, "", types.JSONRPCStringID("0")},
  56. {`{"method": "c", "id": "0", "params": ["a", "10"]}`, "", types.JSONRPCStringID("0")},
  57. }
  58. for i, tt := range tests {
  59. req, _ := http.NewRequest("POST", "http://localhost/", strings.NewReader(tt.payload))
  60. rec := httptest.NewRecorder()
  61. mux.ServeHTTP(rec, req)
  62. res := rec.Result()
  63. // Always expecting back a JSONRPCResponse
  64. assert.True(t, statusOK(res.StatusCode), "#%d: should always return 2XX", i)
  65. blob, err := ioutil.ReadAll(res.Body)
  66. if err != nil {
  67. t.Errorf("#%d: err reading body: %v", i, err)
  68. continue
  69. }
  70. res.Body.Close()
  71. recv := new(types.RPCResponse)
  72. assert.Nil(t, json.Unmarshal(blob, recv), "#%d: expecting successful parsing of an RPCResponse:\nblob: %s", i, blob)
  73. assert.NotEqual(t, recv, new(types.RPCResponse), "#%d: not expecting a blank RPCResponse", i)
  74. assert.Equal(t, tt.expectedId, recv.ID, "#%d: expected ID not matched in RPCResponse", i)
  75. if tt.wantErr == "" {
  76. assert.Nil(t, recv.Error, "#%d: not expecting an error", i)
  77. } else {
  78. assert.True(t, recv.Error.Code < 0, "#%d: not expecting a positive JSONRPC code", i)
  79. // The wanted error is either in the message or the data
  80. assert.Contains(t, recv.Error.Message+recv.Error.Data, tt.wantErr, "#%d: expected substring", i)
  81. }
  82. }
  83. }
  84. func TestJSONRPCID(t *testing.T) {
  85. mux := testMux()
  86. tests := []struct {
  87. payload string
  88. wantErr bool
  89. expectedId interface{}
  90. }{
  91. // good id
  92. {`{"jsonrpc": "2.0", "method": "c", "id": "0", "params": ["a", "10"]}`, false, types.JSONRPCStringID("0")},
  93. {`{"jsonrpc": "2.0", "method": "c", "id": "abc", "params": ["a", "10"]}`, false, types.JSONRPCStringID("abc")},
  94. {`{"jsonrpc": "2.0", "method": "c", "id": 0, "params": ["a", "10"]}`, false, types.JSONRPCIntID(0)},
  95. {`{"jsonrpc": "2.0", "method": "c", "id": 1, "params": ["a", "10"]}`, false, types.JSONRPCIntID(1)},
  96. {`{"jsonrpc": "2.0", "method": "c", "id": 1.3, "params": ["a", "10"]}`, false, types.JSONRPCIntID(1)},
  97. {`{"jsonrpc": "2.0", "method": "c", "id": -1, "params": ["a", "10"]}`, false, types.JSONRPCIntID(-1)},
  98. {`{"jsonrpc": "2.0", "method": "c", "id": null, "params": ["a", "10"]}`, false, nil},
  99. // bad id
  100. {`{"jsonrpc": "2.0", "method": "c", "id": {}, "params": ["a", "10"]}`, true, nil},
  101. {`{"jsonrpc": "2.0", "method": "c", "id": [], "params": ["a", "10"]}`, true, nil},
  102. }
  103. for i, tt := range tests {
  104. req, _ := http.NewRequest("POST", "http://localhost/", strings.NewReader(tt.payload))
  105. rec := httptest.NewRecorder()
  106. mux.ServeHTTP(rec, req)
  107. res := rec.Result()
  108. // Always expecting back a JSONRPCResponse
  109. assert.True(t, statusOK(res.StatusCode), "#%d: should always return 2XX", i)
  110. blob, err := ioutil.ReadAll(res.Body)
  111. if err != nil {
  112. t.Errorf("#%d: err reading body: %v", i, err)
  113. continue
  114. }
  115. res.Body.Close()
  116. recv := new(types.RPCResponse)
  117. err = json.Unmarshal(blob, recv)
  118. assert.Nil(t, err, "#%d: expecting successful parsing of an RPCResponse:\nblob: %s", i, blob)
  119. if !tt.wantErr {
  120. assert.NotEqual(t, recv, new(types.RPCResponse), "#%d: not expecting a blank RPCResponse", i)
  121. assert.Equal(t, tt.expectedId, recv.ID, "#%d: expected ID not matched in RPCResponse", i)
  122. assert.Nil(t, recv.Error, "#%d: not expecting an error", i)
  123. } else {
  124. assert.True(t, recv.Error.Code < 0, "#%d: not expecting a positive JSONRPC code", i)
  125. }
  126. }
  127. }
  128. func TestRPCNotification(t *testing.T) {
  129. mux := testMux()
  130. body := strings.NewReader(`{"jsonrpc": "2.0", "id": ""}`)
  131. req, _ := http.NewRequest("POST", "http://localhost/", body)
  132. rec := httptest.NewRecorder()
  133. mux.ServeHTTP(rec, req)
  134. res := rec.Result()
  135. // Always expecting back a JSONRPCResponse
  136. require.True(t, statusOK(res.StatusCode), "should always return 2XX")
  137. blob, err := ioutil.ReadAll(res.Body)
  138. res.Body.Close()
  139. require.Nil(t, err, "reading from the body should not give back an error")
  140. require.Equal(t, len(blob), 0, "a notification SHOULD NOT be responded to by the server")
  141. }
  142. func TestRPCNotificationInBatch(t *testing.T) {
  143. mux := testMux()
  144. tests := []struct {
  145. payload string
  146. expectCount int
  147. }{
  148. {
  149. `[
  150. {"jsonrpc": "2.0","id": ""},
  151. {"jsonrpc": "2.0","method":"c","id":"abc","params":["a","10"]}
  152. ]`,
  153. 1,
  154. },
  155. {
  156. `[
  157. {"jsonrpc": "2.0","id": ""},
  158. {"jsonrpc": "2.0","method":"c","id":"abc","params":["a","10"]},
  159. {"jsonrpc": "2.0","id": ""},
  160. {"jsonrpc": "2.0","method":"c","id":"abc","params":["a","10"]}
  161. ]`,
  162. 2,
  163. },
  164. }
  165. for i, tt := range tests {
  166. req, _ := http.NewRequest("POST", "http://localhost/", strings.NewReader(tt.payload))
  167. rec := httptest.NewRecorder()
  168. mux.ServeHTTP(rec, req)
  169. res := rec.Result()
  170. // Always expecting back a JSONRPCResponse
  171. assert.True(t, statusOK(res.StatusCode), "#%d: should always return 2XX", i)
  172. blob, err := ioutil.ReadAll(res.Body)
  173. if err != nil {
  174. t.Errorf("#%d: err reading body: %v", i, err)
  175. continue
  176. }
  177. res.Body.Close()
  178. var responses []types.RPCResponse
  179. // try to unmarshal an array first
  180. err = json.Unmarshal(blob, &responses)
  181. if err != nil {
  182. // if we were actually expecting an array, but got an error
  183. if tt.expectCount > 1 {
  184. t.Errorf("#%d: expected an array, couldn't unmarshal it\nblob: %s", i, blob)
  185. continue
  186. } else {
  187. // we were expecting an error here, so let's unmarshal a single response
  188. var response types.RPCResponse
  189. err = json.Unmarshal(blob, &response)
  190. if err != nil {
  191. t.Errorf("#%d: expected successful parsing of an RPCResponse\nblob: %s", i, blob)
  192. continue
  193. }
  194. // have a single-element result
  195. responses = []types.RPCResponse{response}
  196. }
  197. }
  198. if tt.expectCount != len(responses) {
  199. t.Errorf("#%d: expected %d response(s), but got %d\nblob: %s", i, tt.expectCount, len(responses), blob)
  200. continue
  201. }
  202. for _, response := range responses {
  203. assert.NotEqual(t, response, new(types.RPCResponse), "#%d: not expecting a blank RPCResponse", i)
  204. }
  205. }
  206. }
  207. func TestUnknownRPCPath(t *testing.T) {
  208. mux := testMux()
  209. req, _ := http.NewRequest("GET", "http://localhost/unknownrpcpath", nil)
  210. rec := httptest.NewRecorder()
  211. mux.ServeHTTP(rec, req)
  212. res := rec.Result()
  213. // Always expecting back a 404 error
  214. require.Equal(t, http.StatusNotFound, res.StatusCode, "should always return 404")
  215. res.Body.Close()
  216. }
  217. //////////////////////////////////////////////////////////////////////////////
  218. // JSON-RPC over WEBSOCKETS
  219. func TestWebsocketManagerHandler(t *testing.T) {
  220. s := newWSServer()
  221. defer s.Close()
  222. // check upgrader works
  223. d := websocket.Dialer{}
  224. c, dialResp, err := d.Dial("ws://"+s.Listener.Addr().String()+"/websocket", nil)
  225. require.NoError(t, err)
  226. if got, want := dialResp.StatusCode, http.StatusSwitchingProtocols; got != want {
  227. t.Errorf("dialResp.StatusCode = %q, want %q", got, want)
  228. }
  229. // check basic functionality works
  230. req, err := types.MapToRequest(
  231. amino.NewCodec(),
  232. types.JSONRPCStringID("TestWebsocketManager"),
  233. "c",
  234. map[string]interface{}{"s": "a", "i": 10},
  235. )
  236. require.NoError(t, err)
  237. err = c.WriteJSON(req)
  238. require.NoError(t, err)
  239. var resp types.RPCResponse
  240. err = c.ReadJSON(&resp)
  241. require.NoError(t, err)
  242. require.Nil(t, resp.Error)
  243. dialResp.Body.Close()
  244. }
  245. func newWSServer() *httptest.Server {
  246. funcMap := map[string]*rs.RPCFunc{
  247. "c": rs.NewWSRPCFunc(func(ctx *types.Context, s string, i int) (string, error) { return "foo", nil }, "s,i"),
  248. }
  249. wm := rs.NewWebsocketManager(funcMap, amino.NewCodec())
  250. wm.SetLogger(log.TestingLogger())
  251. mux := http.NewServeMux()
  252. mux.HandleFunc("/websocket", wm.WebsocketHandler)
  253. return httptest.NewServer(mux)
  254. }