core: find confirmed block in pending set and candidate set (#419)

5 files changed, 74 insertions, 31 deletions

diff --git a/core/agreement.go b/core/agreement.go
index 62bbe25..c17c59f 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -545,17 +545,22 @@ func (a *agreement) addCandidateBlockNoLock(block *types.Block) {
 	a.candidateBlock[block.Hash] = block
 }
 
-func (a *agreement) findCandidateBlock(hash common.Hash) (*types.Block, bool) {
-	a.lock.RLock()
-	defer a.lock.RUnlock()
-	return a.findCandidateBlockNoLock(hash)
-}
-
 func (a *agreement) findCandidateBlockNoLock(
 	hash common.Hash) (*types.Block, bool) {
 	b, e := a.candidateBlock[hash]
 	return b, e
 }
+
+// find a block in both candidate blocks and pending blocks in leader-selector.
+// A block might be confirmed by others while we can't verify its validity.
+func (a *agreement) findBlockNoLock(hash common.Hash) (*types.Block, bool) {
+	b, e := a.findCandidateBlockNoLock(hash)
+	if !e {
+		b, e = a.data.leader.findPendingBlock(hash)
+	}
+	return b, e
+}
+
 func (a *agreementData) countVote(period uint64, voteType types.VoteType) (
 	blockHash common.Hash, ok bool) {
 	a.lock.RLock()
diff --git a/core/agreement_test.go b/core/agreement_test.go
index 7c15b83..6e2a3f4 100644
--- a/core/agreement_test.go
+++ b/core/agreement_test.go
@@ -118,12 +118,10 @@ func (s *AgreementTestSuite) SetupTest() {
 }
 
 func (s *AgreementTestSuite) newAgreement(
-	numNotarySet, leaderIdx int) (*agreement, types.NodeID) {
+	numNotarySet, leaderIdx int, validLeader validLeaderFn) (*agreement, types.NodeID) {
 	s.Require().True(leaderIdx < numNotarySet)
 	logger := &common.NullLogger{}
-	leader := newLeaderSelector(func(*types.Block) (bool, error) {
-		return true, nil
-	}, logger)
+	leader := newLeaderSelector(validLeader, logger)
 	agreementIdx := len(s.agreement)
 	var leaderNode types.NodeID
 	notarySet := make(map[types.NodeID]struct{})
@@ -174,7 +172,9 @@ func (s *AgreementTestSuite) prepareVote(
 }
 
 func (s *AgreementTestSuite) TestSimpleConfirm() {
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	// FastState
 	a.nextState()
 	// FastVoteState
@@ -222,7 +222,9 @@ func (s *AgreementTestSuite) TestSimpleConfirm() {
 }
 
 func (s *AgreementTestSuite) TestPartitionOnCommitVote() {
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	// FastState
 	a.nextState()
 	// FastVoteState
@@ -266,7 +268,9 @@ func (s *AgreementTestSuite) TestPartitionOnCommitVote() {
 }
 
 func (s *AgreementTestSuite) TestFastConfirmLeader() {
-	a, leaderNode := s.newAgreement(4, 0)
+	a, leaderNode := s.newAgreement(4, 0, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	s.Require().Equal(s.ID, leaderNode)
 	// FastState
 	a.nextState()
@@ -292,7 +296,9 @@ func (s *AgreementTestSuite) TestFastConfirmLeader() {
 }
 
 func (s *AgreementTestSuite) TestFastConfirmNonLeader() {
-	a, leaderNode := s.newAgreement(4, 1)
+	a, leaderNode := s.newAgreement(4, 1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	s.Require().NotEqual(s.ID, leaderNode)
 	// FastState
 	a.nextState()
@@ -321,7 +327,9 @@ func (s *AgreementTestSuite) TestFastConfirmNonLeader() {
 
 func (s *AgreementTestSuite) TestFastForwardCond1() {
 	votes := 0
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	a.data.lockRound = 1
 	a.data.period = 3
 	hash := common.NewRandomHash()
@@ -375,7 +383,9 @@ func (s *AgreementTestSuite) TestFastForwardCond1() {
 
 func (s *AgreementTestSuite) TestFastForwardCond2() {
 	votes := 0
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	a.data.period = 1
 	hash := common.NewRandomHash()
 	for nID := range a.notarySet {
@@ -412,7 +422,9 @@ func (s *AgreementTestSuite) TestFastForwardCond2() {
 func (s *AgreementTestSuite) TestFastForwardCond3() {
 	numVotes := 0
 	votes := []*types.Vote{}
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	a.data.period = 1
 	for nID := range a.notarySet {
 		vote := s.prepareVote(nID, types.VoteCom, common.NewRandomHash(), uint64(2))
@@ -439,7 +451,9 @@ func (s *AgreementTestSuite) TestFastForwardCond3() {
 
 func (s *AgreementTestSuite) TestDecide() {
 	votes := 0
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	a.data.period = 5
 
 	// No decide if com-vote on SKIP.
@@ -467,7 +481,9 @@ func (s *AgreementTestSuite) TestDecide() {
 }
 
 func (s *AgreementTestSuite) TestForkVote() {
-	a, _ := s.newAgreement(4, -1)
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	a.data.period = 2
 	for nID := range a.notarySet {
 		v01 := s.prepareVote(nID, types.VotePreCom, common.NewRandomHash(), 2)
@@ -480,8 +496,10 @@ func (s *AgreementTestSuite) TestForkVote() {
 	}
 }
 
-func (s AgreementTestSuite) TestForkBlock() {
-	a, _ := s.newAgreement(4, -1)
+func (s *AgreementTestSuite) TestForkBlock() {
+	a, _ := s.newAgreement(4, -1, func(*types.Block) (bool, error) {
+		return true, nil
+	})
 	for nID := range a.notarySet {
 		b01 := s.proposeBlock(nID, a.data.leader.hashCRS)
 		b02 := s.proposeBlock(nID, a.data.leader.hashCRS)
@@ -492,6 +510,22 @@ func (s AgreementTestSuite) TestForkBlock() {
 	}
 }
 
+func (s *AgreementTestSuite) TestFindBlockInPendingSet() {
+	a, leaderNode := s.newAgreement(4, 0, func(*types.Block) (bool, error) {
+		return false, nil
+	})
+	block := s.proposeBlock(leaderNode, a.data.leader.hashCRS)
+	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)
+	s.Require().True(exist)
+	s.Require().NotNil(block)
+	// This block is allowed to be found by findBlockNoLock.
+	block, exist = a.findBlockNoLock(block.Hash)
+	s.Require().True(exist)
+	s.Require().NotNil(block)
+}
+
 func TestAgreement(t *testing.T) {
 	suite.Run(t, new(AgreementTestSuite))
 }
diff --git a/core/consensus.go b/core/consensus.go
index 0754e80..b84947a 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -137,7 +137,7 @@ func (recv *consensusBAReceiver) ConfirmBlock(
 		}
 	} else {
 		var exist bool
-		block, exist = recv.agreementModule.findCandidateBlockNoLock(hash)
+		block, exist = recv.agreementModule.findBlockNoLock(hash)
 		if !exist {
 			recv.consensus.logger.Error("Unknown block confirmed",
 				"hash", hash.String()[:6],
diff --git a/core/leader-selector.go b/core/leader-selector.go
index bcfa57f..214b4cb 100644
--- a/core/leader-selector.go
+++ b/core/leader-selector.go
@@ -55,7 +55,7 @@ type leaderSelector struct {
 	numCRS        *big.Int
 	minCRSBlock   *big.Int
 	minBlockHash  common.Hash
-	pendingBlocks []*types.Block
+	pendingBlocks map[common.Hash]*types.Block
 	validLeader   validLeaderFn
 	lock          sync.Mutex
 	logger        common.Logger
@@ -94,13 +94,12 @@ func (l *leaderSelector) restart(crs common.Hash) {
 	l.hashCRS = crs
 	l.minCRSBlock = maxHash
 	l.minBlockHash = common.Hash{}
-	l.pendingBlocks = []*types.Block{}
+	l.pendingBlocks = make(map[common.Hash]*types.Block)
 }
 
 func (l *leaderSelector) leaderBlockHash() common.Hash {
 	l.lock.Lock()
 	defer l.lock.Unlock()
-	newPendingBlocks := []*types.Block{}
 	for _, b := range l.pendingBlocks {
 		ok, dist := l.potentialLeader(b)
 		if !ok {
@@ -109,15 +108,14 @@ func (l *leaderSelector) leaderBlockHash() common.Hash {
 		ok, err := l.validLeader(b)
 		if err != nil {
 			l.logger.Error("Error checking validLeader", "error", err, "block", b)
+			delete(l.pendingBlocks, b.Hash)
 			continue
 		}
 		if ok {
 			l.updateLeader(b, dist)
-		} else {
-			newPendingBlocks = append(newPendingBlocks, b)
+			delete(l.pendingBlocks, b.Hash)
 		}
 	}
-	l.pendingBlocks = newPendingBlocks
 	return l.minBlockHash
 }
 
@@ -140,7 +138,7 @@ func (l *leaderSelector) processBlock(block *types.Block) error {
 		return err
 	}
 	if !ok {
-		l.pendingBlocks = append(l.pendingBlocks, block)
+		l.pendingBlocks[block.Hash] = block
 		return nil
 	}
 	l.updateLeader(block, dist)
@@ -157,3 +155,9 @@ func (l *leaderSelector) updateLeader(block *types.Block, dist *big.Int) {
 	l.minCRSBlock = dist
 	l.minBlockHash = block.Hash
 }
+
+func (l *leaderSelector) findPendingBlock(
+	hash common.Hash) (*types.Block, bool) {
+	b, e := l.pendingBlocks[hash]
+	return b, e
+}
diff --git a/core/leader-selector_test.go b/core/leader-selector_test.go
index 71499d6..9e09279 100644
--- a/core/leader-selector_test.go
+++ b/core/leader-selector_test.go
@@ -170,9 +170,9 @@ func (s *LeaderSelectorTestSuite) TestPotentialLeader() {
 		s.Require().NoError(leader.processBlock(block))
 		if i > 0 {
 			if ok {
-				s.Contains(leader.pendingBlocks, block)
+				s.Contains(leader.pendingBlocks, block.Hash)
 			} else {
-				s.NotContains(leader.pendingBlocks, block)
+				s.NotContains(leader.pendingBlocks, block.Hash)
 			}
 			blocks[block.Hash] = block
 		}