Simplify Leader Quorum Calculations for Commit and Prepare Phases

api/service/synchronize/stagedstreamsync/stage_blockhashes.go

@@ -10,6 +10,7 @@ import (
 	"github.com/harmony-one/harmony/core"
 	"github.com/harmony-one/harmony/internal/utils"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
+	types "github.com/harmony-one/harmony/common/types"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
@@ -228,7 +229,7 @@ func (bh *StageBlockHashes) runBlockHashWorkerLoop(ctx context.Context,
 		}
 
 		// Map to store block hashes fetched from peers
-		peerHashes := sttypes.NewSafeMap[sttypes.StreamID, []common.Hash]()
+		peerHashes := types.NewSafeMap[sttypes.StreamID, []common.Hash]()
 		var wg sync.WaitGroup
 
 		if bh.configs.protocol.NumStreams() < bh.configs.concurrency {
@@ -376,7 +377,7 @@ func (bh *StageBlockHashes) checkFinalHashes(batch []uint64, hashes map[uint64]c
 // calculateFinalBlockHashes Calculates the most frequent block hashes for a given batch and removes streams with invalid hashes.
 // note: final hashes could be zero hashes
 func (bh *StageBlockHashes) calculateFinalBlockHashes(
-	peerHashes *sttypes.SafeMap[sttypes.StreamID, []common.Hash],
+	peerHashes *types.SafeMap[sttypes.StreamID, []common.Hash],
 	batch []uint64,
 ) (map[uint64]common.Hash, map[sttypes.StreamID]struct{}, error) {
 

api/service/synchronize/stagedstreamsync/stage_bodies.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/harmony-one/harmony/core"
 	"github.com/harmony-one/harmony/core/types"
+	ctypes "github.com/harmony-one/harmony/common/types"
 	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/p2p/stream/common/requestmanager"
 	syncProto "github.com/harmony-one/harmony/p2p/stream/protocols/sync"
@@ -191,7 +192,7 @@ func (b *StageBodies) Exec(ctx context.Context, firstCycle bool, invalidBlockRev
 // Failed streams are only punished when synced streams exist; otherwise the
 // stream pool is preserved to avoid cascading removal during systemic issues.
 func (b *StageBodies) identifySyncedStreams(ctx context.Context, s *StageState, targetHeight uint64, excludeIDs []sttypes.StreamID) (streams []sttypes.StreamID, err error) {
-	results := sttypes.NewSafeMap[sttypes.StreamID, error]()
+	results := ctypes.NewSafeMap[sttypes.StreamID, error]()
 	var (
 		wg          sync.WaitGroup
 		syncedCount int32

p2p/stream/types/safe_map.go → common/types/safe_map.go

@@ -1,4 +1,4 @@
-package sttypes
+package types
 
 import (
 	"sync"

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,16 @@ 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 {
+	// Quorum may already be met before this vote when the leader's own multi-BLS
+	// keys (added in announce) alone reach threshold. Track the last handled
+	// block so we fire prepare side-effects once per round.
+	quorumIsMet := consensus.decider().IsQuorumAchieved(quorum.Prepare)
+	if quorumIsMet && recvMsg.BlockNum > consensus.current.GetLastQuorumAchievedBlock(quorum.Prepare) {
 		// NOTE Let it handle its own logs
 		if err := consensus.didReachPrepareQuorum(); err != nil {
 			return
 		}
+		consensus.current.SetLastQuorumAchievedBlock(quorum.Prepare, recvMsg.BlockNum)
 		consensus.switchPhase("onPrepare", FBFTCommit)
 	}
 	//// Read - End
@@ -236,15 +210,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
@@ -316,15 +283,14 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 	//// Read - Start
 	viewID := consensus.getCurBlockViewID()
 
+	// Same multi-BLS case as prepare: leader may already meet commit quorum from
+	// keys added in didReachPrepareQuorum. Fire commit side-effects once per round.
 	quorumIsMet := consensus.decider().IsQuorumAchieved(quorum.Commit)
 	//// Read - End
-
-	quorumAchievedByFirstCommit := hasMultiBlsKeys && isFirstReceivedSignature && quorumWasMet
-	quorumAchievedByThisCommit := !quorumWasMet && quorumIsMet
-
-	if quorumAchievedByFirstCommit || quorumAchievedByThisCommit {
+	if quorumIsMet && recvMsg.BlockNum > consensus.current.GetLastQuorumAchievedBlock(quorum.Commit) {
 		logger.Info().Msg("[OnCommit] 2/3 Enough commits received")
 		consensus.fBFTLog.MarkBlockVerified(blockObj)
+		consensus.current.SetLastQuorumAchievedBlock(quorum.Commit, recvMsg.BlockNum)
 
 		if !blockObj.IsLastBlockInEpoch() {
 			// only do early commit if it's not epoch block to avoid problems

consensus/state.go

@@ -4,6 +4,7 @@ import (
 	"sync/atomic"
 	"unsafe"
 
+	"github.com/harmony-one/harmony/consensus/quorum"
 	bls_cosi "github.com/harmony-one/harmony/crypto/bls"
 	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/rs/zerolog"
@@ -38,6 +39,14 @@ type State struct {
 
 	// ShardID of the consensus
 	ShardID uint32
+
+	// lastPrepareQuorumBlock / lastCommitQuorumBlock record the block number for which
+	// prepare/commit quorum side-effects were already applied. Used to fire those
+	// side-effects once per consensus round, including the multi-BLS case where the
+	// leader's own keys may already meet quorum before the first external vote.
+	// Cleared in resetState so the same blockNum can be retried after view change.
+	lastPrepareQuorumBlock uint64
+	lastCommitQuorumBlock  uint64
 }
 
 func NewState(mode Mode, shardID uint32) State {
@@ -54,7 +63,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 +78,37 @@ func (pm *State) GetBlockNum() uint64 {
 	return pm.getBlockNum()
 }
 
+// GetLastQuorumAchievedBlock returns the last block number for which quorum
+// side-effects were applied for the given phase, or 0 if none.
+func (pm *State) GetLastQuorumAchievedBlock(p quorum.Phase) uint64 {
+	switch p {
+	case quorum.Prepare:
+		return atomic.LoadUint64(&pm.lastPrepareQuorumBlock)
+	case quorum.Commit:
+		return atomic.LoadUint64(&pm.lastCommitQuorumBlock)
+	default:
+		return 0
+	}
+}
+
+// SetLastQuorumAchievedBlock records that quorum side-effects were applied for
+// the given phase at blockNum.
+func (pm *State) SetLastQuorumAchievedBlock(p quorum.Phase, blockNum uint64) {
+	switch p {
+	case quorum.Prepare:
+		atomic.StoreUint64(&pm.lastPrepareQuorumBlock, blockNum)
+	case quorum.Commit:
+		atomic.StoreUint64(&pm.lastCommitQuorumBlock, blockNum)
+	}
+}
+
+// clearLastQuorumAchievedBlocks clears prepare/commit quorum markers so a new
+// consensus round (including same blockNum after view change) can fire again.
+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_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/harmony-one/harmony/consensus"
+	"github.com/harmony-one/harmony/consensus/quorum"
 )
 
 func TestState_SetBlockNum(t *testing.T) {
@@ -16,3 +17,36 @@ func TestState_SetBlockNum(t *testing.T) {
 		t.Errorf("SetBlockNum failed")
 	}
 }
+
+func TestState_LastQuorumAchievedBlock(t *testing.T) {
+	state := consensus.NewState(consensus.Normal, 0)
+
+	if got := state.GetLastQuorumAchievedBlock(quorum.Prepare); got != 0 {
+		t.Fatalf("Prepare last quorum: got %d, want 0", got)
+	}
+	if got := state.GetLastQuorumAchievedBlock(quorum.Commit); got != 0 {
+		t.Fatalf("Commit last quorum: got %d, want 0", got)
+	}
+
+	state.SetLastQuorumAchievedBlock(quorum.Prepare, 10)
+	state.SetLastQuorumAchievedBlock(quorum.Commit, 11)
+
+	if got := state.GetLastQuorumAchievedBlock(quorum.Prepare); got != 10 {
+		t.Fatalf("Prepare last quorum: got %d, want 10", got)
+	}
+	if got := state.GetLastQuorumAchievedBlock(quorum.Commit); got != 11 {
+		t.Fatalf("Commit last quorum: got %d, want 11", got)
+	}
+
+	// Phases are independent; setting one must not clobber the other.
+	state.SetLastQuorumAchievedBlock(quorum.Prepare, 12)
+	if got := state.GetLastQuorumAchievedBlock(quorum.Commit); got != 11 {
+		t.Fatalf("Commit last quorum changed unexpectedly: got %d, want 11", got)
+	}
+
+	// Unknown phases are ignored.
+	state.SetLastQuorumAchievedBlock(quorum.ViewChange, 99)
+	if got := state.GetLastQuorumAchievedBlock(quorum.ViewChange); got != 0 {
+		t.Fatalf("ViewChange last quorum: got %d, want 0", got)
+	}
+}

p2p/host.go

@@ -16,6 +16,7 @@ import (
 	"github.com/harmony-one/abool"
 	prom "github.com/harmony-one/harmony/api/service/prometheus"
 	"github.com/harmony-one/harmony/common/clock"
+	ctypes "github.com/harmony-one/harmony/common/types"
 	bls "github.com/harmony-one/harmony/crypto/bls/core"
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 	"github.com/harmony-one/harmony/internal/utils"
@@ -465,7 +466,7 @@ func NewHost(cfg HostConfig) (Host, error) {
 		joined:                  map[string]*libp2p_pubsub.Topic{},
 		self:                    *self,
 		trustedNodes:            cfg.TrustedNodes,
-		trustedPeerIDs:          sttypes.NewSafeMap[libp2p_peer.ID, struct{}](),
+		trustedPeerIDs:          ctypes.NewSafeMap[libp2p_peer.ID, struct{}](),
 		trustedMinPeers:         cfg.TrustedMinPeers,
 		trustedBootstrapEnabled: cfg.TrustedBootstrapEnabled,
 		dnsStaticNodes:          cfg.DNSStaticNodes,
@@ -580,7 +581,7 @@ type HostV2 struct {
 	streamProtos            []sttypes.Protocol
 	self                    Peer
 	trustedNodes            []string
-	trustedPeerIDs          *sttypes.SafeMap[libp2p_peer.ID, struct{}] // Thread-safe map of trusted peer IDs
+	trustedPeerIDs          *ctypes.SafeMap[libp2p_peer.ID, struct{}] // Thread-safe map of trusted peer IDs
 	trustedMinPeers         int
 	trustedBootstrapEnabled bool
 	dnsStaticNodes          []string

p2p/stream/common/requestmanager/interface_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/rlp"
+	types "github.com/harmony-one/harmony/common/types"
 	"github.com/harmony-one/harmony/p2p/stream/common/streammanager"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
 )
@@ -156,8 +157,8 @@ func makeDummyTestStreams(indexes []int) []sttypes.Stream {
 	return sts
 }
 
-func makeDummyStreamSets(indexes []int) *sttypes.SafeMap[sttypes.StreamID, *stream] {
-	m := sttypes.NewSafeMap[sttypes.StreamID, *stream]()
+func makeDummyStreamSets(indexes []int) *types.SafeMap[sttypes.StreamID, *stream] {
+	m := types.NewSafeMap[sttypes.StreamID, *stream]()
 
 	for _, index := range indexes {
 		st := &testStream{

p2p/stream/common/requestmanager/requestmanager.go

@@ -12,6 +12,7 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/ethereum/go-ethereum/event"
+	types "github.com/harmony-one/harmony/common/types"
 	"github.com/harmony-one/harmony/internal/utils"
 	"github.com/harmony-one/harmony/p2p/stream/common/streammanager"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
@@ -22,10 +23,10 @@ import (
 // TODO: each peer is able to have a queue of requests instead of one request at a time.
 // TODO: add QoS evaluation for each stream
 type requestManager struct {
-	streams   *sttypes.SafeMap[sttypes.StreamID, *stream]  // All streams
-	available *sttypes.SafeMap[sttypes.StreamID, struct{}] // Streams that are available for request
-	pendings  *sttypes.SafeMap[uint64, *request]           // requests that are sent but not received response
-	waitings  requestQueues                                // double linked list of requests that are on the waiting list
+	streams   *types.SafeMap[sttypes.StreamID, *stream]  // All streams
+	available *types.SafeMap[sttypes.StreamID, struct{}] // Streams that are available for request
+	pendings  *types.SafeMap[uint64, *request]           // requests that are sent but not received response
+	waitings  requestQueues                              // double linked list of requests that are on the waiting list
 
 	myProtoID sttypes.ProtoID
 
@@ -71,9 +72,9 @@ func newRequestManager(sm streammanager.ReaderSubscriber, pid sttypes.ProtoID) *
 	logger := utils.Logger().With().Str("module", "request manager").Logger()
 
 	return &requestManager{
-		streams:   sttypes.NewSafeMap[sttypes.StreamID, *stream](),
-		available: sttypes.NewSafeMap[sttypes.StreamID, struct{}](),
-		pendings:  sttypes.NewSafeMap[uint64, *request](),
+		streams:   types.NewSafeMap[sttypes.StreamID, *stream](),
+		available: types.NewSafeMap[sttypes.StreamID, struct{}](),
+		pendings:  types.NewSafeMap[uint64, *request](),
 		waitings:  newRequestQueues(),
 
 		myProtoID: pid,
@@ -598,7 +599,7 @@ func (rm *requestManager) refreshStreams() {
 	}
 }
 
-func checkStreamUpdates(exists *sttypes.SafeMap[sttypes.StreamID, *stream], targets []sttypes.Stream) (added []sttypes.Stream, removed []*stream) {
+func checkStreamUpdates(exists *types.SafeMap[sttypes.StreamID, *stream], targets []sttypes.Stream) (added []sttypes.Stream, removed []*stream) {
 	targetM := make(map[sttypes.StreamID]sttypes.Stream)
 
 	for _, target := range targets {

p2p/stream/common/requestmanager/requestmanager_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	types "github.com/harmony-one/harmony/common/types"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
 	"github.com/pkg/errors"
 )
@@ -418,7 +419,7 @@ func TestRequestManager_Concurrency(t *testing.T) {
 func TestGenReqID(t *testing.T) {
 	retry := 100000
 	rm := &requestManager{
-		pendings: sttypes.NewSafeMap[uint64, *request](),
+		pendings: types.NewSafeMap[uint64, *request](),
 	}
 
 	for i := 0; i != retry; i++ {
@@ -432,7 +433,7 @@ func TestGenReqID(t *testing.T) {
 
 func TestCheckStreamUpdates(t *testing.T) {
 	tests := []struct {
-		exists            *sttypes.SafeMap[sttypes.StreamID, *stream]
+		exists            *types.SafeMap[sttypes.StreamID, *stream]
 		targets           []sttypes.Stream
 		expAddedIndexes   []int
 		expRemovedIndexes []int