From 2f7b8cffd43b7c38cac91e19d7b83798cb43cbbc Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Tue, 4 Aug 2026 22:14:12 +0800 Subject: [PATCH] refactor(consensus): simplify prepare/commit quorum with state-backed markers --- consensus/consensus_service.go | 1 + consensus/leader.go | 50 ++++------------------------------ consensus/state.go | 29 +++++++++++++++++++- consensus/state_quorum_test.go | 38 ++++++++++++++++++++++++++ 4 files changed, 72 insertions(+), 46 deletions(-) create mode 100644 consensus/state_quorum_test.go diff --git a/consensus/consensus_service.go b/consensus/consensus_service.go index cb1ed1b16e..a8c09de8a9 100644 --- a/consensus/consensus_service.go +++ b/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() diff --git a/consensus/leader.go b/consensus/leader.go index 9e901e70c6..71c0095c17 100644 --- a/consensus/leader.go +++ b/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 diff --git a/consensus/state.go b/consensus/state.go index 170de3f3b3..6bc38e4ab6 100644 --- a/consensus/state.go +++ b/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)) } diff --git a/consensus/state_quorum_test.go b/consensus/state_quorum_test.go new file mode 100644 index 0000000000..1aa8bf6210 --- /dev/null +++ b/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) + } +}