[DO NOT MERGE] Test branch for equivalent messages with activation @ epoch 4

cmd/node/config/config.toml

@@ -43,7 +43,7 @@
     # ChainParametersByEpoch defines chain operation configurable values that can be modified based on epochs
     ChainParametersByEpoch = [
         { EnableEpoch = 0, RoundDuration = 6000, ShardConsensusGroupSize = 7, ShardMinNumNodes = 10, MetachainConsensusGroupSize = 10, MetachainMinNumNodes = 10, Hysteresis = 0.2, Adaptivity = false },
-        { EnableEpoch = 8, RoundDuration = 6000, ShardConsensusGroupSize = 10, ShardMinNumNodes = 10, MetachainConsensusGroupSize = 10, MetachainMinNumNodes = 10, Hysteresis = 0.2, Adaptivity = false }
+        { EnableEpoch = 4, RoundDuration = 6000, ShardConsensusGroupSize = 10, ShardMinNumNodes = 10, MetachainConsensusGroupSize = 10, MetachainMinNumNodes = 10, Hysteresis = 0.2, Adaptivity = false }
     ]
 
 [HardwareRequirements]
@@ -632,7 +632,7 @@
 [EpochStartConfig]
     GenesisEpoch = 0
     MinRoundsBetweenEpochs = 20
-    RoundsPerEpoch         = 200
+    RoundsPerEpoch         = 100
     # Min and Max ShuffledOutRestartThreshold represents the minimum and maximum duration of an epoch (in percentage) after a node which
     # has been shuffled out has to restart its process in order to start in a new shard
     MinShuffledOutRestartThreshold = 0.05

cmd/node/config/enableEpochs.toml

@@ -337,10 +337,10 @@
     RelayedTransactionsV3EnableEpoch = 1
 
     # EquivalentMessagesEnableEpoch represents the epoch when the equivalent messages are enabled
-    EquivalentMessagesEnableEpoch = 8 # the chain simulator tests for staking v4 fail if this is set earlier, as they test the transition in epochs 4-7
+    EquivalentMessagesEnableEpoch = 4 # the chain simulator tests for staking v4 fail if this is set earlier, as they test the transition in epochs 4-7
 
     # FixedOrderInConsensusEnableEpoch represents the epoch when the fixed order in consensus is enabled
