prefer headers with proofs

dataRetriever/factory/resolverscontainer/metaResolversContainerFactory.go

@@ -4,13 +4,15 @@ 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/core/throttler"
+
 	"github.com/multiversx/mx-chain-go/common"
 	"github.com/multiversx/mx-chain-go/dataRetriever"
 	"github.com/multiversx/mx-chain-go/dataRetriever/factory/containers"
 	"github.com/multiversx/mx-chain-go/dataRetriever/resolvers"
 	"github.com/multiversx/mx-chain-go/process"
 
 	"github.com/multiversx/mx-chain-core-go/marshal"
+
 	"github.com/multiversx/mx-chain-go/process/factory"
 )
 
@@ -229,6 +231,7 @@ func (mrcf *metaResolversContainerFactory) createShardHeaderResolver(
 			Throttler:        mrcf.throttler,
 		},
 		Headers:              mrcf.dataPools.Headers(),
+		Proofs:               mrcf.dataPools.Proofs(),
 		HdrStorage:           hdrStorer,
 		HeadersNoncesStorage: hdrNonceStore,
 		NonceConverter:       mrcf.uint64ByteSliceConverter,
@@ -292,6 +295,7 @@ func (mrcf *metaResolversContainerFactory) createMetaChainHeaderResolver(
 			Throttler:        mrcf.throttler,
 		},
 		Headers:              mrcf.dataPools.Headers(),
+		Proofs:               mrcf.dataPools.Proofs(),
 		HdrStorage:           hdrStorer,
 		HeadersNoncesStorage: hdrNonceStore,
 		NonceConverter:       mrcf.uint64ByteSliceConverter,

dataRetriever/factory/resolverscontainer/shardResolversContainerFactory.go

@@ -5,6 +5,7 @@ import (
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/core/throttler"
 	"github.com/multiversx/mx-chain-core-go/marshal"
+
 	"github.com/multiversx/mx-chain-go/common"
 	"github.com/multiversx/mx-chain-go/dataRetriever"
 	"github.com/multiversx/mx-chain-go/dataRetriever/factory/containers"
@@ -180,6 +181,7 @@ func (srcf *shardResolversContainerFactory) generateHeaderResolvers() error {
 			Throttler:        srcf.throttler,
 		},
 		Headers:              srcf.dataPools.Headers(),
+		Proofs:               srcf.dataPools.Proofs(),
 		HdrStorage:           hdrStorer,
 		HeadersNoncesStorage: hdrNonceStore,
 		NonceConverter:       srcf.uint64ByteSliceConverter,
@@ -233,6 +235,7 @@ func (srcf *shardResolversContainerFactory) generateMetablockHeaderResolvers() e
 			Throttler:        srcf.throttler,
 		},
 		Headers:              srcf.dataPools.Headers(),
+		Proofs:               srcf.dataPools.Proofs(),
 		HdrStorage:           hdrStorer,
 		HeadersNoncesStorage: hdrNonceStore,
 		NonceConverter:       srcf.uint64ByteSliceConverter,

dataRetriever/resolvers/headerResolver.go

