core: only qualified nodes can participant BA (#540)

* core: only qualified nodes can participant BA

* core: remove leader calculation from node set cache

7 files changed, 104 insertions, 77 deletions

diff --git a/core/agreement-mgr.go b/core/agreement-mgr.go
index 3e0851d..8cb4c2e 100644
--- a/core/agreement-mgr.go
+++ b/core/agreement-mgr.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/dexon-foundation/dexon-consensus/common"
 	"github.com/dexon-foundation/dexon-consensus/core/types"
+	typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
 	"github.com/dexon-foundation/dexon-consensus/core/utils"
 )
 
@@ -34,6 +35,7 @@ var (
 	ErrPreviousRoundIsNotFinished = errors.New("previous round is not finished")
 	ErrRoundOutOfRange            = errors.New("round out of range")
 	ErrInvalidBlock               = errors.New("invalid block")
+	ErrNoValidLeader              = errors.New("no valid leader")
 )
 
 const maxResultCache = 100
@@ -89,7 +91,8 @@ func newAgreementMgrConfig(prev agreementMgrConfig, config *types.Config,
 
 type baRoundSetting struct {
 	round     uint64
-	notarySet map[types.NodeID]struct{}
+	dkgSet    map[types.NodeID]struct{}
+	threshold int
 	ticker    Ticker
 	crs       common.Hash
 }
@@ -173,6 +176,19 @@ func (mgr *agreementMgr) run() {
 	}()
 }
 
+func (mgr *agreementMgr) calcLeader(
+	dkgSet map[types.NodeID]struct{},
+	crs common.Hash, pos types.Position) (
+	types.NodeID, error) {
+	nodeSet := types.NewNodeSetFromMap(dkgSet)
+	leader := nodeSet.GetSubSet(1, types.NewNodeLeaderTarget(
+		crs, pos.Height))
+	for nID := range leader {
+		return nID, nil
+	}
+	return types.NodeID{}, ErrNoValidLeader
+}
+
 func (mgr *agreementMgr) config(round uint64) *agreementMgrConfig {
 	mgr.lock.RLock()
 	defer mgr.lock.RUnlock()
@@ -274,10 +290,6 @@ func (mgr *agreementMgr) processAgreementResult(
 		}
 	} else if result.Position.Newer(aID) {
 		mgr.logger.Info("Fast syncing BA", "position", result.Position)
-		nIDs, err := mgr.cache.GetNotarySet(result.Position.Round)
-		if err != nil {
-			return err
-		}
 		if result.Position.Round < DKGDelayRound {
 			mgr.logger.Debug("Calling Network.PullBlocks for fast syncing BA",
 				"hash", result.BlockHash)
@@ -288,13 +300,19 @@ func (mgr *agreementMgr) processAgreementResult(
 				}
 			}
 		}
-		mgr.logger.Debug("Calling Governance.CRS", "round", result.Position.Round)
-		crs := utils.GetCRSWithPanic(mgr.gov, result.Position.Round, mgr.logger)
-		leader, err := mgr.cache.GetLeaderNode(result.Position)
+		setting := mgr.generateSetting(result.Position.Round)
+		if setting == nil {
+			mgr.logger.Warn("unable to get setting", "round",
+				result.Position.Round)
+			return ErrConfigurationNotReady
+		}
+		leader, err := mgr.calcLeader(setting.dkgSet, setting.crs, result.Position)
 		if err != nil {
 			return err
 		}
-		mgr.baModule.restart(nIDs, result.Position, leader, crs)
+		mgr.baModule.restart(
+			setting.dkgSet, setting.threshold,
+			result.Position, leader, setting.crs)
 		if result.Position.Round >= DKGDelayRound {
 			return mgr.baModule.processAgreementResult(result)
 		}
@@ -322,58 +340,87 @@ func (mgr *agreementMgr) stop() {
 	mgr.waitGroup.Wait()
 }
 
