Merge branch 'development' into EN-4543-Improve-probable-highest-nonc…

…e-reset-and-sync-status

# Conflicts:
#	process/constants.go
#	process/sync/baseSync.go

dataRetriever/dataPool/metaDataPool.go

@@ -53,8 +53,8 @@ func NewMetaDataPool(
 	}, nil
 }
 
-// MetaChainBlocks returns the holder for meta blocks
-func (mdp *metaDataPool) MetaChainBlocks() storage.Cacher {
+// MetaBlocks returns the holder for meta blocks
+func (mdp *metaDataPool) MetaBlocks() storage.Cacher {
 	return mdp.metaBlocks
 }
 

dataRetriever/dataPool/metaDataPool_test.go

@@ -110,15 +110,15 @@ func TestNewMetaDataPool_NilUnsingedPoolNoncesShouldErr(t *testing.T) {
 func TestNewMetaDataPool_ConfigOk(t *testing.T) {
 	t.Parallel()
 
-	metaChainBlocks := &mock.CacherStub{}
+	metaBlocks := &mock.CacherStub{}
 	shardHeaders := &mock.CacherStub{}
 	miniBlocks := &mock.CacherStub{}
 	hdrsNonces := &mock.Uint64SyncMapCacherStub{}
 	transactions := &mock.ShardedDataStub{}
 	unsigned := &mock.ShardedDataStub{}
 
 	tdp, err := dataPool.NewMetaDataPool(
-		metaChainBlocks,
+		metaBlocks,
 		miniBlocks,
 		shardHeaders,
 		hdrsNonces,
@@ -128,7 +128,7 @@ func TestNewMetaDataPool_ConfigOk(t *testing.T) {
 
 	assert.Nil(t, err)
 	//pointer checking
-	assert.True(t, metaChainBlocks == tdp.MetaChainBlocks())
+	assert.True(t, metaBlocks == tdp.MetaBlocks())
 	assert.True(t, shardHeaders == tdp.ShardHeaders())
 	assert.True(t, miniBlocks == tdp.MiniBlocks())
 	assert.True(t, hdrsNonces == tdp.HeadersNonces())

dataRetriever/dataPool/nonceSyncMapCacher.go

@@ -120,6 +120,7 @@ func (nspc *nonceSyncMapCacher) Remove(nonce uint64, shardId uint32) {
 	}
 
 	syncMap.Delete(shardId)
+	nspc.removeNonceFromCacheIfSyncMapIsEmpty(nonce, syncMap)
 }
 
 // RegisterHandler registers a new handler to be called when a new data is added
@@ -166,3 +167,18 @@ func (nspc *nonceSyncMapCacher) IsInterfaceNil() bool {
 	}
 	return false
 }
+
+func (nspc *nonceSyncMapCacher) removeNonceFromCacheIfSyncMapIsEmpty(nonce uint64, syncMap *ShardIdHashSyncMap) {
+	isSyncMapEmpty := true
+	syncMap.Range(func(shardId uint32, hash []byte) bool {
+		if hash != nil {
+			isSyncMapEmpty = false
+			return false
+		}
+		return true
+	})
+
+	if isSyncMapEmpty {
+		nspc.cacher.Remove(nspc.nonceConverter.ToByteSlice(nonce))
+	}
+}

dataRetriever/dataPool/nonceSyncMapCacher_test.go

@@ -193,6 +193,7 @@ func TestNonceSyncMapCacher_RemoveShardIdWhenExists(t *testing.T) {
 			PeekCalled: func(key []byte) (interface{}, bool) {
 				return syncMap, true
 			},
+			RemoveCalled: func(key []byte) {},
 		},
 		mock.NewNonceHashConverterMock())
 

dataRetriever/factory/metachain/resolversContainerFactory.go

@@ -248,7 +248,7 @@ func (rcf *resolversContainerFactory) createMetaChainHeaderResolver(identifier s
 	hdrNonceStore := rcf.store.GetStorer(dataRetriever.MetaHdrNonceHashDataUnit)
 	resolver, err := resolvers.NewHeaderResolver(
 		resolverSender,
-		rcf.dataPools.MetaChainBlocks(),
+		rcf.dataPools.MetaBlocks(),
 		rcf.dataPools.HeadersNonces(),
 		hdrStorer,
 		hdrNonceStore,

dataRetriever/factory/metachain/resolversContainerFactory_test.go

@@ -52,7 +52,7 @@ func createDataPools() dataRetriever.MetaPoolsHolder {
 		MiniBlocksCalled: func() storage.Cacher {
 			return &mock.CacherStub{}
 		},
-		MetaChainBlocksCalled: func() storage.Cacher {
+		MetaBlocksCalled: func() storage.Cacher {
 			return &mock.CacherStub{}
 		},
 		HeadersNoncesCalled: func() dataRetriever.Uint64SyncMapCacher {

dataRetriever/interface.go

@@ -212,7 +212,7 @@ type PoolsHolder interface {
 
 // MetaPoolsHolder defines getter for data pools for metachain
 type MetaPoolsHolder interface {
-	MetaChainBlocks() storage.Cacher
+	MetaBlocks() storage.Cacher
 	MiniBlocks() storage.Cacher
 	ShardHeaders() storage.Cacher
 	HeadersNonces() Uint64SyncMapCacher

dataRetriever/mock/metaPoolsHolderStub.go

@@ -6,7 +6,7 @@ import (
 )
 
 type MetaPoolsHolderStub struct {
-	MetaChainBlocksCalled      func() storage.Cacher
+	MetaBlocksCalled           func() storage.Cacher
 	MiniBlocksCalled           func() storage.Cacher
 	ShardHeadersCalled         func() storage.Cacher
 	HeadersNoncesCalled        func() dataRetriever.Uint64SyncMapCacher
@@ -22,8 +22,8 @@ func (mphs *MetaPoolsHolderStub) UnsignedTransactions() dataRetriever.ShardedDat
 	return mphs.UnsignedTransactionsCalled()
 }
 
-func (mphs *MetaPoolsHolderStub) MetaChainBlocks() storage.Cacher {
-	return mphs.MetaChainBlocksCalled()
+func (mphs *MetaPoolsHolderStub) MetaBlocks() storage.Cacher {
+	return mphs.MetaBlocksCalled()
 }
 
 func (mphs *MetaPoolsHolderStub) MiniBlocks() storage.Cacher {

integrationTests/multiShard/block/interceptedBlocks_test.go

@@ -156,7 +156,7 @@ func TestMetaHeadersAreRequstedOnlyFromMetachain(t *testing.T) {
 	}
 
 	chanReceived := make(chan struct{}, 1000)
-	node4Meta.MetaDataPool.MetaChainBlocks().Put(metaHdrHashFromMetachain, metaHdrFromMetachain)
+	node4Meta.MetaDataPool.MetaBlocks().Put(metaHdrHashFromMetachain, metaHdrFromMetachain)
 	node1Shard0.ShardDataPool.MetaBlocks().Clear()
 	node1Shard0.ShardDataPool.MetaBlocks().RegisterHandler(func(key []byte) {
 		chanReceived <- struct{}{}

integrationTests/multiShard/block/interceptedHeadersSigVerification_test.go

@@ -144,7 +144,7 @@ func TestInterceptedMetaBlockVerifiedWithCorrectConsensusGroup(t *testing.T) {
 
 	// all nodes in metachain do not have the block in pool as interceptor does not validate it with a wrong consensus
 	for _, metaNode := range nodesMap[sharding.MetachainShardId] {
-		v, ok := metaNode.MetaDataPool.MetaChainBlocks().Get(headerHash)
+		v, ok := metaNode.MetaDataPool.MetaBlocks().Get(headerHash)
 		assert.True(t, ok)
 		assert.Equal(t, header, v)
 	}

integrationTests/multiShard/metablock/blocksDissemination_test.go

@@ -132,7 +132,7 @@ func TestHeadersAreResolvedByMetachainAndShard(t *testing.T) {
 	metaHeaderBytes, _ := integrationTests.TestMarshalizer.Marshal(metaHdr)
 	metaHeaderHash := integrationTests.TestHasher.Compute(string(metaHeaderBytes))
 	for i := 0; i < numMetaNodes; i++ {
-		nodes[i+1].MetaDataPool.MetaChainBlocks().HasOrAdd(metaHeaderHash, metaHdr)
+		nodes[i+1].MetaDataPool.MetaBlocks().HasOrAdd(metaHeaderHash, metaHdr)
 	}
 
 	for i := 0; i < maxNumRequests; i++ {
@@ -156,7 +156,7 @@ func TestHeadersAreResolvedByMetachainAndShard(t *testing.T) {
 	metaHeaderBytes2, _ := integrationTests.TestMarshalizer.Marshal(metaHdr2)
 	metaHeaderHash2 := integrationTests.TestHasher.Compute(string(metaHeaderBytes2))
 	for i := 0; i < numMetaNodes; i++ {
-		nodes[i+1].MetaDataPool.MetaChainBlocks().HasOrAdd(metaHeaderHash2, metaHdr2)
+		nodes[i+1].MetaDataPool.MetaBlocks().HasOrAdd(metaHeaderHash2, metaHdr2)
 
 		syncMap := &dataPool.ShardIdHashSyncMap{}
 		syncMap.Store(sharding.MetachainShardId, metaHeaderHash2)

integrationTests/multiShard/smartContract/testInitializer.go

@@ -901,7 +901,7 @@ func createMetaNetNode(
 	tn.accntState = accntAdapter
 	tn.shardId = sharding.MetachainShardId
 
-	dPool.MetaChainBlocks().RegisterHandler(func(key []byte) {
+	dPool.MetaBlocks().RegisterHandler(func(key []byte) {
 		atomic.AddInt32(&tn.metachainHdrRecv, 1)
 	})
 	dPool.ShardHeaders().RegisterHandler(func(key []byte) {

integrationTests/sync/common.go

@@ -157,7 +157,7 @@ func emptyShardDataPool(sdp dataRetriever.PoolsHolder) {
 
 func emptyMetaDataPool(holder dataRetriever.MetaPoolsHolder) {
 	holder.HeadersNonces().Clear()
-	holder.MetaChainBlocks().Clear()
+	holder.MetaBlocks().Clear()
 	holder.MiniBlocks().Clear()
 	holder.ShardHeaders().Clear()
 }

integrationTests/testProcessorNode.go

@@ -607,7 +607,7 @@ func (tpn *TestProcessorNode) addHandlersForCounters() {
 
 	if tpn.ShardCoordinator.SelfId() == sharding.MetachainShardId {
 		tpn.MetaDataPool.ShardHeaders().RegisterHandler(hdrHandlers)
-		tpn.MetaDataPool.MetaChainBlocks().RegisterHandler(metaHandlers)
+		tpn.MetaDataPool.MetaBlocks().RegisterHandler(metaHandlers)
 	} else {
 		txHandler := func(key []byte) {
 			atomic.AddInt32(&tpn.CounterTxRecv, 1)
@@ -768,7 +768,7 @@ func (tpn *TestProcessorNode) GetBlockBody(header *dataBlock.Header) (dataBlock.
 
 // GetMetaHeader returns the first *dataBlock.MetaBlock stored in datapools having the nonce provided as parameter
 func (tpn *TestProcessorNode) GetMetaHeader(nonce uint64) (*dataBlock.MetaBlock, error) {
-	invalidCachers := tpn.MetaDataPool == nil || tpn.MetaDataPool.MetaChainBlocks() == nil || tpn.MetaDataPool.HeadersNonces() == nil
+	invalidCachers := tpn.MetaDataPool == nil || tpn.MetaDataPool.MetaBlocks() == nil || tpn.MetaDataPool.HeadersNonces() == nil
 	if invalidCachers {
 		return nil, errors.New("invalid data pool")
 	}
@@ -783,7 +783,7 @@ func (tpn *TestProcessorNode) GetMetaHeader(nonce uint64) (*dataBlock.MetaBlock,
 		return nil, errors.New(fmt.Sprintf("no hash-nonce hash in HeadersNonces for nonce %d", nonce))
 	}
 
-	headerObject, ok := tpn.MetaDataPool.MetaChainBlocks().Get(headerHash)
+	headerObject, ok := tpn.MetaDataPool.MetaBlocks().Get(headerHash)
 	if !ok {
 		return nil, errors.New(fmt.Sprintf("no header found for hash %s", hex.EncodeToString(headerHash)))
 	}

integrationTests/testProcessorNodeWithMultisigner.go

@@ -262,7 +262,7 @@ func VerifyNodesHaveHeaders(
 
 		for _, metaNode := range nodesMap[sharding.MetachainShardId] {
 			if shHeader == sharding.MetachainShardId {
-				v, ok = metaNode.MetaDataPool.MetaChainBlocks().Get(headerHash)
+				v, ok = metaNode.MetaDataPool.MetaBlocks().Get(headerHash)
 			} else {
 				v, ok = metaNode.MetaDataPool.ShardHeaders().Get(headerHash)
 			}

node/mock/metaPoolsHolderStub.go

@@ -6,7 +6,7 @@ import (
 )
 
 type MetaPoolsHolderStub struct {
-	MetaChainBlocksCalled      func() storage.Cacher
+	MetaBlocksCalled           func() storage.Cacher
 	MiniBlocksCalled           func() storage.Cacher
 	ShardHeadersCalled         func() storage.Cacher
 	HeadersNoncesCalled        func() dataRetriever.Uint64SyncMapCacher
@@ -22,8 +22,8 @@ func (mphs *MetaPoolsHolderStub) UnsignedTransactions() dataRetriever.ShardedDat
 	return mphs.UnsignedTransactionsCalled()
 }
 
-func (mphs *MetaPoolsHolderStub) MetaChainBlocks() storage.Cacher {
-	return mphs.MetaChainBlocksCalled()
+func (mphs *MetaPoolsHolderStub) MetaBlocks() storage.Cacher {
+	return mphs.MetaBlocksCalled()
 }
 
 func (mphs *MetaPoolsHolderStub) MiniBlocks() storage.Cacher {

process/block/baseProcess_test.go

@@ -107,6 +107,7 @@ func initDataPool(testHash []byte) *mock.PoolsHolderStub {
 				HasCalled: func(nonce uint64, shardId uint32) bool {
 					return true
 				},
+				RemoveCalled: func(nonce uint64, shardId uint32) {},
 			}
 		},
 		MetaBlocksCalled: func() storage.Cacher {
@@ -130,6 +131,7 @@ func initDataPool(testHash []byte) *mock.PoolsHolderStub {
 					return nil, false
 				},
 				RegisterHandlerCalled: func(i func(key []byte)) {},
+				RemoveCalled:          func(key []byte) {},
 			}
 		},
 		MiniBlocksCalled: func() storage.Cacher {
@@ -158,6 +160,8 @@ func initDataPool(testHash []byte) *mock.PoolsHolderStub {
 			cs := &mock.CacherStub{}
 			cs.RegisterHandlerCalled = func(i func(key []byte)) {
 			}
+			cs.RemoveCalled = func(key []byte) {
+			}
 			return cs
 		},
 	}
@@ -166,7 +170,7 @@ func initDataPool(testHash []byte) *mock.PoolsHolderStub {
 
 func initMetaDataPool() *mock.MetaPoolsHolderStub {
 	mdp := &mock.MetaPoolsHolderStub{
-		MetaChainBlocksCalled: func() storage.Cacher {
+		MetaBlocksCalled: func() storage.Cacher {
 			return &mock.CacherStub{
 				GetCalled: func(key []byte) (value interface{}, ok bool) {
 					if reflect.DeepEqual(key, []byte("tx1_hash")) {
@@ -187,6 +191,7 @@ func initMetaDataPool() *mock.MetaPoolsHolderStub {
 					return nil, false
 				},
 				RegisterHandlerCalled: func(i func(key []byte)) {},
+				RemoveCalled:          func(key []byte) {},
 			}
 		},
 		MiniBlocksCalled: func() storage.Cacher {

process/block/metablock.go

@@ -447,14 +447,18 @@ func (mp *metaProcessor) CommitBlock(
 
 	headerNoncePool := mp.dataPool.HeadersNonces()
 	if headerNoncePool == nil {
-		err = process.ErrNilDataPoolHolder
+		err = process.ErrNilHeadersNoncesDataPool
 		return err
 	}
 
-	//TODO: Should be analyzed if put in pool is really necessary or not (right now there is no action of removing them)
-	syncMap := &dataPool.ShardIdHashSyncMap{}
-	syncMap.Store(headerHandler.GetShardID(), headerHash)
-	headerNoncePool.Merge(headerHandler.GetNonce(), syncMap)
+	metaBlockPool := mp.dataPool.MetaBlocks()
+	if metaBlockPool == nil {
+		err = process.ErrNilMetaBlockPool
+		return err
+	}
+
+	headerNoncePool.Remove(header.GetNonce(), header.GetShardID())
+	metaBlockPool.Remove(headerHash)
 
 	body, ok := bodyHandler.(*block.MetaBlockBody)
 	if !ok {

process/block/metablock_test.go

@@ -606,7 +606,7 @@ func TestMetaProcessor_CommitBlockNilNoncesDataPoolShouldErr(t *testing.T) {
 	blkc := createTestBlockchain()
 	err := mp.CommitBlock(blkc, hdr, body)
 
-	assert.Equal(t, process.ErrNilDataPoolHolder, err)
+	assert.Equal(t, process.ErrNilHeadersNoncesDataPool, err)
 }
 
 func TestMetaProcessor_CommitBlockNoTxInPoolShouldErr(t *testing.T) {

process/block/shardblock.go

@@ -601,14 +601,18 @@ func (sp *shardProcessor) CommitBlock(
 
 	headerNoncePool := sp.dataPool.HeadersNonces()
 	if headerNoncePool == nil {
-		err = process.ErrNilDataPoolHolder
+		err = process.ErrNilHeadersNoncesDataPool
 		return err
 	}
 
-	//TODO: Should be analyzed if put in pool is really necessary or not (right now there is no action of removing them)
-	syncMap := &dataPool.ShardIdHashSyncMap{}
-	syncMap.Store(headerHandler.GetShardID(), headerHash)
-	headerNoncePool.Merge(headerHandler.GetNonce(), syncMap)
+	headersPool := sp.dataPool.Headers()
+	if headersPool == nil {
+		err = process.ErrNilHeadersDataPool
+		return err
+	}
+
+	headerNoncePool.Remove(header.GetNonce(), header.GetShardID())
+	headersPool.Remove(headerHash)
 
 	body, ok := bodyHandler.(block.Body)
 	if !ok {
@@ -1077,6 +1081,11 @@ func (sp *shardProcessor) receivedMetaBlock(metaBlockHash []byte) {
 		return
 	}
 
+	isMetaBlockOutOfRange := metaBlock.GetNonce() > lastNotarizedHdr.GetNonce()+process.MaxHeadersToRequestInAdvance
+	if isMetaBlockOutOfRange {
+		return
+	}
+
 	sp.txCoordinator.RequestMiniBlocks(metaBlock)
 }
 

process/block/shardblock_test.go

@@ -1723,7 +1723,7 @@ func TestShardProcessor_CommitBlockNilNoncesDataPoolShouldErr(t *testing.T) {
 	blkc := createTestBlockchain()
 	err := sp.CommitBlock(blkc, hdr, body)
 
-	assert.Equal(t, process.ErrNilDataPoolHolder, err)
+	assert.Equal(t, process.ErrNilHeadersNoncesDataPool, err)
 }
 
 func TestShardProcessor_CommitBlockNoTxInPoolShouldErr(t *testing.T) {

process/constants.go

@@ -40,3 +40,6 @@ const MaxNoncesDifference = 5
 // TODO - calculate exactly in case of the VM, for every VM to have a similar constant, operations / seconds
 const MaxGasLimitPerMiniBlock = uint64(100000)
 const MaxRequestsWithTimeoutAllowed = 3
+
+// MaxHeadersToRequestInAdvance defines the maximum number of headers which will be requested in advance if they are missing
+const MaxHeadersToRequestInAdvance = 10

process/factory/metachain/interceptorsContainerFactory.go

@@ -191,7 +191,7 @@ func (icf *interceptorsContainerFactory) generateMetablockInterceptor() ([]strin
 
 	interceptor, err := interceptors.NewMetachainHeaderInterceptor(
 		icf.marshalizer,
-		icf.dataPool.MetaChainBlocks(),
+		icf.dataPool.MetaBlocks(),
 		icf.dataPool.HeadersNonces(),
 		hdrValidator,
 		icf.multiSigner,

process/factory/metachain/interceptorsContainerFactory_test.go

@@ -54,7 +54,7 @@ func createDataPools() dataRetriever.MetaPoolsHolder {
 		MiniBlocksCalled: func() storage.Cacher {
 			return &mock.CacherStub{}
 		},
-		MetaChainBlocksCalled: func() storage.Cacher {
+		MetaBlocksCalled: func() storage.Cacher {
 			return &mock.CacherStub{}
 		},
 		HeadersNoncesCalled: func() dataRetriever.Uint64SyncMapCacher {