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/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/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/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..f6a40df869 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,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 diff --git a/consensus/state.go b/consensus/state.go index 170de3f3b3..eed3a9d4fb 100644 --- a/consensus/state.go +++ b/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)) } 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/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/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{}{}) } 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