Repository: matrixorigin/matrixcube (go)

Issue:
Not all restored snapshot images are cleaned up
Remove all applied snapshot images whenever possible. 

Pull request description:
Fixed snapshot compaction to remove snapshot images whenever possible
**What type of PR is this?**

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #568 

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available


Patch:
diff --git a/logdb/logdb.go b/logdb/logdb.go
index c97d22e..f66507a 100644
--- a/logdb/logdb.go
+++ b/logdb/logdb.go
@@ -146,26 +146,26 @@ func (l *KVLogDB) NewWorkerContext() *WorkerContext {
 }
 
 func (l *KVLogDB) RemoveSnapshot(shardID uint64, index uint64) error {
+	ss, err := l.GetSnapshot(shardID)
+	if err != nil && err != ErrNoSnapshot {
+		return err
+	}
+	if !raft.IsEmptySnap(ss) && ss.Metadata.Index == index {
+		panic("trying to remove the most recent snapshot")
+	}
 	key := keys.GetSnapshotKey(shardID, index, nil)
 	return l.ms.Delete(key, true)
 }
 
 func (l *KVLogDB) GetSnapshot(shardID uint64) (raftpb.Snapshot, error) {
-	fk := keys.GetSnapshotKey(shardID, 0, nil)
-	lk := keys.GetSnapshotKey(shardID, math.MaxUint64, nil)
-	var v []byte
-	if err := l.ms.Scan(fk, lk, func(key, value []byte) (bool, error) {
-		v = value
-		return true, nil
-	}, true); err != nil {
+	snapshots, err := l.GetAllSnapshots(shardID)
+	if err != nil {
 		return raftpb.Snapshot{}, err
 	}
-	if len(v) == 0 {
+	if len(snapshots) == 0 {
 		return raftpb.Snapshot{}, ErrNoSnapshot
 	}
-	var ss raftpb.Snapshot
-	protoc.MustUnmarshal(&ss, v)
-	return ss, nil
+	return snapshots[len(snapshots)-1], nil
 }
 
 func (l *KVLogDB) GetAllSnapshots(shardID uint64) ([]raftpb.Snapshot, error) {
@@ -300,6 +300,18 @@ func (l *KVLogDB) ReadRaftState(shardID uint64,
 // TODO: check whether index below is larger than the max index
 // RemoveEntriesTo deletes all raft log entries between [0, index].
 func (l *KVLogDB) RemoveEntriesTo(shardID uint64, replicaID uint64, index uint64) error {
+	ss, err := l.GetSnapshot(shardID)
+	if err != nil && err != ErrNoSnapshot {
+		return err
+	}
+	if raft.IsEmptySnap(ss) {
+		panic("no snapshot")
+	}
+	if ss.Metadata.Index < index {
+		l.logger.Fatal("removing too many log entries",
+			log.SnapshotField(ss),
+			log.IndexField(index))
+	}
 	startKey := keys.GetRaftLogKey(shardID, 0, nil)
 	endKey := keys.GetRaftLogKey(shardID, index+1, nil)
 	return l.ms.RangeDelete(startKey, endKey, true)
diff --git a/raftstore/replica.go b/raftstore/replica.go
index 09573f1..9d4cf64 100644
--- a/raftstore/replica.go
+++ b/raftstore/replica.go
@@ -197,7 +197,7 @@ func (pr *replica) start() {
 			zap.Error(err))
 	}
 	// pr.initAppliedIndex must be called before pr.initConfState()
-	if err := pr.initAppliedIndex(pr.sm.dataStorage); err != nil {
+	if err := pr.initAppliedIndex(); err != nil {
 		pr.logger.Fatal("failed to initialize applied index",
 			zap.Error(err))
 	}
@@ -289,6 +289,10 @@ func (pr *replica) getShard() Shard {
 	return pr.sm.getShard()
 }
 
+func (pr *replica) getPersistentLogIndex() (uint64, error) {
+	return pr.sm.dataStorage.GetPersistentLogIndex(pr.shardID)
+}
+
 func (pr *replica) getFirstIndex() uint64 {
 	return pr.sm.getFirstIndex()
 }
@@ -299,9 +303,8 @@ func (pr *replica) getShardID() uint64 {
 
 // TODO: move this into the state machine, it should be invoked as a part of the
 // state machine restart procedure.
-// initAppliedIndex load PersistentLogIndex from datastorage, use this index to init raft rawnode.
-func (pr *replica) initAppliedIndex(storage storage.DataStorage) error {
-	persistentLogIndex, err := storage.GetPersistentLogIndex(pr.shardID)
+func (pr *replica) initAppliedIndex() error {
+	persistentLogIndex, err := pr.getPersistentLogIndex()
 	if err != nil {
 		return err
 	}
@@ -330,7 +333,7 @@ func (pr *replica) initAppliedIndex(storage storage.DataStorage) error {
 // initConfState initializes the ConfState of the LogReader which will be
 // applied to the raft module.
 func (pr *replica) initConfState() error {
-	persistentLogIndex, err := pr.sm.dataStorage.GetPersistentLogIndex(pr.shardID)
+	persistentLogIndex, err := pr.getPersistentLogIndex()
 	if err != nil {
 		return err
 	}
diff --git a/raftstore/replica_apply_result.go b/raftstore/replica_apply_result.go
index 17dcddd..4821089 100644
--- a/raftstore/replica_apply_result.go
+++ b/raftstore/replica_apply_result.go
@@ -134,6 +134,9 @@ func (pr *replica) applyUpdateLabels(result updateLabelsResult) {
 }
 
 func (pr *replica) applyCompactionResult(r compactionResult) {
+	if r.index == 0 {
+		return
+	}
 	pr.logger.Info("log compaction called",
 		log.IndexField(r.index))
 	// generate a dummy snapshot so we can run the log compaction.
@@ -141,44 +144,42 @@ func (pr *replica) applyCompactionResult(r compactionResult) {
 	// the LogReader on startup.
 	// such dummy snapshot will never be loaded, as its Index value is not
 	// greater than data storage's persistentLogIndex value.
-	if r.index > 0 {
-		term, err := pr.lr.Term(r.index)
-		if err != nil {
-			pr.logger.Error("failed to get term value",
-				zap.Error(err),
+	term, err := pr.lr.Term(r.index)
+	if err != nil {
+		pr.logger.Error("failed to get term value",
+			zap.Error(err),
+			log.IndexField(r.index))
+		if err == raft.ErrCompacted || err == raft.ErrUnavailable {
+			// skip this compaction operation as we can't establish the marker
+			// position.
+			pr.logger.Info("skipped a compaction request",
 				log.IndexField(r.index))
-			if err == raft.ErrCompacted || err == raft.ErrUnavailable {
-				// skip this compaction operation as we can't establish the marker
-				// position.
-				pr.logger.Info("skipped a compaction request",
-					log.IndexField(r.index))
-				return
-			}
-			panic(err)
-		}
-		// this is a dummy snapshot meaning there is no on disk snapshot image.
-		// we are not supposed to apply such dummy snapshot. the dummy flag is
-		// used for debugging purposes.
-		si := meta.SnapshotInfo{
-			Dummy: true,
+			return
 		}
-		rd := raft.Ready{
-			Snapshot: raftpb.Snapshot{
-				Metadata: raftpb.SnapshotMetadata{
-					Index: r.index,
-					Term:  term,
-				},
-				Data: protoc.MustMarshal(&si),
+		panic(err)
+	}
+	// this is a dummy snapshot meaning there is no on disk snapshot image.
+	// we are not supposed to apply such dummy snapshot. the dummy flag is
+	// used for debugging purposes.
+	si := meta.SnapshotInfo{
+		Dummy: true,
+	}
+	rd := raft.Ready{
+		Snapshot: raftpb.Snapshot{
+			Metadata: raftpb.SnapshotMetadata{
+				Index: r.index,
+				Term:  term,
 			},
-		}
-		wc := pr.logdb.NewWorkerContext()
-		defer wc.Close()
-		if err := pr.logdb.SaveRaftState(pr.shardID, pr.replicaID, rd, wc); err != nil {
-			panic(err)
-		}
-		pr.logger.Info("dummy snapshot saved",
-			log.IndexField(r.index))
+			Data: protoc.MustMarshal(&si),
+		},
+	}
+	wc := pr.logdb.NewWorkerContext()
+	defer wc.Close()
+	if err := pr.logdb.SaveRaftState(pr.shardID, pr.replicaID, rd, wc); err != nil {
+		panic(err)
 	}
+	pr.logger.Info("dummy snapshot saved",
+		log.IndexField(r.index))
 	// update LogReader's range info to make the compacted entries invisible to
 	// raft.
 	if err := pr.lr.Compact(r.index); err != nil {
diff --git a/raftstore/replica_event_loop.go b/raftstore/replica_event_loop.go
index 066bb0a..1d1b0a8 100644
--- a/raftstore/replica_event_loop.go
+++ b/raftstore/replica_event_loop.go
@@ -253,7 +253,7 @@ func (pr *replica) handleInitializedState() (bool, error) {
 		// should never be empty here
 		panic("unexpected empty snapshot")
 	}
-	index, err := pr.sm.dataStorage.GetPersistentLogIndex(pr.shardID)
+	index, err := pr.getPersistentLogIndex()
 	if err != nil {
 		return false, err
 	}
diff --git a/raftstore/replica_snapshot.go b/raftstore/replica_snapshot.go
index 738d16c..8cc9cf1 100644
--- a/raftstore/replica_snapshot.go
+++ b/raftstore/replica_snapshot.go
@@ -119,9 +119,17 @@ func (r *replica) applySnapshot(ss raftpb.Snapshot) error {
 	// r.replica is more like a local cached copy of the replica record.
 	r.replica = *findReplica(r.getShard(), r.storeID)
 	r.sm.updateAppliedIndexTerm(ss.Metadata.Index, ss.Metadata.Term)
-
+	// persistentLogIndex is not guaranteed to be the same as ss.Metadata.Index
+	// as the log entry at ss.Metadata.Index, including a few nearby entries
+	// are entries not visible to the state machine, e.g. NOOP entries or admin
+	// entries. in such cases, we will have to keep both the ss snapshot record
+	// and its on disk snapshot image.
+	persistentLogIndex, err := r.getPersistentLogIndex()
+	if err != nil {
+		return err
+	}
 	// FIXME: change this to an event worker action
-	if err := r.snapshotCompaction(ss); err != nil {
+	if err := r.snapshotCompaction(ss, persistentLogIndex); err != nil {
 		logger.Error("snapshot compaction failed",
 			zap.Error(err))
 		return err
@@ -138,19 +146,24 @@ func (r *replica) applySnapshot(ss raftpb.Snapshot) error {
 }
 
 // TODO: add a test for snapshotCompaction
-func (r *replica) snapshotCompaction(ss raftpb.Snapshot) error {
+func (r *replica) snapshotCompaction(ss raftpb.Snapshot,
+	persistentLogIndex uint64) error {
 	snapshots, err := r.logdb.GetAllSnapshots(r.shardID)
 	if err != nil {
 		return err
 	}
 	for _, cs := range snapshots {
-		// TODO: determine whether to delete ss based on persistentLogIndex value
 		if cs.Metadata.Index < ss.Metadata.Index {
 			if err := r.removeSnapshot(cs, true); err != nil {
 				return err
 			}
 		}
 	}
+	if persistentLogIndex == ss.Metadata.Index {
+		if err := r.removeSnapshot(ss, false); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 


Test patch:
diff --git a/logdb/logdb_test.go b/logdb/logdb_test.go
index 5d986d2..c781fa1 100644
--- a/logdb/logdb_test.go
+++ b/logdb/logdb_test.go
@@ -69,7 +69,7 @@ func runLogDBTest(t *testing.T, tf func(t *testing.T, db *KVLogDB), fs vfs.FS) {
 	defer leaktest.AfterTest(t)()
 	kv := getTestStorage(fs)
 	defer kv.Close()
-	db := NewKVLogDB(kv, log.GetDefaultZapLogger())
+	db := NewKVLogDB(kv, log.GetPanicZapLogger())
 	tf(t, db)
 }
 
@@ -148,23 +148,53 @@ func TestLogDBGetAllSnapshots(t *testing.T) {
 
 func TestLogDBRemoveSnapshot(t *testing.T) {
 	tf := func(t *testing.T, db *KVLogDB) {
-		rd := raft.Ready{
+		rd1 := raft.Ready{
 			Snapshot: raftpb.Snapshot{
 				Metadata: raftpb.SnapshotMetadata{Index: 100},
 			},
 		}
+		rd2 := raft.Ready{
+			Snapshot: raftpb.Snapshot{
+				Metadata: raftpb.SnapshotMetadata{Index: 200},
+			},
+		}
 		wc := db.NewWorkerContext()
-		if err := db.SaveRaftState(testShardID, 100, rd, wc); err != nil {
+		defer wc.Close()
+		if err := db.SaveRaftState(testShardID, 100, rd1, wc); err != nil {
 			t.Fatalf("failed to save raft state, %v", err)
 		}
-		v, err := db.GetSnapshot(testShardID)
+		wc.Reset()
+		if err := db.SaveRaftState(testShardID, 100, rd2, wc); err != nil {
+			t.Fatalf("failed to save raft state, %v", err)
+		}
+		assert.NoError(t, db.RemoveSnapshot(testShardID, rd1.Snapshot.Metadata.Index))
+
+		v, err := db.GetAllSnapshots(testShardID)
 		assert.NoError(t, err)
-		assert.Equal(t, rd.Snapshot, v)
-		assert.NoError(t, db.RemoveSnapshot(testShardID, rd.Snapshot.Metadata.Index))
+		assert.Equal(t, []raftpb.Snapshot{rd2.Snapshot}, v)
+	}
+	fs := vfs.GetTestFS()
+	runLogDBTest(t, tf, fs)
+}
 
-		v, err = db.GetSnapshot(testShardID)
-		assert.Equal(t, ErrNoSnapshot, err)
-		assert.Empty(t, v)
+func TestLogDBRemovingTheMostRecentSnapshotWillPanic(t *testing.T) {
+	tf := func(t *testing.T, db *KVLogDB) {
+		defer func() {
+			if r := recover(); r == nil {
+				t.Fatalf("panic not triggered")
+			}
+		}()
+		rd1 := raft.Ready{
+			Snapshot: raftpb.Snapshot{
+				Metadata: raftpb.SnapshotMetadata{Index: 100},
+			},
+		}
+		wc := db.NewWorkerContext()
+		defer wc.Close()
+		if err := db.SaveRaftState(testShardID, 100, rd1, wc); err != nil {
+			t.Fatalf("failed to save raft state, %v", err)
+		}
+		db.RemoveSnapshot(testShardID, rd1.Snapshot.Metadata.Index)
 	}
 	fs := vfs.GetTestFS()
 	runLogDBTest(t, tf, fs)
@@ -334,6 +364,12 @@ func TestLogDBRemoveEntriesTo(t *testing.T) {
 		rd := raft.Ready{
 			Entries:   []raftpb.Entry{{Index: 4, Term: 1}, {Index: 5, Term: 1}, {Index: 6, Term: 1}},
 			HardState: raftpb.HardState{Commit: 4, Term: 1, Vote: 2},
+			Snapshot: raftpb.Snapshot{
+				Metadata: raftpb.SnapshotMetadata{
+					Index: 5,
+					Term:  1,
+				},
+			},
 		}
 		wc := db.NewWorkerContext()
 		if err := db.SaveRaftState(testShardID, testReplicaID, rd, wc); err != nil {
@@ -358,3 +394,53 @@ func TestLogDBRemoveEntriesTo(t *testing.T) {
 	fs := vfs.GetTestFS()
 	runLogDBTest(t, tf, fs)
 }
+
+func TestLogDBRemovingTooManyLogEntriesWillPanic(t *testing.T) {
+	tf := func(t *testing.T, db *KVLogDB) {
+		defer func() {
+			if r := recover(); r == nil {
+				t.Fatalf("failed to trigger panic")
+			}
+		}()
+		rd := raft.Ready{
+			Entries: []raftpb.Entry{{Index: 4, Term: 1}, {Index: 5, Term: 1}, {Index: 6, Term: 1}},
+			Snapshot: raftpb.Snapshot{
+				Metadata: raftpb.SnapshotMetadata{
+					Index: 4,
+					Term:  1,
+				},
+			},
+		}
+		wc := db.NewWorkerContext()
+		if err := db.SaveRaftState(testShardID, testReplicaID, rd, wc); err != nil {
+			t.Fatalf("failed to save raft state, %v", err)
+		}
+		if err := db.RemoveEntriesTo(testShardID, testReplicaID, 5); err != nil {
+			t.Fatalf("remove entries to failed, %v", err)
+		}
+	}
+	fs := vfs.GetTestFS()
+	runLogDBTest(t, tf, fs)
+}
+
+func TestLogDBRemoveEntriesWithoutSnapshotWillPanic(t *testing.T) {
+	tf := func(t *testing.T, db *KVLogDB) {
+		defer func() {
+			if r := recover(); r == nil {
+				t.Fatalf("failed to trigger panic")
+			}
+		}()
+		rd := raft.Ready{
+			Entries: []raftpb.Entry{{Index: 4, Term: 1}, {Index: 5, Term: 1}, {Index: 6, Term: 1}},
+		}
+		wc := db.NewWorkerContext()
+		if err := db.SaveRaftState(testShardID, testReplicaID, rd, wc); err != nil {
+			t.Fatalf("failed to save raft state, %v", err)
+		}
+		if err := db.RemoveEntriesTo(testShardID, testReplicaID, 5); err != nil {
+			t.Fatalf("remove entries to failed, %v", err)
+		}
+	}
+	fs := vfs.GetTestFS()
+	runLogDBTest(t, tf, fs)
+}
diff --git a/raftstore/replica_event_loop_test.go b/raftstore/replica_event_loop_test.go
index 26afaa3..730e6b5 100644
--- a/raftstore/replica_event_loop_test.go
+++ b/raftstore/replica_event_loop_test.go
@@ -139,10 +139,11 @@ func TestApplyInitialSnapshot(t *testing.T) {
 		assert.Equal(t, shard, sms[0].Metadata.Shard)
 
 		env := r.snapshotter.getRecoverSnapshotEnv(ss)
-		// latest snapshot is not removed
+		// latest snapshot is removed when the persistentLogIndex matches the
+		// ss.Metadata.Index
 		exist, err := fileutil.Exist(env.GetFinalDir(), fs)
 		assert.NoError(t, err)
-		assert.True(t, exist)
+		assert.False(t, exist)
 		// snapshot record is not removed
 		_, err = r.logdb.GetSnapshot(1)
 		assert.NoError(t, err)
diff --git a/raftstore/replica_event_raft_ready_test.go b/raftstore/replica_event_raft_ready_test.go
index d6fb73a..d30c0ca 100644
--- a/raftstore/replica_event_raft_ready_test.go
+++ b/raftstore/replica_event_raft_ready_test.go
@@ -261,7 +261,7 @@ func TestApplyReceivedSnapshot(t *testing.T) {
 		env := r.snapshotter.getRecoverSnapshotEnv(ss)
 		exist, err := fileutil.Exist(env.GetFinalDir(), fs)
 		assert.NoError(t, err)
-		assert.True(t, exist)
+		assert.False(t, exist)
 		_, err = r.logdb.GetSnapshot(1)
 		assert.NoError(t, err)
 	}
diff --git a/raftstore/replica_snapshot_test.go b/raftstore/replica_snapshot_test.go
index cf1dcda..64cd0a0 100644
--- a/raftstore/replica_snapshot_test.go
+++ b/raftstore/replica_snapshot_test.go
@@ -18,6 +18,7 @@ import (
 
 	"github.com/fagongzi/util/protoc"
 	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/raft/v3/raftpb"
 
 	"github.com/matrixorigin/matrixcube/components/log"
@@ -140,14 +141,14 @@ func TestReplicaSnapshotCanBeApplied(t *testing.T) {
 		r.sm = newStateMachine(r.logger, ds, r.logdb, shard, replicaRec, nil, nil)
 		_, err = r.sm.dataStorage.GetInitialStates()
 		assert.NoError(t, err)
-		persistentLogIndex, err := r.sm.dataStorage.GetPersistentLogIndex(shard.ID)
+		persistentLogIndex, err := r.getPersistentLogIndex()
 		assert.NoError(t, err)
 		assert.Equal(t, uint64(0), persistentLogIndex)
 
 		r.replica = Replica{}
 		assert.NoError(t, r.applySnapshot(ss))
 		// applySnapshot will have the persistentLogIndex value updated
-		persistentLogIndex, err = r.sm.dataStorage.GetPersistentLogIndex(shard.ID)
+		persistentLogIndex, err = r.getPersistentLogIndex()
 		assert.NoError(t, err)
 		assert.Equal(t, uint64(100), persistentLogIndex)
 
@@ -166,7 +167,7 @@ func TestReplicaSnapshotCanBeApplied(t *testing.T) {
 		env := r.snapshotter.getRecoverSnapshotEnv(ss)
 		exist, err := fileutil.Exist(env.GetFinalDir(), fs)
 		assert.NoError(t, err)
-		assert.True(t, exist)
+		assert.False(t, exist)
 	}
 	fs := vfs.GetTestFS()
 	runReplicaSnapshotTest(t, fn, fs)
@@ -229,3 +230,76 @@ func TestCreatingOutOfDateSnapshotWillCausePanic(t *testing.T) {
 	fs := vfs.GetTestFS()
 	runReplicaSnapshotTest(t, fn, fs)
 }
+
+func TestSnapshotCompactionWithMatchedPersistentLogIndex(t *testing.T) {
+	testSnapshotCompaction(t, 200, true)
+}
+
+func TestSnapshotCompactionWithMismatchedPersistentLogIndex(t *testing.T) {
+	testSnapshotCompaction(t, 199, false)
+}
+
+func testSnapshotCompaction(t *testing.T, index uint64, matched bool) {
+	fn := func(t *testing.T, r *replica, fs vfs.FS) {
+		// setup key range and old metadata
+		r.store.updateShardKeyRange(r.getShard().Group, r.getShard())
+		r.aware = newTestShardAware(0)
+		r.aware.Created(r.getShard())
+
+		// update shard
+		replicaRec := Replica{ID: 1, ContainerID: 100}
+		shard := Shard{ID: 1, Replicas: []Replica{replicaRec}, Start: []byte{1}, End: []byte{2}}
+		assert.NoError(t, r.sm.dataStorage.SaveShardMetadata([]meta.ShardMetadata{
+			{ShardID: 1, LogIndex: 100, Metadata: meta.ShardLocalState{Shard: shard}},
+		}))
+		ss1, created, err := r.createSnapshot()
+		if err != nil {
+			t.Fatalf("failed to create snapshot %v", err)
+		}
+		assert.Equal(t, uint64(100), ss1.Metadata.Index)
+		assert.True(t, created)
+
+		assert.NoError(t, r.sm.dataStorage.SaveShardMetadata([]meta.ShardMetadata{
+			{ShardID: 1, LogIndex: 200, Metadata: meta.ShardLocalState{Shard: shard}},
+		}))
+		r.sm.updateAppliedIndexTerm(200, 2)
+		ss2, created, err := r.createSnapshot()
+		if err != nil {
+			t.Fatalf("failed to create snapshot %v", err)
+		}
+		assert.Equal(t, uint64(200), ss2.Metadata.Index)
+		assert.True(t, created)
+
+		rd1 := raft.Ready{Snapshot: ss1}
+		rd2 := raft.Ready{Snapshot: ss2}
+		wc := r.logdb.NewWorkerContext()
+		assert.NoError(t, r.logdb.SaveRaftState(1, 1, rd1, wc))
+		wc.Reset()
+		assert.NoError(t, r.logdb.SaveRaftState(1, 1, rd2, wc))
+		snapshots, err := r.logdb.GetAllSnapshots(r.shardID)
+		assert.NoError(t, err)
+		assert.Equal(t, []raftpb.Snapshot{ss1, ss2}, snapshots)
+
+		env1 := r.snapshotter.getRecoverSnapshotEnv(ss1)
+		env2 := r.snapshotter.getRecoverSnapshotEnv(ss2)
+		assert.True(t, env1.FinalDirExists())
+		assert.True(t, env2.FinalDirExists())
+
+		assert.NoError(t, r.snapshotCompaction(ss2, index))
+
+		// when matched, both snapshot images are suppose to be removed
+		// otherwise, the latest image should be kept
+		// the latest snapshot record is always kept
+		assert.False(t, env1.FinalDirExists())
+		if matched {
+			assert.False(t, env2.FinalDirExists())
+		} else {
+			assert.True(t, env2.FinalDirExists())
+		}
+		snapshots, err = r.logdb.GetAllSnapshots(r.shardID)
+		assert.NoError(t, err)
+		assert.Equal(t, []raftpb.Snapshot{ss2}, snapshots)
+	}
+	fs := vfs.GetTestFS()
+	runReplicaSnapshotTest(t, fn, fs)
+}
diff --git a/raftstore/replica_test.go b/raftstore/replica_test.go
index 3471029..3c48e75 100644
--- a/raftstore/replica_test.go
+++ b/raftstore/replica_test.go
@@ -51,7 +51,7 @@ func TestInitAppliedIndex(t *testing.T) {
 
 	pr, err := newReplica(s, Shard{ID: 1}, Replica{ID: 1000}, "test")
 	assert.NoError(t, err)
-	assert.NoError(t, pr.initAppliedIndex(ds))
+	assert.NoError(t, pr.initAppliedIndex())
 	assert.Equal(t, uint64(0), pr.appliedIndex)
 
 	err = ds.SaveShardMetadata([]meta.ShardMetadata{
@@ -78,7 +78,7 @@ func TestInitAppliedIndex(t *testing.T) {
 	ds.Sync([]uint64{2})
 	pr, err = newReplica(s, Shard{ID: 2}, Replica{ID: 2000}, "test")
 	assert.NoError(t, err)
-	assert.NoError(t, pr.initAppliedIndex(ds))
+	assert.NoError(t, pr.initAppliedIndex())
 	assert.Equal(t, uint64(2), pr.appliedIndex)
 }