consensus: make smr run with a single validator. (#284)

* consensus: make smr run with a single validator.

Author: aucusaga

bcs/consensus/xpoa/kernel_contract.go

@@ -33,7 +33,25 @@ func (x *xpoaConsensus) methodEditValidates(contractCtx contract.KContext) (*con
 	if err != nil {
 		return common.NewContractErrResponse(common.StatusBadRequest, "invalid acl: pls check accept value."), err
 	}
-	if !x.isAuthAddress(aks, acceptValue) {
+
+	curValiBytes, err := contractCtx.Get(x.election.bindContractBucket,
+		[]byte(fmt.Sprintf("%d_%s", x.election.consensusVersion, validateKeys)))
+	curVali, err := func() ([]string, error) {
+		if err != nil || curValiBytes == nil {
+			return x.election.initValidators, nil
+		}
+		var curValiKey ProposerInfo
+		err = json.Unmarshal(curValiBytes, &curValiKey)
+		if err != nil {
+			x.log.Error("Unmarshal error")
+			return nil, err
+		}
+		return curValiKey.Address, nil
+	}()
+	if err != nil {
+		return common.NewContractErrResponse(common.StatusBadRequest, err.Error()), err
+	}
+	if !x.isAuthAddress(curVali, aks, acceptValue, x.election.enableBFT) {
 		return common.NewContractErrResponse(common.StatusBadRequest, aclErr.Error()), aclErr
 	}
 
@@ -87,10 +105,18 @@ func (x *xpoaConsensus) methodGetValidates(contractCtx contract.KContext) (*cont
 }
 
 // isAuthAddress 判断输入aks是否能在贪心下仍能满足签名数量>33%(Chained-BFT装载) or 50%(一般情况)
-func (x *xpoaConsensus) isAuthAddress(aks map[string]float64, threshold float64) bool {
+func (x *xpoaConsensus) isAuthAddress(validators []string, aks map[string]float64, threshold float64, enableBFT bool) bool {
+	// 0. 是否是单个候选人
+	if len(validators) == 1 {
+		weight, ok := aks[validators[0]]
+		if !ok {
+			return false
+		}
+		return weight >= threshold
+	}
 	// 1. 判断aks中的地址是否是当前集合地址
 	for addr, _ := range aks {
-		if !Find(addr, x.election.validators) {
+		if !Find(addr, validators) {
 			return false
 		}
 	}
@@ -106,15 +132,14 @@ func (x *xpoaConsensus) isAuthAddress(aks map[string]float64, threshold float64)
 	greedyCount := 0
 	sum := threshold
 	for i := 0; i < len(aks); i++ {
-		if sum > 0 {
-			sum -= s[i].Weight
-			greedyCount++
-			continue
+		if sum <= 0 {
+			break
 		}
-		break
+		sum -= s[i].Weight
+		greedyCount++
 	}
-	if !x.election.enableBFT {
-		return greedyCount >= len(x.election.validators)/2+1
+	if !enableBFT {
+		return greedyCount >= len(validators)/2+1
 	}
-	return CalFault(int64(greedyCount), int64(len(x.election.validators)))
+	return CalFault(int64(greedyCount), int64(len(validators)))
 }

bcs/consensus/xpoa/kernel_contract_test.go

@@ -12,23 +12,15 @@ var (
 		"dpzuVdosQrF2kmzumhVeFQZa1aYcdgFpN": 0.5,
 		"WNWk3ekXeM5M2232dY2uCJmEqWhfQiDYT": 0.5,
 	}
-)
-
-func TestIsAuthAddress(t *testing.T) {
-	cCtx, err := prepare(getXpoaConsensusConf())
-	if err != nil {
-		t.Error("prepare error", "error", err)
-		return
-	}
-	i := NewXpoaConsensus(*cCtx, getConfig(getXpoaConsensusConf()))
-	xpoa, ok := i.(*xpoaConsensus)
-	if !ok {
-		t.Error("transfer err.")
+	aks2 = map[string]float64{
+		"dpzuVdosQrF2kmzumhVeFQZa1aYcdgFpN": 0.5,
+		"WNWk3ekXeM5M2232dY2uCJmEqWhfQiDYT": 0.6,
 	}
-	if !xpoa.isAuthAddress(aks, 0.6) {
-		t.Error("isAuthAddress err.")
+	aks3 = map[string]float64{
+		"dpzuVdosQrF2kmzumhVeFQZa1aYcdgFpN": 0.4,
+		"WNWk3ekXeM5M2232dY2uCJmEqWhfQiDYT": 0.6,
 	}
-}
+)
 
 func NewEditArgs() map[string][]byte {
 	a := make(map[string][]byte)
@@ -85,3 +77,36 @@ func TestMethodGetValidates(t *testing.T) {
 		return
 	}
 }
+
+func TestIsAuthAddress(t *testing.T) {
+	cCtx, err := prepare(getXpoaConsensusConf())
+	if err != nil {
+		t.Error("prepare error", "error", err)
+		return
+	}
+	i := NewXpoaConsensus(*cCtx, getConfig(getXpoaConsensusConf()))
+	xpoa, ok := i.(*xpoaConsensus)
+	if !ok {
+		t.Error("transfer err.")
+		return
+	}
+	v1 := []string{"dpzuVdosQrF2kmzumhVeFQZa1aYcdgFpN", "WNWk3ekXeM5M2232dY2uCJmEqWhfQiDYT"}
+	if !xpoa.isAuthAddress(v1, aks, 0.6, false) {
+		t.Error("isAuthAddress err.")
+		return
+	}
+	v2 := []string{"dpzuVdosQrF2kmzumhVeFQZa1aYcdgFpN"}
+	if xpoa.isAuthAddress(v2, aks2, 0.6, true) {
+		t.Error("isAuthAddress err.")
+		return
+	}
+	v3 := []string{"WNWk3ekXeM5M2232dY2uCJmEqWhfQiDYT"}
+	if !xpoa.isAuthAddress(v3, aks2, 0.6, true) {
+		t.Error("isAuthAddress err.")
+		return
+	}
+	v4 := []string{"dpzuVdosQrF2kmzumhVeFQZa1aYcdgFpN", "WNWk3ekXeM5M2232dY2uCJmEqWhfQiDYT"}
+	if !xpoa.isAuthAddress(v4, aks2, 0.7, true) {
+		t.Error("isAuthAddress err.")
+	}
+}

kernel/consensus/base/driver/chained-bft/smr.go

@@ -219,14 +219,11 @@ func (s *Smr) ResetProposerStatus(tipBlock cctx.BlockInterface,
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
 
-	if bytes.Equal(s.getHighQC().GetProposalId(), tipBlock.GetBlockid()) {
+	if bytes.Equal(s.getHighQC().GetProposalId(), tipBlock.GetBlockid()) &&
+		s.validNewHighQC(tipBlock.GetBlockid(), validators) {
 		// 此处需要获取带签名的完整Justify
 		return false, s.getCompleteHighQC(), nil
 	}
-	// 单个节点不存在投票验证的hotstuff流程，因此返回true
-	if len(validators) == 1 {
-		return false, nil, nil
-	}
 
 	// 从当前TipBlock开始往前追溯，交给smr根据状态进行回滚。
 	// 在本地状态树上找到指代TipBlock的QC，若找不到，则在状态树上找和TipBlock同一分支上的最近值
@@ -384,10 +381,43 @@ func (s *Smr) ProcessProposal(viewNumber int64, proposalID []byte, parentID []by
 	}
 	go s.p2p.SendMessage(createNewBCtx(), netMsg, p2p.WithAccounts(s.removeLocalValidator(validatesIpInfo)))
 	s.localProposal.Store(utils.F(proposalID), proposal.Timestamp)
-	s.log.Debug("smr:ProcessProposal::new proposal has been made", "address", s.address, "proposalID", utils.F(proposalID))
+	// 若为单候选人情况，则此处需要特殊处理，矿工需要给自己提前签名
+	if len(validatesIpInfo) == 1 {
+		s.voteToSelf(viewNumber, proposalID, parentQuorumCert)
+	}
+	s.log.Debug("smr:ProcessProposal::new proposal has been made", "address", s.address, "proposalID", utils.F(proposalID), "target", validatesIpInfo)
 	return nil
 }
 
+func (s *Smr) voteToSelf(viewNumber int64, proposalID []byte, parent storage.QuorumCertInterface) {
+	selfVote := &storage.VoteInfo{
+		ProposalId:   proposalID,
+		ProposalView: viewNumber,
+		ParentId:     parent.GetProposalId(),
+	}
+	selfLedgerInfo := &storage.LedgerCommitInfo{
+		VoteInfoHash: proposalID,
+	}
+	selfQC := storage.NewQuorumCert(selfVote, selfLedgerInfo, nil)
+	selfSign, err := s.cryptoClient.SignVoteMsg(proposalID)
+	if err != nil {
+		s.log.Error("smr::voteProposal::voteToSelf error", "err", err)
+		return
+	}
+	s.qcVoteMsgs.LoadOrStore(utils.F(proposalID), []*chainedBftPb.QuorumCertSign{selfSign})
+	selfNode := &storage.ProposalNode{
+		In: selfQC,
+	}
+	if err := s.qcTree.UpdateQcStatus(selfNode); err != nil {
+		s.log.Error("smr::voteProposal::updateQcStatus error", "err", err)
+		return
+	}
+	// 更新本地smr状态机
+	s.pacemaker.AdvanceView(selfQC)
+	s.qcTree.UpdateHighQC(proposalID)
+	s.log.Debug("smr:voteProposal::done local voting", "address", s.address, "proposalID", utils.F(proposalID))
+}
+
 // reloadJustifyQC 与LibraBFT不同，返回一个指定的parentQC
 func (s *Smr) reloadJustifyQC(parentID []byte) (storage.QuorumCertInterface, error) {
 	// 第一次proposal，highQC==rootQC==genesisQC
@@ -695,6 +725,9 @@ func (s *Smr) validNewHighQC(inProposalId []byte, validators []string) bool {
 	if !ok {
 		return false
 	}
+	if len(validators) == 1 {
+		return len(signs) == len(validators)
+	}
 	return s.saftyrules.CalVotesThreshold(len(signs), len(validators))
 }