fix(seidb): stop evm-logical-digest replay mode from mutating a live memiavl

sei-db/state_db/sc/memiavl/db.go

@@ -27,7 +27,21 @@ import (
 
 const LockFileName = "LOCK"
 
-var errReadOnly = errors.New("db is read-only")
+var (
+	errReadOnly = errors.New("db is read-only")
+
+	// ErrReadOnlyWALCorrupt means a read-only open observed an incomplete,
+	// corrupt, or concurrently recovering changelog. The source WAL is left
+	// untouched; callers can retry after the writer finishes its current WAL
+	// operation.
+	ErrReadOnlyWALCorrupt = errors.New("read-only changelog is incomplete or corrupt")
+
+	// ErrReadOnlyWALUnavailable means the immutable WAL view cannot replay
+	// every version from the selected snapshot through the requested target.
+	// The live writer may have pruned or advanced the changelog while the
+	// reader opened it; callers can retry against a new point-in-time view.
+	ErrReadOnlyWALUnavailable = errors.New("read-only changelog cannot reach the requested version")
+)
 
 // DB implements DB-like functionalities on top of MultiTree:
 // - async snapshot rewriting
@@ -158,9 +172,26 @@ func OpenDB(targetVersion int64, opts Options) (database *DB, _err error) {
 		)
 	}()
 	var (
-		err      error
-		fileLock FileLock
+		err           error
+		fileLock      FileLock
+		mtree         *MultiTree
+		streamHandler wal.ChangelogWAL
 	)
+	defer func() {
+		if _err == nil {
+			return
+		}
+		if streamHandler != nil {
+			_ = streamHandler.Close()
+		}
+		if mtree != nil {
+			_ = mtree.Close()
+		}
+		if fileLock != nil {
+			_ = fileLock.Unlock()
+			_ = fileLock.Destroy()
+		}
+	}()
 	if err := opts.Validate(); err != nil {
 		return nil, fmt.Errorf("invalid commit store options: %w", err)
 	}
@@ -194,19 +225,28 @@ func OpenDB(targetVersion int64, opts Options) (database *DB, _err error) {
 	}
 
 	path := filepath.Join(opts.Dir, snapshot)
-	mtree, err := LoadMultiTree(context.Background(), path, opts)
+	mtree, err = LoadMultiTree(context.Background(), path, opts)
 	if err != nil {
 		return nil, err
 	}
 
 	// Snapshot mmap files are loaded with MADV_RANDOM in OpenSnapshot().
 
-	// MemIAVL owns changelog lifecycle: always open the WAL here.
-	// Even in read-only mode we may need WAL replay to reconstruct non-snapshot versions.
-	streamHandler, err := wal.NewChangelogWAL(utils.GetChangelogPath(opts.Dir), wal.Config{
-		WriteBufferSize: opts.AsyncCommitBuffer,
-	})
+	// MemIAVL owns changelog lifecycle: always open the WAL here. Read-only
+	// callers still need replay to reconstruct non-snapshot versions, but they
+	// must not use the writable opener: it repairs a torn tail by truncating it
+	// and completes interrupted WAL truncations by renaming or removing files.
+	if opts.ReadOnly {
+		streamHandler, err = wal.OpenReadOnlyChangelogWAL(utils.GetChangelogPath(opts.Dir))
+	} else {
+		streamHandler, err = wal.NewChangelogWAL(utils.GetChangelogPath(opts.Dir), wal.Config{
+			WriteBufferSize: opts.AsyncCommitBuffer,
+		})
+	}
 	if err != nil {
+		if opts.ReadOnly && errors.Is(err, wal.ErrCorrupt) {
+			return nil, fmt.Errorf("%w; source WAL was not modified: %w", ErrReadOnlyWALCorrupt, err)
+		}
 		return nil, fmt.Errorf("failed to open changelog WAL: %w", err)
 	}
 