-    FixedOrderInConsensusEnableEpoch = 8 # the chain simulator tests for staking v4 fail if this is set earlier, as they test the transition in epochs 4-7
+    FixedOrderInConsensusEnableEpoch = 4 # the chain simulator tests for staking v4 fail if this is set earlier, as they test the transition in epochs 4-7
 
     # BLSMultiSignerEnableEpoch represents the activation epoch for different types of BLS multi-signers
     BLSMultiSignerEnableEpoch = [

common/common.go

@@ -6,6 +6,7 @@ import (
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
+
 	"github.com/multiversx/mx-chain-go/consensus"
 )
 
@@ -84,3 +85,14 @@ func VerifyProofAgainstHeader(proof data.HeaderProofHandler, header data.HeaderH
 
 	return nil
 }
+
+// GetShardIDs returns a map of shard IDs based on the provided shard coordinator
+func GetShardIDs(numShards uint32) map[uint32]struct{} {
+	shardIdentifiers := make(map[uint32]struct{})
+	for i := uint32(0); i < numShards; i++ {
+		shardIdentifiers[i] = struct{}{}
+	}
+	shardIdentifiers[core.MetachainShardId] = struct{}{}
+
+	return shardIdentifiers
+}

consensus/broadcast/delayedBroadcast.go

@@ -585,7 +585,7 @@ func (dbb *delayedBlockBroadcaster) registerInterceptorsCallbackForShard(
 	rootTopic string,
 	cb func(topic string, hash []byte, data interface{}),
 ) error {
-	shardIDs := dbb.shardIdentifiers()
+	shardIDs := common.GetShardIDs(dbb.shardCoordinator.NumberOfShards())
 	for idx := range shardIDs {
 		// interested only in cross shard data
 		if idx == dbb.shardCoordinator.SelfId() {
@@ -603,16 +603,6 @@ func (dbb *delayedBlockBroadcaster) registerInterceptorsCallbackForShard(
 	return nil
 }
 
-func (dbb *delayedBlockBroadcaster) shardIdentifiers() map[uint32]struct{} {
-	shardIdentifiers := make(map[uint32]struct{})
-	for i := uint32(0); i < dbb.shardCoordinator.NumberOfShards(); i++ {
-		shardIdentifiers[i] = struct{}{}
-	}
-	shardIdentifiers[core.MetachainShardId] = struct{}{}
-
-	return shardIdentifiers
-}
-
 func (dbb *delayedBlockBroadcaster) interceptedHeader(_ string, headerHash []byte, header interface{}) {
 	headerHandler, ok := header.(data.HeaderHandler)
 	if !ok {

epochStart/shardchain/trigger.go

@@ -570,6 +570,7 @@ func (t *trigger) receivedProof(headerProof data.HeaderProofHandler) {
 		return
 	}
 
+	log.Debug("received proof in trigger", "proof for header hash", headerProof.GetHeaderHash())
 	t.mutTrigger.Lock()
 	defer t.mutTrigger.Unlock()
 
@@ -584,6 +585,11 @@ func (t *trigger) receivedProof(headerProof data.HeaderProofHandler) {
 // receivedMetaBlock is a callback function when a new metablock was received
 // upon receiving checks if trigger can be updated
 func (t *trigger) receivedMetaBlock(headerHandler data.HeaderHandler, metaBlockHash []byte) {
+	if headerHandler.GetShardID() != core.MetachainShardId {
+		return
+	}
+
+	log.Debug("received meta header in trigger", "header hash", metaBlockHash)
 	if t.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, headerHandler.GetEpoch()) {
 		proof, err := t.proofsPool.GetProof(headerHandler.GetShardID(), metaBlockHash)
 		if err != nil {
@@ -607,10 +613,12 @@ func (t *trigger) checkMetaHeaderForEpochTriggerEquivalentProofs(headerHandler d
 	if !ok {
 		return
 	}
+	log.Debug("trigger.checkMetaHeaderForEpochTriggerEquivalentProofs", "metaHdr epoch", metaHdr.GetEpoch(), "metaBlockHash", metaBlockHash)
 	if !t.shouldUpdateTrigger(metaHdr, metaBlockHash) {
 		return
 	}
 
+	log.Debug("trigger.updateTriggerHeaderData")
 	t.updateTriggerHeaderData(metaHdr, metaBlockHash)
 	t.updateTriggerFromMeta()
 }

process/block/baseProcess.go

@@ -124,8 +124,10 @@ type baseProcessor struct {
 	nonceOfFirstCommittedBlock    core.OptionalUint64
 	extraDelayRequestBlockInfo    time.Duration
 
-	proofsPool     dataRetriever.ProofsPool
-	chanNextHeader chan bool
+	proofsPool                     dataRetriever.ProofsPool
+	mutRequestedAttestingNoncesMap sync.RWMutex
+	requestedAttestingNoncesMap    map[string]uint64
+	allProofsReceived              chan bool
 }
 
 type bootStorerDataArgs struct {
@@ -666,10 +668,51 @@ func (bp *baseProcessor) verifyFees(header data.HeaderHandler) error {
 	return nil
 }
 
+func (bp *baseProcessor) filterHeadersWithoutProofs() (map[string]*hdrInfo, error) {
+	removedNonces := make(map[uint32]map[uint64]struct{})
+	noncesWithProofs := make(map[uint32]map[uint64]struct{})
+	shardIDs := common.GetShardIDs(bp.shardCoordinator.NumberOfShards())
+	for shard := range shardIDs {
+		removedNonces[shard] = make(map[uint64]struct{})
+		noncesWithProofs[shard] = make(map[uint64]struct{})
+	}
+	filteredHeadersInfo := make(map[string]*hdrInfo)
+
+	for hdrHash, headerInfo := range bp.hdrsForCurrBlock.hdrHashAndInfo {
+		if bp.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, headerInfo.hdr.GetEpoch()) {
+			if bp.hasMissingProof(headerInfo, hdrHash) {
+				removedNonces[headerInfo.hdr.GetShardID()][headerInfo.hdr.GetNonce()] = struct{}{}
+				continue
+			}
+
+			noncesWithProofs[headerInfo.hdr.GetShardID()][headerInfo.hdr.GetNonce()] = struct{}{}
+			filteredHeadersInfo[hdrHash] = bp.hdrsForCurrBlock.hdrHashAndInfo[hdrHash]
+			continue
+		}
+
+		filteredHeadersInfo[hdrHash] = bp.hdrsForCurrBlock.hdrHashAndInfo[hdrHash]
+	}
+
+	for shard, nonces := range removedNonces {
+		for nonce := range nonces {
+			if _, ok := noncesWithProofs[shard][nonce]; !ok {
+				return nil, fmt.Errorf("%w for shard %d and nonce %d", process.ErrMissingHeaderProof, shard, nonce)
+			}
+		}
+	}
+
+	return filteredHeadersInfo, nil
+}
+
 func (bp *baseProcessor) computeHeadersForCurrentBlock(usedInBlock bool) (map[uint32][]data.HeaderHandler, error) {
 	hdrsForCurrentBlock := make(map[uint32][]data.HeaderHandler)
 
-	for hdrHash, headerInfo := range bp.hdrsForCurrBlock.hdrHashAndInfo {
+	hdrHashAndInfo, err := bp.filterHeadersWithoutProofs()
+	if err != nil {
+		return nil, err
+	}
+
+	for hdrHash, headerInfo := range hdrHashAndInfo {
 		if headerInfo.usedInBlock != usedInBlock {
 			continue
 		}
@@ -747,7 +790,7 @@ func (bp *baseProcessor) sortHeaderHashesForCurrentBlockByNonce(usedInBlock bool
 }
 
 func (bp *baseProcessor) hasMissingProof(headerInfo *hdrInfo, hdrHash string) bool {
-	isFlagEnabledForHeader := bp.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, headerInfo.hdr.GetEpoch()) && headerInfo.hdr.GetNonce() > 1
+	isFlagEnabledForHeader := bp.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, headerInfo.hdr.GetEpoch()) && headerInfo.hdr.GetNonce() >= 1
 	if !isFlagEnabledForHeader {
 		return false
 	}
@@ -2299,56 +2342,96 @@ func (bp *baseProcessor) getHeaderHash(header data.HeaderHandler) ([]byte, error
 	return bp.hasher.Compute(string(marshalledHeader)), nil
 }
 
-func (bp *baseProcessor) checkProofRequestingNextHeaderBlockingIfMissing(
+func (bp *baseProcessor) checkProofRequestingNextHeaderIfMissing(
 	headerShard uint32,
 	headerHash []byte,
 	headerNonce uint64,
-) error {
+) {
 	if bp.proofsPool.HasProof(headerShard, headerHash) {
-		return nil
+		return
 	}
 
-	log.Trace("could not find proof for header, requesting the next one",
+	log.Debug("could not find proof for header, requesting the next one",
 		"current hash", hex.EncodeToString(headerHash),
 		"header shard", headerShard)
-	err := bp.requestNextHeaderBlocking(headerNonce+1, headerShard)
-	if err != nil {
-		return err
-	}
-
-	if bp.proofsPool.HasProof(headerShard, headerHash) {
-		return nil
-	}
 
-	return fmt.Errorf("%w for header hash %s", process.ErrMissingHeaderProof, hex.EncodeToString(headerHash))
+	bp.requestNextHeader(headerHash, headerNonce+1, headerShard)
 }
 
-func (bp *baseProcessor) requestNextHeaderBlocking(nonce uint64, shardID uint32) error {
-	headersPool := bp.dataPool.Headers()
-
-	_ = core.EmptyChannel(bp.chanNextHeader)
+func (bp *baseProcessor) requestNextHeader(currentHeaderHash []byte, nonce uint64, shardID uint32) {
+	bp.mutRequestedAttestingNoncesMap.Lock()
+	bp.requestedAttestingNoncesMap[string(currentHeaderHash)] = nonce
+	bp.mutRequestedAttestingNoncesMap.Unlock()
 
 	if shardID == core.MetachainShardId {
 		go bp.requestHandler.RequestMetaHeaderByNonce(nonce)
 	} else {
 		go bp.requestHandler.RequestShardHeaderByNonce(shardID, nonce)
 	}
+}
 
-	err := bp.waitForNextHeader()
-	if err != nil {
-		return err
+func (bp *baseProcessor) waitAllMissingProofs(waitTime time.Duration) error {
+	bp.mutRequestedAttestingNoncesMap.RLock()
+	isWaitingForProofs := len(bp.requestedAttestingNoncesMap) > 0
+	bp.mutRequestedAttestingNoncesMap.RUnlock()
+	if !isWaitingForProofs {
+		return nil
 	}
 
-	_, _, err = process.GetShardHeaderFromPoolWithNonce(nonce, shardID, headersPool)
-
-	return err
-}
-
-func (bp *baseProcessor) waitForNextHeader() error {
 	select {
-	case <-bp.chanNextHeader:
+	case <-bp.allProofsReceived:
 		return nil
-	case <-time.After(bp.extraDelayRequestBlockInfo):
+	case <-time.After(waitTime):
+		bp.mutRequestedAttestingNoncesMap.RLock()
+		defer bp.mutRequestedAttestingNoncesMap.RUnlock()
+
+		logMessage := ""
+		for hash := range bp.requestedAttestingNoncesMap {
+			logMessage += fmt.Sprintf("\nhash = %s", hex.EncodeToString([]byte(hash)))
+		}
+
+		log.Debug("baseProcessor.waitAllMissingProofs still pending proofs for headers" + logMessage)
+
 		return process.ErrTimeIsOut
 	}
 }
+
+func (bp *baseProcessor) checkReceivedHeaderIfAttestingIsNeeded(headerHandler data.HeaderHandler) {
+	if !common.ShouldBlockHavePrevProof(headerHandler, bp.enableEpochsHandler, common.EquivalentMessagesFlag) {
+		return
+	}
+
+	bp.mutRequestedAttestingNoncesMap.RLock()
+	isWaitingForProofs := len(bp.requestedAttestingNoncesMap) > 0
+	bp.mutRequestedAttestingNoncesMap.RUnlock()
+	if !isWaitingForProofs {
+		return
+	}
+
+	allProofsReceived := bp.checkReceivedHeaderAndUpdateMissingAttesting(headerHandler)
+	if allProofsReceived {
+		bp.allProofsReceived <- true
+	}
+}
+
+func (bp *baseProcessor) checkReceivedHeaderAndUpdateMissingAttesting(headerHandler data.HeaderHandler) bool {
+	bp.mutRequestedAttestingNoncesMap.Lock()
+	defer bp.mutRequestedAttestingNoncesMap.Unlock()
+
+	receivedShard := headerHandler.GetShardID()
+	prevHash := headerHandler.GetPrevHash()
+	_, isHeaderWithoutProof := bp.requestedAttestingNoncesMap[string(prevHash)]
+	if !isHeaderWithoutProof {
+		log.Debug("received header does not have previous hash any of the requested ones")
+		return len(bp.requestedAttestingNoncesMap) == 0
+	}
+
+	if !bp.proofsPool.HasProof(receivedShard, prevHash) {
+		log.Debug("received next header but proof is still missing", "hash", hex.EncodeToString(prevHash))
+		return len(bp.requestedAttestingNoncesMap) == 0
+	}
+
+	delete(bp.requestedAttestingNoncesMap, string(prevHash))
+
+	return len(bp.requestedAttestingNoncesMap) == 0
+}

process/block/metablock.go

@@ -343,7 +343,7 @@ func (mp *metaProcessor) ProcessBlock(
 		}
 	}
 
-	err = mp.checkProofsForShardData(header)
+	err = mp.checkProofsForShardData(header, haveTime())
 	if err != nil {
 		return err
 	}
@@ -418,11 +418,16 @@ func (mp *metaProcessor) ProcessBlock(
 	return nil
 }
 
-func (mp *metaProcessor) checkProofsForShardData(header *block.MetaBlock) error {
+func (mp *metaProcessor) checkProofsForShardData(header *block.MetaBlock, waitTime time.Duration) error {
 	if !(mp.enableEpochsHandler.IsFlagEnabledInEpoch(common.EquivalentMessagesFlag, header.Epoch) && header.GetNonce() > 1) {
 		return nil
 	}
 
+	mp.mutRequestedAttestingNoncesMap.Lock()
+	mp.requestedAttestingNoncesMap = make(map[string]uint64)
+	mp.mutRequestedAttestingNoncesMap.Unlock()
+	_ = core.EmptyChannel(mp.allProofsReceived)
+
 	for _, shardData := range header.ShardInfo {
 		// TODO: consider the validation of the proof:
 		//	compare the one from proofsPool with what shardData.CurrentSignature and shardData.CurrentPubKeysBitmap hold
@@ -437,10 +442,7 @@ func (mp *metaProcessor) checkProofsForShardData(header *block.MetaBlock) error
 			continue
 		}
 
-		err := mp.checkProofRequestingNextHeaderBlockingIfMissing(shardData.ShardID, shardData.HeaderHash, shardData.Nonce)
-		if err != nil {
-			return err
-		}
+		mp.checkProofRequestingNextHeaderIfMissing(shardData.ShardID, shardData.HeaderHash, shardData.Nonce)
 
 		if !common.ShouldBlockHavePrevProof(shardHeader.hdr, mp.enableEpochsHandler, common.EquivalentMessagesFlag) {
 			continue
@@ -464,7 +466,7 @@ func (mp *metaProcessor) checkProofsForShardData(header *block.MetaBlock) error
 		}
 	}
 
-	return nil
+	return mp.waitAllMissingProofs(waitTime)
 }
 
 func (mp *metaProcessor) processEpochStartMetaBlock(
@@ -2073,6 +2075,8 @@ func (mp *metaProcessor) receivedShardHeader(headerHandler data.HeaderHandler, s
 		"hash", shardHeaderHash,
 	)
 
+	mp.checkReceivedHeaderIfAttestingIsNeeded(headerHandler)
+
 	mp.hdrsForCurrBlock.mutHdrsForBlock.Lock()
 
 	haveMissingShardHeaders := mp.hdrsForCurrBlock.missingHdrs > 0 || mp.hdrsForCurrBlock.missingFinalityAttestingHdrs > 0