VoteSet can handle conflicting votes. TODO: add more tests

consensus/state.go

@@ -1400,7 +1400,7 @@ func (cs *ConsensusState) tryAddVote(vote *types.Vote, peerKey string) error {
 		// If it's otherwise invalid, punish peer.
 		if err == ErrVoteHeightMismatch {
 			return err
-		} else if _, ok := err.(*types.ErrVoteConflictingSignature); ok {
+		} else if _, ok := err.(*types.ErrVoteConflictingVotes); ok {
 			if peerKey == "" {
 				log.Warn("Found conflicting vote from ourselves. Did you unsafe_reset a validator?", "height", vote.Height, "round", vote.Round, "type", vote.Type)
 				return err

types/vote.go

@@ -1,6 +1,7 @@
 package types
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"io"
@@ -18,13 +19,13 @@ var (
 	ErrVoteInvalidBlockHash        = errors.New("Invalid block hash")
 )
 
-type ErrVoteConflictingSignature struct {
+type ErrVoteConflictingVotes struct {
 	VoteA *Vote
 	VoteB *Vote
 }
 
-func (err *ErrVoteConflictingSignature) Error() string {
-	return "Conflicting round vote signature"
+func (err *ErrVoteConflictingVotes) Error() string {
+	return "Conflicting votes"
 }
 
 // Represents a prevote, precommit, or commit vote from validators for consensus.
@@ -75,3 +76,10 @@ func (vote *Vote) String() string {
 		vote.Height, vote.Round, vote.Type, typeString,
 		Fingerprint(vote.BlockHash), vote.Signature)
 }
+
+// Does not check signature, but checks for equality of block
+// NOTE: May be from different validators, and signature may be incorrect.
+func (vote *Vote) SameBlockAs(other *Vote) bool {
+	return bytes.Equal(vote.BlockHash, other.BlockHash) &&
+		vote.BlockPartsHeader.Equals(other.BlockPartsHeader)
+}

types/vote_set.go

@@ -10,26 +10,54 @@ import (
 	"github.com/tendermint/go-wire"
 )
 
-// VoteSet helps collect signatures from validators at each height+round
-// for a predefined vote type.
-// Note that there three kinds of votes: prevotes, precommits, and commits.
-// A commit of prior rounds can be added added in lieu of votes/precommits.
-// NOTE: Assumes that the sum total of voting power does not exceed MaxUInt64.
+/*
+	VoteSet helps collect signatures from validators at each height+round for a
+	predefined vote type.
+
+	We need VoteSet to be able to keep track of conflicting votes when validators
+	double-sign.  Yet, we can't keep track of *all* the votes seen, as that could
+	be a DoS attack vector.
+
+	There are two storage areas for votes.
+	1. voteSet.votes
+	2. voteSet.votesByBlock
+
+	`.votes` is the "canonical" list of votes.  It always has at least one vote,
+	if a vote from a validator had been seen at all.  Usually it keeps track of
+	the first vote seen, but when a 2/3 majority is found, votes for that get
+	priority and are copied over from `.votesByBlock`.
+
+	`.votesByBlock` keeps track of a list of votes for a particular block.  There
+	are two ways a &blockVotes{} gets created in `.votesByBlock`.
+	1. the first vote seen by a validator was for the particular block.
+	2. a peer claims to have seen 2/3 majority for the particular block.
+
+	Since the first vote from a validator will always get added in `.votesByBlock`
+	, all votes in `.votes` will have a corresponding entry in `.votesByBlock`.
+
+	When a &blockVotes{} in `.votesByBlock` reaches a 2/3 majority quorum, its
+	votes are copied into `.votes`.
+
+	All this is memory bounded because conflicting votes only get added if a peer
+	told us to track that block, each peer only gets to tell us 1 such block, and,
+	there's only a limited number of peers.
+
+	NOTE: Assumes that the sum total of voting power does not exceed MaxUInt64.
+*/
 type VoteSet struct {
 	chainID string
 	height  int
 	round   int
 	type_   byte
 
-	mtx              sync.Mutex
-	valSet           *ValidatorSet
-	votes            []*Vote          // validator index -> vote
-	votesBitArray    *BitArray        // validator index -> has vote?
-	votesByBlock     map[string]int64 // string(blockHash)+string(blockParts) -> vote sum.
-	totalVotes       int64
-	maj23Hash        []byte
-	maj23PartsHeader PartSetHeader
-	maj23Exists      bool
+	mtx           sync.Mutex
+	valSet        *ValidatorSet
+	votesBitArray *BitArray
+	votes         []*Vote                // Primary votes to share
+	sum           int64                  // Sum of voting power for seen votes, discounting conflicts
+	maj23         *blockInfo             // First 2/3 majority seen
+	votesByBlock  map[string]*blockVotes // string(blockHash|blockParts) -> blockVotes
+	peerMaj23s    map[string]*blockInfo  // Maj23 for each peer
 }
 
 // Constructs a new VoteSet struct used to accumulate votes for given height/round.
@@ -43,10 +71,12 @@ func NewVoteSet(chainID string, height int, round int, type_ byte, valSet *Valid
 		round:         round,
 		type_:         type_,
 		valSet:        valSet,
-		votes:         make([]*Vote, valSet.Size()),
 		votesBitArray: NewBitArray(valSet.Size()),
-		votesByBlock:  make(map[string]int64),
-		totalVotes:    0,
+		votes:         make([]*Vote, valSet.Size()),
+		sum:           0,
+		maj23:         nil,
+		votesByBlock:  make(map[string]*blockVotes, valSet.Size()),
+		peerMaj23s:    make(map[string]*blockInfo),
 	}
 }
 
@@ -86,9 +116,12 @@ func (voteSet *VoteSet) Size() int {
 	}
 }
 
-// Returns added=true
-// Otherwise returns err=ErrVote[UnexpectedStep|InvalidIndex|InvalidAddress|InvalidSignature|InvalidBlockHash|ConflictingSignature]
+// Returns added=true if vote is valid and new.
+// Otherwise returns err=ErrVote[
+//		UnexpectedStep | InvalidIndex | InvalidAddress |
+//		InvalidSignature | InvalidBlockHash | ConflictingVotes ]
 // Duplicate votes return added=false, err=nil.
+// Conflicting votes return added=*, err=ErrVoteConflictingVotes.
 // NOTE: vote should not be mutated after adding.
 func (voteSet *VoteSet) AddVote(vote *Vote) (added bool, err error) {
 	voteSet.mtx.Lock()
@@ -97,11 +130,13 @@ func (voteSet *VoteSet) AddVote(vote *Vote) (added bool, err error) {
 	return voteSet.addVote(vote)
 }
 
+// NOTE: Validates as much as possible before attempting to verify the signature.
 func (voteSet *VoteSet) addVote(vote *Vote) (added bool, err error) {
 	valIndex := vote.ValidatorIndex
 	valAddr := vote.ValidatorAddress
+	blockKey := getBlockKey(vote)
 
-	// Ensure thta validator index was set
+	// Ensure that validator index was set
 	if valIndex < 0 || len(valAddr) == 0 {
 		panic("Validator index or address was not set in vote.")
 	}
@@ -124,20 +159,12 @@ func (voteSet *VoteSet) addVote(vote *Vote) (added bool, err error) {
 		return false, ErrVoteInvalidValidatorAddress
 	}
 
-	// If vote already exists, return false.
-	if existingVote := voteSet.votes[valIndex]; existingVote != nil {
-		if bytes.Equal(existingVote.BlockHash, vote.BlockHash) {
-			return false, nil
+	// If we already know of this vote, return false.
+	if existing, ok := voteSet.getVote(valIndex, blockKey); ok {
+		if existing.Signature.Equals(vote.Signature) {
+			return false, nil // duplicate
 		} else {
-			// Check signature.
-			if !val.PubKey.VerifyBytes(SignBytes(voteSet.chainID, vote), vote.Signature) {
-				// Bad signature.
-				return false, ErrVoteInvalidSignature
-			}
-			return false, &ErrVoteConflictingSignature{
-				VoteA: existingVote,
-				VoteB: vote,
-			}
+			return false, ErrVoteInvalidSignature // NOTE: assumes deterministic signatures
 		}
 	}
 
@@ -147,23 +174,139 @@ func (voteSet *VoteSet) addVote(vote *Vote) (added bool, err error) {
 		return false, ErrVoteInvalidSignature
 	}
 
-	// Add vote.
-	voteSet.votes[valIndex] = vote
-	voteSet.votesBitArray.SetIndex(valIndex, true)
-	blockKey := string(vote.BlockHash) + string(wire.BinaryBytes(vote.BlockPartsHeader))
-	totalBlockHashVotes := voteSet.votesByBlock[blockKey] + val.VotingPower
-	voteSet.votesByBlock[blockKey] = totalBlockHashVotes
-	voteSet.totalVotes += val.VotingPower
+	// Add vote and get conflicting vote if any
+	added, conflicting := voteSet.addVerifiedVote(vote, blockKey, val.VotingPower)
+	if conflicting != nil {
+		return added, &ErrVoteConflictingVotes{
+			VoteA: conflicting,
+			VoteB: vote,
+		}
+	} else {
+		if !added {
+			PanicSanity("Expected to add non-conflicting vote")
+		}
+		return added, nil
+	}
+
+}
+
+// Returns (vote, true) if vote exists for valIndex and blockKey
+func (voteSet *VoteSet) getVote(valIndex int, blockKey string) (vote *Vote, ok bool) {
+	if existing := voteSet.votes[valIndex]; existing != nil && getBlockKey(existing) == blockKey {
+		return existing, true
+	}
+	if existing := voteSet.votesByBlock[blockKey].getByIndex(valIndex); existing != nil {
+		return existing, true
+	}
+	return nil, false
+}
+
+// Assumes signature is valid.
+// If conflicting vote exists, returns it.
+func (voteSet *VoteSet) addVerifiedVote(vote *Vote, blockKey string, votingPower int64) (added bool, conflicting *Vote) {
+	valIndex := vote.ValidatorIndex
+
+	// Already exists in voteSet.votes?
+	if existing := voteSet.votes[valIndex]; existing != nil {
+		if existing.SameBlockAs(vote) {
+			PanicSanity("addVerifiedVote does not expect duplicate votes")
+		} else {
+			conflicting = existing
+		}
+		// Replace vote if blockKey matches voteSet.maj23.
+		if voteSet.maj23 != nil && voteSet.maj23.BlockKey() == blockKey {
+			voteSet.votes[valIndex] = vote
+			voteSet.votesBitArray.SetIndex(valIndex, true)
+		}
+		// Otherwise don't add it to voteSet.votes
+	} else {
+		// Add to voteSet.votes and incr .sum
+		voteSet.votes[valIndex] = vote
+		voteSet.votesBitArray.SetIndex(valIndex, true)
+		voteSet.sum += votingPower
+	}
+
+	votesByBlock, ok := voteSet.votesByBlock[blockKey]
+	if ok {
+		if conflicting != nil && !votesByBlock.peerMaj23 {
+			// There's a conflict and no peer claims that this block is special.
+			return false, conflicting
+		}
+		// We'll add the vote in a bit.
+	} else {
+		// .votesByBlock doesn't exist...
+		if conflicting != nil {
+			// ... and there's a conflicting vote.
+			// We're not even tracking this blockKey, so just forget it.
+			return false, conflicting
+		} else {
+			// ... and there's no conflicting vote.
+			// Start tracking this blockKey
+			votesByBlock = newBlockVotes(false, voteSet.valSet.Size())
+			voteSet.votesByBlock[blockKey] = votesByBlock
+			// We'll add the vote in a bit.
+		}
+	}
+
+	// Before adding to votesByBlock, see if we'll exceed quorum
+	origSum := votesByBlock.sum
+	quorum := voteSet.valSet.TotalVotingPower()*2/3 + 1
+
+	// Add vote to votesByBlock
+	votesByBlock.addVerifiedVote(vote, votingPower)
+
+	// If we just crossed the quorum threshold and have 2/3 majority...
+	if origSum < quorum && quorum <= votesByBlock.sum {
+		// Only consider the first quorum reached
+		if voteSet.maj23 == nil {
+			voteSet.maj23 = getBlockInfo(vote)
+			// And also copy votes over to voteSet.votes
+			for i, vote := range votesByBlock.votes {
+				if vote != nil {
+					voteSet.votes[i] = vote
+				}
+			}
+		}
+	}
+
+	return true, conflicting
+}
+
+// If a peer claims that it has 2/3 majority for given blockKey, call this.
+// NOTE: if there are too many peers, or too much peer churn,
+// this can cause memory issues.
+// TODO: implement ability to remove peers too
+func (voteSet *VoteSet) SetPeerMaj23(peerID string, blockHash []byte, blockPartsHeader PartSetHeader) {
+	voteSet.mtx.Lock()
+	defer voteSet.mtx.Unlock()
+
+	blockInfo := &blockInfo{blockHash, blockPartsHeader}
+	blockKey := blockInfo.BlockKey()
 
-	// If we just nudged it up to two thirds majority, add it.
-	if totalBlockHashVotes > voteSet.valSet.TotalVotingPower()*2/3 &&
-		(totalBlockHashVotes-val.VotingPower) <= voteSet.valSet.TotalVotingPower()*2/3 {
-		voteSet.maj23Hash = vote.BlockHash
-		voteSet.maj23PartsHeader = vote.BlockPartsHeader
-		voteSet.maj23Exists = true
+	// Make sure peer hasn't already told us something.
+	if existing, ok := voteSet.peerMaj23s[peerID]; ok {
+		if existing.Equals(blockInfo) {
+			return // Nothing to do
+		} else {
+			return // TODO bad peer!
+		}
 	}
+	voteSet.peerMaj23s[peerID] = blockInfo
 
-	return true, nil
+	// Create .votesByBlock entry if needed.
+	votesByBlock, ok := voteSet.votesByBlock[blockKey]
+	if ok {
+		if votesByBlock.peerMaj23 {
+			return // Nothing to do
+		} else {
+			votesByBlock.peerMaj23 = true
+			// No need to copy votes, already there.
+		}
+	} else {
+		votesByBlock = newBlockVotes(true, voteSet.valSet.Size())
+		voteSet.votesByBlock[blockKey] = votesByBlock
+		// No need to copy votes, no votes to copy over.
+	}
 }
 
 func (voteSet *VoteSet) BitArray() *BitArray {
@@ -175,6 +318,7 @@ func (voteSet *VoteSet) BitArray() *BitArray {
 	return voteSet.votesBitArray.Copy()
 }
 
+// NOTE: if validator has conflicting votes, picks random.
 func (voteSet *VoteSet) GetByIndex(valIndex int) *Vote {
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
@@ -197,7 +341,7 @@ func (voteSet *VoteSet) HasTwoThirdsMajority() bool {
 	}
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
-	return voteSet.maj23Exists
+	return voteSet.maj23 != nil
 }
 
 func (voteSet *VoteSet) IsCommit() bool {
@@ -209,7 +353,7 @@ func (voteSet *VoteSet) IsCommit() bool {
 	}
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
-	return len(voteSet.maj23Hash) > 0
+	return voteSet.maj23 != nil
 }
 
 func (voteSet *VoteSet) HasTwoThirdsAny() bool {
@@ -218,16 +362,16 @@ func (voteSet *VoteSet) HasTwoThirdsAny() bool {
 	}
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
-	return voteSet.totalVotes > voteSet.valSet.TotalVotingPower()*2/3
+	return voteSet.sum > voteSet.valSet.TotalVotingPower()*2/3
 }
 
 // Returns either a blockhash (or nil) that received +2/3 majority.
-// If there exists no such majority, returns (nil, false).
+// If there exists no such majority, returns (nil, PartSetHeader{}, false).
 func (voteSet *VoteSet) TwoThirdsMajority() (hash []byte, parts PartSetHeader, ok bool) {
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
-	if voteSet.maj23Exists {
-		return voteSet.maj23Hash, voteSet.maj23PartsHeader, true
+	if voteSet.maj23 != nil {
+		return voteSet.maj23.hash, voteSet.maj23.partsHeader, true
 	} else {
 		return nil, PartSetHeader{}, false
 	}
@@ -267,7 +411,7 @@ func (voteSet *VoteSet) StringShort() string {
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
 	return fmt.Sprintf(`VoteSet{H:%v R:%v T:%v +2/3:%v %v}`,
-		voteSet.height, voteSet.round, voteSet.type_, voteSet.maj23Exists, voteSet.votesBitArray)
+		voteSet.height, voteSet.round, voteSet.type_, voteSet.maj23, voteSet.votesBitArray)
 }
 
 //--------------------------------------------------------------------------------
@@ -279,30 +423,61 @@ func (voteSet *VoteSet) MakeCommit() *Commit {
 	}
 	voteSet.mtx.Lock()
 	defer voteSet.mtx.Unlock()
-	if len(voteSet.maj23Hash) == 0 {
+
+	// Make sure we have a 2/3 majority
+	if voteSet.maj23 == nil {
 		PanicSanity("Cannot MakeCommit() unless a blockhash has +2/3")
 	}
-	precommits := make([]*Vote, voteSet.valSet.Size())
-	voteSet.valSet.Iterate(func(valIndex int, val *Validator) bool {
-		vote := voteSet.votes[valIndex]
-		if vote == nil {
-			return false
-		}
-		if !bytes.Equal(vote.BlockHash, voteSet.maj23Hash) {
-			return false
-		}
-		if !vote.BlockPartsHeader.Equals(voteSet.maj23PartsHeader) {
-			return false
-		}
-		precommits[valIndex] = vote
-		return false
-	})
+
+	// For every validator, get the precommit
+	maj23Votes := voteSet.votesByBlock[voteSet.maj23.BlockKey()]
 	return &Commit{
-		Precommits: precommits,
+		Precommits: maj23Votes.votes,
+	}
+}
+
+//--------------------------------------------------------------------------------
+
+/*
+	Votes for a particular block
+	There are two ways a *blockVotes gets created for a blockKey.
+	1. first (non-conflicting) vote of a validator w/ blockKey (peerMaj23=false)
+	2. A peer claims to have a 2/3 majority w/ blockKey (peerMaj23=true)
+*/
+type blockVotes struct {
+	peerMaj23 bool      // peer claims to have maj23
+	bitArray  *BitArray // valIndex -> hasVote?
+	votes     []*Vote   // valIndex -> *Vote
+	sum       int64     // vote sum
+}
+
+func newBlockVotes(peerMaj23 bool, numValidators int) *blockVotes {
+	return &blockVotes{
+		peerMaj23: peerMaj23,
+		bitArray:  NewBitArray(numValidators),
+		votes:     make([]*Vote, numValidators),
+		sum:       0,
 	}
 }
 
-//----------------------------------------
+func (vs *blockVotes) addVerifiedVote(vote *Vote, votingPower int64) {
+	valIndex := vote.ValidatorIndex
+	if existing := vs.votes[valIndex]; existing == nil {
+		vs.bitArray.SetIndex(valIndex, true)
+		vs.votes[valIndex] = vote
+		vs.sum += votingPower
+	}
+}
+
+func (vs *blockVotes) getByIndex(index int) *Vote {
+	if vs == nil {
+		return nil
+	}
+	return vs.votes[index]
+}
+
+//--------------------------------------------------------------------------------
+
 // Common interface between *consensus.VoteSet and types.Commit
 type VoteSetReader interface {
 	Height() int
@@ -313,3 +488,27 @@ type VoteSetReader interface {
 	GetByIndex(int) *Vote
 	IsCommit() bool
 }
+
+//--------------------------------------------------------------------------------
+
+type blockInfo struct {
+	hash        []byte
+	partsHeader PartSetHeader
+}
+
+func (bInfo *blockInfo) Equals(other *blockInfo) bool {
+	return bytes.Equal(bInfo.hash, other.hash) &&
+		bInfo.partsHeader.Equals(other.partsHeader)
+}
+
+func (bInfo *blockInfo) BlockKey() string {
+	return string(bInfo.hash) + string(wire.BinaryBytes(bInfo.partsHeader))
+}
+
+func getBlockInfo(vote *Vote) *blockInfo {
+	return &blockInfo{vote.BlockHash, vote.BlockPartsHeader}
+}
+
+func getBlockKey(vote *Vote) string {
+	return string(vote.BlockHash) + string(wire.BinaryBytes(vote.BlockPartsHeader))
+}