@@ -221,6 +261,23 @@ func OpenDB(targetVersion int64, opts Options) (database *DB, _err error) {
 	if !walHasEntries {
 		walIndexDelta = mtree.WorkingCommitInfo().Version - 1
 	}
+	if opts.ReadOnly && walHasEntries && (targetVersion == 0 || targetVersion > mtree.Version()) {
+		firstIndex, firstErr := streamHandler.FirstOffset()
+		if firstErr != nil {
+			return nil, fmt.Errorf("read changelog first offset: %w", firstErr)
+		}
+		if firstIndex > math.MaxInt64 {
+			return nil, fmt.Errorf("%w: first WAL offset %d overflows int64", ErrReadOnlyWALUnavailable, firstIndex)
+		}
+		firstVersion := int64(firstIndex) + walIndexDelta
+		firstNeeded := utils.NextVersion(mtree.Version(), mtree.initialVersion.Load())
+		if firstVersion > firstNeeded {
+			snapshotVersion := mtree.Version()
+			return nil, fmt.Errorf("%w: selected snapshot version %d needs changelog version %d, "+
+				"but the immutable WAL view starts at version %d",
+				ErrReadOnlyWALUnavailable, snapshotVersion, firstNeeded, firstVersion)
+		}
+	}
 
 	// Replay WAL to catch up to target version (if WAL has entries)
 	if walHasEntries && (targetVersion == 0 || targetVersion > mtree.Version()) {
@@ -230,6 +287,11 @@ func OpenDB(targetVersion int64, opts Options) (database *DB, _err error) {
 		}
 		logger.Info("finished replay and caught up to target version", "version", targetVersion)
 	}
+	if opts.ReadOnly && targetVersion > 0 && mtree.Version() != targetVersion {
+		reached := mtree.Version()
+		return nil, fmt.Errorf("%w: requested %d, reached %d",
+			ErrReadOnlyWALUnavailable, targetVersion, reached)
+	}
 
 	if opts.LoadForOverwriting && targetVersion > 0 {
 		currentSnapshot, err := os.Readlink(currentPath(opts.Dir))
@@ -296,6 +358,10 @@ func OpenDB(targetVersion int64, opts Options) (database *DB, _err error) {
 		snapshotWriterPool:      workerPool,
 		opts:                    opts,
 	}
+	// The DB owns these resources from this point forward.
+	mtree = nil
+	streamHandler = nil
+	fileLock = nil
 
 	// Apply initial stores on a fresh DB (version 0) so they get persisted to WAL.
 	// This creates the trees and populates pendingLogEntry, which will be written
@@ -307,6 +373,7 @@ func OpenDB(targetVersion int64, opts Options) (database *DB, _err error) {
 			upgrades = append(upgrades, &proto.TreeNameUpgrade{Name: name})
 		}
 		if err := db.ApplyUpgrades(upgrades); err != nil {
+			_ = db.Close()
 			return nil, fmt.Errorf("failed to apply initial stores: %w", err)
 		}
 	}

sei-db/state_db/sc/memiavl/db_test.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime/debug"
+	"sort"
 	"strconv"
 	"sync"
 	"testing"
@@ -1116,3 +1117,132 @@ func TestUpdateCurrentSymlinkClearsStaleTmp(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "snapshot-1", target)
 }
