core: optimize handling for bad block (#574)

* core: optimize handling for bad block

* fix test

4 files changed, 62 insertions, 31 deletions

diff --git a/core/agreement-mgr.go b/core/agreement-mgr.go
index ef1730c..17def67 100644
--- a/core/agreement-mgr.go
+++ b/core/agreement-mgr.go
@@ -271,26 +271,34 @@ func (mgr *agreementMgr) notifyRoundEvents(evts []utils.RoundEventParam) error {
 	return nil
 }
 
-func (mgr *agreementMgr) processVote(v *types.Vote) (err error) {
-	if !mgr.recv.isNotary {
-		return nil
-	}
-	if mgr.voteFilter.Filter(v) {
-		return nil
-	}
-	if v.Position.Round == mgr.curRoundSetting.round {
-		if _, exist := mgr.curRoundSetting.dkgSet[v.ProposerID]; !exist {
+func (mgr *agreementMgr) checkProposer(
+	round uint64, proposerID types.NodeID) error {
+	if round == mgr.curRoundSetting.round {
+		if _, exist := mgr.curRoundSetting.dkgSet[proposerID]; !exist {
 			return ErrNotInNotarySet
 		}
-	} else if v.Position.Round == mgr.curRoundSetting.round+1 {
-		setting := mgr.generateSetting(v.Position.Round)
+	} else if round == mgr.curRoundSetting.round+1 {
+		setting := mgr.generateSetting(round)
 		if setting == nil {
 			return ErrConfigurationNotReady
 		}
-		if _, exist := setting.dkgSet[v.ProposerID]; !exist {
+		if _, exist := setting.dkgSet[proposerID]; !exist {
 			return ErrNotInNotarySet
 		}
 	}
+	return nil
+}
+
+func (mgr *agreementMgr) processVote(v *types.Vote) (err error) {
+	if !mgr.recv.isNotary {
+		return nil
+	}
+	if mgr.voteFilter.Filter(v) {
+		return nil
+	}
+	if err := mgr.checkProposer(v.Position.Round, v.ProposerID); err != nil {
+		return err
+	}
 	if err = mgr.baModule.processVote(v); err == nil {
 		mgr.baModule.updateFilter(mgr.voteFilter)
 		mgr.voteFilter.AddVote(v)
@@ -302,6 +310,9 @@ func (mgr *agreementMgr) processVote(v *types.Vote) (err error) {
 }
 
 func (mgr *agreementMgr) processBlock(b *types.Block) error {
+	if err := mgr.checkProposer(b.Position.Round, b.ProposerID); err != nil {
+		return err
+	}
 	return mgr.baModule.processBlock(b)
 }
 
diff --git a/core/agreement-state_test.go b/core/agreement-state_test.go
index 6c06ef1..32ce8ac 100644
--- a/core/agreement-state_test.go
+++ b/core/agreement-state_test.go
@@ -75,6 +75,7 @@ func (s *AgreementStateTestSuite) proposeBlock(
 		Hash:       common.NewRandomHash(),
 	}
 	s.Require().NoError(s.signers[s.ID].SignCRS(block, leader.hashCRS))
+	s.Require().NoError(s.signers[s.ID].SignBlock(block))
 	s.block[block.Hash] = block
 	return block
 }
@@ -213,9 +214,11 @@ func (s *AgreementStateTestSuite) TestPreCommitState() {
 		blocks[i] = s.proposeBlock(a.data.leader)
 		prv, err := ecdsa.NewPrivateKey()
 		s.Require().NoError(err)
+		signer := utils.NewSigner(prv)
 		blocks[i].ProposerID = types.NewNodeID(prv.PublicKey())
-		s.Require().NoError(utils.NewSigner(prv).SignCRS(
+		s.Require().NoError(signer.SignCRS(
 			blocks[i], a.data.leader.hashCRS))
+		s.Require().NoError(signer.SignBlock(blocks[i]))
 		s.Require().NoError(a.processBlock(blocks[i]))
 	}
 
diff --git a/core/agreement.go b/core/agreement.go
index 7830af0..2215394 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -640,19 +640,34 @@ func (a *agreement) confirmedNoLock() bool {
 
 // processBlock is the entry point for processing Block.
 func (a *agreement) processBlock(block *types.Block) error {
+	checkSkip := func() bool {
+		aID := a.agreementID()
+		if block.Position != aID {
+			// Agreement module has stopped.
+			if !isStop(aID) {
+				if aID.Newer(block.Position) {
+					return true
+				}
+			}
+		}
+		return false
+	}
+	if checkSkip() {
+		return nil
+	}
+	if err := utils.VerifyBlockSignature(block); err != nil {
+		return err
+	}
+
 	a.lock.Lock()
 	defer a.lock.Unlock()
 	a.data.blocksLock.Lock()
 	defer a.data.blocksLock.Unlock()
-
 	aID := a.agreementID()
-	if block.Position != aID {
-		// Agreement module has stopped.
-		if !isStop(aID) {
-			if aID.Newer(block.Position) {
-				return nil
-			}
-		}
+	// a.agreementID might change during lock, so we need to checkSkip again.
+	if checkSkip() {
+		return nil
+	} else if aID != block.Position {
 		a.pendingBlock = append(a.pendingBlock, pendingBlock{
 			block:        block,
 			receivedTime: time.Now().UTC(),
diff --git a/core/agreement_test.go b/core/agreement_test.go
index b784560..9332762 100644
--- a/core/agreement_test.go
+++ b/core/agreement_test.go
@@ -46,7 +46,8 @@ func (r *agreementTestReceiver) ProposeVote(vote *types.Vote) {
 func (r *agreementTestReceiver) ProposeBlock() common.Hash {
 	block := r.s.proposeBlock(
 		r.s.agreement[r.agreementIndex].data.ID,
-		r.s.agreement[r.agreementIndex].data.leader.hashCRS)
+		r.s.agreement[r.agreementIndex].data.leader.hashCRS,
+		[]byte{})
 	r.s.blockChan <- block.Hash
 	return block.Hash
 }
@@ -79,17 +80,18 @@ func (r *agreementTestReceiver) ReportForkBlock(b1, b2 *types.Block) {
 }
 
 func (s *AgreementTestSuite) proposeBlock(
-	nID types.NodeID, crs common.Hash) *types.Block {
+	nID types.NodeID, crs common.Hash, payload []byte) *types.Block {
 	block := &types.Block{
 		ProposerID: nID,
 		Position:   types.Position{Height: types.GenesisHeight},
-		Hash:       common.NewRandomHash(),
+		Payload:    payload,
 	}
-	s.block[block.Hash] = block
 	signer, exist := s.signers[block.ProposerID]
 	s.Require().True(exist)
-	s.Require().NoError(signer.SignCRS(
-		block, crs))
+	s.Require().NoError(signer.SignCRS(block, crs))
+	s.Require().NoError(signer.SignBlock(block))
+	s.block[block.Hash] = block
+
 	return block
 }
 
@@ -331,7 +333,7 @@ func (s *AgreementTestSuite) TestFastConfirmNonLeader() {
 	a.nextState()
 	// FastVoteState
 	s.Require().Len(s.blockChan, 0)
-	block := s.proposeBlock(leaderNode, a.data.leader.hashCRS)
+	block := s.proposeBlock(leaderNode, a.data.leader.hashCRS, []byte{})
 	s.Require().Equal(leaderNode, block.ProposerID)
 	s.Require().NoError(a.processBlock(block))
 	// Wait some time for go routine in processBlock to finish.
@@ -537,8 +539,8 @@ func (s *AgreementTestSuite) TestForkVote() {
 func (s *AgreementTestSuite) TestForkBlock() {
 	a, _ := s.newAgreement(4, -1, s.defaultValidLeader)
 	for nID := range a.notarySet {
-		b01 := s.proposeBlock(nID, a.data.leader.hashCRS)
-		b02 := s.proposeBlock(nID, a.data.leader.hashCRS)
+		b01 := s.proposeBlock(nID, a.data.leader.hashCRS, []byte{1})
+		b02 := s.proposeBlock(nID, a.data.leader.hashCRS, []byte{2})
 		s.Require().NoError(a.processBlock(b01))
 		s.Require().IsType(&ErrFork{}, a.processBlock(b02))
 		s.Require().Equal(b01.Hash, <-s.forkBlockChan)
@@ -550,7 +552,7 @@ func (s *AgreementTestSuite) TestFindBlockInPendingSet() {
 	a, leaderNode := s.newAgreement(4, 0, func(*types.Block, common.Hash) (bool, error) {
 		return false, nil
 	})
-	block := s.proposeBlock(leaderNode, a.data.leader.hashCRS)
+	block := s.proposeBlock(leaderNode, a.data.leader.hashCRS, []byte{})
 	s.Require().NoError(a.processBlock(block))
 	// Make sure the block goes to pending pool in leader selector.
 	block, exist := a.data.leader.findPendingBlock(block.Hash)