core: BA 3.0 (#408)

* Add v3 enum

* Add BA leader calculation

* Fast BA

* Add unittest for Fast BA

* Add comment

* Select leader in notarySet

12 files changed, 401 insertions, 55 deletions

diff --git a/core/agreement-mgr.go b/core/agreement-mgr.go
index e8cafbd..1bd077b 100644
--- a/core/agreement-mgr.go
+++ b/core/agreement-mgr.go
@@ -266,7 +266,11 @@ func (mgr *agreementMgr) processAgreementResult(
 				return err
 			}
 		}
-		agreement.restart(nIDs, result.Position, crs)
+		leader, err := mgr.cache.GetLeaderNode(result.Position)
+		if err != nil {
+			return err
+		}
+		agreement.restart(nIDs, result.Position, leader, crs)
 	}
 	return nil
 }
@@ -332,14 +336,12 @@ func (mgr *agreementMgr) runBA(initRound uint64, chainID uint32) {
 			return
 		}
 		// Check if this node in notary set of this chain in this round.
-		nodeSet, err := mgr.cache.GetNodeSet(nextRound)
+		notarySet, err := mgr.cache.GetNotarySet(nextRound, chainID)
 		if err != nil {
 			panic(err)
 		}
 		setting.crs = config.crs
-		setting.notarySet = nodeSet.GetSubSet(
-			int(config.notarySetSize),
-			types.NewNotarySetTarget(config.crs, chainID))
+		setting.notarySet = notarySet
 		_, isNotary = setting.notarySet[mgr.ID]
 		if isNotary {
 			mgr.logger.Info("selected as notary set",
@@ -461,7 +463,11 @@ Loop:
 				Height:  nextHeight,
 			}
 			oldPos = nextPos
-			agr.restart(setting.notarySet, nextPos, setting.crs)
+			leader, err := mgr.cache.GetLeaderNode(nextPos)
+			if err != nil {
+				return err
+			}
+			agr.restart(setting.notarySet, nextPos, leader, setting.crs)
 		default:
 		}
 		if agr.pullVotes() {
diff --git a/core/agreement-state.go b/core/agreement-state.go
index 69700b5..266e442 100644
--- a/core/agreement-state.go
+++ b/core/agreement-state.go
@@ -35,7 +35,10 @@ type agreementStateType int
 
 // agreementStateType enum.
 const (
-	stateInitial agreementStateType = iota
+	stateFast agreementStateType = iota
+	stateFastVote
+	stateFastRollback
+	stateInitial
 	statePreCommit
 	stateCommit
 	stateForward
@@ -58,6 +61,63 @@ type agreementState interface {
 	clocks() int
 }
 
+//----- FastState -----
+type fastState struct {
+	a *agreementData
+}
+
+func newFastState(a *agreementData) *fastState {
+	return &fastState{a: a}
+}
+
+func (s *fastState) state() agreementStateType { return stateFast }
+func (s *fastState) clocks() int               { return 0 }
+func (s *fastState) nextState() (agreementState, error) {
+	if func() bool {
+		s.a.lock.Lock()
+		defer s.a.lock.Unlock()
+		return s.a.isLeader
+	}() {
+		hash := s.a.recv.ProposeBlock()
+		if hash != nullBlockHash {
+			s.a.lock.Lock()
+			defer s.a.lock.Unlock()
+			s.a.recv.ProposeVote(types.NewVote(types.VoteFast, hash, s.a.period))
+		}
+	}
+	return newFastVoteState(s.a), nil
+}
+
+//----- FastVoteState -----
+type fastVoteState struct {
+	a *agreementData
+}
+
+func newFastVoteState(a *agreementData) *fastVoteState {
+	return &fastVoteState{a: a}
+}
+
+func (s *fastVoteState) state() agreementStateType { return stateFastVote }
+func (s *fastVoteState) clocks() int               { return 2 }
+func (s *fastVoteState) nextState() (agreementState, error) {
+	return newFastRollbackState(s.a), nil
+}
+
+//----- FastRollbackState -----
+type fastRollbackState struct {
+	a *agreementData
+}
+
+func newFastRollbackState(a *agreementData) *fastRollbackState {
+	return &fastRollbackState{a: a}
+}
+
+func (s *fastRollbackState) state() agreementStateType { return stateFastRollback }
+func (s *fastRollbackState) clocks() int               { return 1 }
+func (s *fastRollbackState) nextState() (agreementState, error) {
+	return newInitialState(s.a), nil
+}
+
 //----- InitialState -----
 type initialState struct {
 	a *agreementData
@@ -70,10 +130,17 @@ func newInitialState(a *agreementData) *initialState {
 func (s *initialState) state() agreementStateType { return stateInitial }
 func (s *initialState) clocks() int               { return 0 }
 func (s *initialState) nextState() (agreementState, error) {
-	hash := s.a.recv.ProposeBlock()
-	s.a.lock.Lock()
-	defer s.a.lock.Unlock()
-	s.a.recv.ProposeVote(types.NewVote(types.VoteInit, hash, s.a.period))
+	if func() bool {
+		s.a.lock.Lock()
+		defer s.a.lock.Unlock()
+		return !s.a.isLeader
+	}() {
+		// Leader already proposed block in fastState.
+		hash := s.a.recv.ProposeBlock()
+		s.a.lock.Lock()
+		defer s.a.lock.Unlock()
+		s.a.recv.ProposeVote(types.NewVote(types.VoteInit, hash, s.a.period))
+	}
 	return newPreCommitState(s.a), nil
 }
 
diff --git a/core/agreement-state_test.go b/core/agreement-state_test.go
index bb624ea..43557f1 100644
--- a/core/agreement-state_test.go
+++ b/core/agreement-state_test.go
@@ -115,10 +115,73 @@ func (s *AgreementStateTestSuite) newAgreement(numNode int) *agreement {
 		leader,
 		s.signers[s.ID],
 	)
-	agreement.restart(notarySet, types.Position{}, common.NewRandomHash())
+	agreement.restart(notarySet, types.Position{}, types.NodeID{}, common.NewRandomHash())
 	return agreement
 }
 
+func (s *AgreementStateTestSuite) TestFastStateLeader() {
+	a := s.newAgreement(4)
+	state := newFastState(a.data)
+	s.Equal(stateFast, state.state())
+	s.Equal(0, state.clocks())
+
+	// Proposing a new block if it's leader.
+	a.data.period = 1
+	a.data.isLeader = true
+	newState, err := state.nextState()
+	s.Require().NoError(err)
+	s.Require().Len(s.blockChan, 1)
+	proposedBlock := <-s.blockChan
+	s.NotEqual(common.Hash{}, proposedBlock)
+	s.Require().NoError(a.processBlock(s.block[proposedBlock]))
+	s.Require().Len(s.voteChan, 1)
+	proposedVote := <-s.voteChan
+	s.Equal(proposedBlock, proposedVote.BlockHash)
+	s.Equal(types.VoteFast, proposedVote.Type)
+	s.Equal(stateFastVote, newState.state())
+}
+
+func (s *AgreementStateTestSuite) TestFastStateNotLeader() {
+	a := s.newAgreement(4)
+	state := newFastState(a.data)
+	s.Equal(stateFast, state.state())
+	s.Equal(0, state.clocks())
+
+	// Not proposing any block if it's not leader.
+	a.data.period = 1
+	a.data.isLeader = false
+	newState, err := state.nextState()
+	s.Require().NoError(err)
+	s.Require().Len(s.blockChan, 0)
+	s.Equal(stateFastVote, newState.state())
+}
+
+func (s *AgreementStateTestSuite) TestFastVoteState() {
+	a := s.newAgreement(4)
+	state := newFastVoteState(a.data)
+	s.Equal(stateFastVote, state.state())
+	s.Equal(2, state.clocks())
+
+	// The vote proposed is not implemented inside state.
+	a.data.period = 1
+	newState, err := state.nextState()
+	s.Require().NoError(err)
+	s.Require().Len(s.voteChan, 0)
+	s.Equal(stateFastRollback, newState.state())
+}
+
+func (s *AgreementStateTestSuite) TestFastRollbackState() {
+	a := s.newAgreement(4)
+	state := newFastRollbackState(a.data)
+	s.Equal(stateFastRollback, state.state())
+	s.Equal(1, state.clocks())
+
+	a.data.period = 1
+	newState, err := state.nextState()
+	s.Require().NoError(err)
+	s.Equal(stateInitial, newState.state())
+}
+
 func (s *AgreementStateTestSuite) TestInitialState() {
 	a := s.newAgreement(4)
 	state := newInitialState(a.data)
diff --git a/core/agreement.go b/core/agreement.go
index 262c5d5..91341e3 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -91,6 +91,7 @@ type agreementData struct {
 	recv agreementReceiver
 
 	ID           types.NodeID
+	isLeader     bool
 	leader       *leaderSelector
 	lockValue    common.Hash
 	lockRound    uint64
@@ -140,8 +141,8 @@ func newAgreement(
 
 // restart the agreement
 func (a *agreement) restart(
-	notarySet map[types.NodeID]struct{}, aID types.Position, crs common.Hash) {
-
+	notarySet map[types.NodeID]struct{}, aID types.Position, leader types.NodeID,
+	crs common.Hash) {
 	if !func() bool {
 		a.lock.Lock()
 		defer a.lock.Unlock()
@@ -163,12 +164,16 @@ func (a *agreement) restart(
 		a.data.leader.restart(crs)
 		a.data.lockValue = nullBlockHash
 		a.data.lockRound = 0
+		a.data.isLeader = a.data.ID == leader
 		a.fastForward = make(chan uint64, 1)
 		a.hasOutput = false
-		a.state = newInitialState(a.data)
+		a.state = newFastState(a.data)
 		a.notarySet = notarySet
 		a.candidateBlock = make(map[common.Hash]*types.Block)
-		a.aID.Store(aID)
+		a.aID.Store(struct {
+			pos    types.Position
+			leader types.NodeID
+		}{aID, leader})
 		return true
 	}() {
 		return
@@ -225,7 +230,7 @@ func (a *agreement) restart(
 func (a *agreement) stop() {
 	a.restart(make(map[types.NodeID]struct{}), types.Position{
 		ChainID: math.MaxUint32,
-	}, common.Hash{})
+	}, types.NodeID{}, common.Hash{})
 }
 
 func isStop(aID types.Position) bool {
@@ -249,7 +254,18 @@ func (a *agreement) pullVotes() bool {
 
 // agreementID returns the current agreementID.
 func (a *agreement) agreementID() types.Position {
-	return a.aID.Load().(types.Position)
+	return a.aID.Load().(struct {
+		pos    types.Position
+		leader types.NodeID
+	}).pos
+}
+
+// leader returns the current leader.
+func (a *agreement) leader() types.NodeID {
+	return a.aID.Load().(struct {
+		pos    types.Position
+		leader types.NodeID
+	}).leader
 }
 
 // nextState is called at the specific clock time.
@@ -262,6 +278,8 @@ func (a *agreement) nextState() (err error) {
 		a.state = newSleepState(a.data)
 		return
 	}
+	a.lock.Lock()
+	defer a.lock.Unlock()
 	a.state, err = a.state.nextState()
 	return
 }
@@ -335,12 +353,13 @@ func (a *agreement) processVote(vote *types.Vote) error {
 		a.data.votes[vote.Period] = newVoteListMap()
 	}
 	a.data.votes[vote.Period][vote.Type][vote.ProposerID] = vote
-	if !a.hasOutput && vote.Type == types.VoteCom {
+	if !a.hasOutput &&
+		(vote.Type == types.VoteCom || vote.Type == types.VoteFast) {
 		if hash, ok := a.data.countVoteNoLock(vote.Period, vote.Type); ok &&
 			hash != skipBlockHash {
 			a.hasOutput = true
 			a.data.recv.ConfirmBlock(hash,
-				a.data.votes[vote.Period][types.VoteCom])
+				a.data.votes[vote.Period][vote.Type])
 			return nil
 		}
 	} else if a.hasOutput {
@@ -451,6 +470,39 @@ func (a *agreement) processBlock(block *types.Block) error {
 	}
 	a.data.blocks[block.ProposerID] = block
 	a.addCandidateBlockNoLock(block)
+	if (a.state.state() == stateFast || a.state.state() == stateFastVote) &&
+		block.ProposerID == a.leader() {
+		go func() {
+			for func() bool {
+				a.lock.RLock()
+				defer a.lock.RUnlock()
+				if a.state.state() != stateFast && a.state.state() != stateFastVote {
+					return false
+				}
+				block, exist := a.data.blocks[a.leader()]
+				if !exist {
+					return true
+				}
+				a.data.lock.RLock()
+				defer a.data.lock.RUnlock()
+				ok, err := a.data.leader.validLeader(block)
+				if err != nil {
+					fmt.Println("Error checking validLeader for Fast BA",
+						"error", err, "block", block)
+					return false
+				}
+				if ok {
+					a.data.recv.ProposeVote(
+						types.NewVote(types.VoteFast, block.Hash, a.data.period))
+					return false
+				}
+				return true
+			}() {
+				// TODO(jimmy): retry interval should be related to configurations.
+				time.Sleep(250 * time.Millisecond)
+			}
+		}()
+	}
 	return nil
 }
 
diff --git a/core/agreement_test.go b/core/agreement_test.go
index 9bdc4f8..a95e89a 100644
--- a/core/agreement_test.go
+++ b/core/agreement_test.go
@@ -19,6 +19,7 @@ package core
 
 import (
 	"testing"
+	"time"
 
 	"github.com/dexon-foundation/dexon-consensus/common"
 	"github.com/dexon-foundation/dexon-consensus/core/crypto/ecdsa"
@@ -38,7 +39,9 @@ func (r *agreementTestReceiver) ProposeVote(vote *types.Vote) {
 }
 
 func (r *agreementTestReceiver) ProposeBlock() common.Hash {
-	block := r.s.proposeBlock(r.agreementIndex)
+	block := r.s.proposeBlock(
+		r.s.agreement[r.agreementIndex].data.ID,
+		r.s.agreement[r.agreementIndex].data.leader.hashCRS)
 	r.s.blockChan <- block.Hash
 	return block.Hash
 }
@@ -56,14 +59,16 @@ func (r *agreementTestReceiver) PullBlocks(hashes common.Hashes) {
 }
 
 func (s *AgreementTestSuite) proposeBlock(
-	agreementIdx int) *types.Block {
+	nID types.NodeID, crs common.Hash) *types.Block {
 	block := &types.Block{
-		ProposerID: s.ID,
+		ProposerID: nID,
 		Hash:       common.NewRandomHash(),
 	}
 	s.block[block.Hash] = block
-	s.Require().NoError(s.signers[s.ID].SignCRS(
-		block, s.agreement[agreementIdx].data.leader.hashCRS))
+	signer, exist := s.signers[block.ProposerID]
+	s.Require().True(exist)
+	s.Require().NoError(signer.SignCRS(
+		block, crs))
 	return block
 }
 
@@ -93,11 +98,14 @@ func (s *AgreementTestSuite) SetupTest() {
 	s.pulledBlocks = make(map[common.Hash]struct{})
 }
 
-func (s *AgreementTestSuite) newAgreement(numNotarySet int) *agreement {
+func (s *AgreementTestSuite) newAgreement(
+	numNotarySet, leaderIdx int) (*agreement, types.NodeID) {
+	s.Require().True(leaderIdx < numNotarySet)
 	leader := newLeaderSelector(func(*types.Block) (bool, error) {
 		return true, nil
 	}, &common.NullLogger{})
 	agreementIdx := len(s.agreement)
+	var leaderNode types.NodeID
 	notarySet := make(map[types.NodeID]struct{})
 	for i := 0; i < numNotarySet-1; i++ {
 		prvKey, err := ecdsa.NewPrivateKey()
@@ -105,6 +113,12 @@ func (s *AgreementTestSuite) newAgreement(numNotarySet int) *agreement {
 		nID := types.NewNodeID(prvKey.PublicKey())
 		notarySet[nID] = struct{}{}
 		s.signers[nID] = utils.NewSigner(prvKey)
+		if i == leaderIdx-1 {
+			leaderNode = nID
+		}
+	}
+	if leaderIdx == 0 {
+		leaderNode = s.ID
 	}
 	notarySet[s.ID] = struct{}{}
 	agreement := newAgreement(
@@ -116,9 +130,10 @@ func (s *AgreementTestSuite) newAgreement(numNotarySet int) *agreement {
 		leader,
 		s.signers[s.ID],
 	)
-	agreement.restart(notarySet, types.Position{}, common.NewRandomHash())
+	agreement.restart(notarySet, types.Position{},
+		leaderNode, common.NewRandomHash())
 	s.agreement = append(s.agreement, agreement)
-	return agreement
+	return agreement, leaderNode
 }
 
 func (s *AgreementTestSuite) copyVote(
@@ -138,7 +153,13 @@ func (s *AgreementTestSuite) prepareVote(
 }
 
 func (s *AgreementTestSuite) TestSimpleConfirm() {
-	a := s.newAgreement(4)
+	a, _ := s.newAgreement(4, -1)
+	// FastState
+	a.nextState()
+	// FastVoteState
+	a.nextState()
+	// FastRollbackState
+	a.nextState()
 	// InitialState
 	a.nextState()
 	// PreCommitState
@@ -180,7 +201,13 @@ func (s *AgreementTestSuite) TestSimpleConfirm() {
 }
 
 func (s *AgreementTestSuite) TestPartitionOnCommitVote() {
-	a := s.newAgreement(4)
+	a, _ := s.newAgreement(4, -1)
+	// FastState
+	a.nextState()
+	// FastVoteState
+	a.nextState()
+	// FastRollbackState
+	a.nextState()
 	// InitialState
 	a.nextState()
 	// PreCommitState
@@ -217,9 +244,63 @@ func (s *AgreementTestSuite) TestPartitionOnCommitVote() {
 	s.Require().Len(s.voteChan, 0)
 }
 
+func (s *AgreementTestSuite) TestFastConfirmLeader() {
+	a, leaderNode := s.newAgreement(4, 0)
+	s.Require().Equal(s.ID, leaderNode)
+	// FastState
+	a.nextState()
+	// FastVoteState
+	s.Require().Len(s.blockChan, 1)
+	blockHash := <-s.blockChan
+	block, exist := s.block[blockHash]
+	s.Require().True(exist)
+	s.Require().Equal(s.ID, block.ProposerID)
+	s.Require().NoError(a.processBlock(block))
+	s.Require().Len(s.voteChan, 1)
+	vote := <-s.voteChan
+	s.Equal(types.VoteFast, vote.Type)
+	s.Equal(blockHash, vote.BlockHash)
+	for nID := range s.signers {
+		v := s.copyVote(vote, nID)
+		s.Require().NoError(a.processVote(v))
+	}
+	// We have enough of Fast-Votes.
+	s.Require().Len(s.confirmChan, 1)
+	confirmBlock := <-s.confirmChan
+	s.Equal(blockHash, confirmBlock)
+}
+
+func (s *AgreementTestSuite) TestFastConfirmNonLeader() {
+	a, leaderNode := s.newAgreement(4, 1)
+	s.Require().NotEqual(s.ID, leaderNode)
+	// FastState
+	a.nextState()
+	// FastVoteState
+	s.Require().Len(s.blockChan, 0)
+	block := s.proposeBlock(leaderNode, a.data.leader.hashCRS)
+	s.Require().Equal(leaderNode, block.ProposerID)
+	s.Require().NoError(a.processBlock(block))
+	var vote *types.Vote
+	select {
+	case vote = <-s.voteChan:
+	case <-time.After(500 * time.Millisecond):
+		s.FailNow("Should propose vote")
+	}
+	s.Equal(types.VoteFast, vote.Type)
+	s.Equal(block.Hash, vote.BlockHash)
+	for nID := range s.signers {
+		v := s.copyVote(vote, nID)
+		s.Require().NoError(a.processVote(v))
+	}
+	// We have enough of Fast-Votes.
+	s.Require().Len(s.confirmChan, 1)
+	confirmBlock := <-s.confirmChan
+	s.Equal(block.Hash, confirmBlock)
+}
+
 func (s *AgreementTestSuite) TestFastForwardCond1() {
 	votes := 0
-	a := s.newAgreement(4)
+	a, _ := s.newAgreement(4, -1)
 	a.data.lockRound = 1
 	a.data.period = 3
 	hash := common.NewRandomHash()
@@ -273,7 +354,7 @@ func (s *AgreementTestSuite) TestFastForwardCond1() {
 
 func (s *AgreementTestSuite) TestFastForwardCond2() {
 	votes := 0
-	a := s.newAgreement(4)
+	a, _ := s.newAgreement(4, -1)
 	a.data.period = 1
 	hash := common.NewRandomHash()
 	for nID := range a.notarySet {
@@ -310,7 +391,7 @@ func (s *AgreementTestSuite) TestFastForwardCond2() {
 func (s *AgreementTestSuite) TestFastForwardCond3() {
 	numVotes := 0
 	votes := []*types.Vote{}
-	a := s.newAgreement(4)
+	a, _ := s.newAgreement(4, -1)
 	a.data.period = 1
 	for nID := range a.notarySet {
 		vote := s.prepareVote(nID, types.VoteCom, common.NewRandomHash(), uint64(2))
@@ -337,7 +418,7 @@ func (s *AgreementTestSuite) TestFastForwardCond3() {
 
 func (s *AgreementTestSuite) TestDecide() {
 	votes := 0
-	a := s.newAgreement(4)
+	a, _ := s.newAgreement(4, -1)
 	a.data.period = 5
 
 	// No decide if com-vote on SKIP.
diff --git a/core/consensus.go b/core/consensus.go
index 7ceac51..ec7d9fd 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -104,12 +104,15 @@ func (recv *consensusBAReceiver) ProposeBlock() common.Hash {
 		recv.consensus.logger.Error("unable to propose block")
 		return nullBlockHash
 	}
-	if err := recv.consensus.preProcessBlock(block); err != nil {
-		recv.consensus.logger.Error("Failed to pre-process block", "error", err)
-		return common.Hash{}
-	}
-	recv.consensus.logger.Debug("Calling Network.BroadcastBlock", "block", block)
-	recv.consensus.network.BroadcastBlock(block)
+	go func() {
+		if err := recv.consensus.preProcessBlock(block); err != nil {
+			recv.consensus.logger.Error("Failed to pre-process block", "error", err)
+			return
+		}
+		recv.consensus.logger.Debug("Calling Network.BroadcastBlock",
+			"block", block)
+		recv.consensus.network.BroadcastBlock(block)
+	}()
 	return block.Hash
 }
 
diff --git a/core/consensus_test.go b/core/consensus_test.go
index 7748494..938b56d 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -644,16 +644,12 @@ func (s *ConsensusTestSuite) TestSyncBA() {
 	s.Equal(ErrIncorrectVoteSignature, con.ProcessAgreementResult(baResult))
 	s.Require().NoError(signers[0].SignVote(&baResult.Votes[0]))
 
-	for _, signer := range signers {
-		vote := types.NewVote(types.VoteCom, hash, 0)
-		vote.Position = pos
-		s.Require().NoError(signer.SignVote(vote))
-		baResult.Votes = append(baResult.Votes, *vote)
-	}
-	s.Equal(ErrIncorrectVoteProposer, con.ProcessAgreementResult(baResult))
-
 	baResult.Votes = baResult.Votes[:1]
 	s.Equal(ErrNotEnoughVotes, con.ProcessAgreementResult(baResult))
+	for range signers {
+		baResult.Votes = append(baResult.Votes, baResult.Votes[0])
+	}
+	s.Equal(ErrNotEnoughVotes, con.ProcessAgreementResult(baResult))
 }
 
 func TestConsensus(t *testing.T) {
diff --git a/core/types/nodeset.go b/core/types/nodeset.go
index da615e1..21a1e52 100644
--- a/core/types/nodeset.go
+++ b/core/types/nodeset.go
@@ -41,6 +41,7 @@ type subSetTargetType byte
 const (
 	targetNotarySet subSetTargetType = iota
 	targetDKGSet
+	targetNodeLeader
 )
 
 type nodeRank struct {
@@ -72,6 +73,17 @@ func NewNodeSet() *NodeSet {
 	}
 }
 
+// NewNodeSetFromMap creates a new NodeSet from NodeID map.
+func NewNodeSetFromMap(nodes map[NodeID]struct{}) *NodeSet {
+	nIDs := make(map[NodeID]struct{}, len(nodes))
+	for nID := range nodes {
+		nIDs[nID] = struct{}{}
+	}
+	return &NodeSet{
+		IDs: nIDs,
+	}
+}
+
 // NewNotarySetTarget is the target for getting Notary Set.
 func NewNotarySetTarget(crs common.Hash, chainID uint32) *SubSetTarget {
 	binaryChainID := make([]byte, 4)
@@ -80,11 +92,23 @@ func NewNotarySetTarget(crs common.Hash, chainID uint32) *SubSetTarget {
 	return newTarget(targetNotarySet, crs[:], binaryChainID)
 }
 
-// NewDKGSetTarget  is the target for getting DKG Set.
+// NewDKGSetTarget is the target for getting DKG Set.
 func NewDKGSetTarget(crs common.Hash) *SubSetTarget {
 	return newTarget(targetDKGSet, crs[:])
 }
 
+// NewNodeLeaderTarget is the target for getting leader of fast BA.
+func NewNodeLeaderTarget(crs common.Hash, pos Position) *SubSetTarget {
+	binaryRoundID := make([]byte, 8)
+	binary.LittleEndian.PutUint64(binaryRoundID, pos.Round)
+	binaryChainID := make([]byte, 4)
+	binary.LittleEndian.PutUint32(binaryChainID, pos.ChainID)
+	binaryHeight := make([]byte, 8)
+	binary.LittleEndian.PutUint64(binaryHeight, pos.Height)
+	return newTarget(targetNodeLeader, crs[:],
+		binaryRoundID, binaryChainID, binaryHeight)
+}
+
 // Add a NodeID to the set.
 func (ns *NodeSet) Add(ID NodeID) {
 	ns.IDs[ID] = struct{}{}
diff --git a/core/types/vote.go b/core/types/vote.go
index 7601542..97044f5 100644
--- a/core/types/vote.go
+++ b/core/types/vote.go
@@ -32,6 +32,7 @@ const (
 	VoteInit VoteType = iota
 	VotePreCom
 	VoteCom
+	VoteFast
 	// Do not add any type below MaxVoteType.
 	MaxVoteType
 )
diff --git a/core/utils.go b/core/utils.go
index 0c2d155..3b1069e 100644
--- a/core/utils.go
+++ b/core/utils.go
@@ -159,8 +159,10 @@ func VerifyAgreementResult(
 	if len(res.Votes) < len(notarySet)/3*2+1 {
 		return ErrNotEnoughVotes
 	}
-	if len(res.Votes) > len(notarySet) {
-		return ErrIncorrectVoteProposer
+	voted := make(map[types.NodeID]struct{}, len(notarySet))
+	voteType := res.Votes[0].Type
+	if voteType != types.VoteFast && voteType != types.VoteCom {
+		return ErrIncorrectVoteType
 	}
 	for _, vote := range res.Votes {
 		if res.IsEmptyBlock {
@@ -172,7 +174,7 @@ func VerifyAgreementResult(
 				return ErrIncorrectVoteBlockHash
 			}
 		}
-		if vote.Type != types.VoteCom {
+		if vote.Type != voteType {
 			return ErrIncorrectVoteType
 		}
 		if vote.Position != res.Position {
@@ -188,6 +190,10 @@ func VerifyAgreementResult(
 		if !ok {
 			return ErrIncorrectVoteSignature
 		}
+		voted[vote.ProposerID] = struct{}{}
+	}
+	if len(voted) < len(notarySet)/3*2+1 {
+		return ErrNotEnoughVotes
 	}
 	return nil
 }
diff --git a/core/utils/nodeset-cache.go b/core/utils/nodeset-cache.go
index 6d4f7b0..35828b7 100644
--- a/core/utils/nodeset-cache.go
+++ b/core/utils/nodeset-cache.go
@@ -38,9 +38,11 @@ var (
 )
 
 type sets struct {
-	nodeSet   *types.NodeSet
-	notarySet []map[types.NodeID]struct{}
-	dkgSet    map[types.NodeID]struct{}
+	crs        common.Hash
+	nodeSet    *types.NodeSet
+	notarySet  []map[types.NodeID]struct{}
+	dkgSet     map[types.NodeID]struct{}
+	leaderNode []map[uint64]types.NodeID
 }
 
 // NodeSetCacheInterface interface specifies interface used by NodeSetCache.
@@ -146,6 +148,33 @@ func (cache *NodeSetCache) GetDKGSet(
 	return cache.cloneMap(IDs.dkgSet), 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
+	}
+	if pos.ChainID >= uint32(len(IDs.leaderNode)) {
+		return types.NodeID{}, ErrInvalidChainID
+	}
+	cache.lock.Lock()
+	defer cache.lock.Unlock()
+	if _, exist := IDs.leaderNode[pos.ChainID][pos.Height]; !exist {
+		notarySet := types.NewNodeSetFromMap(IDs.notarySet[pos.ChainID])
+		leader :=
+			notarySet.GetSubSet(1, types.NewNodeLeaderTarget(IDs.crs, pos))
+		if len(leader) != 1 {
+			panic(errors.New("length of leader is not one"))
+		}
+		for nID := range leader {
+			IDs.leaderNode[pos.ChainID][pos.Height] = nID
+			break
+		}
+	}
+	return IDs.leaderNode[pos.ChainID][pos.Height], nil
+}
+
 func (cache *NodeSetCache) cloneMap(
 	nIDs map[types.NodeID]struct{}) map[types.NodeID]struct{} {
 	nIDsCopy := make(map[types.NodeID]struct{}, len(nIDs))
@@ -207,15 +236,21 @@ func (cache *NodeSetCache) update(
 		return
 	}
 	nIDs = &sets{
+		crs:       crs,
 		nodeSet:   nodeSet,
 		notarySet: make([]map[types.NodeID]struct{}, cfg.NumChains),
 		dkgSet: nodeSet.GetSubSet(
 			int(cfg.DKGSetSize), types.NewDKGSetTarget(crs)),
+		leaderNode: make([]map[uint64]types.NodeID, cfg.NumChains),
 	}
 	for i := range nIDs.notarySet {
 		nIDs.notarySet[i] = nodeSet.GetSubSet(
 			int(cfg.NotarySetSize), types.NewNotarySetTarget(crs, uint32(i)))
 	}
+	nodesPerChain := cfg.RoundInterval / (cfg.LambdaBA * 4)
+	for i := range nIDs.leaderNode {
+		nIDs.leaderNode[i] = make(map[uint64]types.NodeID, nodesPerChain)
+	}
 
 	cache.rounds[round] = nIDs
 	// Purge older rounds.
diff --git a/core/utils/nodeset-cache_test.go b/core/utils/nodeset-cache_test.go
index 27c8c83..9e6ceee 100644
--- a/core/utils/nodeset-cache_test.go
+++ b/core/utils/nodeset-cache_test.go
@@ -19,6 +19,7 @@ package utils
 
 import (
 	"testing"
+	"time"
 
 	"github.com/dexon-foundation/dexon-consensus/common"
 	"github.com/dexon-foundation/dexon-consensus/core/crypto"
@@ -38,6 +39,8 @@ func (g *nsIntf) Configuration(round uint64) (cfg *types.Config) {
 		NotarySetSize: 7,
 		DKGSetSize:    7,
 		NumChains:     4,
+		LambdaBA:      250 * time.Millisecond,
+		RoundInterval: 60 * time.Second,
 	}
 }
 func (g *nsIntf) CRS(round uint64) (b common.Hash) { return g.crs }
@@ -91,6 +94,15 @@ func (s *NodeSetCacheTestSuite) TestBasicUsage() {
 	dkgSet, err := cache.GetDKGSet(0)
 	req.NoError(err)
 	chk(cache, 0, dkgSet)
+	leaderNode, err := cache.GetLeaderNode(types.Position{
+		Round:   uint64(0),
+		ChainID: uint32(3),
+		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)