refactor(consensus): simplify prepare/commit quorum tracking

consensus/consensus_service.go

@@ -200,6 +200,7 @@ func (consensus *Consensus) resetState() {
 
 	consensus.current.blockHash = [32]byte{}
 	consensus.current.block = []byte{}
+	consensus.current.clearLastQuorumAchievedBlocks()
 	consensus.decider().ResetPrepareAndCommitVotes()
 	if consensus.prepareBitmap != nil {
 		consensus.prepareBitmap.Clear()

consensus/leader.go

@@ -93,19 +93,6 @@ func (consensus *Consensus) announce(block *types.Block) {
 	consensus.switchPhase("Announce", FBFTPrepare)
 }
 
-func (consensus *Consensus) checkFirstReceivedSignature(signerCount int64, phase quorum.Phase) (bool, bool) {
-	hasMultiBlsKeys := len(consensus.priKey) > 0
-	if hasMultiBlsKeys {
-		var myPubkeys []bls.SerializedPublicKey
-		for _, key := range consensus.priKey {
-			myPubkeys = append(myPubkeys, key.Pub.Bytes)
-		}
-		mySignsCount := consensus.decider().GetBallotsCount(phase, myPubkeys)
-		return true, signerCount == mySignsCount
-	}
-	return false, false
-}
-
 // this method is called for each validator sent their vote message
 func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 	// TODO(audit): make FBFT lookup using map instead of looping through all items.
@@ -135,20 +122,6 @@ func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 	}
 
 	signerCount := consensus.decider().SignersCount(quorum.Prepare)
-
-	// check if it is first received signatures
-	// it may multi bls key validators can achieve quorum on first signature
-	hasMultiBlsKeys, isFirstReceivedSignature := consensus.checkFirstReceivedSignature(signerCount, quorum.Prepare)
-
-	quorumPreExisting := consensus.decider().IsQuorumAchieved(quorum.Prepare)
-	//// Read - End
-
-	if quorumPreExisting {
-		// already have enough signatures
-		consensus.getLogger().Debug().
-			Interface("validatorPubKeys", recvMsg.SenderPubkeys).
-			Msg("[OnPrepare] Received Additional Prepare Message")
-	}
 	//// Read - End
 
 	consensus.UpdateLeaderMetrics(float64(signerCount), float64(consensus.getBlockNum()))
@@ -203,15 +176,13 @@ func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) {
 	//// Write - End
 
 	//// Read - Start
-	quorumFromInitialSignature := hasMultiBlsKeys && isFirstReceivedSignature && quorumPreExisting
-	quorumPostNewSignatures := consensus.decider().IsQuorumAchieved(quorum.Prepare)
-	quorumFromNewSignatures := !quorumPreExisting && quorumPostNewSignatures
-
-	if quorumFromInitialSignature || quorumFromNewSignatures {
+	quorumIsMet := consensus.decider().IsQuorumAchieved(quorum.Prepare)
+	if quorumIsMet && recvMsg.BlockNum > consensus.current.getLastPrepareQuorumBlock() {
 		// NOTE Let it handle its own logs
 		if err := consensus.didReachPrepareQuorum(); err != nil {
 			return
 		}
+		consensus.current.setLastPrepareQuorumBlock(recvMsg.BlockNum)
 		consensus.switchPhase("onPrepare", FBFTCommit)
 	}
 	//// Read - End
@@ -236,15 +207,8 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 
 	commitBitmap := consensus.commitBitmap
 
-	// has to be called before verifying signature
-	quorumWasMet := consensus.decider().IsQuorumAchieved(quorum.Commit)
-
 	signerCount := consensus.decider().SignersCount(quorum.Commit)
 
-	// check if it is first received commit
-	// it may multi bls key validators can achieve quorum on first commit
-	hasMultiBlsKeys, isFirstReceivedSignature := consensus.checkFirstReceivedSignature(signerCount, quorum.Commit)
-
 	//// Read - End
 
 	// Verify the signature on commitPayload is correct
@@ -315,16 +279,12 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 
 	//// Read - Start
 	viewID := consensus.getCurBlockViewID()
-
 	quorumIsMet := consensus.decider().IsQuorumAchieved(quorum.Commit)
 	//// Read - End
-
-	quorumAchievedByFirstCommit := hasMultiBlsKeys && isFirstReceivedSignature && quorumWasMet
-	quorumAchievedByThisCommit := !quorumWasMet && quorumIsMet
-
-	if quorumAchievedByFirstCommit || quorumAchievedByThisCommit {
+	if quorumIsMet && recvMsg.BlockNum > consensus.current.getLastCommitQuorumBlock() {
 		logger.Info().Msg("[OnCommit] 2/3 Enough commits received")
 		consensus.fBFTLog.MarkBlockVerified(blockObj)
+		consensus.current.setLastCommitQuorumBlock(recvMsg.BlockNum)
 
 		if !blockObj.IsLastBlockInEpoch() {
 			// only do early commit if it's not epoch block to avoid problems

consensus/state.go

@@ -38,6 +38,11 @@ type State struct {
 
 	// ShardID of the consensus
 	ShardID uint32
+
+	// lastPrepareQuorumBlock / lastCommitQuorumBlock are the block numbers for which
+	// prepare/commit quorum side-effects have been applied.
+	lastPrepareQuorumBlock uint64
+	lastCommitQuorumBlock  uint64
 }
 
 func NewState(mode Mode, shardID uint32) State {
@@ -54,7 +59,7 @@ func (pm *State) getBlockNum() uint64 {
 	return atomic.LoadUint64(&pm.blockNum)
 }
 
-// SetBlockNum sets the blockNum in consensus object, called at node bootstrap
+// setBlockNum sets the blockNum in consensus object, called at node bootstrap
 func (pm *State) setBlockNum(blockNum uint64) {
 	atomic.StoreUint64(&pm.blockNum, blockNum)
 }
@@ -69,6 +74,28 @@ func (pm *State) GetBlockNum() uint64 {
 	return pm.getBlockNum()
 }
 
+func (pm *State) getLastPrepareQuorumBlock() uint64 {
+	return atomic.LoadUint64(&pm.lastPrepareQuorumBlock)
+}
+
+func (pm *State) setLastPrepareQuorumBlock(blockNum uint64) {
+	atomic.StoreUint64(&pm.lastPrepareQuorumBlock, blockNum)
+}
+
+func (pm *State) getLastCommitQuorumBlock() uint64 {
+	return atomic.LoadUint64(&pm.lastCommitQuorumBlock)
+}
+
+func (pm *State) setLastCommitQuorumBlock(blockNum uint64) {
+	atomic.StoreUint64(&pm.lastCommitQuorumBlock, blockNum)
+}
+
+// clearLastQuorumAchievedBlocks clears prepare/commit quorum markers.
+func (pm *State) clearLastQuorumAchievedBlocks() {
+	atomic.StoreUint64(&pm.lastPrepareQuorumBlock, 0)
+	atomic.StoreUint64(&pm.lastCommitQuorumBlock, 0)
+}
+
 func (pm *State) getLeaderPubKey() *bls_cosi.PublicKeyWrapper {
 	return (*bls_cosi.PublicKeyWrapper)(atomic.LoadPointer(&pm.leaderPubKey))
 }

consensus/state_quorum_test.go

@@ -0,0 +1,38 @@
+package consensus
+
+import "testing"
+
+func TestState_LastQuorumAchievedBlock(t *testing.T) {
+	state := NewState(Normal, 0)
+
+	if got := state.getLastPrepareQuorumBlock(); got != 0 {
+		t.Fatalf("Prepare last quorum: got %d, want 0", got)
+	}
+	if got := state.getLastCommitQuorumBlock(); got != 0 {
+		t.Fatalf("Commit last quorum: got %d, want 0", got)
+	}
+
+	state.setLastPrepareQuorumBlock(10)
+	state.setLastCommitQuorumBlock(11)
+
+	if got := state.getLastPrepareQuorumBlock(); got != 10 {
+		t.Fatalf("Prepare last quorum: got %d, want 10", got)
+	}
+	if got := state.getLastCommitQuorumBlock(); got != 11 {
+		t.Fatalf("Commit last quorum: got %d, want 11", got)
+	}
+
+	// Phases are independent; setting one must not clobber the other.
+	state.setLastPrepareQuorumBlock(12)
+	if got := state.getLastCommitQuorumBlock(); got != 11 {
+		t.Fatalf("Commit last quorum changed unexpectedly: got %d, want 11", got)
+	}
+
+	state.clearLastQuorumAchievedBlocks()
+	if got := state.getLastPrepareQuorumBlock(); got != 0 {
+		t.Fatalf("Prepare last quorum after clear: got %d, want 0", got)
+	}
+	if got := state.getLastCommitQuorumBlock(); got != 0 {
+		t.Fatalf("Commit last quorum after clear: got %d, want 0", got)
+	}
+}