complete multistore

store/rootmulti/store.go

@@ -37,6 +37,7 @@ type Store struct {
 	keysByName   map[string]types.StoreKey
 	stores       map[types.StoreKey]types.CommitKVStore
 	listeners    map[types.StoreKey][]types.WriteListener
+	removalMap   map[types.StoreKey]bool
 
 	initialVersion int64
 	lastCommitInfo *types.CommitInfo
@@ -72,8 +73,18 @@ func (rs *Store) Commit() types.CommitID {
 	}
 
 	rs.lastCommitInfo = commitStores(version, rs.stores, nil)
-
-	// TODO persist to disk
+	defer rs.flushCommitInfo(rs.lastCommitInfo)
+
+	// remove remnants of removed stores
+	for sk := range rs.removalMap {
+		if _, ok := rs.stores[sk]; ok {
+			delete(rs.stores, sk)
+			delete(rs.storesParams, sk)
+			delete(rs.keysByName, sk.Name())
+		}
+	}
+	// reset the removalMap
+	rs.removalMap = make(map[types.StoreKey]bool)
 
 	return types.CommitID{
 		Version: version,
@@ -129,7 +140,8 @@ func (rs *Store) CacheMultiStore() types.CacheMultiStore {
 // Implements interface MultiStore
 // used to createQueryContext, abci_query or grpc query service.
 func (rs *Store) CacheMultiStoreWithVersion(version int64) (types.CacheMultiStore, error) {
-	panic("rootmulti Store don't support historical query service")
+	// TO DO, we still need this to query current state
+	return nil, nil
 }
 
 // Implements interface MultiStore
@@ -229,18 +241,7 @@ func (rs *Store) LoadLatestVersion() error {
 // Implements interface CommitMultiStore
 // used by node startup with UpgradeStoreLoader
 func (rs *Store) LoadLatestVersionAndUpgrade(upgrades *types.StoreUpgrades) error {
-	cInfo := &types.CommitInfo{}
-	bz, err := os.ReadFile(filepath.Join(rs.dir, CommitInfoFileName))
-	if err != nil {
-		// if file not exists, assume empty db
-		if !os.IsNotExist(err) {
-			return errors.Wrap(err, "fail to read commit info file")
-		}
-	} else {
-		if err := cInfo.Unmarshal(bz); err != nil {
-			return errors.Wrap(err, "failed unmarshal commit info")
-		}
-	}
+	cInfo := rs.getCommitInfo()
 
 	infos := make(map[string]types.StoreInfo)
 	// convert StoreInfos slice to map
@@ -252,31 +253,45 @@ func (rs *Store) LoadLatestVersionAndUpgrade(upgrades *types.StoreUpgrades) erro
 	newStores := make(map[types.StoreKey]types.CommitKVStore)
 
 	storesKeys := make([]types.StoreKey, 0, len(rs.storesParams))
+
+	rootStoreVersion := cInfo.Version
+
 	for key := range rs.storesParams {
 		storesKeys = append(storesKeys, key)
 	}
-	// deterministic iteration order for upgrades
-	sort.Slice(storesKeys, func(i, j int) bool {
-		return storesKeys[i].Name() < storesKeys[j].Name()
-	})
 
-	var (
-		commitID types.CommitID
-	)
+	if upgrades != nil {
+		// deterministic iteration order for upgrades
+		// (as the underlying store may change and
+		// upgrades make store changes where the execution order may matter)
+		sort.Slice(storesKeys, func(i, j int) bool {
+			return storesKeys[i].Name() < storesKeys[j].Name()
+		})
+	}
+
 	for _, key := range storesKeys {
+		var (
+			commitID     types.CommitID
+			storeVersion int64
+		)
 		storeParams := rs.storesParams[key]
 
 		if info, ok := infos[key.Name()]; !ok {
 			commitID = info.CommitId
+			storeVersion = commitID.Version
 		} else {
 			commitID = types.CommitID{}
 		}
 
-		// If it has been added, set the initial version
-		if upgrades.IsAdded(key.Name()) || upgrades.RenamedFrom(key.Name()) != "" {
-			storeParams.initialVersion = uint64(cInfo.Version) + 1
-		} else if commitID.Version != cInfo.Version && storeParams.typ == types.StoreTypeIAVL {
-			return fmt.Errorf("version of store %s mismatch root store's version; expected %d got %d", key.Name(), cInfo.Version, commitID.Version)
+		// check if this type of store maintains versions or not
+		if isVersionedStore(storeParams.typ) {
+			// if the store is to be created at this height, set initial version
+			if isNewlyCreatedStore(upgrades, key) {
+				storeParams.initialVersion = uint64(cInfo.Version) + 1
+				// if the store existed at the latest height, check if store version equal to root store version
+			} else if storeVersion != rootStoreVersion {
+				return fmt.Errorf("version of store %s mismatch root store's version; expected %d got %d", key.Name(), cInfo.Version, commitID.Version)
+			}
 		}
 
 		store, err := rs.loadCommitStoreFromParams(key, commitID, storeParams)
@@ -298,6 +313,19 @@ func (rs *Store) LoadLatestVersionAndUpgrade(upgrades *types.StoreUpgrades) erro
 	return nil
 }
 
+// isNewlyCreatedStore check if the store is to be created at this height
+func isNewlyCreatedStore(upgrades *types.StoreUpgrades, key types.StoreKey) bool {
+	return upgrades.IsAdded(key.Name()) || upgrades.RenamedFrom(key.Name()) != ""
+}
+
+// isVersionedStore check if this type of store has version or not
+func isVersionedStore(storeType types.StoreType) bool {
+	if storeType == types.StoreTypeIAVL {
+		return true
+	}
+	return false
+}
+
 func (rs *Store) loadCommitStoreFromParams(key types.StoreKey, id types.CommitID, params storeParams) (types.CommitKVStore, error) {
 	switch params.typ {
 	case types.StoreTypeMulti:
@@ -306,7 +334,7 @@ func (rs *Store) loadCommitStoreFromParams(key types.StoreKey, id types.CommitID
 		dir := filepath.Join(rs.dir, key.Name())
 		return memiavlstore.LoadStoreWithInitialVersion(dir, rs.logger, int64(params.initialVersion))
 	case types.StoreTypeDB:
-		panic("recursive MultiStores not yet supported")
+		panic("db store not yet supported")
 	case types.StoreTypeTransient:
 		_, ok := key.(*types.TransientStoreKey)
 		if !ok {
@@ -330,7 +358,10 @@ func (rs *Store) loadCommitStoreFromParams(key types.StoreKey, id types.CommitID
 // Implements interface CommitMultiStore
 // not used in sdk
 func (rs *Store) LoadVersionAndUpgrade(ver int64, upgrades *types.StoreUpgrades) error {
-	panic("rootmulti store don't support LoadVersionAndUpgrade")
+	if ver != 0 {
+		return errors.New("rootmulti store only support loading the latest version")
+	}
+	return rs.LoadLatestVersionAndUpgrade(upgrades)
 }
 
 // Implements interface CommitMultiStore
@@ -451,3 +482,28 @@ func commitStores(version int64, storeMap map[types.StoreKey]types.CommitKVStore
 		StoreInfos: storeInfos,
 	}
 }
+
+func (rs *Store) flushCommitInfo(cInfo *types.CommitInfo) {
+	file, err := os.OpenFile(filepath.Join(rs.dir, CommitInfoFileName), os.O_WRONLY, 0)
+	if err != nil {
+		panic(err)
+	}
+	cInfoBz, err := cInfo.Marshal()
+	_, err = file.Write(cInfoBz)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func (rs *Store) getCommitInfo() *types.CommitInfo {
+	cInfo := &types.CommitInfo{}
+	cInfoBz, err := os.ReadFile(filepath.Join(rs.dir, CommitInfoFileName))
+	if err != nil {
+		panic(err)
+	} else {
+		if err := cInfo.Unmarshal(cInfoBz); err != nil {
+			panic(err)
+		}
+	}
+	return cInfo
+}