fix(mockpv): detect and fix data-race in MockPV (#262)

* fix: extend MockPV synchronization for several exportable methods
* fix: race condition in TestReactorValidatorSetChanges

internal/consensus/reactor_test.go

@@ -11,6 +11,7 @@ import (
  "time"
 
  "github.com/fortytw2/leaktest"
+ "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/mock"
  "github.com/stretchr/testify/require"
  dbm "github.com/tendermint/tm-db"
@@ -603,7 +604,7 @@ func TestReactorValidatorSetChanges(t *testing.T) {
  // it includes the commit for block 4, which should have the updated validator set
  waitForBlockWithUpdatedValsAndValidateIt(t, nPeers, addOneVal.quorumHash, blocksSubs, states)
 
- validate(states)
+ validate(t, states)
 
  waitForAndValidateBlock(t, nPeers, activeVals, blocksSubs, states, addTwoVals.txs...)
  waitForAndValidateBlockWithTx(t, nPeers, activeVals, blocksSubs, states, addTwoVals.txs...)
@@ -614,15 +615,15 @@ func TestReactorValidatorSetChanges(t *testing.T) {
 
  waitForBlockWithUpdatedValsAndValidateIt(t, nPeers, addTwoVals.quorumHash, blocksSubs, states)
 
- validate(states)
+ validate(t, states)
 
  waitForAndValidateBlock(t, nPeers, activeVals, blocksSubs, states, removeTwoVals.txs...)
  waitForAndValidateBlockWithTx(t, nPeers, activeVals, blocksSubs, states, removeTwoVals.txs...)
  waitForAndValidateBlock(t, nPeers, activeVals, blocksSubs, states)
 
  waitForBlockWithUpdatedValsAndValidateIt(t, nPeers, removeTwoVals.quorumHash, blocksSubs, states)
 
- validate(states)
+ validate(t, states)
 }
 
 func makeProTxHashMap(proTxHashes []crypto.ProTxHash) map[string]struct{} {
@@ -754,31 +755,15 @@ func generatePrivValUpdate(proTxHashes []crypto.ProTxHash) (*privValUpdate, erro
  return &privVal, nil
 }
 
-func validate(states []*State) {
- currValidatorCount := len(states[0].Validators.Validators)
- currValidators := states[0].Validators
- currHeight, _ := states[0].GetValidators()
- height, validators := states[0].GetValidatorSet()
- if height != currHeight {
- panic("they should all have the same heights")
- }
- if len(validators.Validators) != currValidatorCount {
- panic("they should all have the same initial validator count")
- }
- if !currValidators.Equals(validators) {
- panic("all validators should be the same")
- }
+func validate(t *testing.T, states []*State) {
+
+ currHeight, currValidators := states[0].GetValidatorSet()
+ currValidatorCount := currValidators.Size()
 
- for _, state := range states {
+ for validatorID, state := range states {
  height, validators := state.GetValidatorSet()
- if height != currHeight {
- panic("they should all have the same heights")
- }
- if len(validators.Validators) != currValidatorCount {
- panic("they should all have the same initial validator count")
- }
- if !currValidators.Equals(validators) {
- panic("all validators should be the same")
- }
+ assert.Equal(t, currHeight, height, "validator_id=%d", validatorID)
+ assert.Equal(t, currValidatorCount, len(validators.Validators), "validator_id=%d", validatorID)
+ assert.True(t, currValidators.Equals(validators), "validator_id=%d", validatorID)
  }
 }

types/priv_validator.go

@@ -179,6 +179,8 @@ func NewMockPVWithParams(
 
 // GetPubKey implements PrivValidator.
 func (pv *MockPV) GetPubKey(ctx context.Context, quorumHash crypto.QuorumHash) (crypto.PubKey, error) {
+ pv.mtx.RLock()
+ defer pv.mtx.RUnlock()
  if keys, ok := pv.PrivateKeys[quorumHash.String()]; ok {
  return keys.PubKey, nil
  }
@@ -194,6 +196,8 @@ func (pv *MockPV) GetProTxHash(ctx context.Context) (crypto.ProTxHash, error) {
 }
 
 func (pv *MockPV) GetFirstQuorumHash(ctx context.Context) (crypto.QuorumHash, error) {
+ pv.mtx.RLock()
+ defer pv.mtx.RUnlock()
  for quorumHashString := range pv.PrivateKeys {
  return hex.DecodeString(quorumHashString)
  }
@@ -202,21 +206,29 @@ func (pv *MockPV) GetFirstQuorumHash(ctx context.Context) (crypto.QuorumHash, er
 
 // GetThresholdPublicKey ...
 func (pv *MockPV) GetThresholdPublicKey(ctx context.Context, quorumHash crypto.QuorumHash) (crypto.PubKey, error) {
+ pv.mtx.RLock()
+ defer pv.mtx.RUnlock()
  return pv.PrivateKeys[quorumHash.String()].ThresholdPublicKey, nil
 }
 
 // GetPrivateKey ...
 func (pv *MockPV) GetPrivateKey(ctx context.Context, quorumHash crypto.QuorumHash) (crypto.PrivKey, error) {
+ pv.mtx.RLock()
+ defer pv.mtx.RUnlock()
  return pv.PrivateKeys[quorumHash.String()].PrivKey, nil
 }
 
 // ThresholdPublicKeyForQuorumHash ...
 func (pv *MockPV) ThresholdPublicKeyForQuorumHash(ctx context.Context, quorumHash crypto.QuorumHash) (crypto.PubKey, error) {
+ pv.mtx.RLock()
+ defer pv.mtx.RUnlock()
  return pv.PrivateKeys[quorumHash.String()].ThresholdPublicKey, nil
 }
 
 // GetHeight ...
 func (pv *MockPV) GetHeight(ctx context.Context, quorumHash crypto.QuorumHash) (int64, error) {
+ pv.mtx.RLock()
+ defer pv.mtx.RUnlock()
  if intString, ok := pv.FirstHeightOfQuorums[quorumHash.String()]; ok {
  return strconv.ParseInt(intString, 10, 64)
  }
@@ -232,6 +244,8 @@ func (pv *MockPV) SignVote(
  vote *tmproto.Vote,
  stateID StateID,
  logger log.Logger) error {
+ pv.mtx.Lock()
+ defer pv.mtx.Unlock()
  useChainID := chainID
  if pv.breakVoteSigning {
  useChainID = "incorrect-chain-id"
@@ -276,6 +290,8 @@ func (pv *MockPV) SignProposal(
  quorumHash crypto.QuorumHash,
  proposal *tmproto.Proposal,
 ) ([]byte, error) {
+ pv.mtx.Lock()
+ defer pv.mtx.Unlock()
  useChainID := chainID
  if pv.breakProposalSigning {
  useChainID = "incorrect-chain-id"
@@ -309,8 +325,8 @@ func (pv *MockPV) UpdatePrivateKey(
 ) {
  // fmt.Printf("mockpv node %X setting a new key %X at height %d\n", pv.ProTxHash,
  // privateKey.PubKey().Bytes(), height)
- pv.mtx.RLock()
- defer pv.mtx.RUnlock()
+ pv.mtx.Lock()
+ defer pv.mtx.Unlock()
  pv.PrivateKeys[quorumHash.String()] = crypto.QuorumKeys{
  PrivKey: privateKey,
  PubKey: privateKey.PubKey(),