Merge pull request #839 from tendermint/bugfix/pubsub-failures

Fix nondeterministic tests failures related to pubsub

consensus/reactor_test.go

@@ -296,7 +296,12 @@ func TestReactorWithTimeoutCommit(t *testing.T) {
 
 func waitForAndValidateBlock(t *testing.T, n int, activeVals map[string]struct{}, eventChans []chan interface{}, css []*ConsensusState, txs ...[]byte) {
 	timeoutWaitGroup(t, n, func(wg *sync.WaitGroup, j int) {
-		newBlockI := <-eventChans[j]
+		defer wg.Done()
+
+		newBlockI, ok := <-eventChans[j]
+		if !ok {
+			return
+		}
 		newBlock := newBlockI.(types.TMEventData).Unwrap().(types.EventDataNewBlock).Block
 		t.Logf("Got block height=%v validator=%v", newBlock.Height, j)
 		err := validateBlock(newBlock, activeVals)
@@ -308,16 +313,20 @@ func waitForAndValidateBlock(t *testing.T, n int, activeVals map[string]struct{}
 				t.Fatal(err)
 			}
 		}
-		wg.Done()
 	}, css)
 }
 
 func waitForBlockWithUpdatedValsAndValidateIt(t *testing.T, n int, updatedVals map[string]struct{}, eventChans []chan interface{}, css []*ConsensusState) {
 	timeoutWaitGroup(t, n, func(wg *sync.WaitGroup, j int) {
+		defer wg.Done()
+
 		var newBlock *types.Block
 	LOOP:
 		for {
-			newBlockI := <-eventChans[j]
+			newBlockI, ok := <-eventChans[j]
+			if !ok {
+				return
+			}
 			newBlock = newBlockI.(types.TMEventData).Unwrap().(types.EventDataNewBlock).Block
 			if newBlock.LastCommit.Size() == len(updatedVals) {
 				t.Logf("Block with new validators height=%v validator=%v", newBlock.Height, j)
@@ -331,8 +340,6 @@ func waitForBlockWithUpdatedValsAndValidateIt(t *testing.T, n int, updatedVals m
 		if err != nil {
 			t.Fatal(err)
 		}
-
-		wg.Done()
 	}, css)
 }