Feat/sync missing trie nodes

common/constants.go

@@ -770,9 +770,6 @@ const HardforkResolversIdentifier = "hardfork resolver"
 // EpochStartInterceptorsIdentifier represents the identifier that is used in the start-in-epoch process
 const EpochStartInterceptorsIdentifier = "epoch start interceptor"
 
-// GetNodeFromDBErrorString represents the string which is returned when a getting node from DB returns an error
-const GetNodeFromDBErrorString = "getNodeFromDB error"
-
 // TimeoutGettingTrieNodes defines the timeout in trie sync operation if no node is received
 const TimeoutGettingTrieNodes = 2 * time.Minute // to consider syncing a very large trie node of 64MB at ~1MB/s
 

common/errors.go

@@ -7,3 +7,6 @@ var ErrInvalidTimeout = errors.New("invalid timeout value")
 
 // ErrNilWasmChangeLocker signals that a nil wasm change locker has been provided
 var ErrNilWasmChangeLocker = errors.New("nil wasm change locker")
+
+// ErrNilStateSyncNotifierSubscriber signals that a nil state sync notifier subscriber has been provided
+var ErrNilStateSyncNotifierSubscriber = errors.New("nil state sync notifier subscriber")

common/interface.go

@@ -54,6 +54,12 @@ type TrieStats interface {
 	GetTrieStats(address string, rootHash []byte) (*statistics.TrieStatsDTO, error)
 }
 