@@ -1,12 +1,14 @@
 package resolvers
 
 import (
+	"bytes"
 	"fmt"
 	"runtime/debug"
 	"sync"
 
 	"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-core-go/data/typeConverters"
 	logger "github.com/multiversx/mx-chain-logger-go"
 
@@ -25,6 +27,7 @@ var _ dataRetriever.HeaderResolver = (*HeaderResolver)(nil)
 type ArgHeaderResolver struct {
 	ArgBaseResolver
 	Headers              dataRetriever.HeadersPool
+	Proofs               dataRetriever.ProofsPool
 	HdrStorage           storage.Storer
 	HeadersNoncesStorage storage.Storer
 	NonceConverter       typeConverters.Uint64ByteSliceConverter
@@ -38,6 +41,7 @@ type HeaderResolver struct {
 	baseStorageResolver
 	messageProcessor
 	headers          dataRetriever.HeadersPool
+	proofs           dataRetriever.ProofsPool
 	hdrNoncesStorage storage.Storer
 	nonceConverter   typeConverters.Uint64ByteSliceConverter
 	mutEpochHandler  sync.RWMutex
@@ -58,6 +62,7 @@ func NewHeaderResolver(arg ArgHeaderResolver) (*HeaderResolver, error) {
 			TopicResolverSender: arg.SenderResolver,
 		},
 		headers:             arg.Headers,
+		proofs:              arg.Proofs,
 		baseStorageResolver: createBaseStorageResolver(arg.HdrStorage, arg.IsFullHistoryNode),
 		hdrNoncesStorage:    arg.HeadersNoncesStorage,
 		nonceConverter:      arg.NonceConverter,
@@ -82,6 +87,9 @@ func checkArgHeaderResolver(arg ArgHeaderResolver) error {
 	if check.IfNil(arg.Headers) {
 		return dataRetriever.ErrNilHeadersDataPool
 	}
+	if check.IfNil(arg.Proofs) {
+		return dataRetriever.ErrNilProofsPool
+	}
 	if check.IfNil(arg.HdrStorage) {
 		return dataRetriever.ErrNilHeadersStorage
 	}
@@ -207,12 +215,13 @@ func (hdrRes *HeaderResolver) resolveHeaderFromNonce(rd *dataRetriever.RequestDa
 }
 
 func (hdrRes *HeaderResolver) searchInCache(nonce uint64) ([]byte, error) {
-	headers, _, err := hdrRes.headers.GetHeadersByNonceAndShardId(nonce, hdrRes.TargetShardID())
+	headers, hashes, err := hdrRes.headers.GetHeadersByNonceAndShardId(nonce, hdrRes.TargetShardID())
 	if err != nil {
 		return nil, err
 	}
 
-	hdr := headers[len(headers)-1]
+	// serve the proven header, not a more recently received non-final fork at the same nonce
+	hdr := hdrRes.selectProvenOrLastHeader(nonce, headers, hashes)
 	buff, err := hdrRes.marshalizer.Marshal(hdr)
 	if err != nil {
 		return nil, err
@@ -221,6 +230,27 @@ func (hdrRes *HeaderResolver) searchInCache(nonce uint64) ([]byte, error) {
 	return buff, nil
 }
 
+func (hdrRes *HeaderResolver) selectProvenOrLastHeader(
+	nonce uint64,
+	headers []data.HeaderHandler,
+	hashes [][]byte,
+) data.HeaderHandler {
+	lastHeader := headers[len(headers)-1]
+
+	proof, err := hdrRes.proofs.GetProofByNonce(nonce, hdrRes.TargetShardID())
+	if err != nil {
+		return lastHeader
+	}
+
+	for i, hash := range hashes {
+		if bytes.Equal(hash, proof.GetHeaderHash()) {
+			return headers[i]
+		}
+	}
+
+	return lastHeader
+}
+
 // resolveHeaderFromHash resolves a header using its key (header hash)
 func (hdrRes *HeaderResolver) resolveHeaderFromHash(rd *dataRetriever.RequestData) ([]byte, error) {
 	value, err := hdrRes.headers.GetHeaderByHash(rd.Value)

dataRetriever/resolvers/headerResolver_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/multiversx/mx-chain-go/dataRetriever/mock"
 	"github.com/multiversx/mx-chain-go/dataRetriever/resolvers"
 	"github.com/multiversx/mx-chain-go/p2p"
+	dataRetrieverMocks "github.com/multiversx/mx-chain-go/testscommon/dataRetriever"
 	"github.com/multiversx/mx-chain-go/testscommon/p2pmocks"
 	storageStubs "github.com/multiversx/mx-chain-go/testscommon/storage"
 )
@@ -35,6 +36,7 @@ func createMockArgHeaderResolver() resolvers.ArgHeaderResolver {
 	return resolvers.ArgHeaderResolver{
 		ArgBaseResolver:      createMockArgBaseResolver(),
 		Headers:              &mock.HeadersCacherStub{},
+		Proofs:               &dataRetrieverMocks.ProofsPoolMock{},
 		HdrStorage:           &storageStubs.StorerStub{},
 		HeadersNoncesStorage: &storageStubs.StorerStub{},
 		NonceConverter:       mock.NewNonceHashConverterMock(),
@@ -66,6 +68,17 @@ func TestNewHeaderResolver_NilHeadersPoolShouldErr(t *testing.T) {
 	assert.Nil(t, hdrRes)
 }
 
+func TestNewHeaderResolver_NilProofsPoolShouldErr(t *testing.T) {
+	t.Parallel()
+
+	arg := createMockArgHeaderResolver()
+	arg.Proofs = nil
+	hdrRes, err := resolvers.NewHeaderResolver(arg)
+
+	assert.Equal(t, dataRetriever.ErrNilProofsPool, err)
+	assert.Nil(t, hdrRes)
+}
+
 func TestNewHeaderResolver_NilHeadersStorageShouldErr(t *testing.T) {
 	t.Parallel()
 
@@ -724,6 +737,99 @@ func TestHeaderResolver_ProcessReceivedMessageRequestNonceTypeNotFoundInHdrNonce
 	assert.Len(t, msgID, 0)
 }
 
+func TestHeaderResolver_ProcessReceivedMessageRequestNonceTypePrefersProvenHeaderFromPool(t *testing.T) {
+	t.Parallel()
+
+	requestedNonce := uint64(67)
+	hashX := []byte("hashX-proven")
+	hashY := []byte("hashY-fork")
+	headerX := &block.Header{Nonce: requestedNonce, Round: 1}
+	headerY := &block.Header{Nonce: requestedNonce, Round: 2}
+
+	headers := &mock.HeadersCacherStub{}
+	headers.GetHeaderByNonceAndShardIdCalled = func(hdrNonce uint64, shardId uint32) ([]data.HeaderHandler, [][]byte, error) {
+		// Y is the last-seen (more recently received) header for the nonce
+		return []data.HeaderHandler{headerX, headerY}, [][]byte{hashX, hashY}, nil
+	}
+
+	arg := createMockArgHeaderResolver()
+	var sentBuff []byte
+	arg.SenderResolver = &mock.TopicResolverSenderStub{
+		SendCalled: func(buff []byte, peer core.PeerID, source p2p.MessageHandler) error {
+			sentBuff = buff
+			return nil
+		},
+	}
+	arg.Headers = headers
+	arg.HeadersNoncesStorage = &storageStubs.StorerStub{
+		SearchFirstCalled: func(key []byte) ([]byte, error) {
+			return nil, errKeyNotFound // not committed yet -> serve from cache
+		},
+	}
+	arg.Proofs = &dataRetrieverMocks.ProofsPoolMock{
+		GetProofByNonceCalled: func(headerNonce uint64, shardID uint32) (data.HeaderProofHandler, error) {
+			return &block.HeaderProof{HeaderHash: hashX, HeaderNonce: headerNonce}, nil
+		},
+	}
+	hdrRes, _ := resolvers.NewHeaderResolver(arg)
+
+	_, err := hdrRes.ProcessReceivedMessage(
+		createRequestMsg(dataRetriever.NonceType, arg.NonceConverter.ToByteSlice(requestedNonce)),
+		fromConnectedPeerId,
+		&p2pmocks.MessengerStub{},
+	)
+
+	assert.Nil(t, err)
+	expectedBuff, _ := arg.Marshaller.Marshal(headerX)
+	assert.Equal(t, expectedBuff, sentBuff)
+}
+
+func TestHeaderResolver_ProcessReceivedMessageRequestNonceTypeNoProofServesLastSeenFromPool(t *testing.T) {
+	t.Parallel()
+
+	requestedNonce := uint64(67)
+	hashX := []byte("hashX")
+	hashY := []byte("hashY")
+	headerX := &block.Header{Nonce: requestedNonce, Round: 1}
+	headerY := &block.Header{Nonce: requestedNonce, Round: 2}
+
+	headers := &mock.HeadersCacherStub{}
+	headers.GetHeaderByNonceAndShardIdCalled = func(hdrNonce uint64, shardId uint32) ([]data.HeaderHandler, [][]byte, error) {
+		return []data.HeaderHandler{headerX, headerY}, [][]byte{hashX, hashY}, nil
+	}
+
+	arg := createMockArgHeaderResolver()
+	var sentBuff []byte
+	arg.SenderResolver = &mock.TopicResolverSenderStub{
+		SendCalled: func(buff []byte, peer core.PeerID, source p2p.MessageHandler) error {
+			sentBuff = buff
+			return nil
+		},
+	}
+	arg.Headers = headers
+	arg.HeadersNoncesStorage = &storageStubs.StorerStub{
+		SearchFirstCalled: func(key []byte) ([]byte, error) {
+			return nil, errKeyNotFound
+		},
+	}
+	arg.Proofs = &dataRetrieverMocks.ProofsPoolMock{
+		GetProofByNonceCalled: func(headerNonce uint64, shardID uint32) (data.HeaderProofHandler, error) {
+			return nil, errKeyNotFound // no proof known -> fall back to last-seen
+		},
+	}
+	hdrRes, _ := resolvers.NewHeaderResolver(arg)
+
+	_, err := hdrRes.ProcessReceivedMessage(
+		createRequestMsg(dataRetriever.NonceType, arg.NonceConverter.ToByteSlice(requestedNonce)),
+		fromConnectedPeerId,
+		&p2pmocks.MessengerStub{},
+	)
+
+	assert.Nil(t, err)
+	expectedBuff, _ := arg.Marshaller.Marshal(headerY)
+	assert.Equal(t, expectedBuff, sentBuff)
+}
+
 func TestHeaderResolver_ProcessReceivedMessageRequestNonceTypeFoundInHdrNoncePoolCheckRetErr(t *testing.T) {
 	t.Parallel()
 

process/sync/baseSync.go

@@ -804,7 +804,8 @@ func (boot *baseBootstrap) doJobOnSyncBlockFail(bodyHandler data.BodyHandler, he
 	shouldAllowRollback := boot.shouldAllowRollback(lastCommittedBlock, lastCommittedBlockHash)
 
 	shouldRollBack := isProcessWithError || isSyncWithErrorsLimitReachedInProperRound
-	if shouldRollBack && shouldAllowRollback {
+	didRollBack := shouldRollBack && shouldAllowRollback
+	if didRollBack {
 		if !check.IfNil(headerHandler) {
 			hash := boot.removeHeaderFromPools(headerHandler)
 			boot.forkDetector.RemoveHeader(headerHandler.GetNonce(), hash)
@@ -820,6 +821,38 @@ func (boot *baseBootstrap) doJobOnSyncBlockFail(bodyHandler data.BodyHandler, he
 			boot.removeHeadersHigherThanNonceFromPool(boot.getNonceForCurrentBlock())
 		}
 	}
+
+	// stuck fetching the next header (no processing, no rollback): drop a non-final fork header whose proof
+	// will never arrive so it gets re-requested
+	if check.IfNil(headerHandler) && !didRollBack {
+		boot.removeBlockingUnprovenNextHeader(allowedSyncWithErrorsLimitReached)
+	}
+}
+
+func (boot *baseBootstrap) removeBlockingUnprovenNextHeader(limitReached bool) {
+	if !limitReached {
+		return
+	}
+
+	nonce := boot.getNonceForNextBlock()
+	hdr, hash, err := boot.getHeaderFromPoolWithNonce(nonce)
+	if err != nil {
+		return
+	}
+	if boot.hasProof(hash, hdr) {
+		return
+	}
+
+	log.Debug("removeBlockingUnprovenNextHeader: removing unproven header blocking sync, will re-request",
+		"shard", hdr.GetShardID(),
+		"nonce", nonce,
+		"hash", hash,
+	)
+
+	boot.headers.RemoveHeaderByHash(hash)
+	boot.forkDetector.RemoveHeader(nonce, hash)
+	// reset so the re-requested header gets a full window before it could be removed in turn
+	boot.resetSyncedWithErrorsForNonce(nonce)
 }
 
 func (boot *baseBootstrap) incrementSyncedWithErrorsForNonce(nonce uint64) uint32 {
@@ -831,6 +864,12 @@ func (boot *baseBootstrap) incrementSyncedWithErrorsForNonce(nonce uint64) uint3
 	return numSyncedWithErrors
 }
 
+func (boot *baseBootstrap) resetSyncedWithErrorsForNonce(nonce uint64) {
+	boot.mutNonceSyncedWithErrors.Lock()
+	delete(boot.mapNonceSyncedWithErrors, nonce)
+	boot.mutNonceSyncedWithErrors.Unlock()
+}
+
 func (boot *baseBootstrap) prepareForSyncAtBoostrapIfNeeded() error {
 	// this will be triggered only once, after a full node restart.
 	// it is needed for the case when the node will go through bootstrap process and start

process/sync/metablock.go

@@ -226,7 +226,7 @@ func (boot *MetaBootstrap) requestEpochStartBlockIfStuck() {
 
 	targetNonce := currentNonce + 1
 
-	header, headerHash, err := process.GetMetaHeaderFromPoolWithNonce(targetNonce, boot.headers)
+	header, headerHash, err := boot.getMetaHeaderForNoncePreferProven(targetNonce)
 	if err == nil && !check.IfNil(header) {
 		if boot.proofs.HasProof(core.MetachainShardId, headerHash) {
 			return
@@ -251,6 +251,18 @@ func (boot *MetaBootstrap) requestEpochStartBlockIfStuck() {
 	boot.requestHandler.RequestEquivalentProofByNonce(core.MetachainShardId, targetNonce)
 }
 
+func (boot *MetaBootstrap) getMetaHeaderForNoncePreferProven(nonce uint64) (data.HeaderHandler, []byte, error) {
+	proof, err := boot.proofs.GetProofByNonce(nonce, core.MetachainShardId)
+	if err == nil {
+		provenHeader, errHdr := boot.headers.GetHeaderByHash(proof.GetHeaderHash())
+		if errHdr == nil {
+			return provenHeader, proof.GetHeaderHash(), nil
+		}
+	}
+
+	return process.GetMetaHeaderFromPoolWithNonce(nonce, boot.headers)
+}
+
 func (boot *MetaBootstrap) setLastEpochStartRound() {
 	hdr := boot.chainHandler.GetCurrentBlockHeader()
 	if check.IfNil(hdr) || hdr.GetEpoch() < 1 {