feat: add absentee votes to the certificate

consensus/consensus_test.go

@@ -71,6 +71,7 @@ func (o *OverrideStringer) String() string {
 
 func setup(t *testing.T) *testData {
 	t.Helper()
+	queryVoteInitialTimeout = 2 * time.Hour
 
 	return setupWithSeed(t, testsuite.GenerateSeed())
 }
@@ -121,21 +122,21 @@ func setupWithSeed(t *testing.T, seed int64) *testData {
 		genDoc:       genDoc,
 		consMessages: consMessages,
 	}
-	broadcaster := func(sender crypto.Address, msg message.Message) {
+	broadcasterFunc := func(sender crypto.Address, msg message.Message) {
 		fmt.Printf("received a message %s: %s\n", msg.Type(), msg.String())
 		td.consMessages = append(td.consMessages, consMessage{
 			sender:  sender,
 			message: msg,
 		})
 	}
 	td.consX = newConsensus(testConfig(), stX, valKeys[tIndexX],
-		valKeys[tIndexX].PublicKey().AccountAddress(), broadcaster, newConcreteMediator())
+		valKeys[tIndexX].PublicKey().AccountAddress(), broadcasterFunc, newConcreteMediator())
 	td.consY = newConsensus(testConfig(), stY, valKeys[tIndexY],
-		valKeys[tIndexY].PublicKey().AccountAddress(), broadcaster, newConcreteMediator())
+		valKeys[tIndexY].PublicKey().AccountAddress(), broadcasterFunc, newConcreteMediator())
 	td.consB = newConsensus(testConfig(), stB, valKeys[tIndexB],
-		valKeys[tIndexB].PublicKey().AccountAddress(), broadcaster, newConcreteMediator())
+		valKeys[tIndexB].PublicKey().AccountAddress(), broadcasterFunc, newConcreteMediator())
 	td.consP = newConsensus(testConfig(), stP, valKeys[tIndexP],
-		valKeys[tIndexP].PublicKey().AccountAddress(), broadcaster, newConcreteMediator())
+		valKeys[tIndexP].PublicKey().AccountAddress(), broadcasterFunc, newConcreteMediator())
 
 	// -------------------------------
 	// Better logging during testing

consensus/cp_prevote.go

@@ -7,6 +7,8 @@ import (
 	"github.com/pactus-project/pactus/types/vote"
 )
 
+var queryVoteInitialTimeout = 2 * time.Second
+
 type cpPreVoteState struct {
 	*changeProposer
 }
@@ -31,7 +33,7 @@ func (s *cpPreVoteState) decide() {
 			just := &vote.JustInitOne{}
 			s.signAddCPPreVote(hash.UndefHash, s.cpRound, 1, just)
 		}
