core: Fix BA Fork issues (#104)

4 files changed, 49 insertions, 69 deletions

diff --git a/core/agreement-state.go b/core/agreement-state.go
index b948417..d7225d6 100644
--- a/core/agreement-state.go
+++ b/core/agreement-state.go
@@ -69,19 +69,14 @@ func (s *prepareState) nextState() (agreementState, error) {
 	hash := common.Hash{}
 	if s.a.period == 1 {
 		hash = s.a.blockProposer().Hash
+		s.a.recv.ProposeBlock(hash)
 	} else {
 		var proposed bool
-		hash, proposed = s.a.countVote(s.a.period-1, types.VotePass)
+		_, proposed = s.a.countVote(s.a.period-1, types.VotePass)
 		if !proposed {
 			return nil, ErrNoEnoughVoteInPrepareState
 		}
-		if hash == nullBlockHash {
-			hash = s.a.blocks[s.a.ID].Hash
-		} else {
-			delete(s.a.blocks, s.a.ID)
-		}
 	}
-	s.a.recv.ProposeBlock(hash)
 	return newAckState(s.a), nil
 }
 func (s *prepareState) receiveVote() error { return nil }
@@ -259,30 +254,9 @@ func (s *pass2State) receiveVote() error {
 	if s.voted {
 		return nil
 	}
-	ackHash, ok := s.a.countVote(s.a.period, types.VoteAck)
-	if ok && ackHash != nullBlockHash {
-		s.a.recv.ProposeVote(&types.Vote{
-			Type:      types.VotePass,
-			BlockHash: ackHash,
-			Period:    s.a.period,
-		})
-		s.voted = true
-	} else if s.a.period > 1 {
-		if _, exist :=
-			s.a.votes[s.a.period][types.VoteConfirm][s.a.ID]; !exist {
-			hash, ok := s.a.countVote(s.a.period-1, types.VotePass)
-			if ok && hash == nullBlockHash {
-				s.a.recv.ProposeVote(&types.Vote{
-					Type:      types.VotePass,
-					BlockHash: hash,
-					Period:    s.a.period,
-				})
-				s.voted = true
-			}
-		}
-	}
 	hash, ok := s.a.countVote(s.a.period, types.VotePass)
 	if ok {
+		s.voted = true
 		s.enoughPassVote <- hash
 	}
 	return nil
diff --git a/core/agreement-state_test.go b/core/agreement-state_test.go
index 1591625..bc8dc8e 100644
--- a/core/agreement-state_test.go
+++ b/core/agreement-state_test.go
@@ -151,9 +151,6 @@ func (s *AgreementStateTestSuite) TestPrepareState() {
 
 	newState, err = state.nextState()
 	s.Require().Nil(err)
-	s.Require().True(len(s.blockChan) > 0)
-	hash := <-s.blockChan
-	s.Equal(proposedBlock, hash)
 	s.Equal(stateAck, newState.state())
 
 	// For period >= 2, if the pass-vote for block v not equal to {}
@@ -172,9 +169,6 @@ func (s *AgreementStateTestSuite) TestPrepareState() {
 
 	newState, err = state.nextState()
 	s.Require().Nil(err)
-	s.Require().True(len(s.blockChan) > 0)
-	hash = <-s.blockChan
-	s.Equal(block.Hash, hash)
 	s.Equal(stateAck, newState.state())
 }
 
@@ -382,10 +376,6 @@ func (s *AgreementStateTestSuite) TestPass2State() {
 		s.Require().Nil(a.processVote(vote))
 	}
 	s.Require().Nil(state.receiveVote())
-	s.Require().True(len(s.voteChan) > 0)
-	vote := <-s.voteChan
-	s.Equal(types.VotePass, vote.Type)
-	s.Equal(block.Hash, vote.BlockHash)
 	// Only propose one vote.
 	s.Require().Nil(state.receiveVote())
 	s.Require().True(len(s.voteChan) == 0)
@@ -400,14 +390,9 @@ func (s *AgreementStateTestSuite) TestPass2State() {
 		vote := s.prepareVote(vID, types.VotePass, common.Hash{}, 1)
 		s.Require().Nil(a.processVote(vote))
 	}
-	vote = s.prepareVote(s.ID, types.VoteAck, common.Hash{}, 2)
+	vote := s.prepareVote(s.ID, types.VoteAck, common.Hash{}, 2)
 	s.Require().Nil(a.processVote(vote))
 	s.Require().Nil(state.receiveVote())
-	s.Require().True(len(s.voteChan) > 0)
-	vote = <-s.voteChan
-	s.Equal(types.VotePass, vote.Type)
-	s.Equal(common.Hash{}, vote.BlockHash)
-
 	// Test terminate.
 	ok := make(chan struct{})
 	go func() {
diff --git a/core/agreement.go b/core/agreement.go
index e741145..86cb6fc 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -32,7 +32,6 @@ import (
 var (
 	ErrNotValidator           = fmt.Errorf("not a validaotr")
 	ErrIncorrectVoteSignature = fmt.Errorf("incorrect vote signature")
-	ErrForkVote               = fmt.Errorf("fork vote")
 )
 
 // ErrFork for fork error in agreement.
@@ -46,6 +45,17 @@ func (e *ErrFork) Error() string {
 		e.vID.String(), e.old, e.new)
 }
 
+// ErrForkVote for fork vote error in agreement.
+type ErrForkVote struct {
+	vID      types.ValidatorID
+	old, new *types.Vote
+}
+
+func (e *ErrForkVote) Error() string {
+	return fmt.Sprintf("fork vote is found for %s, old %s, new %s",
+		e.vID.String(), e.old, e.new)
+}
+
 type blockProposerFn func() *types.Block
 
 func newVoteListMap() []map[types.ValidatorID]*types.Vote {
@@ -148,6 +158,7 @@ func (a *agreement) restart(validators types.ValidatorIDs, aID types.Position) {
 		a.data.blocks = make(map[types.ValidatorID]*types.Block)
 		a.data.requiredVote = len(validators)/3*2 + 1
 		a.data.leader.restart()
+		a.data.defaultBlock = common.Hash{}
 		a.hasOutput = false
 		a.state = newPrepareState(a.data)
 		a.validators = make(map[types.ValidatorID]struct{})
@@ -210,6 +221,9 @@ func (a *agreement) agreementID() types.Position {
 
 // nextState is called at the spcifi clock time.
 func (a *agreement) nextState() (err error) {
+	if err = a.state.receiveVote(); err != nil {
+		return
+	}
 	a.state, err = a.state.nextState()
 	return
 }
@@ -230,14 +244,17 @@ func (a *agreement) sanityCheck(vote *types.Vote) error {
 	if !ok {
 		return ErrIncorrectVoteSignature
 	}
+	return nil
+}
 
+func (a *agreement) checkForkVote(vote *types.Vote) error {
 	if err := func() error {
 		a.data.votesLock.RLock()
 		defer a.data.votesLock.RUnlock()
 		if votes, exist := a.data.votes[vote.Period]; exist {
 			if oldVote, exist := votes[vote.Type][vote.ProposerID]; exist {
 				if vote.BlockHash != oldVote.BlockHash {
-					return ErrForkVote
+					return &ErrForkVote{vote.ProposerID, oldVote, vote}
 				}
 			}
 		}
@@ -272,6 +289,10 @@ func (a *agreement) processVote(vote *types.Vote) error {
 		})
 		return nil
 	}
+	if err := a.checkForkVote(vote); err != nil {
+		return err
+	}
+
 	if func() bool {
 		a.data.votesLock.Lock()
 		defer a.data.votesLock.Unlock()
@@ -303,12 +324,6 @@ func (a *agreement) prepareBlock(
 func (a *agreement) processBlock(block *types.Block) error {
 	a.data.blocksLock.Lock()
 	defer a.data.blocksLock.Unlock()
-	if b, exist := a.data.blocks[block.ProposerID]; exist {
-		if b.Hash != block.Hash {
-			return &ErrFork{block.ProposerID, b.Hash, block.Hash}
-		}
-		return nil
-	}
 	if block.Position != a.agreementID() {
 		a.pendingBlock = append(a.pendingBlock, pendingBlock{
 			block:        block,
@@ -316,6 +331,12 @@ func (a *agreement) processBlock(block *types.Block) error {
 		})
 		return nil
 	}
+	if b, exist := a.data.blocks[block.ProposerID]; exist {
+		if b.Hash != block.Hash {
+			return &ErrFork{block.ProposerID, b.Hash, block.Hash}
+		}
+		return nil
+	}
 	if err := a.data.leader.processBlock(block); err != nil {
 		return err
 	}
diff --git a/core/consensus.go b/core/consensus.go
index ef39795..85b6e38 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -72,12 +72,12 @@ type consensusReceiver struct {
 
 func (recv *consensusReceiver) ProposeVote(vote *types.Vote) {
 	if err := recv.consensus.prepareVote(recv.chainID, vote); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 		return
 	}
 	go func() {
 		if err := recv.consensus.ProcessVote(vote); err != nil {
-			fmt.Println(err)
+			log.Println(err)
 			return
 		}
 		recv.consensus.network.BroadcastVote(vote)
@@ -87,12 +87,12 @@ func (recv *consensusReceiver) ProposeVote(vote *types.Vote) {
 func (recv *consensusReceiver) ProposeBlock(hash common.Hash) {
 	block, exist := recv.consensus.baModules[recv.chainID].findCandidateBlock(hash)
 	if !exist {
-		fmt.Println(ErrUnknownBlockProposed)
-		fmt.Println(hash)
+		log.Println(ErrUnknownBlockProposed)
+		log.Println(hash)
 		return
 	}
 	if err := recv.consensus.PreProcessBlock(block); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 		return
 	}
 	recv.consensus.network.BroadcastBlock(block)
@@ -101,11 +101,11 @@ func (recv *consensusReceiver) ProposeBlock(hash common.Hash) {
 func (recv *consensusReceiver) ConfirmBlock(hash common.Hash) {
 	block, exist := recv.consensus.baModules[recv.chainID].findCandidateBlock(hash)
 	if !exist {
-		fmt.Println(ErrUnknownBlockConfirmed, hash)
+		log.Println(ErrUnknownBlockConfirmed, hash)
 		return
 	}
 	if err := recv.consensus.ProcessBlock(block); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 		return
 	}
 	recv.restart <- struct{}{}
@@ -294,10 +294,10 @@ func (con *Consensus) RunLegacy() {
 		},
 	}
 	if err := con.PrepareGenesisBlock(genesisBlock, time.Now().UTC()); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 	}
 	if err := con.ProcessBlock(genesisBlock); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 	}
 	con.network.BroadcastBlock(genesisBlock)
 
@@ -315,10 +315,10 @@ ProposingBlockLoop:
 			},
 		}
 		if err := con.PrepareBlock(block, time.Now().UTC()); err != nil {
-			fmt.Println(err)
+			log.Println(err)
 		}
 		if err := con.ProcessBlock(block); err != nil {
-			fmt.Println(err)
+			log.Println(err)
 		}
 		con.network.BroadcastBlock(block)
 	}
@@ -344,15 +344,15 @@ func (con *Consensus) processMsg(
 		switch val := msg.(type) {
 		case *types.Block:
 			if err := blockProcesser(val); err != nil {
-				fmt.Println(err)
+				log.Println(err)
 			}
 		case *types.NotaryAck:
 			if err := con.ProcessNotaryAck(val); err != nil {
-				fmt.Println(err)
+				log.Println(err)
 			}
 		case *types.Vote:
 			if err := con.ProcessVote(val); err != nil {
-				fmt.Println(err)
+				log.Println(err)
 			}
 		}
 	}
@@ -367,11 +367,11 @@ func (con *Consensus) proposeBlock(chainID uint32) *types.Block {
 		},
 	}
 	if err := con.PrepareBlock(block, time.Now().UTC()); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 		return nil
 	}
 	if err := con.baModules[chainID].prepareBlock(block, con.prvKey); err != nil {
-		fmt.Println(err)
+		log.Println(err)
 		return nil
 	}
 	return block