Repository: matrixorigin/matrixcube (go)

Issue:
Gap in log entries error when restarting MatrixCube
```
{"level":"FATAL","time":"2021/12/16 15:35:55.868399 +0800","name":"cube.store","caller":"raftstore/log_reader.go:333","msg":"gap in log entries","store":"node0","store-id":1,"shard-id":5,"replica-id":9,"id":"logreader [00005:00009] index 1571 term 3 length 1","marker":1571,"lr.len":1,"first":2250,"len":39}
```

Pull request description:
Never remove the latest snapshot record from LogDB
**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 #556 

**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/raftstore/replica_event_loop.go b/raftstore/replica_event_loop.go
index 3da69c5..7df0fa5 100644
--- a/raftstore/replica_event_loop.go
+++ b/raftstore/replica_event_loop.go
@@ -252,18 +252,28 @@ func (pr *replica) handleInitializedState() (bool, error) {
 		// should never be empty here
 		panic("unexpected empty snapshot")
 	}
-	index, _ := pr.sm.getAppliedIndexTerm()
+	index, err := pr.sm.dataStorage.GetPersistentLogIndex(pr.shardID)
+	if err != nil {
+		return false, err
+	}
 	pr.logger.Info("initial snapshot available",
-		zap.Uint64("applied-index", index),
+		zap.Uint64("persistent-log-index", index),
 		zap.Uint64("snapshot-index", ss.Metadata.Index))
 	if ss.Metadata.Index > index {
+		pr.logger.Info("applying initial snapshot",
+			zap.Uint64("index", ss.Metadata.Index))
 		if err := pr.applySnapshot(ss); err != nil {
 			return false, err
 		}
 		pr.pushedIndex = ss.Metadata.Index
 	} else {
-		// snapshot is out of date, just remove it
-		if err := pr.removeSnapshot(ss, true); err != nil {
+		// snapshot is out of date, remove the disk image as we will never apply
+		// the snapshot in the future. keep the logdb record as LogReader needs
+		// this snapshot record to establish the starting point of the persisted
+		// raft log entries.
+		pr.logger.Info("skipped applying initial snapshot",
+			zap.Uint64("index", ss.Metadata.Index))
+		if err := pr.removeSnapshot(ss, false); err != nil {
 			return false, err
 		}
 	}
diff --git a/raftstore/store_bootstrap.go b/raftstore/store_bootstrap.go
index 7993b54..8f168ca 100644
--- a/raftstore/store_bootstrap.go
+++ b/raftstore/store_bootstrap.go
@@ -67,6 +67,7 @@ func (s *store) doBootstrapCluster(bootstrap bool) {
 	s.initMeta()
 
 	if s.mustLoadStoreMetadata() {
+		s.mustPutStore()
 		return
 	}
 
@@ -127,13 +128,16 @@ func (s *store) doBootstrapCluster(bootstrap bool) {
 		}
 	}
 
+	s.mustPutStore()
+	s.startHandleResourceHeartbeat()
+}
+
+func (s *store) mustPutStore() {
 	if err := s.pd.GetClient().PutContainer(s.meta); err != nil {
 		s.logger.Fatal("failed to put container to prophet",
 			s.storeField(),
 			zap.Error(err))
 	}
-
-	s.startHandleResourceHeartbeat()
 }
 
 func (s *store) mustSaveStoreMetadata() {


Test patch:
diff --git a/raftstore/replica_event_loop_test.go b/raftstore/replica_event_loop_test.go
index 44d5191..26afaa3 100644
--- a/raftstore/replica_event_loop_test.go
+++ b/raftstore/replica_event_loop_test.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/matrixorigin/matrixcube/components/log"
 	"github.com/matrixorigin/matrixcube/logdb"
+	"github.com/matrixorigin/matrixcube/pb/meta"
 	"github.com/matrixorigin/matrixcube/pb/rpc"
 	"github.com/matrixorigin/matrixcube/storage"
 	"github.com/matrixorigin/matrixcube/storage/kv"
@@ -117,6 +118,8 @@ func TestApplyInitialSnapshot(t *testing.T) {
 		base := kv.NewBaseStorage(dsMem, fs)
 		ds := kv.NewKVDataStorage(base, nil)
 		defer ds.Close()
+		_, err = ds.GetInitialStates()
+		assert.NoError(t, err)
 		replicaRec := Replica{ID: 1, ContainerID: 100}
 		shard := Shard{ID: 1, Replicas: []Replica{replicaRec}}
 		r.sm = newStateMachine(r.logger, ds, r.logdb, shard, replicaRec, nil, nil)
@@ -140,6 +143,7 @@ func TestApplyInitialSnapshot(t *testing.T) {
 		exist, err := fileutil.Exist(env.GetFinalDir(), fs)
 		assert.NoError(t, err)
 		assert.True(t, exist)
+		// snapshot record is not removed
 		_, err = r.logdb.GetSnapshot(1)
 		assert.NoError(t, err)
 	}
@@ -147,6 +151,57 @@ func TestApplyInitialSnapshot(t *testing.T) {
 	runReplicaSnapshotTest(t, fn, fs)
 }
 
+func TestInitialSnapshotRecordIsNeverRemoved(t *testing.T) {
+	fn := func(t *testing.T, r *replica, fs vfs.FS) {
+		ss, created, err := r.createSnapshot()
+		if err != nil {
+			t.Fatalf("failed to create snapshot %v", err)
+		}
+		assert.Equal(t, uint64(100), ss.Metadata.Index)
+		assert.True(t, created)
+
+		rd := raft.Ready{Snapshot: ss}
+		assert.NoError(t, r.logdb.SaveRaftState(1, 1, rd, r.logdb.NewWorkerContext()))
+		// reset the data storage
+		dsMem := mem.NewStorage()
+		base := kv.NewBaseStorage(dsMem, fs)
+		ds := kv.NewKVDataStorage(base, nil)
+		defer ds.Close()
+
+		assert.NoError(t, ds.SaveShardMetadata([]meta.ShardMetadata{
+			{
+				ShardID:  1,
+				LogIndex: 101,
+				Metadata: meta.ShardLocalState{
+					Shard: Shard{ID: 1},
+				},
+			},
+		}))
+		ds.Sync(nil)
+		_, err = ds.GetInitialStates()
+		assert.NoError(t, err)
+
+		replicaRec := Replica{ID: 1, ContainerID: 100}
+		shard := Shard{ID: 1, Replicas: []Replica{replicaRec}}
+		r.sm = newStateMachine(r.logger, ds, r.logdb, shard, replicaRec, nil, nil)
+		assert.False(t, r.initialized)
+		_, err = r.handleEvent(r.logdb.NewWorkerContext())
+		assert.NoError(t, err)
+		assert.True(t, r.initialized)
+		// snapshot record is not removed
+		ss, err = r.logdb.GetSnapshot(1)
+		assert.NoError(t, err)
+		assert.Equal(t, uint64(100), ss.Metadata.Index)
+		env := r.snapshotter.getRecoverSnapshotEnv(ss)
+		// snapshot image is removed
+		exist, err := fileutil.Exist(env.GetFinalDir(), fs)
+		assert.NoError(t, err)
+		assert.False(t, exist)
+	}
+	fs := vfs.GetTestFS()
+	runReplicaSnapshotTest(t, fn, fs)
+}
+
 func TestDoCheckCompactLog(t *testing.T) {
 	defer leaktest.AfterTest(t)()