+
+func TestReadOnlyOpenRejectsTornWALWithoutRepair(t *testing.T) {
+	dir := t.TempDir()
+	db, err := OpenDB(0, Options{
+		Dir:             dir,
+		CreateIfMissing: true,
+		InitialStores:   []string{"test"},
+	})
+	require.NoError(t, err)
+	for i := 0; i < 3; i++ {
+		require.NoError(t, db.ApplyChangeSets([]*proto.NamedChangeSet{{
+			Name:      "test",
+			Changeset: ChangeSets[i],
+		}}))
+		_, err := db.Commit()
+		require.NoError(t, err)
+	}
+	require.NoError(t, db.Close())
+
+	segment := lastMemiAVLWALSegment(t, dir)
+	file, err := os.OpenFile(filepath.Clean(segment), os.O_WRONLY|os.O_APPEND, 0)
+	require.NoError(t, err)
+	_, err = file.Write([]byte{0x10})
+	require.NoError(t, err)
+	require.NoError(t, file.Close())
+	before, err := os.ReadFile(filepath.Clean(segment))
+	require.NoError(t, err)
+
+	_, err = OpenDB(0, Options{Dir: dir, ReadOnly: true})
+	require.ErrorIs(t, err, ErrReadOnlyWALCorrupt)
+	after, readErr := os.ReadFile(filepath.Clean(segment))
+	require.NoError(t, readErr)
+	require.Equal(t, before, after, "read-only open must leave a torn live tail untouched")
+
+	repaired, err := OpenDB(0, Options{Dir: dir})
+	require.NoError(t, err, "the writable owner must retain the existing tail-repair behavior")
+	require.Equal(t, int64(3), repaired.Version())
+	require.NoError(t, repaired.Close())
+}
+
+func TestReadOnlyOpenRejectsWALGap(t *testing.T) {
+	dir := t.TempDir()
+	db, err := OpenDB(0, Options{
+		Dir:             dir,
+		CreateIfMissing: true,
+		InitialStores:   []string{"test"},
+	})
+	require.NoError(t, err)
+	for i := 0; i < 3; i++ {
+		require.NoError(t, db.ApplyChangeSets([]*proto.NamedChangeSet{{
+			Name:      "test",
+			Changeset: ChangeSets[i],
+		}}))
+		_, err := db.Commit()
+		require.NoError(t, err)
+	}
+	require.NoError(t, db.GetWAL().TruncateBefore(2))
+	require.NoError(t, db.Close())
+
+	_, err = OpenDB(3, Options{Dir: dir, ReadOnly: true})
+	require.ErrorIs(t, err, ErrReadOnlyWALUnavailable)
+	require.Contains(t, err.Error(), "needs changelog version 1")
+	require.Contains(t, err.Error(), "starts at version 2")
+}
+
+func TestReadOnlyOpenRejectsShortWAL(t *testing.T) {
+	dir := t.TempDir()
+	db, err := OpenDB(0, Options{
+		Dir:             dir,
+		CreateIfMissing: true,
+		InitialStores:   []string{"test"},
+	})
+	require.NoError(t, err)
+	for i := 0; i < 3; i++ {
+		require.NoError(t, db.ApplyChangeSets([]*proto.NamedChangeSet{{
+			Name:      "test",
+			Changeset: ChangeSets[i],
+		}}))
+		_, err := db.Commit()
+		require.NoError(t, err)
+	}
+	require.NoError(t, db.GetWAL().TruncateAfter(2))
+	require.NoError(t, db.Close())
+
+	_, err = OpenDB(3, Options{Dir: dir, ReadOnly: true})
+	require.ErrorIs(t, err, ErrReadOnlyWALUnavailable)
+	require.Contains(t, err.Error(), "requested 3, reached 2")
+}
+
+func TestOpenDBFailureReleasesFileLock(t *testing.T) {
+	dir := t.TempDir()
+	db, err := OpenDB(0, Options{
+		Dir:             dir,
+		CreateIfMissing: true,
+		InitialStores:   []string{"test"},
+	})
+	require.NoError(t, err)
+	require.NoError(t, db.ApplyChangeSets([]*proto.NamedChangeSet{{
+		Name:      "test",
+		Changeset: ChangeSets[0],
+	}}))
+	_, err = db.Commit()
+	require.NoError(t, err)
+	require.NoError(t, db.Close())
+
+	require.NoError(t, os.Remove(currentPath(dir)))
+	_, err = OpenDB(1, Options{Dir: dir, LoadForOverwriting: true})
+	require.ErrorContains(t, err, "fail to read current version")
+
+	lock, err := LockFile(filepath.Join(dir, LockFileName))
+	require.NoError(t, err, "failed OpenDB must release its exclusive lock")
+	require.NoError(t, lock.Unlock())
+	require.NoError(t, lock.Destroy())
+}
+
+func lastMemiAVLWALSegment(t *testing.T, dir string) string {
+	t.Helper()
+	entries, err := os.ReadDir(utils.GetChangelogPath(dir))
+	require.NoError(t, err)
+	var names []string
+	for _, entry := range entries {
+		if !entry.IsDir() && len(entry.Name()) == 20 {
+			names = append(names, entry.Name())
+		}
+	}
+	require.NotEmpty(t, names)
+	sort.Strings(names)
+	return filepath.Join(utils.GetChangelogPath(dir), names[len(names)-1])
+}

sei-db/tools/cmd/seidb/operations/evm_logical_digest.go

