rpc: create buffered subscriptions on /subscribe (#4521)

Closes #3935

CHANGELOG_PENDING.md

@@ -24,6 +24,7 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 
 ### BUG FIXES:
 
+- [rpc] \#3935 Create buffered subscriptions on `/subscribe` (@melekes)
 - [rpc] [\#4493](https://github.com/tendermint/tendermint/pull/4493) Keep the original subscription "id" field when new RPCs come in (@michaelfig)
 - [rpc] [\#4437](https://github.com/tendermint/tendermint/pull/4437) Fix tx_search pagination with ordered results (@erikgrinaker)
 - [rpc] [\#4406](https://github.com/tendermint/tendermint/pull/4406) Fix issue with multiple subscriptions on the websocket (@antho1404)

rpc/core/README.md

@@ -5,3 +5,14 @@
 Requests that return multiple items will be paginated to 30 items by default.
 You can specify further pages with the ?page parameter. You can also set a
 custom page size up to 100 with the ?per_page parameter.
+
+## Subscribing to events
+
+The user can subscribe to events emitted by Tendermint, using `/subscribe`. If
+the maximum number of clients is reached or the client has too many
+subscriptions, an error will be returned. The subscription timeout is 5 sec.
+Each subscription has a buffer to accommodate short bursts of events or some
+slowness in clients. If the buffer gets full, the subscription will be canceled
+("client is not pulling messages fast enough"). If Tendermint exits, all
+subscriptions are canceled ("Tendermint exited"). The user can unsubscribe
+using either `/unsubscribe` or `/unsubscribe_all`.

rpc/core/events.go

@@ -12,6 +12,11 @@ import (
 	rpctypes "github.com/tendermint/tendermint/rpc/lib/types"
 )
 
+const (
+	// Buffer on the Tendermint (server) side to allow some slowness in clients.
+	subBufferSize = 100
+)
+
 // Subscribe for events via WebSocket.
 // More: https://docs.tendermint.com/master/rpc/#/Websocket/subscribe
 func Subscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultSubscribe, error) {
@@ -33,7 +38,7 @@ func Subscribe(ctx *rpctypes.Context, query string) (*ctypes.ResultSubscribe, er
 	subCtx, cancel := context.WithTimeout(ctx.Context(), SubscribeTimeout)
 	defer cancel()
 
-	sub, err := eventBus.Subscribe(subCtx, addr, q)
+	sub, err := eventBus.Subscribe(subCtx, addr, q, subBufferSize)
 	if err != nil {
 		return nil, err
 	}