From db0464b5301ef2c3366b942abe18873a84058c41 Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Wed, 11 Dec 2024 17:46:38 +0800 Subject: [PATCH 1/6] move SafeMap to common folder --- {p2p/stream => common}/types/safe_map.go | 2 +- .../common/requestmanager/interface_test.go | 5 +++-- .../common/requestmanager/requestmanager.go | 17 +++++++++-------- .../requestmanager/requestmanager_test.go | 5 +++-- p2p/stream/protocols/sync/message/msg.pb.go | 5 +++-- 5 files changed, 19 insertions(+), 15 deletions(-) rename {p2p/stream => common}/types/safe_map.go (99%) diff --git a/p2p/stream/types/safe_map.go b/common/types/safe_map.go similarity index 99% rename from p2p/stream/types/safe_map.go rename to common/types/safe_map.go index e4a5e559c5..001438da76 100644 --- a/p2p/stream/types/safe_map.go +++ b/common/types/safe_map.go @@ -1,4 +1,4 @@ -package sttypes +package types import ( "sync" diff --git a/p2p/stream/common/requestmanager/interface_test.go b/p2p/stream/common/requestmanager/interface_test.go index 5133329db2..b0e42aa14b 100644 --- a/p2p/stream/common/requestmanager/interface_test.go +++ b/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{ diff --git a/p2p/stream/common/requestmanager/requestmanager.go b/p2p/stream/common/requestmanager/requestmanager.go index aca8150a49..a267287e96 100644 --- a/p2p/stream/common/requestmanager/requestmanager.go +++ b/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 { diff --git a/p2p/stream/common/requestmanager/requestmanager_test.go b/p2p/stream/common/requestmanager/requestmanager_test.go index b077e6d597..49afe67654 100644 --- a/p2p/stream/common/requestmanager/requestmanager_test.go +++ b/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 diff --git a/p2p/stream/protocols/sync/message/msg.pb.go b/p2p/stream/protocols/sync/message/msg.pb.go index db37f7c7da..2205810507 100644 --- a/p2p/stream/protocols/sync/message/msg.pb.go +++ b/p2p/stream/protocols/sync/message/msg.pb.go @@ -7,10 +7,11 @@ package message import ( - protoreflect "google.golang.org/protobuf/reflect/protoreflect" - protoimpl "google.golang.org/protobuf/runtime/protoimpl" reflect "reflect" sync "sync" + + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" ) const ( From 043e7533ae6924b7e5f3412dfaa9c0a34ffc1b64 Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Wed, 11 Dec 2024 17:48:32 +0800 Subject: [PATCH 2/6] Simplify Quorum Calculations for Commit and Prepare Phases --- consensus/leader.go | 38 +++++++------------------------------- consensus/view_change.go | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 31 deletions(-) diff --git a/consensus/leader.go b/consensus/leader.go index 9e901e70c6..063c1e9c4d 100644 --- a/consensus/leader.go +++ b/consensus/leader.go @@ -135,20 +135,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 +189,14 @@ 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) + lastQuorumAchievedBlock := consensus.current.GetLastQuorumAchievedBlock(quorum.Prepare) + if quorumIsMet && recvMsg.BlockNum > lastQuorumAchievedBlock { // 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 +221,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 @@ -318,13 +296,11 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) { quorumIsMet := consensus.decider().IsQuorumAchieved(quorum.Commit) //// Read - End - - quorumAchievedByFirstCommit := hasMultiBlsKeys && isFirstReceivedSignature && quorumWasMet - quorumAchievedByThisCommit := !quorumWasMet && quorumIsMet - - if quorumAchievedByFirstCommit || quorumAchievedByThisCommit { + lastQuorumAchievedBlock := consensus.current.GetLastQuorumAchievedBlock(quorum.Commit) + if quorumIsMet && blockObj.NumberU64() > lastQuorumAchievedBlock { logger.Info().Msg("[OnCommit] 2/3 Enough commits received") consensus.fBFTLog.MarkBlockVerified(blockObj) + consensus.current.SetLastQuorumAchievedBlock(quorum.Commit, blockObj.NumberU64()) if !blockObj.IsLastBlockInEpoch() { // only do early commit if it's not epoch block to avoid problems diff --git a/consensus/view_change.go b/consensus/view_change.go index cc0fbd0d49..f49b869f1c 100644 --- a/consensus/view_change.go +++ b/consensus/view_change.go @@ -8,6 +8,7 @@ import ( "github.com/ethereum/go-ethereum/common" msg_pb "github.com/harmony-one/harmony/api/proto/message" "github.com/harmony-one/harmony/block" + types "github.com/harmony-one/harmony/common/types" "github.com/harmony-one/harmony/consensus/engine" "github.com/harmony-one/harmony/consensus/quorum" "github.com/harmony-one/harmony/crypto/bls" @@ -24,6 +25,28 @@ import ( // MaxViewIDDiff limits the received view ID to only 249 further from the current view ID const MaxViewIDDiff = 249 +// State contains current mode and current viewID +type State struct { + mode uint32 + + // current view id in normal mode + // it changes per successful consensus + blockViewID uint64 + + // view changing id is used during view change mode + // it is the next view id + viewChangingID uint64 + + quorumAchievedBlock *types.SafeMap[quorum.Phase, uint64] +} + +func NewState(mode Mode) State { + return State{ + mode: uint32(mode), + quorumAchievedBlock: types.NewSafeMap[quorum.Phase, uint64](), + } +} + // Mode return the current node mode func (pm *State) Mode() Mode { return Mode(atomic.LoadUint32(&pm.mode)) @@ -57,6 +80,23 @@ func (pm *State) SetViewChangingID(id uint64) { atomic.StoreUint64(&pm.viewChangingID, id) } +// GetLastQuorumAchievedBlock retrieves the block number of the last block +// that achieved quorum for the specified phase. +// If no quorum has been achieved for the given phase, it returns 0. +func (pm *State) GetLastQuorumAchievedBlock(p quorum.Phase) uint64 { + lqab, exists := pm.quorumAchievedBlock.Get(p) + if !exists { + return 0 + } + return lqab +} + +// SetLastQuorumAchievedBlock updates the block number of the last block +// that achieved quorum for the specified phase. +func (pm *State) SetLastQuorumAchievedBlock(p quorum.Phase, blockNum uint64) { + pm.quorumAchievedBlock.Set(p, blockNum) +} + // GetViewChangeDuraion return the duration of the current view change // It increase in the power of difference betweeen view changing ID and current view ID func (pm *State) GetViewChangeDuraion() time.Duration { From 6bb270ab32845316d2ee87e80eb0278a006ede62 Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Thu, 12 Dec 2024 14:33:11 +0800 Subject: [PATCH 3/6] revert changes by goimports in msg.pb.go --- p2p/stream/protocols/sync/message/msg.pb.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/p2p/stream/protocols/sync/message/msg.pb.go b/p2p/stream/protocols/sync/message/msg.pb.go index 2205810507..db37f7c7da 100644 --- a/p2p/stream/protocols/sync/message/msg.pb.go +++ b/p2p/stream/protocols/sync/message/msg.pb.go @@ -7,11 +7,10 @@ package message import ( - reflect "reflect" - sync "sync" - protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" ) const ( From 2b5264d3ee0860d1dcf8a68c99c7d1a6b5edc84e Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Thu, 12 Dec 2024 15:06:36 +0800 Subject: [PATCH 4/6] remove unsed function checkFirstReceivedSignature --- consensus/leader.go | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/consensus/leader.go b/consensus/leader.go index 063c1e9c4d..f36915e175 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. From 6bc141b19750e0bafae680f788c77e92b2011214 Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Fri, 18 Jul 2025 21:36:32 +0800 Subject: [PATCH 5/6] fix conflict issues of the consensus state and safe maps --- .../stagedstreamsync/stage_blockhashes.go | 5 ++- consensus/state.go | 32 ++++++++++++--- consensus/view_change.go | 40 ------------------- .../common/streammanager/streammanager.go | 9 +++-- 4 files changed, 35 insertions(+), 51 deletions(-) diff --git a/api/service/synchronize/stagedstreamsync/stage_blockhashes.go b/api/service/synchronize/stagedstreamsync/stage_blockhashes.go index 896c012c19..a2a2c0b475 100644 --- a/api/service/synchronize/stagedstreamsync/stage_blockhashes.go +++ b/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) { diff --git a/consensus/state.go b/consensus/state.go index 170de3f3b3..5044a4148a 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -4,6 +4,8 @@ import ( "sync/atomic" "unsafe" + "github.com/harmony-one/harmony/common/types" + "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,13 +40,16 @@ type State struct { // ShardID of the consensus ShardID uint32 + + quorumAchievedBlock *types.SafeMap[quorum.Phase, uint64] } func NewState(mode Mode, shardID uint32) State { state := State{ - mode: uint32(mode), - ShardID: shardID, - phase: atomic.Value{}, + mode: uint32(mode), + ShardID: shardID, + phase: atomic.Value{}, + quorumAchievedBlock: types.NewSafeMap[quorum.Phase, uint64](), } state.phase.Store(FBFTAnnounce) return state @@ -54,12 +59,12 @@ func (pm *State) getBlockNum() uint64 { return atomic.LoadUint64(&pm.blockNum) } -// SetBlockNum sets the blockNum in consensus object, called at node bootstrap +// setBlockNum sets the FBFT blockNum in consensus object, called at node bootstrap func (pm *State) setBlockNum(blockNum uint64) { atomic.StoreUint64(&pm.blockNum, blockNum) } -// SetBlockNum sets the blockNum in consensus object, called at node bootstrap +// SetBlockNum sets the FBFT blockNum in consensus object, called at node bootstrap func (pm *State) SetBlockNum(blockNum uint64) { pm.setBlockNum(blockNum) } @@ -69,6 +74,23 @@ func (pm *State) GetBlockNum() uint64 { return pm.getBlockNum() } +// GetLastQuorumAchievedBlock retrieves the block number of the last block +// that achieved quorum for the specified phase. +// If no quorum has been achieved for the given phase, it returns 0. +func (pm *State) GetLastQuorumAchievedBlock(p quorum.Phase) uint64 { + lqab, exists := pm.quorumAchievedBlock.Get(p) + if !exists { + return 0 + } + return lqab +} + +// SetLastQuorumAchievedBlock updates the block number of the last block +// that achieved quorum for the specified phase. +func (pm *State) SetLastQuorumAchievedBlock(p quorum.Phase, blockNum uint64) { + pm.quorumAchievedBlock.Set(p, blockNum) +} + func (pm *State) getLeaderPubKey() *bls_cosi.PublicKeyWrapper { return (*bls_cosi.PublicKeyWrapper)(atomic.LoadPointer(&pm.leaderPubKey)) } diff --git a/consensus/view_change.go b/consensus/view_change.go index f49b869f1c..cc0fbd0d49 100644 --- a/consensus/view_change.go +++ b/consensus/view_change.go @@ -8,7 +8,6 @@ import ( "github.com/ethereum/go-ethereum/common" msg_pb "github.com/harmony-one/harmony/api/proto/message" "github.com/harmony-one/harmony/block" - types "github.com/harmony-one/harmony/common/types" "github.com/harmony-one/harmony/consensus/engine" "github.com/harmony-one/harmony/consensus/quorum" "github.com/harmony-one/harmony/crypto/bls" @@ -25,28 +24,6 @@ import ( // MaxViewIDDiff limits the received view ID to only 249 further from the current view ID const MaxViewIDDiff = 249 -// State contains current mode and current viewID -type State struct { - mode uint32 - - // current view id in normal mode - // it changes per successful consensus - blockViewID uint64 - - // view changing id is used during view change mode - // it is the next view id - viewChangingID uint64 - - quorumAchievedBlock *types.SafeMap[quorum.Phase, uint64] -} - -func NewState(mode Mode) State { - return State{ - mode: uint32(mode), - quorumAchievedBlock: types.NewSafeMap[quorum.Phase, uint64](), - } -} - // Mode return the current node mode func (pm *State) Mode() Mode { return Mode(atomic.LoadUint32(&pm.mode)) @@ -80,23 +57,6 @@ func (pm *State) SetViewChangingID(id uint64) { atomic.StoreUint64(&pm.viewChangingID, id) } -// GetLastQuorumAchievedBlock retrieves the block number of the last block -// that achieved quorum for the specified phase. -// If no quorum has been achieved for the given phase, it returns 0. -func (pm *State) GetLastQuorumAchievedBlock(p quorum.Phase) uint64 { - lqab, exists := pm.quorumAchievedBlock.Get(p) - if !exists { - return 0 - } - return lqab -} - -// SetLastQuorumAchievedBlock updates the block number of the last block -// that achieved quorum for the specified phase. -func (pm *State) SetLastQuorumAchievedBlock(p quorum.Phase, blockNum uint64) { - pm.quorumAchievedBlock.Set(p, blockNum) -} - // GetViewChangeDuraion return the duration of the current view change // It increase in the power of difference betweeen view changing ID and current view ID func (pm *State) GetViewChangeDuraion() time.Duration { diff --git a/p2p/stream/common/streammanager/streammanager.go b/p2p/stream/common/streammanager/streammanager.go index 69303a118d..a2dfc3e271 100644 --- a/p2p/stream/common/streammanager/streammanager.go +++ b/p2p/stream/common/streammanager/streammanager.go @@ -11,6 +11,7 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/harmony-one/abool" "github.com/harmony-one/harmony/internal/utils" + types "github.com/harmony-one/harmony/common/types" sttypes "github.com/harmony-one/harmony/p2p/stream/types" "github.com/libp2p/go-libp2p/core/network" libp2p_peer "github.com/libp2p/go-libp2p/core/peer" @@ -48,7 +49,7 @@ type streamManager struct { // protocol ID (e.g. different version) streams *streamSet // tracks removed streams with cooldown - removedStreams *sttypes.SafeMap[sttypes.StreamID, *RemovalInfo] + removedStreams *types.SafeMap[sttypes.StreamID, *RemovalInfo] // reserved streams reservedStreams *streamSet isTrustedPeer func(libp2p_peer.ID) bool @@ -83,7 +84,7 @@ type streamManager struct { // trustedStreams tracks stream IDs of successfully established trusted peer streams // and their location (main or reserved list) for efficient counting // Value: true = main list, false = reserved list - trustedStreams *sttypes.SafeMap[sttypes.StreamID, bool] + trustedStreams *types.SafeMap[sttypes.StreamID, bool] // Atomic counters for trusted streams - optimized for O(1) counting numTrustedStreamsMain int64 // Count of trusted streams in main list numTrustedStreamsReserved int64 // Count of trusted streams in reserved list @@ -187,7 +188,7 @@ func newStreamManager(pid sttypes.ProtoID, host host, pf peerFinder, handleStrea config: c, streams: newStreamSet(), reservedStreams: newStreamSet(), - removedStreams: sttypes.NewSafeMap[sttypes.StreamID, *RemovalInfo](), + removedStreams: types.NewSafeMap[sttypes.StreamID, *RemovalInfo](), isTrustedPeer: c.IsTrustedPeer, getTrustedPeers: c.GetTrustedPeers, host: host, @@ -206,7 +207,7 @@ func newStreamManager(pid sttypes.ProtoID, host host, pf peerFinder, handleStrea setupSem: make(chan struct{}, setupConcurrency), trustedPeersInitiated: c.TrustedPeersInitiated, trustedPeersProcessed: abool.New(), - trustedStreams: sttypes.NewSafeMap[sttypes.StreamID, bool](), + trustedStreams: types.NewSafeMap[sttypes.StreamID, bool](), } // Initialize all stream metrics with this protocol ID From 79f50ace71edfd9d0d7bd0a151c7eddd64f545a6 Mon Sep 17 00:00:00 2001 From: GheisMohammadi Date: Mon, 3 Aug 2026 19:33:03 +0800 Subject: [PATCH 6/6] fix(consensus): make prepare/commit quorum tracking safe across view changes --- .../stagedstreamsync/stage_bodies.go | 3 +- consensus/consensus_service.go | 1 + consensus/leader.go | 13 +++-- consensus/state.go | 52 +++++++++++++------ consensus/state_test.go | 34 ++++++++++++ p2p/host.go | 5 +- p2p/stream/common/requestmanager/types.go | 9 ++-- 7 files changed, 88 insertions(+), 29 deletions(-) diff --git a/api/service/synchronize/stagedstreamsync/stage_bodies.go b/api/service/synchronize/stagedstreamsync/stage_bodies.go index f66a24d2f5..f3512c525a 100644 --- a/api/service/synchronize/stagedstreamsync/stage_bodies.go +++ b/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 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 f36915e175..f6a40df869 100644 --- a/consensus/leader.go +++ b/consensus/leader.go @@ -176,9 +176,11 @@ func (consensus *Consensus) onPrepare(recvMsg *FBFTMessage) { //// Write - End //// Read - Start + // 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) - lastQuorumAchievedBlock := consensus.current.GetLastQuorumAchievedBlock(quorum.Prepare) - if quorumIsMet && recvMsg.BlockNum > lastQuorumAchievedBlock { + if quorumIsMet && recvMsg.BlockNum > consensus.current.GetLastQuorumAchievedBlock(quorum.Prepare) { // NOTE Let it handle its own logs if err := consensus.didReachPrepareQuorum(); err != nil { return @@ -281,13 +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 - lastQuorumAchievedBlock := consensus.current.GetLastQuorumAchievedBlock(quorum.Commit) - if quorumIsMet && blockObj.NumberU64() > lastQuorumAchievedBlock { + 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, blockObj.NumberU64()) + consensus.current.SetLastQuorumAchievedBlock(quorum.Commit, 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 5044a4148a..eed3a9d4fb 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -4,7 +4,6 @@ import ( "sync/atomic" "unsafe" - "github.com/harmony-one/harmony/common/types" "github.com/harmony-one/harmony/consensus/quorum" bls_cosi "github.com/harmony-one/harmony/crypto/bls" "github.com/harmony-one/harmony/internal/utils" @@ -41,15 +40,20 @@ type State struct { // ShardID of the consensus ShardID uint32 - quorumAchievedBlock *types.SafeMap[quorum.Phase, uint64] + // 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 { state := State{ - mode: uint32(mode), - ShardID: shardID, - phase: atomic.Value{}, - quorumAchievedBlock: types.NewSafeMap[quorum.Phase, uint64](), + mode: uint32(mode), + ShardID: shardID, + phase: atomic.Value{}, } state.phase.Store(FBFTAnnounce) return state @@ -59,12 +63,12 @@ func (pm *State) getBlockNum() uint64 { return atomic.LoadUint64(&pm.blockNum) } -// setBlockNum sets the FBFT 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) } -// SetBlockNum sets the FBFT 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) { pm.setBlockNum(blockNum) } @@ -74,21 +78,35 @@ func (pm *State) GetBlockNum() uint64 { return pm.getBlockNum() } -// GetLastQuorumAchievedBlock retrieves the block number of the last block -// that achieved quorum for the specified phase. -// If no quorum has been achieved for the given phase, it returns 0. +// 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 { - lqab, exists := pm.quorumAchievedBlock.Get(p) - if !exists { + switch p { + case quorum.Prepare: + return atomic.LoadUint64(&pm.lastPrepareQuorumBlock) + case quorum.Commit: + return atomic.LoadUint64(&pm.lastCommitQuorumBlock) + default: return 0 } - return lqab } -// SetLastQuorumAchievedBlock updates the block number of the last block -// that achieved quorum for the specified phase. +// SetLastQuorumAchievedBlock records that quorum side-effects were applied for +// the given phase at blockNum. func (pm *State) SetLastQuorumAchievedBlock(p quorum.Phase, blockNum uint64) { - pm.quorumAchievedBlock.Set(p, blockNum) + 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 { diff --git a/consensus/state_test.go b/consensus/state_test.go index 5263705184..9e940d9de7 100644 --- a/consensus/state_test.go +++ b/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) + } +} diff --git a/p2p/host.go b/p2p/host.go index 23dbb33322..22db4aebd1 100644 --- a/p2p/host.go +++ b/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 diff --git a/p2p/stream/common/requestmanager/types.go b/p2p/stream/common/requestmanager/types.go index e9ccbf42b1..54c71a43ae 100644 --- a/p2p/stream/common/requestmanager/types.go +++ b/p2p/stream/common/requestmanager/types.go @@ -7,6 +7,7 @@ import ( "sync/atomic" "time" + types "github.com/harmony-one/harmony/common/types" sttypes "github.com/harmony-one/harmony/p2p/stream/types" "github.com/pkg/errors" ) @@ -100,8 +101,8 @@ type request struct { raw *interface{} // options priority reqPriority - whitelist *sttypes.SafeMap[sttypes.StreamID, struct{}] // allowed streams - blacklist *sttypes.SafeMap[sttypes.StreamID, struct{}] // banned streams} + whitelist *types.SafeMap[sttypes.StreamID, struct{}] // allowed streams + blacklist *types.SafeMap[sttypes.StreamID, struct{}] // banned streams} } func (req *request) ReqID() uint64 { @@ -140,7 +141,7 @@ func (req *request) isStreamAllowed(stid sttypes.StreamID) bool { func (req *request) addBlacklistedStream(stid sttypes.StreamID) { if req.blacklist == nil { - req.blacklist = sttypes.NewSafeMap[sttypes.StreamID, struct{}]() + req.blacklist = types.NewSafeMap[sttypes.StreamID, struct{}]() } req.blacklist.Set(stid, struct{}{}) } @@ -177,7 +178,7 @@ func (req *request) blacklistIDs() []sttypes.StreamID { func (req *request) addWhiteListStream(stid sttypes.StreamID) { if req.whitelist == nil { - req.whitelist = sttypes.NewSafeMap[sttypes.StreamID, struct{}]() + req.whitelist = types.NewSafeMap[sttypes.StreamID, struct{}]() } req.whitelist.Set(stid, struct{}{}) }