core: Add PullVotes to Network. (#281)

7 files changed, 64 insertions, 71 deletions

diff --git a/core/agreement-state.go b/core/agreement-state.go
index 77195ac..426b062 100644
--- a/core/agreement-state.go
+++ b/core/agreement-state.go
@@ -39,7 +39,7 @@ const (
 	statePreCommit
 	stateCommit
 	stateForward
-	stateRepeatVote
+	statePullVote
 )
 
 var nullBlockHash = common.Hash{}
@@ -127,54 +127,44 @@ func (s *commitState) nextState() (agreementState, error) {
 	} else {
 		hash = skipBlockHash
 	}
-	vote := &types.Vote{
+	s.a.recv.ProposeVote(&types.Vote{
 		Type:      types.VoteCom,
 		BlockHash: hash,
 		Period:    s.a.period,
-	}
-	s.a.recv.ProposeVote(vote)
-	return newForwardState(s.a, vote), nil
+	})
+	return newForwardState(s.a), nil
 }
 
 // ----- ForwardState -----
 type forwardState struct {
-	a    *agreementData
-	vote *types.Vote
+	a *agreementData
 }
 
-func newForwardState(a *agreementData, vote *types.Vote) *forwardState {
-	return &forwardState{
-		a:    a,
-		vote: vote,
-	}
+func newForwardState(a *agreementData) *forwardState {
+	return &forwardState{a: a}
 }
 
 func (s *forwardState) state() agreementStateType { return stateForward }
 func (s *forwardState) clocks() int               { return 4 }
 
 func (s *forwardState) nextState() (agreementState, error) {
-	return newRepeatVoteState(s.a, s.vote), nil
+	return newPullVoteState(s.a), nil
 }
 
-// ----- RepeatVoteState -----
-// repeateVoteState is a special state to ensure the assumption in the consensus
+// ----- PullVoteState -----
+// pullVoteState is a special state to ensure the assumption in the consensus
 // algorithm that every vote will eventually arrive for all nodes.
-type repeatVoteState struct {
-	a    *agreementData
-	vote *types.Vote
+type pullVoteState struct {
+	a *agreementData
 }
 
-func newRepeatVoteState(a *agreementData, vote *types.Vote) *repeatVoteState {
-	return &repeatVoteState{
-		a:    a,
-		vote: vote,
-	}
+func newPullVoteState(a *agreementData) *pullVoteState {
+	return &pullVoteState{a: a}
 }
 
-func (s *repeatVoteState) state() agreementStateType { return stateRepeatVote }
-func (s *repeatVoteState) clocks() int               { return 4 }
+func (s *pullVoteState) state() agreementStateType { return statePullVote }
+func (s *pullVoteState) clocks() int               { return 4 }
 
-func (s *repeatVoteState) nextState() (agreementState, error) {
-	s.a.recv.ProposeVote(s.vote)
+func (s *pullVoteState) nextState() (agreementState, error) {
 	return s, nil
 }
diff --git a/core/agreement-state_test.go b/core/agreement-state_test.go
index 862ac3e..b7ba829 100644
--- a/core/agreement-state_test.go
+++ b/core/agreement-state_test.go
@@ -85,7 +85,7 @@ func (s *AgreementStateTestSuite) prepareVote(
 
 func (s *AgreementStateTestSuite) SetupTest() {
 	prvKey, err := ecdsa.NewPrivateKey()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.ID = types.NewNodeID(prvKey.PublicKey())
 	s.auths = map[types.NodeID]*Authenticator{
 		s.ID: NewAuthenticator(prvKey),
@@ -103,7 +103,7 @@ func (s *AgreementStateTestSuite) newAgreement(numNode int) *agreement {
 	notarySet := make(map[types.NodeID]struct{})
 	for i := 0; i < numNode-1; i++ {
 		prvKey, err := ecdsa.NewPrivateKey()
-		s.Require().Nil(err)
+		s.Require().NoError(err)
 		nID := types.NewNodeID(prvKey.PublicKey())
 		notarySet[nID] = struct{}{}
 		s.auths[nID] = NewAuthenticator(prvKey)
@@ -132,11 +132,11 @@ func (s *AgreementStateTestSuite) TestInitialState() {
 	// Proposing a new block.
 	a.data.period = 1
 	newState, err := state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.blockChan, 1)
 	proposedBlock := <-s.blockChan
 	s.NotEqual(common.Hash{}, proposedBlock)
-	s.Require().Nil(a.processBlock(s.block[proposedBlock]))
+	s.Require().NoError(a.processBlock(s.block[proposedBlock]))
 	s.Equal(statePreCommit, newState.state())
 }
 
@@ -150,18 +150,18 @@ func (s *AgreementStateTestSuite) TestPreCommitState() {
 	for i := range blocks {
 		blocks[i] = s.proposeBlock(a.data.leader)
 		prv, err := ecdsa.NewPrivateKey()
-		s.Require().Nil(err)
+		s.Require().NoError(err)
 		blocks[i].ProposerID = types.NewNodeID(prv.PublicKey())
-		s.Require().Nil(NewAuthenticator(prv).SignCRS(
+		s.Require().NoError(NewAuthenticator(prv).SignCRS(
 			blocks[i], a.data.leader.hashCRS))
-		s.Require().Nil(a.processBlock(blocks[i]))
+		s.Require().NoError(a.processBlock(blocks[i]))
 	}
 
 	// If lockvalue == null, propose preCom-vote for the leader block.
 	a.data.lockValue = nullBlockHash
 	a.data.period = 1
 	newState, err := state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.voteChan, 1)
 	vote := <-s.voteChan
 	s.Equal(types.VotePreCom, vote.Type)
@@ -173,7 +173,7 @@ func (s *AgreementStateTestSuite) TestPreCommitState() {
 	hash := common.NewRandomHash()
 	a.data.lockValue = hash
 	newState, err = state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.voteChan, 1)
 	vote = <-s.voteChan
 	s.Equal(types.VotePreCom, vote.Type)
@@ -191,13 +191,13 @@ func (s *AgreementStateTestSuite) TestCommitState() {
 	// propose a com-vote for block v.
 	a.data.period = 1
 	block := s.proposeBlock(a.data.leader)
-	s.Require().Nil(a.processBlock(block))
+	s.Require().NoError(a.processBlock(block))
 	for nID := range a.notarySet {
 		vote := s.prepareVote(nID, types.VotePreCom, block.Hash, 1)
-		s.Require().Nil(a.processVote(vote))
+		s.Require().NoError(a.processVote(vote))
 	}
 	newState, err := state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.voteChan, 1)
 	s.Equal(block.Hash, a.data.lockValue)
 	s.Equal(uint64(1), a.data.lockRound)
@@ -209,7 +209,7 @@ func (s *AgreementStateTestSuite) TestCommitState() {
 	// Else, com-vote on SKIP.
 	a.data.period = 2
 	newState, err = state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.voteChan, 1)
 	vote = <-s.voteChan
 	s.Equal(types.VoteCom, vote.Type)
@@ -220,10 +220,10 @@ func (s *AgreementStateTestSuite) TestCommitState() {
 	a.data.period = 3
 	for nID := range a.notarySet {
 		vote := s.prepareVote(nID, types.VotePreCom, skipBlockHash, 3)
-		s.Require().Nil(a.processVote(vote))
+		s.Require().NoError(a.processVote(vote))
 	}
 	newState, err = state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.voteChan, 1)
 	vote = <-s.voteChan
 	s.Equal(types.VoteCom, vote.Type)
@@ -233,36 +233,26 @@ func (s *AgreementStateTestSuite) TestCommitState() {
 
 func (s *AgreementStateTestSuite) TestForwardState() {
 	a := s.newAgreement(4)
-	vote := &types.Vote{
-		BlockHash: common.NewRandomHash(),
-	}
-	state := newForwardState(a.data, vote)
+	state := newForwardState(a.data)
 	s.Equal(stateForward, state.state())
 	s.Equal(4, state.clocks())
 
 	newState, err := state.nextState()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.Require().Len(s.voteChan, 0)
-	s.Equal(stateRepeatVote, newState.state())
+	s.Equal(statePullVote, newState.state())
 }
 
-func (s *AgreementStateTestSuite) TestRepeatVoteState() {
+func (s *AgreementStateTestSuite) TestPullVoteState() {
 	a := s.newAgreement(4)
-	vote := &types.Vote{
-		BlockHash: common.NewRandomHash(),
-	}
-	state := newRepeatVoteState(a.data, vote)
-	s.Equal(stateRepeatVote, state.state())
+	state := newPullVoteState(a.data)
+	s.Equal(statePullVote, state.state())
 	s.Equal(4, state.clocks())
 
-	for i := 0; i < 5; i++ {
-		newState, err := state.nextState()
-		s.Require().Nil(err)
-		s.Require().Len(s.voteChan, 1)
-		proposedVote := <-s.voteChan
-		s.Equal(vote, proposedVote)
-		s.Equal(stateRepeatVote, newState.state())
-	}
+	newState, err := state.nextState()
+	s.Require().NoError(err)
+	s.Require().Len(s.voteChan, 0)
+	s.Equal(statePullVote, newState.state())
 }
 
 func TestAgreementState(t *testing.T) {
diff --git a/core/agreement.go b/core/agreement.go
index 8c2218b..3162b2e 100644
--- a/core/agreement.go
+++ b/core/agreement.go
@@ -223,6 +223,11 @@ func (a *agreement) clocks() int {
 	return a.state.clocks()
 }
 
+// pullVotes returns if current agreement requires more votes to continue.
+func (a *agreement) pullVotes() bool {
+	return a.state.state() == statePullVote
+}
+
 // agreementID returns the current agreementID.
 func (a *agreement) agreementID() types.Position {
 	a.lock.RLock()
diff --git a/core/agreement_test.go b/core/agreement_test.go
index 0e3814e..0541b44 100644
--- a/core/agreement_test.go
+++ b/core/agreement_test.go
@@ -80,7 +80,7 @@ type AgreementTestSuite struct {
 
 func (s *AgreementTestSuite) SetupTest() {
 	prvKey, err := ecdsa.NewPrivateKey()
-	s.Require().Nil(err)
+	s.Require().NoError(err)
 	s.ID = types.NewNodeID(prvKey.PublicKey())
 	s.auths = map[types.NodeID]*Authenticator{
 		s.ID: NewAuthenticator(prvKey),
@@ -100,7 +100,7 @@ func (s *AgreementTestSuite) newAgreement(numNotarySet int) *agreement {
 	notarySet := make(map[types.NodeID]struct{})
 	for i := 0; i < numNotarySet-1; i++ {
 		prvKey, err := ecdsa.NewPrivateKey()
-		s.Require().Nil(err)
+		s.Require().NoError(err)
 		nID := types.NewNodeID(prvKey.PublicKey())
 		notarySet[nID] = struct{}{}
 		s.auths[nID] = NewAuthenticator(prvKey)
@@ -217,13 +217,8 @@ func (s *AgreementTestSuite) TestPartitionOnCommitVote() {
 	s.Equal(uint64(1), a.data.lockRound)
 	// RepeateVoteState
 	a.nextState()
-	// The agreement does not receive others commit vote, it will keep re-sending.
-	for i := 0; i < 5; i++ {
-		a.nextState()
-		s.Require().Len(s.voteChan, 1)
-		proposedVote := <-s.voteChan
-		s.Equal(vote, proposedVote)
-	}
+	s.True(a.pullVotes())
+	s.Require().Len(s.voteChan, 0)
 }
 
 func (s *AgreementTestSuite) TestFastForwardCond1() {
diff --git a/core/consensus.go b/core/consensus.go
index 394ae36..15ecf67 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -471,6 +471,12 @@ BALoop:
 			agreement.restart(nIDs, nextPos)
 		default:
 		}
+		if agreement.pullVotes() {
+			pos := agreement.agreementID()
+			con.logger.Debug("Calling Network.PullVotes for syncing votes",
+				"position", pos)
+			con.network.PullVotes(pos)
+		}
 		err := agreement.nextState()
 		if err != nil {
 			con.logger.Error("Failed to proceed to next state",
diff --git a/core/consensus_test.go b/core/consensus_test.go
index e79028f..1ba9fa0 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -41,6 +41,10 @@ type network struct {
 func (n *network) PullBlocks(common.Hashes) {
 }
 
+// PullVotes tries to pull votes from the DEXON network.
+func (n *network) PullVotes(types.Position) {
+}
+
 // BroadcastVote broadcasts vote to all nodes in DEXON network.
 func (n *network) BroadcastVote(vote *types.Vote) {
 	n.conn.broadcast(n.nID, vote)
diff --git a/core/interfaces.go b/core/interfaces.go
index 4f6ad45..d6c1baf 100644
--- a/core/interfaces.go
+++ b/core/interfaces.go
@@ -62,6 +62,9 @@ type Network interface {
 	// PullBlocks tries to pull blocks from the DEXON network.
 	PullBlocks(hashes common.Hashes)
 
+	// PullVotes tries to pull votes from the DEXON network.
+	PullVotes(position types.Position)
+
 	// BroadcastVote broadcasts vote to all nodes in DEXON network.
 	BroadcastVote(vote *types.Vote)