core: sync BA (#264)

3 files changed, 72 insertions, 19 deletions

diff --git a/core/agreement.go b/core/agreement.go
index cc7af4f..13f39ce 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -169,7 +169,9 @@ func (a *agreement) restart(
 		defer a.lock.Unlock()
 		newPendingBlock := make([]pendingBlock, 0)
 		for _, pending := range a.pendingBlock {
-			if pending.block.Position == aID {
+			if aID.Newer(&pending.block.Position) {
+				continue
+			} else if pending.block.Position == aID {
 				replayBlock = append(replayBlock, pending.block)
 			} else if pending.receivedTime.After(expireTime) {
 				newPendingBlock = append(newPendingBlock, pending)
@@ -184,7 +186,9 @@ func (a *agreement) restart(
 		defer a.lock.Unlock()
 		newPendingVote := make([]pendingVote, 0)
 		for _, pending := range a.pendingVote {
-			if pending.vote.Position == aID {
+			if aID.Newer(&pending.vote.Position) {
+				continue
+			} else if pending.vote.Position == aID {
 				replayVote = append(replayVote, pending.vote)
 			} else if pending.receivedTime.After(expireTime) {
 				newPendingVote = append(newPendingVote, pending)
@@ -273,7 +277,11 @@ func (a *agreement) processVote(vote *types.Vote) error {
 	if err := a.sanityCheck(vote); err != nil {
 		return err
 	}
-	if vote.Position != a.agreementID() {
+	aID := a.agreementID()
+	if vote.Position != aID {
+		if aID.Newer(&vote.Position) {
+			return nil
+		}
 		a.lock.Lock()
 		defer a.lock.Unlock()
 		a.pendingVote = append(a.pendingVote, pendingVote{
@@ -372,7 +380,12 @@ func (a *agreement) done() <-chan struct{} {
 func (a *agreement) processBlock(block *types.Block) error {
 	a.data.blocksLock.Lock()
 	defer a.data.blocksLock.Unlock()
-	if block.Position != a.agreementID() {
+
+	aID := a.agreementID()
+	if block.Position != aID {
+		if aID.Newer(&block.Position) {
+			return nil
+		}
 		a.pendingBlock = append(a.pendingBlock, pendingBlock{
 			block:        block,
 			receivedTime: time.Now().UTC(),
diff --git a/core/consensus.go b/core/consensus.go
index bd311ba..23b33ed 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -47,6 +47,8 @@ var (
 		"incorrect agreement result position")
 	ErrNotEnoughVotes = fmt.Errorf(
 		"not enought votes")
+	ErrIncorrectVoteBlockHash = fmt.Errorf(
+		"incorrect vote block hash")
 	ErrIncorrectVoteProposer = fmt.Errorf(
 		"incorrect vote proposer")
 	ErrIncorrectBlockRandomnessResult = fmt.Errorf(
@@ -743,27 +745,19 @@ func (con *Consensus) ProcessVote(vote *types.Vote) (err error) {
 // ProcessAgreementResult processes the randomness request.
 func (con *Consensus) ProcessAgreementResult(
 	rand *types.AgreementResult) error {
-	if rand.Position.Round == 0 {
-		return nil
-	}
-	if !con.ccModule.blockRegistered(rand.BlockHash) {
-		return nil
-	}
-	if DiffUint64(con.round, rand.Position.Round) > 1 {
-		return nil
-	}
-	if len(rand.Votes) <= int(con.currentConfig.NotarySetSize/3*2) {
-		return ErrNotEnoughVotes
-	}
-	if rand.Position.ChainID >= con.currentConfig.NumChains {
-		return ErrIncorrectAgreementResultPosition
-	}
+	// Sanity Check.
 	notarySet, err := con.nodeSetCache.GetNotarySet(
 		rand.Position.Round, rand.Position.ChainID)
 	if err != nil {
 		return err
 	}
+	if len(rand.Votes) < len(notarySet)/3*2+1 {
+		return ErrNotEnoughVotes
+	}
 	for _, vote := range rand.Votes {
+		if vote.BlockHash != rand.BlockHash {
+			return ErrIncorrectVoteBlockHash
+		}
 		if _, exist := notarySet[vote.ProposerID]; !exist {
 			return ErrIncorrectVoteProposer
 		}
@@ -775,6 +769,37 @@ func (con *Consensus) ProcessAgreementResult(
 			return ErrIncorrectVoteSignature
 		}
 	}
+	// Syncing BA Module.
+	agreement := con.baModules[rand.Position.ChainID]
+	aID := agreement.agreementID()
+	if rand.Position.Newer(&aID) {
+		con.logger.Info("Syncing BA", "position", rand.Position)
+		nodes, err := con.nodeSetCache.GetNodeSet(rand.Position.Round)
+		if err != nil {
+			return err
+		}
+		con.logger.Debug("Calling Network.PullBlocks for syncing BA",
+			"hash", rand.BlockHash)
+		con.network.PullBlocks(common.Hashes{rand.BlockHash})
+		nIDs := nodes.GetSubSet(
+			int(con.gov.Configuration(rand.Position.Round).NotarySetSize),
+			types.NewNotarySetTarget(
+				con.gov.CRS(rand.Position.Round), rand.Position.ChainID))
+		for _, vote := range rand.Votes {
+			agreement.processVote(&vote)
+		}
+		agreement.restart(nIDs, rand.Position)
+	}
+	// Calculating randomness.
+	if rand.Position.Round == 0 {
+		return nil
+	}
+	if !con.ccModule.blockRegistered(rand.BlockHash) {
+		return nil
+	}
+	if DiffUint64(con.round, rand.Position.Round) > 1 {
+		return nil
+	}
 	// Sanity check done.
 	if !con.cfgModule.touchTSigHash(rand.BlockHash) {
 		return nil
diff --git a/core/consensus_test.go b/core/consensus_test.go
index b8d25d9..7aa2493 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -183,6 +183,21 @@ func (s *ConsensusTestSuite) prepareConsensus(
 		dMoment, app, gov, db, network, prvKey, &common.NullLogger{})
 	con.ccModule.init(&types.Block{})
 	conn.setCon(nID, con)
+	round := uint64(0)
+	nodes, err := con.nodeSetCache.GetNodeSet(round)
+	s.Require().NoError(err)
+	for i, agreement := range con.baModules {
+		chainID := uint32(i)
+		nIDs := nodes.GetSubSet(
+			int(gov.Configuration(round).NotarySetSize),
+			types.NewNotarySetTarget(
+				gov.CRS(round), chainID))
+		agreement.restart(nIDs, types.Position{
+			Round:   round,
+			ChainID: chainID,
+			Height:  uint64(0),
+		})
+	}
 	return app, con
 }