optimize consensus nodes sync block (#1137)

* add log in pendingblock

* modify print log

* modify sync block

* modify check start sync  commit block num

* start CommitTimeout if more endorsed after commitDone

1. start CommitBlockTimeout when committer receiving more endorses
2. commit-sig overwrites old endorse-sig

* optimize sync block

* modify sync block

* Revert "modify check start sync  commit block num"

This reverts commit 04d88d5595b1e2c962ba329a32b58157e524ad3b.

* Revert "modify sync block"

This reverts commit aff87f3547e6a6c7ebbdbfe1c456555b7cff6fea.

* go format files

* modify log level and remove unused code

Co-authored-by: Edmond <[email protected]>

consensus/vbft/block_pool.go

@@ -158,7 +158,7 @@ func (pool *BlockPool) newBlockProposal(msg *blockProposalMsg) error {
 		Signature:        msg.Block.Block.Header.SigData[0],
 		ForEmpty:         false,
 	}
-	pool.addBlockEndorsementLocked(msg.GetBlockNum(), proposer, eSig)
+	pool.addBlockEndorsementLocked(msg.GetBlockNum(), proposer, eSig, false)
 	return nil
 }
 
@@ -262,10 +262,10 @@ func (pool *BlockPool) setProposalEndorsed(proposal *blockProposalMsg, forEmpty
 	return nil
 }
 
-func (pool *BlockPool) addBlockEndorsementLocked(blkNum uint32, endorser uint32, eSig *CandidateEndorseSigInfo) {
+func (pool *BlockPool) addBlockEndorsementLocked(blkNum uint32, endorser uint32, eSig *CandidateEndorseSigInfo, commitment bool) {
 	candidate := pool.getCandidateInfoLocked(blkNum)
 
-	if eSigs, present := candidate.EndorseSigs[endorser]; present {
+	if eSigs, present := candidate.EndorseSigs[endorser]; present && !commitment {
 		for _, eSig := range eSigs {
 			if eSig.ForEmpty {
 				// has endorsed for empty, ignore new endorsement
@@ -303,7 +303,7 @@ func (pool *BlockPool) newBlockEndorsement(msg *blockEndorseMsg) {
 		Signature:        msg.EndorserSig,
 		ForEmpty:         msg.EndorseForEmpty,
 	}
-	pool.addBlockEndorsementLocked(msg.GetBlockNum(), msg.Endorser, eSig)
+	pool.addBlockEndorsementLocked(msg.GetBlockNum(), msg.Endorser, eSig, false)
 }
 
 //
@@ -474,15 +474,15 @@ func (pool *BlockPool) newBlockCommitment(msg *blockCommitMsg) error {
 			Signature:        sig,
 			ForEmpty:         msg.CommitForEmpty,
 		}
-		pool.addBlockEndorsementLocked(blkNum, endorser, eSig)
+		pool.addBlockEndorsementLocked(blkNum, endorser, eSig, false)
 	}
 
 	// add committer sig
 	pool.addBlockEndorsementLocked(blkNum, msg.Committer, &CandidateEndorseSigInfo{
 		EndorsedProposer: msg.BlockProposer,
 		Signature:        msg.CommitterSig,
 		ForEmpty:         msg.CommitForEmpty,
-	})
+	}, true)
 
 	// add msg to commit-msgs
 	candidate.CommitMsgs = append(candidate.CommitMsgs, msg)

consensus/vbft/chain_store.go

@@ -62,6 +62,7 @@ func OpenBlockStore(db *ledger.Ledger, serverPid *actor.PID) (*ChainStore, error
 	if err != nil {
 		return nil, err
 	}
+	log.Debugf("chainstore openblockstore pendingBlocks height:%d,", chainstore.chainedBlockNum)
 	chainstore.pendingBlocks[chainstore.chainedBlockNum] = &PendingBlock{block: block, execResult: &store.ExecuteResult{WriteSet: writeSet, MerkleRoot: merkleRoot}, hasSubmitted: true}
 	return chainstore, nil
 }
@@ -107,6 +108,7 @@ func (self *ChainStore) ReloadFromLedger() {
 				newPending[blkNum] = blk
 			}
 		}
+		log.Debug("chainstore ReloadFromLedger pendingBlocks")
 		// update pending blocks
 		self.pendingBlocks = newPending
 	}
@@ -135,6 +137,7 @@ func (self *ChainStore) AddBlock(block *Block) error {
 		log.Errorf("chainstore AddBlock GetBlockExecResult: %s", err)
 		return fmt.Errorf("chainstore AddBlock GetBlockExecResult: %s", err)
 	}
+	log.Debugf("chainstore addblock pendingBlocks height:%d,block height:%d", blkNum, block.getBlockNum())
 	self.pendingBlocks[blkNum] = &PendingBlock{block: block, execResult: &execResult, hasSubmitted: false}
 
 	if self.pid != nil {
@@ -157,6 +160,7 @@ func (self *ChainStore) submitBlock(blkNum uint32) error {
 			return fmt.Errorf("ledger add submitBlk (%d, %d, %d) failed: %s", blkNum, self.GetChainedBlockNum(), self.db.GetCurrentBlockHeight(), err)
 		}
 		if _, present := self.pendingBlocks[blkNum-1]; present {
+			log.Infof("chainstore submitBlock delete pendingBlocks height:%d", blkNum-1)
 			delete(self.pendingBlocks, blkNum-1)
 		}
 		submitBlk.hasSubmitted = true

consensus/vbft/node_sync.go

@@ -406,24 +406,13 @@ func (self *PeerSyncer) run() {
 		}
 
 		var proposalBlock *Block
-		for _, p := range self.server.msgPool.GetProposalMsgs(blkNum) {
-			m, ok := p.(*blockProposalMsg)
-			if !ok {
-				panic("")
-			}
-			if m.Block.getProposer() == blkProposers[blkNum] {
-				proposalBlock = m.Block
-				break
-			}
-		}
-
+		proposalBlock, _ = self.server.blockPool.getSealedBlock(blkNum)
 		if proposalBlock == nil {
 			if proposalBlock, err = self.requestBlock(blkNum); err != nil {
 				log.Errorf("failed to get block %d from peer %d: %s", blkNum, self.peerIdx, err)
 				return
 			}
 		}
-
 		if err := self.fetchedBlock(blkNum, proposalBlock); err != nil {
 			log.Errorf("failed to commit block %d from peer syncer %d to syncer: %s",
 				blkNum, self.peerIdx, err)

consensus/vbft/service.go

@@ -1267,6 +1267,11 @@ func (self *Server) processMsgEvent() error {
 
 				// if had committed for current round, skip the following steps
 				if self.blockPool.committedForBlock(msgBlkNum) {
+					// get more endorse msg after committed, trigger seal-block-timeout
+					if err := self.timer.StartCommitTimer(msgBlkNum); err != nil {
+						log.Errorf("server %d start commit timer for %d(%d), block %d, err: %s",
+							self.Index, pMsg.Endorser, pMsg.EndorsedProposer, msgBlkNum, err)
+					}
 					return nil
 				}