fix writing on channel when a block is requested in order to attest the current one

process/block/baseProcess.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/multiversx/mx-chain-core-go/core"
-	"github.com/multiversx/mx-chain-core-go/core/atomic"
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
 	"github.com/multiversx/mx-chain-core-go/data/block"
@@ -125,9 +124,10 @@ type baseProcessor struct {
 	nonceOfFirstCommittedBlock    core.OptionalUint64
 	extraDelayRequestBlockInfo    time.Duration
 
-	proofsPool             dataRetriever.ProofsPool
-	chanNextHeader         chan bool
-	isWaitingForNextHeader atomic.Flag
+	proofsPool                     dataRetriever.ProofsPool
+	mutRequestedAttestingNoncesMap sync.RWMutex
+	requestedAttestingNoncesMap    map[string]uint64
+	allProofsReceived              chan bool
 }
 
 type bootStorerDataArgs struct {
@@ -2342,58 +2342,95 @@ 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.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)
-	bp.isWaitingForNextHeader.SetValue(true)
+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() 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:
-		bp.isWaitingForNextHeader.Reset()
+	case <-bp.allProofsReceived:
 		return nil
 	case <-time.After(bp.extraDelayRequestBlockInfo):
+		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) checkReceivedHeaderAndUpdateMissingAttesting(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
+	}
+
+	bp.mutRequestedAttestingNoncesMap.Lock()
+
+	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")
+		bp.mutRequestedAttestingNoncesMap.Unlock()
+		return
+	}
+
+	if !bp.proofsPool.HasProof(receivedShard, prevHash) {
+		log.Debug("received next header but proof is still missing", "hash", hex.EncodeToString(prevHash))
+		bp.mutRequestedAttestingNoncesMap.Unlock()
+		return
+	}
+
+	delete(bp.requestedAttestingNoncesMap, string(prevHash))
+
+	allProofsReceived := len(bp.requestedAttestingNoncesMap) == 0
+	bp.mutRequestedAttestingNoncesMap.Unlock()
+
+	if allProofsReceived {
+		bp.allProofsReceived <- true
+	}
+}

process/block/metablock.go

@@ -423,6 +423,11 @@ func (mp *metaProcessor) checkProofsForShardData(header *block.MetaBlock) error
 		return nil
 	}
 
+	mp.mutRequestedAttestingNoncesMap.Lock()
+	mp.requestedAttestingNoncesMap = make(map[string]uint64)
+	_ = core.EmptyChannel(mp.allProofsReceived)
+	mp.mutRequestedAttestingNoncesMap.Unlock()
+
 	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()
 }
 
 func (mp *metaProcessor) processEpochStartMetaBlock(
@@ -2073,11 +2075,7 @@ func (mp *metaProcessor) receivedShardHeader(headerHandler data.HeaderHandler, s
 		"hash", shardHeaderHash,
 	)
 
-	if mp.isWaitingForNextHeader.IsSet() {
-		log.Trace("received shard header attesting the previous one")
-		mp.chanNextHeader <- true
-		return
-	}
+	mp.checkReceivedHeaderAndUpdateMissingAttesting(headerHandler)
 
 	mp.hdrsForCurrBlock.mutHdrsForBlock.Lock()
 

process/block/shardblock.go

@@ -293,6 +293,11 @@ func (sp *shardProcessor) ProcessBlock(
 		return process.ErrAccountStateDirty
 	}
 
+	sp.mutRequestedAttestingNoncesMap.Lock()
+	sp.requestedAttestingNoncesMap = make(map[string]uint64)
+	_ = core.EmptyChannel(sp.allProofsReceived)
+	sp.mutRequestedAttestingNoncesMap.Unlock()
+
 	// check proofs for cross notarized metablocks
 	for _, metaBlockHash := range header.GetMetaBlockHashes() {
 		hInfo, ok := sp.hdrsForCurrBlock.hdrHashAndInfo[string(metaBlockHash)]
@@ -304,10 +309,12 @@ func (sp *shardProcessor) ProcessBlock(
 			continue
 		}
 
-		err = sp.checkProofRequestingNextHeaderBlockingIfMissing(core.MetachainShardId, metaBlockHash, hInfo.hdr.GetNonce())
-		if err != nil {
-			return err
-		}
+		sp.checkProofRequestingNextHeaderIfMissing(core.MetachainShardId, metaBlockHash, hInfo.hdr.GetNonce())
+	}
+
+	err = sp.waitAllMissingProofs()
+	if err != nil {
+		return err
 	}
 
 	defer func() {
@@ -1756,11 +1763,7 @@ func (sp *shardProcessor) receivedMetaBlock(headerHandler data.HeaderHandler, me
 		"hash", metaBlockHash,
 	)
 
-	if sp.isWaitingForNextHeader.IsSet() {
-		log.Trace("received meta header attesting the previous one")
-		sp.chanNextHeader <- true
-		return
-	}
+	sp.checkReceivedHeaderAndUpdateMissingAttesting(headerHandler)
 
 	sp.hdrsForCurrBlock.mutHdrsForBlock.Lock()