Problem: version mismatch happen occasionally (backport #1759) (#1765)

* Problem: version mismatch happen occationally

Solution:
- make sure don't load iavl version ahead of versiondb

* changelog

* Update app/app.go



* memiavl support write old version idopodently

* add unit test

* fix upgrade store loader

* reproduce in integration test

* fix memiavl

* better test

* fix lint

---------

Signed-off-by: yihuang <[email protected]>

Author: yihuang

.github/workflows/sims.yml

@@ -136,7 +136,7 @@ jobs:
         run: go version
       - name: Install runsim
         run: export GO111MODULE="on" && go install github.com/cosmos/tools/cmd/[email protected]
-      - uses: actions/cache@v2.1.6
+      - uses: actions/cache@v4
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-go-runsim-binary
@@ -183,7 +183,7 @@ jobs:
             go.mod
             go.sum
         if: github.event_name == 'push' || github.event_name == 'pull_request'
-      - uses: actions/cache@v2.1.6
+      - uses: actions/cache@v4
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-go-runsim-binary
@@ -239,7 +239,7 @@ jobs:
             go.mod
             go.sum
         if: github.event_name == 'push' || github.event_name == 'pull_request'
-      - uses: actions/cache@v2.1.6
+      - uses: actions/cache@v4
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-go-runsim-binary
@@ -295,7 +295,7 @@ jobs:
             go.mod
             go.sum
         if: github.event_name == 'push' || github.event_name == 'pull_request'
-      - uses: actions/cache@v2.1.6
+      - uses: actions/cache@v4
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-go-runsim-binary
@@ -351,7 +351,7 @@ jobs:
             go.mod
             go.sum
         if: github.event_name == 'push' || github.event_name == 'pull_request'
-      - uses: actions/cache@v2.1.6
+      - uses: actions/cache@v4
         with:
           path: ~/go/bin
           key: ${{ runner.os }}-go-runsim-binary

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## UNRELEASED
+
+### Bug Fixes
+
+* [#1759](https://github.com/crypto-org-chain/cronos/pull/1759) Fix version mismatch happen occasionally.
+
 *Feb 28, 2025*
 
 ## v1.4.4

app/app.go

@@ -959,9 +959,27 @@ func New(
 		panic(err)
 	}
 
+	// wire up the versiondb's `StreamingService` and `MultiStore`.
+	if cast.ToBool(appOpts.Get("versiondb.enable")) {
+		var err error
+		app.qms, err = app.setupVersionDB(homePath, keys, tkeys, memKeys, okeys)
+		if err != nil {
+			panic(err)
+		}
+	}
+
+	var qmsVersion int64
+	if app.qms != nil {
+		qmsVersion = app.qms.LatestVersion()
+	}
+
 	// RegisterUpgradeHandlers is used for registering any on-chain upgrades.
 	// Make sure it's called after `app.mm` and `app.configurator` are set.
-	app.RegisterUpgradeHandlers(app.appCodec)
+	storeLoaderOverritten := app.RegisterUpgradeHandlers(app.appCodec, qmsVersion)
+	if !storeLoaderOverritten {
+		// Register the default store loader
+		app.SetStoreLoader(MaxVersionStoreLoader(qmsVersion))
+	}
 
 	// add test gRPC service for testing gRPC queries in isolation
 	// testdata.RegisterQueryServer(app.GRPCQueryRouter(), testdata.QueryImpl{})
@@ -992,15 +1010,6 @@ func New(
 	app.MountMemoryStores(memKeys)
 	app.MountObjectStores(okeys)
 
-	// wire up the versiondb's `StreamingService` and `MultiStore`.
-	if cast.ToBool(appOpts.Get("versiondb.enable")) {
-		var err error
-		app.qms, err = app.setupVersionDB(homePath, keys, tkeys, memKeys, okeys)
-		if err != nil {
-			panic(err)
-		}
-	}
-
 	// initialize BaseApp
 	app.SetInitChainer(app.InitChainer)
 	app.SetPreBlocker(app.PreBlocker)
@@ -1045,12 +1054,13 @@ func New(
 			tmos.Exit(err.Error())
 		}
 
-		if app.qms != nil {
-			v1 := app.qms.LatestVersion()
-			v2 := app.LastBlockHeight()
-			if v1 > 0 && v1 < v2 {
+		if qmsVersion > 0 {
+			// it should not happens since we constraint the loaded iavl version to not exceed the versiondb version,
+			// still keep the check for safety.
+			iavlVersion := app.LastBlockHeight()
+			if qmsVersion < iavlVersion {
 				// try to prevent gap being created in versiondb
-				tmos.Exit(fmt.Sprintf("versiondb version %d lag behind iavl version %d", v1, v2))
+				tmos.Exit(fmt.Sprintf("versiondb version %d lag behind iavl version %d", qmsVersion, iavlVersion))
 			}
 		}
 

app/storeloader.go

@@ -0,0 +1,37 @@
+package app
+
+import (
+	storetypes "cosmossdk.io/store/types"
+	upgradetypes "cosmossdk.io/x/upgrade/types"
+	"github.com/cosmos/cosmos-sdk/baseapp"
+)
+
+// MaxVersionStoreLoader will be used when there's versiondb to cap the loaded iavl version
+func MaxVersionStoreLoader(version int64) baseapp.StoreLoader {
+	if version == 0 {
+		return baseapp.DefaultStoreLoader
+	}
+
+	return func(ms storetypes.CommitMultiStore) error {
+		return ms.LoadVersion(version)
+	}
+}
+
+// MaxVersionUpgradeStoreLoader is used to prepare baseapp with a fixed StoreLoader
+func MaxVersionUpgradeStoreLoader(version int64, upgradeHeight int64, storeUpgrades *storetypes.StoreUpgrades) baseapp.StoreLoader {
+	if version == 0 {
+		return upgradetypes.UpgradeStoreLoader(upgradeHeight, storeUpgrades)
+	}
+
+	return func(ms storetypes.CommitMultiStore) error {
+		if upgradeHeight == ms.LastCommitID().Version+1 {
+			// Check if the current commit version and upgrade height matches
+			if len(storeUpgrades.Renamed) > 0 || len(storeUpgrades.Deleted) > 0 || len(storeUpgrades.Added) > 0 {
+				return ms.LoadLatestVersionAndUpgrade(storeUpgrades)
+			}
+		}
+
+		// Otherwise load default store loader
+		return MaxVersionStoreLoader(version)(ms)
+	}
+}

app/upgrades.go

@@ -17,7 +17,8 @@ import (
 	evmtypes "github.com/evmos/ethermint/x/evm/types"
 )
 
-func (app *App) RegisterUpgradeHandlers(cdc codec.BinaryCodec) {
+// RegisterUpgradeHandlers returns if store loader is overridden
+func (app *App) RegisterUpgradeHandlers(cdc codec.BinaryCodec, maxVersion int64) bool {
 	planName := "v1.4"
 	app.UpgradeKeeper.SetUpgradeHandler(planName, func(ctx context.Context, plan upgradetypes.Plan, fromVM module.VersionMap) (module.VersionMap, error) {
 		m, err := app.ModuleManager.RunMigrations(ctx, app.configurator, fromVM)
@@ -54,14 +55,18 @@ func (app *App) RegisterUpgradeHandlers(cdc codec.BinaryCodec) {
 	}
 	if !app.UpgradeKeeper.IsSkipHeight(upgradeInfo.Height) {
 		if upgradeInfo.Name == planName {
-			app.SetStoreLoader(upgradetypes.UpgradeStoreLoader(upgradeInfo.Height, &storetypes.StoreUpgrades{
+			app.SetStoreLoader(MaxVersionUpgradeStoreLoader(maxVersion, upgradeInfo.Height, &storetypes.StoreUpgrades{
 				Added: []string{
 					icahosttypes.StoreKey,
 				},
 				Deleted: []string{"icaauth"},
 			}))
+
+			return true
 		}
 	}
+
+	return false
 }
 
 func UpdateExpeditedParams(ctx context.Context, gov govkeeper.Keeper) error {

integration_tests/shell.nix

@@ -17,5 +17,6 @@ pkgs.mkShell {
   shellHook = ''
     mkdir ./coverage
     export GOCOVERDIR=./coverage
+    export TMPDIR=/tmp
   '';
 }

integration_tests/test_versiondb.py

@@ -5,7 +5,7 @@
 from pystarport import ports
 
 from .network import Cronos
-from .utils import ADDRS, send_transaction, wait_for_port
+from .utils import ADDRS, send_transaction, w3_wait_for_new_blocks, wait_for_port
 
 
 def test_versiondb_migration(cronos: Cronos):
@@ -37,6 +37,9 @@ def test_versiondb_migration(cronos: Cronos):
     balance1 = w3.eth.get_balance(community)
     block1 = w3.eth.block_number
 
+    # wait for a few blocks
+    w3_wait_for_new_blocks(w3, 5)
+
     # stop the network first
     print("stop all nodes")
     print(cronos.supervisorctl("stop", "all"))
@@ -45,7 +48,10 @@ def test_versiondb_migration(cronos: Cronos):
 
     changeset_dir = tempfile.mkdtemp(dir=cronos.base_dir)
     print("dump to:", changeset_dir)
-    print(cli1.changeset_dump(changeset_dir))
+
+    # only restore to an intermidiate version to test version mismatch behavior
+    print(cli1.changeset_dump(changeset_dir, end_version=block1 + 1))
+
     snapshot_dir = tempfile.mkdtemp(dir=cronos.base_dir)
     print("verify and save to snapshot:", snapshot_dir)
     _, commit_info = cli0.changeset_verify(changeset_dir, save_snapshot=snapshot_dir)

memiavl/db.go

@@ -80,6 +80,9 @@ type DB struct {
 	mtx sync.Mutex
 	// worker goroutine IdleTimeout = 5s
 	snapshotWriterPool *pond.WorkerPool
+
+	// reusable write batch
+	wbatch wal.Batch
 }
 
 type Options struct {
@@ -440,8 +443,13 @@ func (db *DB) checkBackgroundSnapshotRewrite() error {
 		db.snapshotRewriteCancel = nil
 
 		if result.mtree == nil {
-			// background snapshot rewrite failed
-			return fmt.Errorf("background snapshot rewriting failed: %w", result.err)
+			if result.err != nil {
+				// background snapshot rewrite failed
+				return fmt.Errorf("background snapshot rewriting failed: %w", result.err)
+			}
+
+			// background snapshot rewrite don't success, but no error to propagate, ignore it.
+			return nil
 		}
 
 		// wait for potential pending wal writings to finish, to make sure we catch up to latest state.
@@ -556,11 +564,17 @@ func (db *DB) Commit() (int64, error) {
 			// async wal writing
 			db.walChan <- &entry
 		} else {
-			bz, err := entry.data.Marshal()
+			lastIndex, err := db.wal.LastIndex()
 			if err != nil {
 				return 0, err
 			}
-			if err := db.wal.Write(entry.index, bz); err != nil {
+
+			db.wbatch.Clear()
+			if err := writeEntry(&db.wbatch, db.logger, lastIndex, &entry); err != nil {
+				return 0, err
+			}
+
+			if err := db.wal.WriteBatch(&db.wbatch); err != nil {
 				return 0, err
 			}
 		}
@@ -591,13 +605,17 @@ func (db *DB) initAsyncCommit() {
 				break
 			}
 
+			lastIndex, err := db.wal.LastIndex()
+			if err != nil {
+				walQuit <- err
+				return
+			}
+
 			for _, entry := range entries {
-				bz, err := entry.data.Marshal()
-				if err != nil {
+				if err := writeEntry(&batch, db.logger, lastIndex, entry); err != nil {
 					walQuit <- err
 					return
 				}
-				batch.Write(entry.index, bz)
 			}
 
 			if err := db.wal.WriteBatch(&batch); err != nil {
@@ -749,7 +767,8 @@ func (db *DB) rewriteSnapshotBackground() error {
 
 		cloned.logger.Info("start rewriting snapshot", "version", cloned.Version())
 		if err := cloned.RewriteSnapshotWithContext(ctx); err != nil {
-			ch <- snapshotResult{err: err}
+			// write error log but don't stop the client, it could happen when load an old version.
+			cloned.logger.Error("failed to rewrite snapshot", "err", err)
 			return
 		}
 		cloned.logger.Info("finished rewriting snapshot", "version", cloned.Version())
@@ -1093,3 +1112,17 @@ func channelBatchRecv[T any](ch <-chan *T) []*T {
 
 	return result
 }
+
+func writeEntry(batch *wal.Batch, logger Logger, lastIndex uint64, entry *walEntry) error {
+	bz, err := entry.data.Marshal()
+	if err != nil {
+		return err
+	}
+
+	if entry.index <= lastIndex {
+		logger.Error("commit old version idempotently", "lastIndex", lastIndex, "version", entry.index)
+	} else {
+		batch.Write(entry.index, bz)
+	}
+	return nil
+}