-		s.scheduleTimeout(2*time.Second, s.height, s.round, tickerTargetQueryVotes)
+		s.scheduleTimeout(queryVoteInitialTimeout, s.height, s.round, tickerTargetQueryVotes)
 	} else {
 		cpMainVotes := s.log.CPMainVoteVoteSet(s.round)
 		if cpMainVotes.HasAnyVoteFor(s.cpRound-1, vote.CPValueOne) {

consensus/manager.go

@@ -19,6 +19,7 @@ type manager struct {
 	// the current block's consensus is complete.
 	upcomingVotes     []*vote.Vote         // Map to cache votes for future block heights
 	upcomingProposals []*proposal.Proposal // Map to cache proposals for future block heights
+	state             state.Facade
 }
 
 // NewManager creates a new manager instance that manages a set of consensus instances,
@@ -35,6 +36,7 @@ func NewManager(
 		instances:         make([]Consensus, len(valKeys)),
 		upcomingVotes:     make([]*vote.Vote, 0),
 		upcomingProposals: make([]*proposal.Proposal, 0),
+		state:             state,
 	}
 	mediatorConcrete := newConcreteMediator()
 
@@ -60,7 +62,7 @@ func (mgr *manager) Start() error {
 func (mgr *manager) Stop() {
 }
 
-// Instances returns all consensus instances that are read-only and
+// Instances return all consensus instances that are read-only and
 // can be safely accessed without modifying their state.
 func (mgr *manager) Instances() []Reader {
 	readers := make([]Reader, len(mgr.instances))
@@ -76,7 +78,7 @@ func (mgr *manager) PickRandomVote(round int16) *vote.Vote {
 	return cons.PickRandomVote(round)
 }
 
-// RoundProposal returns the proposal for a specific round from a random consensus instance.
+// Proposal returns the proposal for a specific round from a random consensus instance.
 func (mgr *manager) Proposal() *proposal.Proposal {
 	cons := mgr.getBestInstance()
 	return cons.Proposal()
@@ -154,7 +156,7 @@ func (mgr *manager) AddVote(v *vote.Vote) {
 	curHeight, _ := inst.HeightRound()
 	switch {
 	case v.Height() < curHeight:
-		// discard the old vote
+		_ = mgr.state.UpdateLastCertificate(v)
 
 	case v.Height() > curHeight:
 		mgr.upcomingVotes = append(mgr.upcomingVotes, v)

state/state.go

@@ -123,12 +123,12 @@
 	}
 
 	logger.Debug("try to restore the last state")
-	committee, err := st.lastInfo.RestoreLastInfo(st.store, st.params.CommitteeSize)
+	committeeInstance, err := st.lastInfo.RestoreLastInfo(st.store, st.params.CommitteeSize)
 	if err != nil {
 		return err
 	}
 
-	st.committee = committee
+	st.committee = committeeInstance
 
 	logger.Info("last state restored",
 		"last height", st.lastInfo.BlockHeight(),
@@ -153,11 +153,11 @@
 		return err
 	}
 
-	committee, err := committee.NewCommittee(vals, st.params.CommitteeSize, vals[0].Address())
+	committeeInstance, err := committee.NewCommittee(vals, st.params.CommitteeSize, vals[0].Address())
 	if err != nil {
 		return err
 	}
-	st.committee = committee
+	st.committee = committeeInstance
 	st.lastInfo.UpdateBlockTime(genDoc.GenesisTime())
 
 	return nil
@@ -285,19 +285,24 @@
 		return err
 	}
 
+	if !util.Contains(lastCert.Absentees(), val.Number()) {
+		return InvalidVoteForCertificateError{
+			Vote: v,
+		}
+	}
+
 	err = v.Verify(val.PublicKey())
 	if err != nil {
 		return err
 	}
 
-	if util.Contains(lastCert.Absentees(), val.Number()) {
-		lastCert.AddSignature(val.Number(), v.Signature())
-		st.lastInfo.UpdateCertificate(lastCert)
+	// prevent race condition
+	cloneLastCert := lastCert.Clone()
 
-		return nil
-	}
+	cloneLastCert.AddSignature(val.Number(), v.Signature())
+	st.lastInfo.UpdateCertificate(cloneLastCert)
 
-	return InvalidVoteForCertificateError{Vote: v}
+	return nil
 }
 
 func (st *state) createSubsidyTx(rewardAddr crypto.Address, fee int64) *tx.Tx {
@@ -318,7 +323,7 @@
 	txs := st.txPool.PrepareBlockTransactions()
 	txs = util.Trim(txs, maxTransactionsPerBlock-1)
 	for i := 0; i < txs.Len(); i++ {
-		// Only one subsidy transaction per block
+		// Only one subsidy transaction per blk
 		if txs[i].IsSubsidyTx() {
 			st.logger.Error("found duplicated subsidy transaction", "tx", txs[i])
 			st.txPool.RemoveTx(txs[i].ID())
@@ -343,7 +348,7 @@
 	txs.Prepend(subsidyTx)
 	preSeed := st.lastInfo.SortitionSeed()
 
-	block := block.MakeBlock(
+	blk := block.MakeBlock(
 		st.params.BlockVersion,
 		st.proposeNextBlockTime(),
 		txs,
@@ -353,7 +358,7 @@
 		preSeed.GenerateNext(valKey.PrivateKey()),
 		valKey.Address())
 
-	return block, nil
+	return blk, nil
 }
 
 func (st *state) ValidateBlock(blk *block.Block) error {
@@ -389,7 +394,7 @@
 	}
 
 	// There are two modules that can commit a block: Consensus and Sync.
-	// Consensus engine is ours, we have full control over that and we know when
+	// The Consensus engine is ours, we have full control over that, and we know when
 	// and why a block should be committed.
 	// On the other hand, Sync module receives new blocks from the network and
 	// tries to commit them.
@@ -453,7 +458,7 @@
 	st.evaluateSortition()
 
 	// -----------------------------------
-	// At this point we can assign new sandbox to tx pool
+	// At this point we can assign a new sandbox to tx pool
 	st.txPool.SetNewSandboxAndRecheck(st.concreteSandbox())
 
 	// -----------------------------------
@@ -635,11 +640,11 @@
 }
 
 func (st *state) CommittedTx(id tx.ID) *store.CommittedTx {
-	tx, err := st.store.Transaction(id)
+	transaction, err := st.store.Transaction(id)
 	if err != nil {
 		st.logger.Trace("searching transaction in local store failed", "id", id, "error", err)
 	}
-	return tx
+	return transaction
 }
 
 func (st *state) BlockHash(height uint32) hash.Hash {
@@ -704,17 +709,17 @@
 	st.eventCh <- blockEvent
 
 	for i := 1; i < blk.Transactions().Len(); i++ {
-		tx := blk.Transactions().Get(i)
+		transaction := blk.Transactions().Get(i)
 
-		accChangeEvent := event.CreateAccountChangeEvent(tx.Payload().Signer(), height)
+		accChangeEvent := event.CreateAccountChangeEvent(transaction.Payload().Signer(), height)
 		st.eventCh <- accChangeEvent
 
-		if tx.Payload().Receiver() != nil {
-			accChangeEvent := event.CreateAccountChangeEvent(*tx.Payload().Receiver(), height)
+		if transaction.Payload().Receiver() != nil {
+			accChangeEvent := event.CreateAccountChangeEvent(*transaction.Payload().Receiver(), height)
 			st.eventCh <- accChangeEvent
 		}
 
-		TxEvent := event.CreateTransactionEvent(tx.ID(), height)
+		TxEvent := event.CreateTransactionEvent(transaction.ID(), height)
 		st.eventCh <- TxEvent
 	}
 }

state/state_test.go

@@ -349,38 +349,35 @@ func TestUpdateLastCertificate(t *testing.T) {
 	td.commitBlockForAllStates(t, blk, cert)
 
 	invValKey := td.RandValKey()
-	notActiveValKey := td.RandValKey()
-	valNum := int32(4) // [0..3] are in the committee now
-	val := validator.NewValidator(notActiveValKey.PublicKey(), valNum)
-	td.state1.store.UpdateValidator(val)
 
 	v1 := vote.NewPrepareVote(blk.Hash(), cert.Height(), cert.Round(), td.valKey3.Address())
 	v2 := vote.NewPrecommitVote(blk.Hash(), cert.Height()+1, cert.Round(), td.valKey3.Address())
 	v3 := vote.NewPrecommitVote(blk.Hash(), cert.Height(), cert.Round()-1, td.valKey3.Address())
 	v4 := vote.NewPrecommitVote(blk.Hash(), cert.Height(), cert.Round(), td.valKey4.Address())
 	v5 := vote.NewPrecommitVote(blk.Hash(), cert.Height(), cert.Round(), invValKey.Address())
-	v6 := vote.NewPrecommitVote(blk.Hash(), cert.Height(), cert.Round(), notActiveValKey.Address())
+	v6 := vote.NewPrecommitVote(blk.Hash(), cert.Height(), cert.Round(), td.valKey1.Address())
 	v7 := vote.NewPrecommitVote(blk.Hash(), cert.Height(), cert.Round(), td.valKey4.Address())
 
 	td.HelperSignVote(td.valKey3, v1)
 	td.HelperSignVote(td.valKey3, v2)
 	td.HelperSignVote(td.valKey3, v3)
 	td.HelperSignVote(invValKey, v4)
 	td.HelperSignVote(invValKey, v5)
-	td.HelperSignVote(notActiveValKey, v6)
+	td.HelperSignVote(td.valKey1, v6)
 	td.HelperSignVote(td.valKey4, v7)
 
 	tests := []struct {
-		vote *vote.Vote
-		err  error
+		vote   *vote.Vote
+		err    error
+		reason string
 	}{
-		{v1, InvalidVoteForCertificateError{Vote: v1}},
-		{v2, InvalidVoteForCertificateError{Vote: v2}},
-		{v3, InvalidVoteForCertificateError{Vote: v3}},
-		{v4, crypto.ErrInvalidSignature},
-		{v5, store.ErrNotFound},
-		{v6, InvalidVoteForCertificateError{Vote: v6}},
-		{v7, nil},
+		{v1, InvalidVoteForCertificateError{Vote: v1}, "invalid vote type"},
+		{v2, InvalidVoteForCertificateError{Vote: v2}, "invalid height"},
+		{v3, InvalidVoteForCertificateError{Vote: v3}, "invalid round"},
+		{v4, crypto.ErrInvalidSignature, "invalid signature"},
+		{v5, store.ErrNotFound, "unknown validator"},
+		{v6, InvalidVoteForCertificateError{Vote: v6}, "not absentee"},
+		{v7, nil, "ok"},
 	}
 
 	for i, test := range tests {

types/certificate/certificate.go

@@ -103,6 +103,18 @@ func (cert *Certificate) Hash() hash.Hash {
 	return hash.CalcHash(w.Bytes())
 }
 
+func (cert *Certificate) Clone() *Certificate {
+	return &Certificate{
+		data: certificateData{
+			Height:     cert.Height(),
+			Round:      cert.Round(),
+			Committers: cert.Committers(),
+			Absentees:  cert.Absentees(),
+			Signature:  cert.Signature(),
+		},
+	}
+}
+
 // SerializeSize returns the number of bytes it would take to serialize the block.
 func (cert *Certificate) SerializeSize() int {
 	sz := 6 + // height (4) + round(2)

types/certificate/certificate_test.go

@@ -308,3 +308,13 @@ func TestAddSignature(t *testing.T) {
 	assert.Empty(t, cert.Absentees())
 	assert.NoError(t, cert.Validate(blockHeight, []*validator.Validator{val1, val2, val3, val4}, signBytes))
 }
+
+func TestClone(t *testing.T) {
+	ts := testsuite.NewTestSuite(t)
+
+	cert1 := ts.GenerateTestCertificate(ts.RandHeight())
+	cert2 := cert1.Clone()
+
+	cert1.AddSignature(cert1.Absentees()[0], ts.RandBLSSignature())
+	assert.NotEqual(t, cert1, cert2)
+}