+func (mgr *agreementMgr) generateSetting(round uint64) *baRoundSetting {
+	curConfig := mgr.config(round)
+	if curConfig == nil {
+		return nil
+	}
+	var dkgSet map[types.NodeID]struct{}
+	if round >= DKGDelayRound {
+		_, qualidifed, err := typesDKG.CalcQualifyNodes(
+			mgr.gov.DKGMasterPublicKeys(round),
+			mgr.gov.DKGComplaints(round),
+			utils.GetDKGThreshold(mgr.gov.Configuration(round)),
+		)
+		if err != nil {
+			mgr.logger.Error("Failed to get gpk", "round", round, "error", err)
+			return nil
+		}
+		dkgSet = qualidifed
+	}
+	if len(dkgSet) == 0 {
+		var err error
+		dkgSet, err = mgr.cache.GetNotarySet(round)
+		if err != nil {
+			mgr.logger.Error("Failed to get notarySet", "round", round)
+			return nil
+		}
+	}
+	return &baRoundSetting{
+		crs:    curConfig.crs,
+		dkgSet: dkgSet,
+		round:  round,
+		threshold: utils.GetBAThreshold(&types.Config{
+			NotarySetSize: curConfig.notarySetSize}),
+	}
+}
+
 func (mgr *agreementMgr) runBA(initRound uint64) {
 	// These are round based variables.
 	var (
 		currentRound uint64
 		nextRound    = initRound
 		curConfig    = mgr.config(initRound)
-		setting      = baRoundSetting{}
+		setting      = &baRoundSetting{}
 		tickDuration time.Duration
+		ticker       Ticker
 	)
 
 	// Check if this routine needs to awake in this round and prepare essential
 	// variables when yes.
-	checkRound := func() (isNotary bool) {
+	checkRound := func() (isDKG bool) {
 		defer func() {
 			currentRound = nextRound
 			nextRound++
 		}()
 		// Wait until the configuartion for next round is ready.
 		for {
-			if curConfig = mgr.config(nextRound); curConfig != nil {
+			if setting = mgr.generateSetting(nextRound); setting != nil {
 				break
 			} else {
 				mgr.logger.Debug("Round is not ready", "round", nextRound)
 				time.Sleep(1 * time.Second)
 			}
 		}
-		// Check if this node in notary set of this chain in this round.
-		notarySet, err := mgr.cache.GetNotarySet(nextRound)
-		if err != nil {
-			panic(err)
-		}
-		setting.crs = curConfig.crs
-		setting.notarySet = notarySet
-		setting.round = nextRound
-		_, isNotary = setting.notarySet[mgr.ID]
-		if isNotary {
-			mgr.logger.Info("Selected as notary set",
+		_, isDKG = setting.dkgSet[mgr.ID]
+		if isDKG {
+			mgr.logger.Info("Selected as dkg set",
 				"ID", mgr.ID,
 				"round", nextRound)
 		} else {
-			mgr.logger.Info("Not selected as notary set",
+			mgr.logger.Info("Not selected as dkg set",
 				"ID", mgr.ID,
 				"round", nextRound)
 		}
 		// Setup ticker
 		if tickDuration != curConfig.lambdaBA {
-			if setting.ticker != nil {
-				setting.ticker.Stop()
+			if ticker != nil {
+				ticker.Stop()
 			}
-			setting.ticker = newTicker(mgr.gov, nextRound, TickerBA)
+			ticker = newTicker(mgr.gov, nextRound, TickerBA)
 			tickDuration = curConfig.lambdaBA
 		}
+		setting.ticker = ticker
 		return
 	}
 Loop:
@@ -403,7 +450,7 @@ Loop:
 			mgr.recv.npks = nil
 			mgr.recv.psigSigner = nil
 		}
-		if err := mgr.baRoutineForOneRound(&setting); err != nil {
+		if err := mgr.baRoutineForOneRound(setting); err != nil {
 			mgr.logger.Error("BA routine failed",
 				"error", err,
 				"nodeID", mgr.ID)
@@ -473,13 +520,13 @@ func (mgr *agreementMgr) baRoutineForOneRound(
 		}
 		oldPos = nextPos
 		var leader types.NodeID
-		leader, err = mgr.cache.GetLeaderNode(nextPos)
+		leader, err = mgr.calcLeader(setting.dkgSet, setting.crs, nextPos)
 		if err != nil {
 			return
 		}
 		time.Sleep(nextTime.Sub(time.Now()))
 		setting.ticker.Restart()
-		agr.restart(setting.notarySet, nextPos, leader, setting.crs)
+		agr.restart(setting.dkgSet, setting.threshold, nextPos, leader, setting.crs)
 		return
 	}
 Loop:
diff --git a/core/agreement-state_test.go b/core/agreement-state_test.go
index 0b8605d..cbcbc63 100644
--- a/core/agreement-state_test.go
+++ b/core/agreement-state_test.go
@@ -125,7 +125,11 @@ func (s *AgreementStateTestSuite) newAgreement(numNode int) *agreement {
 		s.signers[s.ID],
 		logger,
 	)
-	agreement.restart(notarySet, types.Position{Height: types.GenesisHeight},
+	agreement.restart(notarySet,
+		utils.GetBAThreshold(&types.Config{
+			NotarySetSize: uint32(len(notarySet)),
+		}),
+		types.Position{Height: types.GenesisHeight},
 		types.NodeID{}, common.NewRandomHash())
 	return agreement
 }
diff --git a/core/agreement.go b/core/agreement.go
index 3745848..b122a4d 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -160,7 +160,8 @@ func newAgreement(
 
 // restart the agreement
 func (a *agreement) restart(
-	notarySet map[types.NodeID]struct{}, aID types.Position, leader types.NodeID,
+	notarySet map[types.NodeID]struct{},
+	threshold int, aID types.Position, leader types.NodeID,
 	crs common.Hash) {
 	if !func() bool {
 		a.lock.Lock()
@@ -181,7 +182,7 @@ func (a *agreement) restart(
 		a.data.votes[1] = newVoteListMap()
 		a.data.period = 2
 		a.data.blocks = make(map[types.NodeID]*types.Block)
-		a.data.requiredVote = len(notarySet)*2/3 + 1
+		a.data.requiredVote = threshold
 		a.data.leader.restart(crs)
 		a.data.lockValue = types.SkipBlockHash
 		a.data.lockIter = 0
@@ -289,9 +290,11 @@ func (a *agreement) restart(
 }
 
 func (a *agreement) stop() {
-	a.restart(make(map[types.NodeID]struct{}), types.Position{
-		Height: math.MaxUint64,
-	}, types.NodeID{}, common.Hash{})
+	a.restart(make(map[types.NodeID]struct{}), int(math.MaxInt32),
+		types.Position{
+			Height: math.MaxUint64,
+		},
+		types.NodeID{}, common.Hash{})
 }
 
 func isStop(aID types.Position) bool {
diff --git a/core/agreement_test.go b/core/agreement_test.go
index 33e398c..9afc0f2 100644
--- a/core/agreement_test.go
+++ b/core/agreement_test.go
@@ -157,7 +157,9 @@ func (s *AgreementTestSuite) newAgreement(
 		s.signers[s.ID],
 		logger,
 	)
-	agreement.restart(notarySet, s.agreementID, leaderNode,
+	agreement.restart(notarySet, utils.GetBAThreshold(&types.Config{
+		NotarySetSize: uint32(len(notarySet)),
+	}), s.agreementID, leaderNode,
 		common.NewRandomHash())
 	s.agreement = append(s.agreement, agreement)
 	return agreement, leaderNode
diff --git a/core/utils/nodeset-cache.go b/core/utils/nodeset-cache.go
index 89ebd24..89dcfc8 100644
--- a/core/utils/nodeset-cache.go
+++ b/core/utils/nodeset-cache.go
@@ -36,10 +36,9 @@ var (
 )
 
 type sets struct {
-	crs        common.Hash
-	nodeSet    *types.NodeSet
-	notarySet  map[types.NodeID]struct{}
-	leaderNode map[uint64]types.NodeID
+	crs       common.Hash
+	nodeSet   *types.NodeSet
+	notarySet map[types.NodeID]struct{}
 }
 
 // NodeSetCacheInterface interface specifies interface used by NodeSetCache.
@@ -133,30 +132,6 @@ func (cache *NodeSetCache) GetNotarySet(
 	return cache.cloneMap(IDs.notarySet), nil
 }
 
-// GetLeaderNode returns the BA leader of the position.
-func (cache *NodeSetCache) GetLeaderNode(pos types.Position) (
-	types.NodeID, error) {
-	IDs, err := cache.getOrUpdate(pos.Round)
-	if err != nil {
-		return types.NodeID{}, err
-	}
-	cache.lock.Lock()
-	defer cache.lock.Unlock()
-	if _, exist := IDs.leaderNode[pos.Height]; !exist {
-		notarySet := types.NewNodeSetFromMap(IDs.notarySet)
-		leader := notarySet.GetSubSet(1, types.NewNodeLeaderTarget(
-			IDs.crs, pos.Height))
-		if len(leader) != 1 {
-			panic(errors.New("length of leader is not one"))
-		}
-		for nID := range leader {
-			IDs.leaderNode[pos.Height] = nID
-			break
-		}
-	}
-	return IDs.leaderNode[pos.Height], nil
-}
-
 // Purge a specific round.
 func (cache *NodeSetCache) Purge(rID uint64) {
 	cache.lock.Lock()
@@ -238,10 +213,9 @@ func (cache *NodeSetCache) update(round uint64) (nIDs *sets, err error) {
 		return
 	}
 	nIDs = &sets{
-		crs:        crs,
-		nodeSet:    nodeSet,
-		notarySet:  make(map[types.NodeID]struct{}),
-		leaderNode: make(map[uint64]types.NodeID, cfg.RoundLength),
+		crs:       crs,
+		nodeSet:   nodeSet,
+		notarySet: make(map[types.NodeID]struct{}),
 	}
 	nIDs.notarySet = nodeSet.GetSubSet(
 		int(cfg.NotarySetSize), types.NewNotarySetTarget(crs))
diff --git a/core/utils/nodeset-cache_test.go b/core/utils/nodeset-cache_test.go
index 0ee3883..b9052c8 100644
--- a/core/utils/nodeset-cache_test.go
+++ b/core/utils/nodeset-cache_test.go
@@ -90,14 +90,6 @@ func (s *NodeSetCacheTestSuite) TestBasicUsage() {
 	notarySet, err := cache.GetNotarySet(0)
 	req.NoError(err)
 	chk(cache, 0, notarySet)
-	leaderNode, err := cache.GetLeaderNode(types.Position{
-		Round:  uint64(0),
-		Height: uint64(10),
-	})
-	req.NoError(err)
-	chk(cache, 0, map[types.NodeID]struct{}{
-		leaderNode: struct{}{},
-	})
 	// Try to get round 1.
 	nodeSet1, err := cache.GetNodeSet(1)
 	req.NoError(err)
diff --git a/core/utils/utils.go b/core/utils/utils.go
index e6739ce..1a372c7 100644
--- a/core/utils/utils.go
+++ b/core/utils/utils.go
@@ -144,6 +144,11 @@ func GetDKGValidThreshold(config *types.Config) int {
 	return int(config.NotarySetSize * 5 / 6)
 }
 
+// GetBAThreshold return threshold for BA votes.
+func GetBAThreshold(config *types.Config) int {
+	return int(config.NotarySetSize*2/3 + 1)
+}
+
 // GetNextRoundValidationHeight returns the block height to check if the next
 // round is ready.
 func GetNextRoundValidationHeight(begin, length uint64) uint64 {