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.

273 lines
9.4 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
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. {`{"method": "c", "id": "0", "params": a}`, "invalid character", types.JSONRPCStringID("")}, // id not captured in JSON parsing failures
  49. {`{"method": "c", "id": "0", "params": ["a"]}`, "got 1", types.JSONRPCStringID("0")},
  50. {`{"method": "c", "id": "0", "params": ["a", "b"]}`, "invalid character", types.JSONRPCStringID("0")},
  51. {`{"method": "c", "id": "0", "params": [1, 1]}`, "of type string", types.JSONRPCStringID("0")},
  52. // good
  53. {`{"jsonrpc": "2.0", "method": "c", "id": "0", "params": null}`, "", types.JSONRPCStringID("0")},
  54. {`{"method": "c", "id": "0", "params": {}}`, "", types.JSONRPCStringID("0")},
  55. {`{"method": "c", "id": "0", "params": ["a", "10"]}`, "", types.JSONRPCStringID("0")},
  56. }
  57. for i, tt := range tests {
  58. req, _ := http.NewRequest("POST", "http://localhost/", strings.NewReader(tt.payload))
  59. rec := httptest.NewRecorder()
  60. mux.ServeHTTP(rec, req)
  61. res := rec.Result()
  62. // Always expecting back a JSONRPCResponse
  63. assert.True(t, statusOK(res.StatusCode), "#%d: should always return 2XX", i)
  64. blob, err := ioutil.ReadAll(res.Body)
  65. if err != nil {
  66. t.Errorf("#%d: err reading body: %v", i, err)
  67. continue
  68. }
  69. recv := new(types.RPCResponse)
  70. assert.Nil(t, json.Unmarshal(blob, recv), "#%d: expecting successful parsing of an RPCResponse:\nblob: %s", i, blob)
  71. assert.NotEqual(t, recv, new(types.RPCResponse), "#%d: not expecting a blank RPCResponse", i)
  72. assert.Equal(t, tt.expectedId, recv.ID, "#%d: expected ID not matched in RPCResponse", i)
  73. if tt.wantErr == "" {
  74. assert.Nil(t, recv.Error, "#%d: not expecting an error", i)
  75. } else {
  76. assert.True(t, recv.Error.Code < 0, "#%d: not expecting a positive JSONRPC code", i)
  77. // The wanted error is either in the message or the data
  78. assert.Contains(t, recv.Error.Message+recv.Error.Data, tt.wantErr, "#%d: expected substring", i)
  79. }
  80. }
  81. }
  82. func TestJSONRPCID(t *testing.T) {
  83. mux := testMux()
  84. tests := []struct {
  85. payload string
  86. wantErr bool
  87. expectedId interface{}
  88. }{
  89. // good id
  90. {`{"jsonrpc": "2.0", "method": "c", "id": "0", "params": ["a", "10"]}`, false, types.JSONRPCStringID("0")},
  91. {`{"jsonrpc": "2.0", "method": "c", "id": "abc", "params": ["a", "10"]}`, false, types.JSONRPCStringID("abc")},
  92. {`{"jsonrpc": "2.0", "method": "c", "id": 0, "params": ["a", "10"]}`, false, types.JSONRPCIntID(0)},
  93. {`{"jsonrpc": "2.0", "method": "c", "id": 1, "params": ["a", "10"]}`, false, types.JSONRPCIntID(1)},
  94. {`{"jsonrpc": "2.0", "method": "c", "id": 1.3, "params": ["a", "10"]}`, false, types.JSONRPCIntID(1)},
  95. {`{"jsonrpc": "2.0", "method": "c", "id": -1, "params": ["a", "10"]}`, false, types.JSONRPCIntID(-1)},
  96. {`{"jsonrpc": "2.0", "method": "c", "id": null, "params": ["a", "10"]}`, false, nil},
  97. // bad id
  98. {`{"jsonrpc": "2.0", "method": "c", "id": {}, "params": ["a", "10"]}`, true, nil},
  99. {`{"jsonrpc": "2.0", "method": "c", "id": [], "params": ["a", "10"]}`, true, nil},
  100. }
  101. for i, tt := range tests {
  102. req, _ := http.NewRequest("POST", "http://localhost/", strings.NewReader(tt.payload))
  103. rec := httptest.NewRecorder()
  104. mux.ServeHTTP(rec, req)
  105. res := rec.Result()
  106. // Always expecting back a JSONRPCResponse
  107. assert.True(t, statusOK(res.StatusCode), "#%d: should always return 2XX", i)
  108. blob, err := ioutil.ReadAll(res.Body)
  109. if err != nil {
  110. t.Errorf("#%d: err reading body: %v", i, err)
  111. continue
  112. }
  113. recv := new(types.RPCResponse)
  114. err = json.Unmarshal(blob, recv)
  115. assert.Nil(t, err, "#%d: expecting successful parsing of an RPCResponse:\nblob: %s", i, blob)
  116. if !tt.wantErr {
  117. assert.NotEqual(t, recv, new(types.RPCResponse), "#%d: not expecting a blank RPCResponse", i)
  118. assert.Equal(t, tt.expectedId, recv.ID, "#%d: expected ID not matched in RPCResponse", i)
  119. assert.Nil(t, recv.Error, "#%d: not expecting an error", i)
  120. } else {
  121. assert.True(t, recv.Error.Code < 0, "#%d: not expecting a positive JSONRPC code", i)
  122. }
  123. }
  124. }
  125. func TestRPCNotification(t *testing.T) {
  126. mux := testMux()
  127. body := strings.NewReader(`{"jsonrpc": "2.0", "id": ""}`)
  128. req, _ := http.NewRequest("POST", "http://localhost/", body)
  129. rec := httptest.NewRecorder()
  130. mux.ServeHTTP(rec, req)
  131. res := rec.Result()
  132. // Always expecting back a JSONRPCResponse
  133. require.True(t, statusOK(res.StatusCode), "should always return 2XX")
  134. blob, err := ioutil.ReadAll(res.Body)
  135. require.Nil(t, err, "reading from the body should not give back an error")
  136. require.Equal(t, len(blob), 0, "a notification SHOULD NOT be responded to by the server")
  137. }
  138. func TestRPCNotificationInBatch(t *testing.T) {
  139. mux := testMux()
  140. tests := []struct {
  141. payload string
  142. expectCount int
  143. }{
  144. {
  145. `[
  146. {"jsonrpc": "2.0","id": ""},
  147. {"jsonrpc": "2.0","method":"c","id":"abc","params":["a","10"]}
  148. ]`,
  149. 1,
  150. },
  151. {
  152. `[
  153. {"jsonrpc": "2.0","id": ""},
  154. {"jsonrpc": "2.0","method":"c","id":"abc","params":["a","10"]},
  155. {"jsonrpc": "2.0","id": ""},
  156. {"jsonrpc": "2.0","method":"c","id":"abc","params":["a","10"]}
  157. ]`,
  158. 2,
  159. },
  160. }
  161. for i, tt := range tests {
  162. req, _ := http.NewRequest("POST", "http://localhost/", strings.NewReader(tt.payload))
  163. rec := httptest.NewRecorder()
  164. mux.ServeHTTP(rec, req)
  165. res := rec.Result()
  166. // Always expecting back a JSONRPCResponse
  167. assert.True(t, statusOK(res.StatusCode), "#%d: should always return 2XX", i)
  168. blob, err := ioutil.ReadAll(res.Body)
  169. if err != nil {
  170. t.Errorf("#%d: err reading body: %v", i, err)
  171. continue
  172. }
  173. var responses []types.RPCResponse
  174. // try to unmarshal an array first
  175. err = json.Unmarshal(blob, &responses)
  176. if err != nil {
  177. // if we were actually expecting an array, but got an error
  178. if tt.expectCount > 1 {
  179. t.Errorf("#%d: expected an array, couldn't unmarshal it\nblob: %s", i, blob)
  180. continue
  181. } else {
  182. // we were expecting an error here, so let's unmarshal a single response
  183. var response types.RPCResponse
  184. err = json.Unmarshal(blob, &response)
  185. if err != nil {
  186. t.Errorf("#%d: expected successful parsing of an RPCResponse\nblob: %s", i, blob)
  187. continue
  188. }
  189. // have a single-element result
  190. responses = []types.RPCResponse{response}
  191. }
  192. }
  193. if tt.expectCount != len(responses) {
  194. t.Errorf("#%d: expected %d response(s), but got %d\nblob: %s", i, tt.expectCount, len(responses), blob)
  195. continue
  196. }
  197. for _, response := range responses {
  198. assert.NotEqual(t, response, new(types.RPCResponse), "#%d: not expecting a blank RPCResponse", i)
  199. }
  200. }
  201. }
  202. func TestUnknownRPCPath(t *testing.T) {
  203. mux := testMux()
  204. req, _ := http.NewRequest("GET", "http://localhost/unknownrpcpath", nil)
  205. rec := httptest.NewRecorder()
  206. mux.ServeHTTP(rec, req)
  207. res := rec.Result()
  208. // Always expecting back a 404 error
  209. require.Equal(t, http.StatusNotFound, res.StatusCode, "should always return 404")
  210. }
  211. //////////////////////////////////////////////////////////////////////////////
  212. // JSON-RPC over WEBSOCKETS
  213. func TestWebsocketManagerHandler(t *testing.T) {
  214. s := newWSServer()
  215. defer s.Close()
  216. // check upgrader works
  217. d := websocket.Dialer{}
  218. c, dialResp, err := d.Dial("ws://"+s.Listener.Addr().String()+"/websocket", nil)
  219. require.NoError(t, err)
  220. if got, want := dialResp.StatusCode, http.StatusSwitchingProtocols; got != want {
  221. t.Errorf("dialResp.StatusCode = %q, want %q", got, want)
  222. }
  223. // check basic functionality works
  224. req, err := types.MapToRequest(amino.NewCodec(), types.JSONRPCStringID("TestWebsocketManager"), "c", map[string]interface{}{"s": "a", "i": 10})
  225. require.NoError(t, err)
  226. err = c.WriteJSON(req)
  227. require.NoError(t, err)
  228. var resp types.RPCResponse
  229. err = c.ReadJSON(&resp)
  230. require.NoError(t, err)
  231. require.Nil(t, resp.Error)
  232. }
  233. func newWSServer() *httptest.Server {
  234. funcMap := map[string]*rs.RPCFunc{
  235. "c": rs.NewWSRPCFunc(func(ctx *types.Context, s string, i int) (string, error) { return "foo", nil }, "s,i"),
  236. }
  237. wm := rs.NewWebsocketManager(funcMap, amino.NewCodec())
  238. wm.SetLogger(log.TestingLogger())
  239. mux := http.NewServeMux()
  240. mux.HandleFunc("/websocket", wm.WebsocketHandler)
  241. return httptest.NewServer(mux)
  242. }