@@ -81,12 +81,14 @@ const (
 //     that exact height (or --height 0 for the current symlink). This is the
 //     preferred mode whenever the target height lines up with an existing
 //     snapshot boundary.
-//   - replay (SLOW): opens a read-only DB, replays the changelog up to
-//     --height, then walks the in-memory/mmap tree. Roughly an order of
-//     magnitude slower than snapshot (changelog replay + per-leaf tree walk
-//     instead of a sequential file read). Use it only when no snapshot exists
-//     at the target height — e.g. nodes whose snapshot rewrite lags the tip, so
-//     an arbitrary comparison height has no snapshot-<height> on disk.
+//   - replay (SLOW): opens a non-mutating read-only WAL view, replays the
+//     changelog up to --height, then walks the in-memory/mmap tree. Roughly an
+//     order of magnitude slower than snapshot (changelog replay + per-leaf tree
+//     walk instead of a sequential file read). If a live writer leaves a torn
+//     tail in view, replay fails and asks the operator to rerun instead of
+//     repairing the source WAL. Use it only when no snapshot exists at the
+//     target height — e.g. nodes whose snapshot rewrite lags the tip, so an
+//     arbitrary comparison height has no snapshot-<height> on disk.
 //
 // The flatkv side is always a pebble WAL-replay-to-height and is fast
 // regardless. So when comparing across nodes, pick a height that is an existing
@@ -1111,8 +1113,26 @@ func openMemiAVLReplayReadOnly(dbDir string, height int64) (*memiavl.DB, error)
 		ZeroCopy: true,
 	})
 	if err != nil {
+		if errors.Is(err, memiavl.ErrReadOnlyWALCorrupt) {
+			return nil, fmt.Errorf("memiavl changelog tail is incomplete, corrupt, or changing; "+
+				"live WAL was not modified; rerun the command, and if the error persists after stopping seid, "+
+				"repair the WAL offline: %w", err)
+		}
+		if errors.Is(err, memiavl.ErrReadOnlyWALUnavailable) {
+			return nil, fmt.Errorf("the immutable memiavl changelog view could not reach height %d; "+
+				"live WAL was not modified; rerun the command: %w", height, err)
+		}
 		return nil, fmt.Errorf("open memiavl read-only replay: %w", err)
 	}
+	if height > 0 && db.Version() != height {
+		versionErr := fmt.Errorf("memiavl replay version mismatch: requested %d, reached %d; "+
+			"the live changelog did not provide a complete path to the target; rerun the command",
+			height, db.Version())
+		if closeErr := db.Close(); closeErr != nil {
+			return nil, errors.Join(versionErr, fmt.Errorf("close memiavl read-only replay: %w", closeErr))
+		}
+		return nil, versionErr
+	}
 	return db, nil
 }
 

sei-db/tools/cmd/seidb/operations/memiavl_open_test.go

@@ -0,0 +1,61 @@
+package operations
+
+import (
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/sei-protocol/sei-chain/sei-db/common/keys"
+	"github.com/sei-protocol/sei-chain/sei-db/common/utils"
+	"github.com/sei-protocol/sei-chain/sei-db/proto"
+)
+
+func TestOpenMemiAVLReplayReadOnlyReportsRetryWithoutRepair(t *testing.T) {
+	homeDir := t.TempDir()
+	store := newTestMemiavlStore(t, homeDir)
+	require.NoError(t, store.ApplyChangeSets([]*proto.NamedChangeSet{{
+		Name:      keys.EVMStoreKey,
+		Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{noncePair(addrN(0xA1), 1)}},
+	}}))
+	_, err := store.Commit()
+	require.NoError(t, err)
+	require.NoError(t, store.Close())
+
+	dbDir := utils.GetCosmosSCStorePath(homeDir)
+	segment := lastOperationsMemiAVLWALSegment(t, dbDir)
+	file, err := os.OpenFile(filepath.Clean(segment), os.O_WRONLY|os.O_APPEND, 0)
+	require.NoError(t, err)
+	_, err = file.Write([]byte{0x10})
+	require.NoError(t, err)
+	require.NoError(t, file.Close())
+	before, err := os.ReadFile(filepath.Clean(segment))
+	require.NoError(t, err)
+
+	_, err = openMemiAVLReplayReadOnly(dbDir, 0)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "live WAL was not modified")
+	require.Contains(t, err.Error(), "rerun the command")
+
+	after, readErr := os.ReadFile(filepath.Clean(segment))
+	require.NoError(t, readErr)
+	require.Equal(t, before, after)
+}
+
+func lastOperationsMemiAVLWALSegment(t *testing.T, dbDir string) string {
+	t.Helper()
+	changelogDir := utils.GetChangelogPath(dbDir)
+	entries, err := os.ReadDir(changelogDir)
+	require.NoError(t, err)
+	var names []string
+	for _, entry := range entries {
+		if !entry.IsDir() && len(entry.Name()) == 20 {
+			names = append(names, entry.Name())
+		}
+	}
+	require.NotEmpty(t, names)
+	sort.Strings(names)
+	return filepath.Join(changelogDir, names[len(names)-1])
+}