drain channels before Unsubscribe(All)

Read https://github.com/tendermint/tendermint/blob/55362ed76630f3e1ebec159a598f6a9fb5892cb1/libs/pubsub/pubsub.go#L13
for the detailed explanation of the issue.

We'll need to fix it someday. Make sure to keep an eye on
https://github.com/tendermint/tendermint/blob/master/docs/architecture/adr-033-pubsub.md

CHANGELOG_PENDING.md

@@ -31,3 +31,4 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
 - [p2p] \#2771 Fix `peer-id` label name in prometheus metrics
 - [abci] unlock mutex in localClient so even when app panics (e.g. during CheckTx), consensus continue working
 - [abci] fix DATA RACE in localClient
+- [rpc] drain channel before calling Unsubscribe(All) in /broadcast_tx_commit

consensus/replay_file.go

@@ -58,7 +58,12 @@ func (cs *ConsensusState) ReplayFile(file string, console bool) error {
 	if err != nil {
 		return errors.Errorf("failed to subscribe %s to %v", subscriber, types.EventQueryNewRoundStep)
 	}
-	defer cs.eventBus.Unsubscribe(ctx, subscriber, types.EventQueryNewRoundStep)
+	defer func() {
+		for range newStepCh {
+			// drain newStepCh to make sure we don't block
+		}
+		cs.eventBus.Unsubscribe(ctx, subscriber, types.EventQueryNewRoundStep)
+	}()
 
 	// just open the file for reading, no need to use wal
 	fp, err := os.OpenFile(file, os.O_RDONLY, 0600)
@@ -221,7 +226,12 @@ func (pb *playback) replayConsoleLoop() int {
 			if err != nil {
 				cmn.Exit(fmt.Sprintf("failed to subscribe %s to %v", subscriber, types.EventQueryNewRoundStep))
 			}
-			defer pb.cs.eventBus.Unsubscribe(ctx, subscriber, types.EventQueryNewRoundStep)
+			defer func() {
+				for range newStepCh {
+					// drain newStepCh to make sure we don't block
+				}
+				pb.cs.eventBus.Unsubscribe(ctx, subscriber, types.EventQueryNewRoundStep)
+			}()
 
 			if len(tokens) == 1 {
 				if err := pb.replayReset(1, newStepCh); err != nil {

rpc/client/helpers.go

@@ -69,7 +69,12 @@ func WaitForOneEvent(c EventsClient, evtTyp string, timeout time.Duration) (type
 	}
 
 	// make sure to unregister after the test is over
-	defer c.UnsubscribeAll(ctx, subscriber)
+	defer func() {
+		for range evts {
+			// drain evts to make sure we don't block
+		}
+		c.UnsubscribeAll(ctx, subscriber)
+	}()
 
 	select {
 	case evt := <-evts:

rpc/core/mempool.go

@@ -161,7 +161,12 @@ func BroadcastTxCommit(tx types.Tx) (*ctypes.ResultBroadcastTxCommit, error) {
 		logger.Error("Error on broadcastTxCommit", "err", err)
 		return nil, fmt.Errorf("Error on broadcastTxCommit: %v", err)
 	}
-	defer eventBus.Unsubscribe(context.Background(), "mempool", q)
+	defer func() {
+		for range deliverTxResCh {
+			// drain deliverTxResCh to make sure we don't block
+		}
+		eventBus.Unsubscribe(context.Background(), "mempool", q)
+	}()
 
 	// broadcast the tx and register checktx callback
 	checkTxResCh := make(chan *abci.Response, 1)