+// StorageMarker is used to mark the given storer as synced and active
+type StorageMarker interface {
+	MarkStorerAsSyncedAndActive(storer StorageManager)
+	IsInterfaceNil() bool
+}
+
 // KeyBuilder is used for building trie keys as you traverse the trie
 type KeyBuilder interface {
 	BuildKey(keyPart []byte)
@@ -70,9 +76,8 @@ type DataTrieHandler interface {
 
 // StorageManager manages all trie storage operations
 type StorageManager interface {
-	Get(key []byte) ([]byte, error)
+	TrieStorageInteractor
 	GetFromCurrentEpoch(key []byte) ([]byte, error)
-	Put(key []byte, val []byte) error
 	PutInEpoch(key []byte, val []byte, epoch uint32) error
 	PutInEpochWithoutCache(key []byte, val []byte, epoch uint32) error
 	TakeSnapshot(address string, rootHash []byte, mainTrieRootHash []byte, iteratorChannels *TrieIteratorChannels, missingNodesChan chan []byte, stats SnapshotStatisticsHandler, epoch uint32)
@@ -83,7 +88,6 @@ type StorageManager interface {
 	EnterPruningBufferingMode()
 	ExitPruningBufferingMode()
 	AddDirtyCheckpointHashes([]byte, ModifiedHashes) bool
-	Remove(hash []byte) error
 	SetEpochForPutOperation(uint32)
 	ShouldTakeSnapshot() bool
 	GetBaseTrieStorageManager() StorageManager
@@ -92,8 +96,14 @@ type StorageManager interface {
 	IsInterfaceNil() bool
 }
 
-// DBWriteCacher is used to cache changes made to the trie, and only write to the database when it's needed
-type DBWriteCacher interface {
+// TrieStorageInteractor defines the methods used for interacting with the trie storage
+type TrieStorageInteractor interface {
+	BaseStorer
+	GetIdentifier() string
+}
+
+// BaseStorer define the base methods needed for a storer
+type BaseStorer interface {
 	Put(key, val []byte) error
 	Get(key []byte) ([]byte, error)
 	Remove(key []byte) error
@@ -103,7 +113,7 @@ type DBWriteCacher interface {
 
 // SnapshotDbHandler is used to keep track of how many references a snapshot db has
 type SnapshotDbHandler interface {
-	DBWriteCacher
+	BaseStorer
 	IsInUse() bool
 	DecreaseNumReferences()
 	IncreaseNumReferences()
@@ -364,3 +374,16 @@ type ManagedPeersHolder interface {
 	IsMultiKeyMode() bool
 	IsInterfaceNil() bool
 }
+
+// MissingTrieNodesNotifier defines the operations of an entity that notifies about missing trie nodes
+type MissingTrieNodesNotifier interface {
+	RegisterHandler(handler StateSyncNotifierSubscriber) error
+	AsyncNotifyMissingTrieNode(hash []byte)
+	IsInterfaceNil() bool
+}
+
+// StateSyncNotifierSubscriber defines the operations of an entity that subscribes to a missing trie nodes notifier
+type StateSyncNotifierSubscriber interface {
+	MissingDataTrieNodeFound(hash []byte)
+	IsInterfaceNil() bool
+}

common/logging/errorLogging.go

@@ -1,8 +1,8 @@
 package logging
 
 import (
+	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/core/check"
-	chainErrors "github.com/multiversx/mx-chain-go/errors"
 	logger "github.com/multiversx/mx-chain-logger-go"
 )
 
@@ -24,7 +24,7 @@ func logErrAsLevelExceptAsDebugIfClosingError(logInstance logger.Logger, logLeve
 		return
 	}
 
-	if chainErrors.IsClosingError(err) {
+	if core.IsClosingError(err) {
 		logLevel = logger.LogDebug
 	}
 

dataRetriever/factory/resolverscontainer/metaResolversContainerFactory.go

@@ -8,7 +8,6 @@ import (
 	"github.com/multiversx/mx-chain-go/dataRetriever"
 	"github.com/multiversx/mx-chain-go/dataRetriever/factory/containers"
 	"github.com/multiversx/mx-chain-go/dataRetriever/resolvers"
-	triesFactory "github.com/multiversx/mx-chain-go/trie/factory"
 
 	"github.com/multiversx/mx-chain-core-go/marshal"
 	"github.com/multiversx/mx-chain-go/process/factory"
@@ -142,7 +141,7 @@ func (mrcf *metaResolversContainerFactory) AddShardTrieNodeResolvers(container d
 		identifierTrieNodes := factory.AccountTrieNodesTopic + shardC.CommunicationIdentifier(idx)
 		resolver, err := mrcf.createTrieNodesResolver(
 			identifierTrieNodes,
-			triesFactory.UserAccountTrie,
+			dataRetriever.UserAccountsUnit.String(),
 			idx,
 		)
 		if err != nil {
@@ -295,7 +294,7 @@ func (mrcf *metaResolversContainerFactory) generateTrieNodesResolvers() error {
 	identifierTrieNodes := factory.AccountTrieNodesTopic + core.CommunicationIdentifierBetweenShards(core.MetachainShardId, core.MetachainShardId)
 	resolver, err := mrcf.createTrieNodesResolver(
 		identifierTrieNodes,
-		triesFactory.UserAccountTrie,
+		dataRetriever.UserAccountsUnit.String(),
 		core.MetachainShardId,
 	)
 	if err != nil {
@@ -308,7 +307,7 @@ func (mrcf *metaResolversContainerFactory) generateTrieNodesResolvers() error {
 	identifierTrieNodes = factory.ValidatorTrieNodesTopic + core.CommunicationIdentifierBetweenShards(core.MetachainShardId, core.MetachainShardId)
 	resolver, err = mrcf.createTrieNodesResolver(
 		identifierTrieNodes,
-		triesFactory.PeerAccountTrie,
+		dataRetriever.PeerAccountsUnit.String(),
 		core.MetachainShardId,
 	)
 	if err != nil {

dataRetriever/factory/resolverscontainer/metaResolversContainerFactory_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/multiversx/mx-chain-go/testscommon/p2pmocks"
 	storageStubs "github.com/multiversx/mx-chain-go/testscommon/storage"
 	trieMock "github.com/multiversx/mx-chain-go/testscommon/trie"
-	triesFactory "github.com/multiversx/mx-chain-go/trie/factory"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -82,8 +81,8 @@ func createStoreForMeta() dataRetriever.StorageService {
 
 func createTriesHolderForMeta() common.TriesHolder {
 	triesHolder := state.NewDataTriesHolder()
-	triesHolder.Put([]byte(triesFactory.UserAccountTrie), &trieMock.TrieStub{})
-	triesHolder.Put([]byte(triesFactory.PeerAccountTrie), &trieMock.TrieStub{})
+	triesHolder.Put([]byte(dataRetriever.UserAccountsUnit.String()), &trieMock.TrieStub{})
+	triesHolder.Put([]byte(dataRetriever.PeerAccountsUnit.String()), &trieMock.TrieStub{})
 	return triesHolder
 }
 

dataRetriever/factory/resolverscontainer/shardResolversContainerFactory.go

@@ -9,7 +9,6 @@ import (
 	"github.com/multiversx/mx-chain-go/dataRetriever/factory/containers"
 	"github.com/multiversx/mx-chain-go/dataRetriever/resolvers"
 	"github.com/multiversx/mx-chain-go/process/factory"
-	triesFactory "github.com/multiversx/mx-chain-go/trie/factory"
 )
 
 var _ dataRetriever.ResolversContainerFactory = (*shardResolversContainerFactory)(nil)
@@ -232,7 +231,7 @@ func (srcf *shardResolversContainerFactory) generateTrieNodesResolvers() error {
 	identifierTrieNodes := factory.AccountTrieNodesTopic + shardC.CommunicationIdentifier(core.MetachainShardId)
 	resolver, err := srcf.createTrieNodesResolver(
 		identifierTrieNodes,
-		triesFactory.UserAccountTrie,
+		dataRetriever.UserAccountsUnit.String(),
 		core.MetachainShardId,
 	)
 	if err != nil {

dataRetriever/factory/resolverscontainer/shardResolversContainerFactory_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/multiversx/mx-chain-go/testscommon/p2pmocks"
 	storageStubs "github.com/multiversx/mx-chain-go/testscommon/storage"
 	trieMock "github.com/multiversx/mx-chain-go/testscommon/trie"
-	triesFactory "github.com/multiversx/mx-chain-go/trie/factory"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -88,8 +87,8 @@ func createStoreForShard() dataRetriever.StorageService {
 
 func createTriesHolderForShard() common.TriesHolder {
 	triesHolder := state.NewDataTriesHolder()
-	triesHolder.Put([]byte(triesFactory.UserAccountTrie), &trieMock.TrieStub{})
-	triesHolder.Put([]byte(triesFactory.PeerAccountTrie), &trieMock.TrieStub{})
+	triesHolder.Put([]byte(dataRetriever.UserAccountsUnit.String()), &trieMock.TrieStub{})
+	triesHolder.Put([]byte(dataRetriever.PeerAccountsUnit.String()), &trieMock.TrieStub{})
 	return triesHolder
 }
 

dataRetriever/factory/storageRequestersContainer/baseRequestersContainerFactory.go

@@ -231,6 +231,7 @@ func (brcf *baseRequestersContainerFactory) createMiniBlocksRequester(responseTo
 func (brcf *baseRequestersContainerFactory) newImportDBTrieStorage(
 	mainStorer storage.Storer,
 	checkpointsStorer storage.Storer,
+	storageIdentifier dataRetriever.UnitType,
 ) (common.StorageManager, dataRetriever.TrieDataGetter, error) {
 	pathManager, err := storageFactory.CreatePathManager(
 		storageFactory.ArgCreatePathManager{
@@ -261,6 +262,7 @@ func (brcf *baseRequestersContainerFactory) newImportDBTrieStorage(
 		MaxTrieLevelInMem:  brcf.generalConfig.StateTriesConfig.MaxStateTrieLevelInMemory,
 		SnapshotsEnabled:   brcf.snapshotsEnabled,
 		IdleProvider:       disabled.NewProcessStatusHandler(),
+		Identifier:         storageIdentifier.String(),
 	}
 	return trieFactoryInstance.Create(args)
 }

dataRetriever/factory/storageRequestersContainer/metaRequestersContainerFactory.go

@@ -194,6 +194,7 @@ func (mrcf *metaRequestersContainerFactory) generateTrieNodesRequesters() error
 	storageManager, userAccountsDataTrie, err := mrcf.newImportDBTrieStorage(
 		userAccountsStorer,
 		userAccountsCheckpointStorer,
+		dataRetriever.UserAccountsUnit,
 	)
 	if err != nil {
 		return fmt.Errorf("%w while creating user accounts data trie storage getter", err)
@@ -230,6 +231,7 @@ func (mrcf *metaRequestersContainerFactory) generateTrieNodesRequesters() error
 	storageManager, peerAccountsDataTrie, err := mrcf.newImportDBTrieStorage(
 		peerAccountsStorer,
 		peerAccountsCheckpointStorer,
+		dataRetriever.PeerAccountsUnit,
 	)
 	if err != nil {
 		return fmt.Errorf("%w while creating peer accounts data trie storage getter", err)

dataRetriever/factory/storageRequestersContainer/shardRequestersContainerFactory.go

@@ -169,6 +169,7 @@ func (srcf *shardRequestersContainerFactory) generateTrieNodesRequesters() error
 	storageManager, userAccountsDataTrie, err := srcf.newImportDBTrieStorage(
 		userAccountsStorer,
 		userAccountsCheckpointStorer,
+		dataRetriever.UserAccountsUnit,
 	)
 	if err != nil {
 		return fmt.Errorf("%w while creating user accounts data trie storage getter", err)

dataRetriever/storageRequesters/trieNodeRequester_test.go

@@ -11,8 +11,8 @@ import (
 	"github.com/multiversx/mx-chain-core-go/data/endProcess"
 	"github.com/multiversx/mx-chain-go/dataRetriever"
 	"github.com/multiversx/mx-chain-go/dataRetriever/mock"
-	"github.com/multiversx/mx-chain-go/testscommon"
 	"github.com/multiversx/mx-chain-go/testscommon/p2pmocks"
+	"github.com/multiversx/mx-chain-go/testscommon/storageManager"
 	trieMock "github.com/multiversx/mx-chain-go/testscommon/trie"
 	"github.com/stretchr/testify/assert"
 )
@@ -23,7 +23,7 @@ func createMockTrieRequesterArguments() ArgTrieRequester {
 		ResponseTopicName:        "",
 		Marshalizer:              &mock.MarshalizerStub{},
 		TrieDataGetter:           &trieMock.TrieStub{},
-		TrieStorageManager:       &testscommon.StorageManagerStub{},
+		TrieStorageManager:       &storageManager.StorageManagerStub{},
 		ManualEpochStartNotifier: &mock.ManualEpochStartNotifierStub{},
 		ChanGracefullyClose:      make(chan endProcess.ArgEndProcess, 1),
 		DelayBeforeGracefulClose: 0,

epochStart/bootstrap/process.go

@@ -1066,7 +1066,7 @@ func (e *epochStartBootstrap) syncUserAccountsState(rootHash []byte) error {
 	}
 
 	e.mutTrieStorageManagers.RLock()
-	trieStorageManager := e.trieStorageManagers[factory.UserAccountTrie]
+	trieStorageManager := e.trieStorageManagers[dataRetriever.UserAccountsUnit.String()]
 	e.mutTrieStorageManagers.RUnlock()
 
 	argsUserAccountsSyncer := syncer.ArgsNewUserAccountsSyncer{
@@ -1081,7 +1081,6 @@ func (e *epochStartBootstrap) syncUserAccountsState(rootHash []byte) error {
 			MaxHardCapForMissingNodes:         e.maxHardCapForMissingNodes,
 			TrieSyncerVersion:                 e.trieSyncerVersion,
 			CheckNodesOnDisk:                  e.checkNodesOnDisk,
-			StorageMarker:                     storageMarker.NewTrieStorageMarker(),
 			UserAccountsSyncStatisticsHandler: e.trieSyncStatisticsProvider,
 			AppStatusHandler:                  e.statusHandler,
 		},
@@ -1094,7 +1093,7 @@ func (e *epochStartBootstrap) syncUserAccountsState(rootHash []byte) error {
 		return err
 	}
 
-	err = accountsDBSyncer.SyncAccounts(rootHash)
+	err = accountsDBSyncer.SyncAccounts(rootHash, storageMarker.NewTrieStorageMarker())
 	if err != nil {
 		return err
 	}
@@ -1139,7 +1138,7 @@ func (e *epochStartBootstrap) createStorageService(
 
 func (e *epochStartBootstrap) syncValidatorAccountsState(rootHash []byte) error {
 	e.mutTrieStorageManagers.RLock()
-	peerTrieStorageManager := e.trieStorageManagers[factory.PeerAccountTrie]
+	peerTrieStorageManager := e.trieStorageManagers[dataRetriever.PeerAccountsUnit.String()]
 	e.mutTrieStorageManagers.RUnlock()
 
 	argsValidatorAccountsSyncer := syncer.ArgsNewValidatorAccountsSyncer{
@@ -1154,7 +1153,6 @@ func (e *epochStartBootstrap) syncValidatorAccountsState(rootHash []byte) error
 			MaxHardCapForMissingNodes:         e.maxHardCapForMissingNodes,
 			TrieSyncerVersion:                 e.trieSyncerVersion,
 			CheckNodesOnDisk:                  e.checkNodesOnDisk,
-			StorageMarker:                     storageMarker.NewTrieStorageMarker(),
 			UserAccountsSyncStatisticsHandler: statistics.NewTrieSyncStatistics(),
 			AppStatusHandler:                  disabledCommon.NewAppStatusHandler(),
 		},
@@ -1164,7 +1162,7 @@ func (e *epochStartBootstrap) syncValidatorAccountsState(rootHash []byte) error
 		return err
 	}
 
-	err = accountsDBSyncer.SyncAccounts(rootHash)
+	err = accountsDBSyncer.SyncAccounts(rootHash, storageMarker.NewTrieStorageMarker())
 	if err != nil {
 		return err
 	}

epochStart/interface.go

@@ -115,12 +115,6 @@ type PendingMiniBlocksSyncHandler interface {
 	IsInterfaceNil() bool
 }
 
-// AccountsDBSyncer defines the methods for the accounts db syncer
-type AccountsDBSyncer interface {
-	SyncAccounts(rootHash []byte) error
-	IsInterfaceNil() bool
-}
-
 // StartOfEpochMetaSyncer defines the methods to synchronize epoch start meta block from the network when nothing is known
 type StartOfEpochMetaSyncer interface {
 	SyncEpochStartMeta(waitTime time.Duration) (data.MetaHeaderHandler, error)

epochStart/metachain/baseRewards_test.go

@@ -1136,11 +1136,11 @@ func getBaseRewardsArguments() BaseRewardsCreatorArgs {
 	hasher := sha256.NewSha256()
 	marshalizer := &marshal.GogoProtoMarshalizer{}
 
-	storageManagerArgs, options := storage.GetStorageManagerArgsAndOptions()
+	storageManagerArgs := storage.GetStorageManagerArgs()
 	storageManagerArgs.Marshalizer = marshalizer
 	storageManagerArgs.Hasher = hasher
 
-	trieFactoryManager, _ := trie.CreateTrieStorageManager(storageManagerArgs, options)
+	trieFactoryManager, _ := trie.CreateTrieStorageManager(storageManagerArgs, storage.GetStorageManagerOptions())
 	userAccountsDB := createAccountsDB(hasher, marshalizer, factory.NewAccountCreator(), trieFactoryManager)
 	shardCoordinator := mock.NewMultiShardsCoordinatorMock(2)
 	shardCoordinator.CurrentShard = core.MetachainShardId

epochStart/metachain/systemSCs_test.go

@@ -910,13 +910,13 @@ func createAccountsDB(
 func createFullArgumentsForSystemSCProcessing(enableEpochsConfig config.EnableEpochs, trieStorer storage.Storer) (ArgsNewEpochStartSystemSCProcessing, vm.SystemSCContainer) {
 	hasher := sha256.NewSha256()
 	marshalizer := &marshal.GogoProtoMarshalizer{}
-	storageManagerArgs, options := stateMock.GetStorageManagerArgsAndOptions()
+	storageManagerArgs := stateMock.GetStorageManagerArgs()
 	storageManagerArgs.Marshalizer = marshalizer
 	storageManagerArgs.Hasher = hasher
 	storageManagerArgs.MainStorer = trieStorer
 	storageManagerArgs.CheckpointsStorer = trieStorer
 
-	trieFactoryManager, _ := trie.CreateTrieStorageManager(storageManagerArgs, options)
+	trieFactoryManager, _ := trie.CreateTrieStorageManager(storageManagerArgs, stateMock.GetStorageManagerOptions())
 	userAccountsDB := createAccountsDB(hasher, marshalizer, factory.NewAccountCreator(), trieFactoryManager)
 	peerAccountsDB := createAccountsDB(hasher, marshalizer, factory.NewPeerAccountCreator(), trieFactoryManager)
 	en := forking.NewGenericEpochNotifier()
@@ -953,23 +953,24 @@ func createFullArgumentsForSystemSCProcessing(enableEpochsConfig config.EnableEp
 	nodesSetup := &mock.NodesSetupStub{}
 
 	argsHook := hooks.ArgBlockChainHook{
-		Accounts:              userAccountsDB,
-		PubkeyConv:            &testscommon.PubkeyConverterMock{},
-		StorageService:        &storageStubs.ChainStorerStub{},
-		BlockChain:            blockChain,
-		ShardCoordinator:      &mock.ShardCoordinatorStub{},
-		Marshalizer:           marshalizer,
-		Uint64Converter:       &mock.Uint64ByteSliceConverterMock{},
-		BuiltInFunctions:      vmcommonBuiltInFunctions.NewBuiltInFunctionContainer(),
-		NFTStorageHandler:     &testscommon.SimpleNFTStorageHandlerStub{},
-		GlobalSettingsHandler: &testscommon.ESDTGlobalSettingsHandlerStub{},
-		DataPool:              testDataPool,
-		CompiledSCPool:        testDataPool.SmartContracts(),
-		EpochNotifier:         en,
-		EnableEpochsHandler:   enableEpochsHandler,
-		NilCompiledSCStore:    true,
-		GasSchedule:           gasScheduleNotifier,
-		Counter:               &testscommon.BlockChainHookCounterStub{},
+		Accounts:                 userAccountsDB,
+		PubkeyConv:               &testscommon.PubkeyConverterMock{},
+		StorageService:           &storageStubs.ChainStorerStub{},
+		BlockChain:               blockChain,
+		ShardCoordinator:         &mock.ShardCoordinatorStub{},
+		Marshalizer:              marshalizer,
+		Uint64Converter:          &mock.Uint64ByteSliceConverterMock{},
+		BuiltInFunctions:         vmcommonBuiltInFunctions.NewBuiltInFunctionContainer(),
+		NFTStorageHandler:        &testscommon.SimpleNFTStorageHandlerStub{},
+		GlobalSettingsHandler:    &testscommon.ESDTGlobalSettingsHandlerStub{},
+		DataPool:                 testDataPool,
+		CompiledSCPool:           testDataPool.SmartContracts(),
+		EpochNotifier:            en,
+		EnableEpochsHandler:      enableEpochsHandler,
+		NilCompiledSCStore:       true,
+		GasSchedule:              gasScheduleNotifier,
+		Counter:                  &testscommon.BlockChainHookCounterStub{},
+		MissingTrieNodesNotifier: &testscommon.MissingTrieNodesNotifierStub{},
 	}
 
 	blockChainHookImpl, _ := hooks.NewBlockChainHookImpl(argsHook)