Repository: matrixorigin/matrixcube (go)

Issue:
Limit what committed raft log entries to apply
Need to skip committed raft log entries that have been previously covered in a snapshot. 

Pull request description:
Fix #488 limit what committed raft log entries to apply
**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 #488 

**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.go b/raftstore/replica.go
index 5645095..dc490b2 100644
--- a/raftstore/replica.go
+++ b/raftstore/replica.go
@@ -86,8 +86,11 @@ type replica struct {
 	actions              *task.Queue
 	items                []interface{}
 	appliedIndex         uint64
-	stats                *replicaStats
-	metrics              localMetrics
+	// pushedIndex is the log index that has been passed to the state machine to
+	// be applied
+	pushedIndex uint64
+	stats       *replicaStats
+	metrics     localMetrics
 
 	initialized bool
 	closedC     chan struct{}
@@ -291,6 +294,7 @@ func (pr *replica) initAppliedIndex(storage storage.DataStorage) error {
 
 	pr.sm.updateAppliedIndexTerm(persistentLogIndex, 0)
 	pr.appliedIndex = persistentLogIndex
+	pr.pushedIndex = persistentLogIndex
 	pr.logger.Info("applied index loaded",
 		zap.Uint64("applied-index", pr.appliedIndex))
 	return nil
diff --git a/raftstore/replica_event_raft_ready.go b/raftstore/replica_event_raft_ready.go
index 2f61074..94f015b 100644
--- a/raftstore/replica_event_raft_ready.go
+++ b/raftstore/replica_event_raft_ready.go
@@ -152,11 +152,35 @@ func (pr *replica) saveRaftState(rd raft.Ready, wc *logdb.WorkerContext) error {
 	return nil
 }
 
+func (pr *replica) entriesToApply(entries []raftpb.Entry) []raftpb.Entry {
+	if len(entries) == 0 {
+		return entries
+	}
+	lastIndex := entries[len(entries)-1].Index
+	firstIndex := entries[0].Index
+	if lastIndex <= pr.pushedIndex {
+		pr.logger.Fatal("all entries older than current state",
+			zap.Uint64("firstIndex", firstIndex),
+			zap.Uint64("lastIndex", lastIndex),
+			zap.Uint64("expected", pr.pushedIndex+1))
+	}
+	if firstIndex > pr.pushedIndex+1 {
+		pr.logger.Fatal("entry hole found",
+			zap.Uint64("firstIndex", firstIndex),
+			zap.Uint64("expected", pr.pushedIndex+1))
+	}
+	if pr.pushedIndex-firstIndex+1 < uint64(len(entries)) {
+		return entries[pr.pushedIndex-firstIndex+1:]
+	}
+	return []raftpb.Entry{}
+}
+
 func (pr *replica) applyCommittedEntries(rd raft.Ready) error {
 	if !raft.IsEmptySnap(rd.Snapshot) {
 		if err := pr.applySnapshot(rd.Snapshot); err != nil {
 			return err
 		}
+		pr.pushedIndex = rd.Snapshot.Metadata.Index
 		pr.logger.Info("snapshot applied into the replica")
 	}
 	for _, entry := range rd.CommittedEntries {
@@ -289,10 +313,14 @@ func (pr *replica) updateMessageMetrics(msg raftpb.Message) {
 }
 
 func (pr *replica) doApplyCommittedEntries(entries []raftpb.Entry) error {
-	pr.sm.applyCommittedEntries(entries)
-	if pr.sm.isRemoved() {
-		// local replica is removed, keep the shard
-		pr.store.destroyReplica(pr.shardID, false, true, "removed by config change")
+	entries = pr.entriesToApply(entries)
+	if len(entries) > 0 {
+		pr.pushedIndex = entries[len(entries)-1].Index
+		pr.sm.applyCommittedEntries(entries)
+		if pr.sm.isRemoved() {
+			// local replica is removed, keep the shard
+			pr.store.destroyReplica(pr.shardID, false, true, "removed by config change")
+		}
 	}
 	return nil
 }


Test patch:
diff --git a/raftstore/integration_snapshot_test.go b/raftstore/integration_snapshot_test.go
index a6e151a..254bea3 100644
--- a/raftstore/integration_snapshot_test.go
+++ b/raftstore/integration_snapshot_test.go
@@ -71,6 +71,15 @@ func TestCompactionAndSnapshot(t *testing.T) {
 	assert.NoError(t, kv.Set("k3", "v3", snapshotTestTimeout))
 	assert.NoError(t, kv.Set("k4", "v4", snapshotTestTimeout))
 
+	// sync the data storage to move the persistent log index.
+	// otherwise compaction request at index 3 will be ignored
+	for i := 0; i < 3; i++ {
+		store := c.GetStore(i).(*store)
+		if pr := store.getReplica(shardID, true); pr != nil {
+			pr.sm.dataStorage.Sync([]uint64{shardID})
+		}
+	}
+
 	compactionCompleted := false
 	for i := 0; i < 2; i++ {
 		store := c.GetStore(i).(*store)
@@ -86,11 +95,13 @@ func TestCompactionAndSnapshot(t *testing.T) {
 				}
 				panic(err)
 			}
-
 			assert.True(t, hasLog(2))
+
 			pr.addAdminRequest(rpc.AdminCmdType_CompactLog, &rpc.CompactLogRequest{
 				CompactIndex: 3,
 			})
+
+			// wait for compaction to complete
 			for i := 0; i < 10; i++ {
 				if hasLog(2) {
 					time.Sleep(time.Second)
diff --git a/raftstore/replica_event_raft_ready_test.go b/raftstore/replica_event_raft_ready_test.go
index db0d0a8..5fadef8 100644
--- a/raftstore/replica_event_raft_ready_test.go
+++ b/raftstore/replica_event_raft_ready_test.go
@@ -22,6 +22,7 @@ import (
 	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/raft/v3/raftpb"
 
+	"github.com/matrixorigin/matrixcube/components/log"
 	"github.com/matrixorigin/matrixcube/logdb"
 	"github.com/matrixorigin/matrixcube/pb/meta"
 	"github.com/matrixorigin/matrixcube/storage/kv"
@@ -274,3 +275,46 @@ func TestApplyReceivedSnapshot(t *testing.T) {
 	fs := vfs.GetTestFS()
 	runReplicaSnapshotTest(t, fn, fs)
 }
+
+func TestEntriesToApply(t *testing.T) {
+	tests := []struct {
+		inputIndex   uint64
+		inputLength  uint64
+		crash        bool
+		resultIndex  uint64
+		resultLength uint64
+		pushedIndex  uint64
+	}{
+		{1, 5, true, 0, 0, 10},
+		{1, 10, false, 0, 0, 10},
+		{1, 11, false, 11, 1, 10},
+		{1, 20, false, 11, 10, 10},
+		{10, 6, false, 11, 5, 10},
+		{11, 5, false, 11, 5, 10},
+		{12, 5, true, 0, 0, 10},
+	}
+	for idx, tt := range tests {
+		func() {
+			defer func() {
+				if r := recover(); r == nil && tt.crash {
+					t.Fatalf("%d, didn't panic", idx)
+				}
+			}()
+			inputs := make([]raftpb.Entry, 0)
+			for i := tt.inputIndex; i < tt.inputIndex+tt.inputLength; i++ {
+				inputs = append(inputs, raftpb.Entry{Index: i})
+			}
+			r := &replica{pushedIndex: tt.pushedIndex, logger: log.GetPanicZapLogger()}
+			results := r.entriesToApply(inputs)
+			if uint64(len(results)) != tt.resultLength {
+				t.Errorf("%d, result len %d, want %d", idx, len(results), tt.resultLength)
+			}
+			if len(results) > 0 {
+				if results[0].Index != tt.resultIndex {
+					t.Errorf("%d, first result index %d, want %d",
+						idx, results[0].Index, tt.resultIndex)
+				}
+			}
+		}()
+	}
+}