core: Leader selector will retry sanityCheck (#261)

8 files changed, 118 insertions, 27 deletions

diff --git a/core/agreement-state_test.go b/core/agreement-state_test.go
index 8a2c016..37d9ae2 100644
--- a/core/agreement-state_test.go
+++ b/core/agreement-state_test.go
@@ -95,7 +95,9 @@ func (s *AgreementStateTestSuite) SetupTest() {
 }
 
 func (s *AgreementStateTestSuite) newAgreement(numNode int) *agreement {
-	leader := newLeaderSelector(common.NewRandomHash())
+	leader := newLeaderSelector(common.NewRandomHash(), func(*types.Block) bool {
+		return true
+	})
 	notarySet := make(map[types.NodeID]struct{})
 	for i := 0; i < numNode-1; i++ {
 		prvKey, err := ecdsa.NewPrivateKey()
diff --git a/core/agreement_test.go b/core/agreement_test.go
index 83405ad..3b97a10 100644
--- a/core/agreement_test.go
+++ b/core/agreement_test.go
@@ -84,7 +84,9 @@ func (s *AgreementTestSuite) SetupTest() {
 }
 
 func (s *AgreementTestSuite) newAgreement(numNotarySet int) *agreement {
-	leader := newLeaderSelector(common.NewRandomHash())
+	leader := newLeaderSelector(common.NewRandomHash(), func(*types.Block) bool {
+		return true
+	})
 	agreementIdx := len(s.agreement)
 	notarySet := make(map[types.NodeID]struct{})
 	for i := 0; i < numNotarySet-1; i++ {
diff --git a/core/consensus.go b/core/consensus.go
index 05f93e1..3e9c816 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -337,6 +337,10 @@ func NewConsensus(
 		roundToNotify: roundToNotify,
 	}
 
+	validLeader := func(block *types.Block) bool {
+		return lattice.SanityCheck(block) == nil
+	}
+
 	con.baModules = make([]*agreement, config.NumChains)
 	con.receivers = make([]*consensusBAReceiver, config.NumChains)
 	for i := uint32(0); i < config.NumChains; i++ {
@@ -350,7 +354,7 @@ func NewConsensus(
 			con.ID,
 			recv,
 			nodes.IDs,
-			newLeaderSelector(crs),
+			newLeaderSelector(crs, validLeader),
 			con.authModule,
 		)
 		// Hacky way to make agreement module self contained.
@@ -815,9 +819,10 @@ func (con *Consensus) ProcessBlockRandomnessResult(
 
 // preProcessBlock performs Byzantine Agreement on the block.
 func (con *Consensus) preProcessBlock(b *types.Block) (err error) {
-	// TODO(jimmy-dexon): add failed block to pool.
 	if err = con.lattice.SanityCheck(b); err != nil {
-		return
+		if err != ErrRetrySanityCheckLater {
+			return
+		}
 	}
 	if err = con.baModules[b.Position.ChainID].processBlock(b); err != nil {
 		return err
diff --git a/core/lattice.go b/core/lattice.go
index 06005c1..68b05c2 100644
--- a/core/lattice.go
+++ b/core/lattice.go
@@ -18,6 +18,7 @@
 package core
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -26,6 +27,11 @@ import (
 	"github.com/dexon-foundation/dexon-consensus-core/core/types"
 )
 
+// Errors for sanity check error.
+var (
+	ErrRetrySanityCheckLater = fmt.Errorf("retry sanity check later")
+)
+
 // Lattice represents a unit to produce a global ordering from multiple chains.
 type Lattice struct {
 	lock       sync.RWMutex
@@ -138,6 +144,9 @@ func (s *Lattice) SanityCheck(b *types.Block) (err error) {
 		s.lock.RLock()
 		defer s.lock.RUnlock()
 		if err = s.data.sanityCheck(b); err != nil {
+			if _, ok := err.(*ErrAckingBlockNotExists); ok {
+				err = ErrRetrySanityCheckLater
+			}
 			s.logger.Error("Sanity Check failed", "error", err)
 			return
 		}
@@ -147,10 +156,11 @@ func (s *Lattice) SanityCheck(b *types.Block) (err error) {
 	}
 	// Verify data in application layer.
 	s.logger.Debug("Calling Application.VerifyBlock", "block", b)
-	// TODO(jimmy-dexon): handle types.VerifyRetryLater.
-	if s.app.VerifyBlock(b) == types.VerifyInvalidBlock {
+	switch s.app.VerifyBlock(b) {
+	case types.VerifyInvalidBlock:
 		err = ErrInvalidBlock
-		return err
+	case types.VerifyRetryLater:
+		err = ErrRetrySanityCheckLater
 	}
 	return
 }
diff --git a/core/lattice_test.go b/core/lattice_test.go
index e14321d..ec40d05 100644
--- a/core/lattice_test.go
+++ b/core/lattice_test.go
@@ -55,7 +55,7 @@ func (mgr *testLatticeMgr) processBlock(b *types.Block) (err error) {
 		delivered []*types.Block
 	)
 	if err = mgr.lattice.SanityCheck(b); err != nil {
-		if _, ok := err.(*ErrAckingBlockNotExists); ok {
+		if err == ErrRetrySanityCheckLater {
 			err = nil
 		} else {
 			return
diff --git a/core/leader-selector.go b/core/leader-selector.go
index 23b9bb1..bfaa19c 100644
--- a/core/leader-selector.go
+++ b/core/leader-selector.go
@@ -20,6 +20,7 @@ package core
 import (
 	"fmt"
 	"math/big"
+	"sync"
 
 	"github.com/dexon-foundation/dexon-consensus-core/common"
 	"github.com/dexon-foundation/dexon-consensus-core/core/crypto"
@@ -31,6 +32,8 @@ var (
 	ErrIncorrectCRSSignature = fmt.Errorf("incorrect CRS signature")
 )
 
+type validLeaderFn func(*types.Block) bool
+
 // Some constant value.
 var (
 	maxHash *big.Int
@@ -47,20 +50,24 @@ func init() {
 }
 
 type leaderSelector struct {
-	hashCRS      common.Hash
-	numCRS       *big.Int
-	minCRSBlock  *big.Int
-	minBlockHash common.Hash
+	hashCRS       common.Hash
+	numCRS        *big.Int
+	minCRSBlock   *big.Int
+	minBlockHash  common.Hash
+	pendingBlocks []*types.Block
+	validLeader   validLeaderFn
+	lock          sync.Mutex
 }
 
 func newLeaderSelector(
-	crs common.Hash) *leaderSelector {
+	crs common.Hash, validLeader validLeaderFn) *leaderSelector {
 	numCRS := big.NewInt(0)
 	numCRS.SetBytes(crs[:])
 	return &leaderSelector{
 		numCRS:      numCRS,
 		hashCRS:     crs,
 		minCRSBlock: maxHash,
+		validLeader: validLeader,
 	}
 }
 
@@ -80,11 +87,25 @@ func (l *leaderSelector) probability(sig crypto.Signature) float64 {
 }
 
 func (l *leaderSelector) restart() {
+	l.lock.Lock()
+	defer l.lock.Unlock()
 	l.minCRSBlock = maxHash
 	l.minBlockHash = common.Hash{}
+	l.pendingBlocks = []*types.Block{}
 }
 
 func (l *leaderSelector) leaderBlockHash() common.Hash {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	newPendingBlocks := []*types.Block{}
+	for _, b := range l.pendingBlocks {
+		if l.validLeader(b) {
+			l.updateLeader(b)
+		} else {
+			newPendingBlocks = append(newPendingBlocks, b)
+		}
+	}
+	l.pendingBlocks = newPendingBlocks
 	return l.minBlockHash
 }
 
@@ -96,11 +117,20 @@ func (l *leaderSelector) processBlock(block *types.Block) error {
 	if !ok {
 		return ErrIncorrectCRSSignature
 	}
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	if !l.validLeader(block) {
+		l.pendingBlocks = append(l.pendingBlocks, block)
+		return nil
+	}
+	l.updateLeader(block)
+	return nil
+}
+func (l *leaderSelector) updateLeader(block *types.Block) {
 	dist := l.distance(block.CRSSignature)
 	cmp := l.minCRSBlock.Cmp(dist)
 	if cmp > 0 || (cmp == 0 && block.Hash.Less(l.minBlockHash)) {
 		l.minCRSBlock = dist
 		l.minBlockHash = block.Hash
 	}
-	return nil
 }
diff --git a/core/leader-selector_test.go b/core/leader-selector_test.go
index e2f7070..0927fe0 100644
--- a/core/leader-selector_test.go
+++ b/core/leader-selector_test.go
@@ -29,35 +29,49 @@ import (
 
 type LeaderSelectorTestSuite struct {
 	suite.Suite
+	mockValidLeaderDefault bool
+	mockValidLeaderDB      map[common.Hash]bool
+	mockValidLeader        validLeaderFn
+}
+
+func (s *LeaderSelectorTestSuite) SetupTest() {
+	s.mockValidLeaderDefault = true
+	s.mockValidLeaderDB = make(map[common.Hash]bool)
+	s.mockValidLeader = func(b *types.Block) bool {
+		if ret, exist := s.mockValidLeaderDB[b.Hash]; exist {
+			return ret
+		}
+		return s.mockValidLeaderDefault
+	}
 }
 
 func (s *LeaderSelectorTestSuite) newLeader() *leaderSelector {
-	return newLeaderSelector(common.NewRandomHash())
+	return newLeaderSelector(common.NewRandomHash(), s.mockValidLeader)
 }
 
 func (s *LeaderSelectorTestSuite) TestDistance() {
 	leader := s.newLeader()
 	hash := common.NewRandomHash()
 	prv, err := ecdsa.NewPrivateKey()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	sig, err := prv.Sign(hash)
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	dis := leader.distance(sig)
-	s.True(dis.Cmp(maxHash) == -1)
+	s.Equal(-1, dis.Cmp(maxHash))
 }
 
 func (s *LeaderSelectorTestSuite) TestProbability() {
 	leader := s.newLeader()
 	prv1, err := ecdsa.NewPrivateKey()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	prv2, err := ecdsa.NewPrivateKey()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	for {
 		hash := common.NewRandomHash()
 		sig1, err := prv1.Sign(hash)
-		s.Require().Nil(err)
+		s.Require().NoError(err)
 		sig2, err := prv2.Sign(hash)
-		s.Require().Nil(err)
+		s.Require().NoError(err)
 		dis1 := leader.distance(sig1)
 		dis2 := leader.distance(sig2)
 		prob1 := leader.probability(sig1)
@@ -83,14 +97,14 @@ func (s *LeaderSelectorTestSuite) TestLeaderBlockHash() {
 	blocks := make(map[common.Hash]*types.Block)
 	for i := 0; i < 10; i++ {
 		prv, err := ecdsa.NewPrivateKey()
-		s.Require().Nil(err)
+		s.Require().NoError(err)
 		block := &types.Block{
 			ProposerID: types.NewNodeID(prv.PublicKey()),
 			Hash:       common.NewRandomHash(),
 		}
 		s.Require().NoError(
 			NewAuthenticator(prv).SignCRS(block, leader.hashCRS))
-		s.Require().Nil(leader.processBlock(block))
+		s.Require().NoError(leader.processBlock(block))
 		blocks[block.Hash] = block
 	}
 	blockHash := leader.leaderBlockHash()
@@ -102,8 +116,36 @@ func (s *LeaderSelectorTestSuite) TestLeaderBlockHash() {
 			continue
 		}
 		dist := leader.distance(block.CRSSignature)
-		s.True(leaderDist.Cmp(dist) == -1)
+		s.Equal(-1, leaderDist.Cmp(dist))
+	}
+}
+
+func (s *LeaderSelectorTestSuite) TestValidLeaderFn() {
+	leader := s.newLeader()
+	blocks := make(map[common.Hash]*types.Block)
+	for i := 0; i < 10; i++ {
+		prv, err := ecdsa.NewPrivateKey()
+		s.Require().NoError(err)
+		block := &types.Block{
+			ProposerID: types.NewNodeID(prv.PublicKey()),
+			Hash:       common.NewRandomHash(),
+		}
+		s.Require().NoError(
+			NewAuthenticator(prv).SignCRS(block, leader.hashCRS))
+		s.Require().NoError(leader.processBlock(block))
+		blocks[block.Hash] = block
+	}
+	blockHash := leader.leaderBlockHash()
+
+	s.mockValidLeaderDB[blockHash] = false
+	leader.restart()
+	for _, b := range blocks {
+		s.Require().NoError(leader.processBlock(b))
 	}
+	s.NotEqual(blockHash, leader.leaderBlockHash())
+	s.mockValidLeaderDB[blockHash] = true
+	s.Equal(blockHash, leader.leaderBlockHash())
+	s.Len(leader.pendingBlocks, 0)
 }
 
 func TestLeaderSelector(t *testing.T) {
diff --git a/integration_test/node.go b/integration_test/node.go
index 0fb661b..8b2931b 100644
--- a/integration_test/node.go
+++ b/integration_test/node.go
@@ -188,7 +188,7 @@ func (n *Node) processBlock(b *types.Block) (err error) {
 		delivered []*types.Block
 	)
 	if err = n.lattice.SanityCheck(b); err != nil {
-		if _, ok := err.(*core.ErrAckingBlockNotExists); ok {
+		if err == core.ErrRetrySanityCheckLater {
 			err = nil
 		} else {
 			return