sync txs and miniblocks at bootstrap

process/sync/baseSync.go

@@ -23,7 +23,11 @@ import (
 	"github.com/multiversx/mx-chain-core-go/marshal"
 	logger "github.com/multiversx/mx-chain-logger-go"
 
+	"github.com/multiversx/mx-chain-go/epochStart"
+	"github.com/multiversx/mx-chain-go/epochStart/bootstrap/disabled"
 	"github.com/multiversx/mx-chain-go/process/asyncExecution/queue"
+	"github.com/multiversx/mx-chain-go/update"
+	updateSync "github.com/multiversx/mx-chain-go/update/sync"
 
 	"github.com/multiversx/mx-chain-go/common"
 	"github.com/multiversx/mx-chain-go/consensus"
@@ -50,6 +54,7 @@ var _ closing.Closer = (*baseBootstrap)(nil)
 // sleepTime defines the time in milliseconds between each iteration made in syncBlocks method
 const sleepTime = 50 * time.Millisecond
 const minimumProcessWaitTime = time.Millisecond * 100
+const defaultTimeToWaitForRequestedData = 5 * time.Minute
 
 // hdrInfo hold the data related to a header
 type hdrInfo struct {
@@ -146,6 +151,9 @@ type baseBootstrap struct {
 	preparedForSyncAtBootstrap   bool
 
 	repopulateTokensSupplies bool
+
+	miniBlocksSyncer epochStart.PendingMiniBlocksSyncHandler
+	txSyncer         update.TransactionsSyncHandler
 }
 
 func (boot *baseBootstrap) getProcessWaitTime(round uint64) time.Duration {
@@ -818,7 +826,7 @@ func (boot *baseBootstrap) prepareForSyncAtBoostrapIfNeeded() error {
 		"currHeader nonce", currentHeader.GetNonce(),
 	)
 
-	err := boot.prepareForSyncIfNeeded(syncingNonce)
+	err := boot.prepareForSyncIfNeeded(syncingNonce, true)
 	if err != nil {
 		return err
 	}
@@ -990,7 +998,7 @@ func (boot *baseBootstrap) prepareForLegacySyncIfNeeded() error {
 // Finally, if everything works, the block will be committed and added into the processing queue.
 // And all this mechanism will be reiterated for the next block.
 func (boot *baseBootstrap) syncBlockV3(body data.BodyHandler, header data.HeaderHandler) error {
-	err := boot.prepareForSyncIfNeeded(header.GetNonce())
+	err := boot.prepareForSyncIfNeeded(header.GetNonce(), false)
 	if err != nil {
 		return err
 	}
@@ -1045,7 +1053,38 @@ func (boot *baseBootstrap) syncBlockV3(body data.BodyHandler, header data.Header
 	return nil
 }
 
-func (boot *baseBootstrap) prepareForSyncIfNeeded(syncingNonce uint64) error {
+func (boot *baseBootstrap) syncMiniBlocksAndTxsForHeader(
+	header data.HeaderHandler,
+) error {
+	boot.miniBlocksSyncer.ClearFields()
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeToWaitForRequestedData)
+	err := boot.miniBlocksSyncer.SyncPendingMiniBlocks(header.GetMiniBlockHeaderHandlers(), ctx)
+	cancel()
+	if err != nil {
+		return err
+	}
+
+	miniBlocks, err := boot.miniBlocksSyncer.GetMiniBlocks()
+	if err != nil {
+		return err
+	}
+
+	// sync all txs into pools
+
+	ctx, cancel = context.WithTimeout(context.Background(), defaultTimeToWaitForRequestedData)
+	err = boot.txSyncer.SyncTransactionsFor(miniBlocks, header.GetEpoch(), ctx)
+	cancel()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (boot *baseBootstrap) prepareForSyncIfNeeded(
+	syncingNonce uint64,
+	withTxs bool,
+) error {
 	if boot.preparedForSync {
 		return nil
 	}
@@ -1067,6 +1106,13 @@ func (boot *baseBootstrap) prepareForSyncIfNeeded(syncingNonce uint64) error {
 			return errGetBody
 		}
 
+		if withTxs {
+			err = boot.syncMiniBlocksAndTxsForHeader(currentHeader)
+			if err != nil {
+				return err
+			}
+		}
+
 		err = boot.saveProposedTxsToPool(currentHeader, currentBody)
 		if err != nil {
 			return err
@@ -1103,6 +1149,13 @@ func (boot *baseBootstrap) prepareForSyncIfNeeded(syncingNonce uint64) error {
 			return errGetBody
 		}
 
+		if withTxs {
+			err = boot.syncMiniBlocksAndTxsForHeader(hdr)
+			if err != nil {
+				return err
+			}
+		}
+
 		err = boot.saveProposedTxsToPool(hdr, body)
 		if err != nil {
 			return err
@@ -2209,3 +2262,31 @@ func (boot *baseBootstrap) getHeaderMiniBlocks(
 func (boot *baseBootstrap) IsInterfaceNil() bool {
 	return boot == nil
 }
+
+func (boot *baseBootstrap) createTxSyncer() error {
+	var err error
+
+	syncMiniBlocksArgs := updateSync.ArgsNewPendingMiniBlocksSyncer{
+		Storage:        disabled.CreateMemUnit(),
+		Cache:          boot.dataPool.MiniBlocks(),
+		Marshalizer:    boot.marshalizer,
+		RequestHandler: boot.requestHandler,
+	}
+	boot.miniBlocksSyncer, err = updateSync.NewPendingMiniBlocksSyncer(syncMiniBlocksArgs)
+	if err != nil {
+		return err
+	}
+
+	syncTxsArgs := updateSync.ArgsNewTransactionsSyncer{
+		DataPools:      boot.dataPool,
+		Storages:       boot.store,
+		Marshaller:     boot.marshalizer,
+		RequestHandler: boot.requestHandler,
+	}
+	boot.txSyncer, err = updateSync.NewTransactionsSyncer(syncTxsArgs)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

process/sync/metablock.go

@@ -115,6 +115,11 @@ func NewMetaBootstrap(arguments ArgMetaBootstrapper) (*MetaBootstrap, error) {
 		return nil, err
 	}
 
+	err = base.createTxSyncer()
+	if err != nil {
+		return nil, err
+	}
+
 	base.init()
 
 	return &boot, nil

process/sync/shardblock.go

@@ -101,6 +101,11 @@ func NewShardBootstrap(arguments ArgShardBootstrapper) (*ShardBootstrap, error)
 		return nil, err
 	}
 
+	err = base.createTxSyncer()
+	if err != nil {
+		return nil, err
+	}
+
 	base.init()
 
 	return &boot, nil

process/sync/shardblock_test.go

@@ -222,6 +222,7 @@ func createFullStore() dataRetriever.StorageService {
 	store.AddStorer(dataRetriever.ScheduledSCRsUnit, generateTestUnit())
 	store.AddStorer(dataRetriever.UserAccountsUnit, generateTestUnit())
 	store.AddStorer(dataRetriever.PeerAccountsUnit, generateTestUnit())
+	store.AddStorer(dataRetriever.UnsignedTransactionUnit, generateTestUnit())
 	return store
 }
 
@@ -678,7 +679,7 @@ func TestNewShardBootstrap_OkValsShouldWork(t *testing.T) {
 
 	assert.False(t, check.IfNil(bs))
 	assert.Nil(t, err)
-	assert.Equal(t, 2, wasCalled)
+	assert.Equal(t, 3, wasCalled)
 	assert.False(t, bs.IsInterfaceNil())
 	assert.True(t, bs.IsInImportMode())
 
@@ -1937,7 +1938,8 @@ func TestBootstrap_GetTxBodyHavingHashFoundInStorageShouldWork(t *testing.T) {
 		},
 	}
 
-	bs, _ := sync.NewShardBootstrap(args)
+	bs, err := sync.NewShardBootstrap(args)
+	require.Nil(t, err)
 	gotMbsAndHashes, _ := bs.GetMiniBlocks(requestedHash)
 
 	assert.Equal(t, mbsAndHashes, gotMbsAndHashes)