Repository: matrixorigin/matrixcube (go)

Issue:
LogReader is incorrectly initialized
There are several issues identified during snapshot related tests with AOE folks - 

1. LogReader range is incorrectly initialized
2. LogReader is not updated once snapshot is applied
3. replica.replica is not updated when applying a snapshot

Pull request:
Fix LogReader initialization in the presence of snapshotting
**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 #514 

**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 1904cc6..3d47638 100644
--- a/logdb/logdb.go
+++ b/logdb/logdb.go
@@ -102,7 +102,7 @@ type LogDB interface {
 		size uint64, shardID uint64, replicaID uint64, low uint64,
 		high uint64, maxSize uint64) ([]raftpb.Entry, uint64, error)
 	// ReadRaftState returns the persistented raft state found in Log DB.
-	ReadRaftState(shardID uint64, replicaID uint64) (RaftState, error)
+	ReadRaftState(shardID uint64, replicaID uint64, snapshotIndex uint64) (RaftState, error)
 	// RemoveEntriesTo removes entries with indexes between (0, index].
 	RemoveEntriesTo(shardID uint64, replicaID uint64, index uint64) error
 	// GetSnapshot returns the most recent snapshot metadata for the specified
@@ -273,26 +273,12 @@ func (l *KVLogDB) IterateEntries(ents []raftpb.Entry,
 	return ents, size, nil
 }
 
-func (l *KVLogDB) ReadRaftState(shardID uint64, replicaID uint64) (RaftState, error) {
-	target := keys.GetRaftLogKey(shardID, 0, nil)
-	key, _, err := l.ms.Seek(target)
-	if err != nil {
-		return RaftState{}, err
-	}
-	if len(key) == 0 || !keys.IsRaftLogKey(key) {
-		return RaftState{}, ErrNoSavedLog
-	}
-	startIndex, err := keys.GetRaftLogIndex(key)
+func (l *KVLogDB) ReadRaftState(shardID uint64,
+	replicaID uint64, snapshotIndex uint64) (RaftState, error) {
+	firstIndex, length, err := l.getRange(shardID, replicaID, snapshotIndex)
 	if err != nil {
 		return RaftState{}, err
 	}
-	maxIndex, err := l.getMaxIndex(shardID, replicaID)
-	if err != nil {
-		return RaftState{}, err
-	}
-	if maxIndex < startIndex {
-		panic("invalid maxIndex or startIndex")
-	}
 
 	var st raftpb.HardState
 	v, err := l.ms.Get(keys.GetHardStateKey(shardID, replicaID, nil))
@@ -303,10 +289,11 @@ func (l *KVLogDB) ReadRaftState(shardID uint64, replicaID uint64) (RaftState, er
 		return RaftState{}, ErrNoSavedLog
 	}
 	protoc.MustUnmarshal(&st, v)
+
 	return RaftState{
 		State:      st,
-		FirstIndex: startIndex,
-		EntryCount: maxIndex - startIndex + 1,
+		FirstIndex: firstIndex,
+		EntryCount: length,
 	}, nil
 }
 
@@ -318,6 +305,38 @@ func (l *KVLogDB) RemoveEntriesTo(shardID uint64, replicaID uint64, index uint64
 	return l.ms.RangeDelete(startKey, endKey, true)
 }
 
+func (l *KVLogDB) getRange(shardID uint64,
+	replicaID uint64, snapshotIndex uint64) (uint64, uint64, error) {
+	maxIndex, err := l.getMaxIndex(shardID, replicaID)
+	if err == ErrNoSavedLog {
+		return snapshotIndex, 0, nil
+	}
+	if err != nil {
+		return 0, 0, err
+	}
+	if snapshotIndex == maxIndex {
+		return snapshotIndex, 0, nil
+	}
+
+	target := keys.GetRaftLogKey(shardID, snapshotIndex, nil)
+	key, _, err := l.ms.Seek(target)
+	if err != nil {
+		return 0, 0, err
+	}
+	if len(key) == 0 || !keys.IsRaftLogKey(key) {
+		return snapshotIndex, 0, nil
+	}
+	startIndex, err := keys.GetRaftLogIndex(key)
+	if err != nil {
+		return 0, 0, err
+	}
+	if startIndex == 0 && maxIndex != 0 {
+		l.logger.Fatal("maxIndex recorded, but no log entry found",
+			zap.Uint64("max-index", maxIndex))
+	}
+	return startIndex, maxIndex - startIndex + 1, nil
+}
+
 func (l *KVLogDB) getMaxIndex(shardID uint64, replicaID uint64) (uint64, error) {
 	v, err := l.ms.Get(keys.GetMaxIndexKey(shardID, nil))
 	if err != nil {
diff --git a/raftstore/replica.go b/raftstore/replica.go
index d6f847e..f653738 100644
--- a/raftstore/replica.go
+++ b/raftstore/replica.go
@@ -16,6 +16,7 @@ package raftstore
 import (
 	"context"
 	"fmt"
+	"math"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -296,14 +297,14 @@ func (pr *replica) getShardID() uint64 {
 // 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)
+	dummySnapshot, err := pr.getLogReaderMarkerState()
 	if err != nil {
 		return err
 	}
-
-	pr.sm.updateAppliedIndexTerm(persistentLogIndex, 0)
-	pr.appliedIndex = persistentLogIndex
-	pr.pushedIndex = persistentLogIndex
+	index, term := dummySnapshot.Metadata.Index, dummySnapshot.Metadata.Term
+	pr.sm.updateAppliedIndexTerm(index, term)
+	pr.appliedIndex = index
+	pr.pushedIndex = index
 	pr.logger.Info("applied index loaded",
 		zap.Uint64("applied-index", pr.appliedIndex))
 	return nil
@@ -343,9 +344,77 @@ func (pr *replica) initConfState() error {
 	return nil
 }
 
+func (pr *replica) getLogReaderMarkerState() (raftpb.Snapshot, error) {
+	// FIXME: this is an ugly hack
+	// the fundamental issue here is that we can't directly tell what is the term
+	// value of the persistentLogIndex and thus couldn't establish an marker point
+	// for the LogReader. getTerm() works around this issue by querying the
+	// underlying LogDB (LogReader.Term() is not ready yet as we are trying to
+	// correctly initialize LogReader). This in turn forces us not to cleanup
+	// all applied raft logs and snapshots.
+	getTerm := func(index uint64) (uint64, error) {
+		if index == 0 {
+			return 0, nil
+		}
+
+		snapshots, err := pr.logdb.GetAllSnapshots(pr.shardID)
+		if err != nil {
+			return 0, err
+		}
+		for _, cs := range snapshots {
+			if cs.Metadata.Index == index {
+				return cs.Metadata.Term, nil
+			}
+		}
+
+		ents, _, err := pr.logdb.IterateEntries(nil, 0, pr.shardID, pr.replicaID,
+			index, index+1, math.MaxUint64)
+		if err != nil {
+			return 0, err
+		}
+		if ents[0].Index != index {
+			panic("unexpected entry")
+		}
+		return ents[0].Term, nil
+	}
+
+	persistentLogIndex, err := pr.sm.dataStorage.GetPersistentLogIndex(pr.shardID)
+	if err != nil {
+		return raftpb.Snapshot{}, err
+	}
+	ss, err := pr.logdb.GetSnapshot(pr.shardID)
+	if err != nil && err != logdb.ErrNoSnapshot {
+		return raftpb.Snapshot{}, err
+	}
+
+	if raft.IsEmptySnap(ss) || ss.Metadata.Index < persistentLogIndex {
+		term, err := getTerm(persistentLogIndex)
+		if err != nil {
+			return raftpb.Snapshot{}, err
+		}
+		return raftpb.Snapshot{
+			Metadata: raftpb.SnapshotMetadata{
+				Index: persistentLogIndex,
+				Term:  term,
+			},
+		}, nil
+	} else if ss.Metadata.Index >= persistentLogIndex {
+		return ss, nil
+	}
+	panic("not suppose to reach here")
+}
+
 // initLogState returns a boolean flag indicating whether this is a new node.
 func (pr *replica) initLogState() (bool, error) {
-	rs, err := pr.logdb.ReadRaftState(pr.shardID, pr.replicaID)
+	dummySnapshot, err := pr.getLogReaderMarkerState()
+	if err != nil {
+		return false, err
+	}
+	if err := pr.lr.ApplySnapshot(dummySnapshot); err != nil {
+		return false, err
+	}
+	rs, err := pr.logdb.ReadRaftState(pr.shardID,
+		pr.replicaID, dummySnapshot.Metadata.Index)
 	if errors.Is(err, logdb.ErrNoSavedLog) {
 		return true, nil
 	}
diff --git a/raftstore/replica_event_proposal.go b/raftstore/replica_event_proposal.go
index 595290a..15361a7 100644
--- a/raftstore/replica_event_proposal.go
+++ b/raftstore/replica_event_proposal.go
@@ -214,7 +214,10 @@ func (pr *replica) proposeConfChange(c batch) bool {
 	}
 
 	if pr.rn.PendingConfIndex() > pr.appliedIndex {
-		pr.logger.Error(ErrPendingConfigChange.Error())
+		pr.logger.Error("pending config change",
+			zap.Error(ErrPendingConfigChange),
+			zap.Uint64("conf-index", pr.rn.PendingConfIndex()),
+			zap.Uint64("applied-index", pr.appliedIndex))
 		c.respOtherError(ErrPendingConfigChange)
 		return false
 	}
diff --git a/raftstore/replica_snapshot.go b/raftstore/replica_snapshot.go
index aa6d5d9..d6af842 100644
--- a/raftstore/replica_snapshot.go
+++ b/raftstore/replica_snapshot.go
@@ -101,8 +101,13 @@ func (r *replica) applySnapshot(ss raftpb.Snapshot) error {
 			zap.Error(err))
 		return err
 	}
+	r.appliedIndex = ss.Metadata.Index
+	r.lr.ApplySnapshot(ss)
 	r.sm.updateShard(md.Metadata.Shard)
+	// 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)
+
 	// FIXME: change this to an event worker action
 	if err := r.snapshotCompaction(ss); err != nil {
 		logger.Error("snapshot compaction failed",
@@ -141,6 +146,8 @@ func (r *replica) removeSnapshot(ss raftpb.Snapshot, removeFromLogDB bool) error
 	}
 	env := r.snapshotter.getRecoverSnapshotEnv(ss)
 	if env.FinalDirExists() {
+		r.logger.Info("removing snapshot dir",
+			zap.String("dir", env.GetFinalDir()))
 		if err := env.RemoveFinalDir(); err != nil {
 			logger.Error("failed to remove snapshot final directory",
 				zap.Error(err))


Test patch:
diff --git a/logdb/logdb_test.go b/logdb/logdb_test.go
index a24c1aa..87b8e57 100644
--- a/logdb/logdb_test.go
+++ b/logdb/logdb_test.go
@@ -241,7 +241,7 @@ func TestLogDBGetMaxIndex(t *testing.T) {
 
 func TestLogDBReadRaftStateReturnsNoSavedLogErrorWhenStateIsNeverSaved(t *testing.T) {
 	tf := func(t *testing.T, db *KVLogDB) {
-		if _, err := db.ReadRaftState(testShardID, testReplicaID); !errors.Is(err, ErrNoSavedLog) {
+		if _, err := db.ReadRaftState(testShardID, testReplicaID, 0); !errors.Is(err, ErrNoSavedLog) {
 			t.Fatalf("failed to return the expected error")
 		}
 	}
@@ -259,7 +259,7 @@ func TestLogDBSaveRaftState(t *testing.T) {
 		if err := db.SaveRaftState(testShardID, testReplicaID, rd, wc); err != nil {
 			t.Fatalf("failed to save raft state, %v", err)
 		}
-		rs, err := db.ReadRaftState(testShardID, testReplicaID)
+		rs, err := db.ReadRaftState(testShardID, testReplicaID, 0)
 		if errors.Is(err, ErrNoSavedLog) {
 			t.Fatalf("failed to get raft state, %v", err)
 		}
diff --git a/raftstore/replica_create_test.go b/raftstore/replica_create_test.go
index 1b0ee34..39c0008 100644
--- a/raftstore/replica_create_test.go
+++ b/raftstore/replica_create_test.go
@@ -60,7 +60,7 @@ func testShardCreateWithSaveMetadataWithSync(t *testing.T, sync bool) {
 		assert.True(t, s.DataStorageByGroup(0).(storage.StatsKeeper).Stats().SyncCount == 0)
 	}
 
-	stat, err := s.logdb.ReadRaftState(1, 2)
+	stat, err := s.logdb.ReadRaftState(1, 2, 0)
 	assert.NoError(t, err)
 	assert.Equal(t, uint64(1), stat.FirstIndex)
 	assert.Equal(t, uint64(1), stat.EntryCount)
diff --git a/raftstore/replica_event_loop_test.go b/raftstore/replica_event_loop_test.go
index e5c27e3..44d5191 100644
--- a/raftstore/replica_event_loop_test.go
+++ b/raftstore/replica_event_loop_test.go
@@ -117,8 +117,8 @@ func TestApplyInitialSnapshot(t *testing.T) {
 		base := kv.NewBaseStorage(dsMem, fs)
 		ds := kv.NewKVDataStorage(base, nil)
 		defer ds.Close()
-		shard := Shard{ID: 1}
-		replicaRec := Replica{ID: 1}
+		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)
@@ -128,7 +128,7 @@ func TestApplyInitialSnapshot(t *testing.T) {
 		assert.True(t, r.initialized)
 		assert.Equal(t, ss.Metadata.Index, r.sm.metadataMu.index)
 		assert.Equal(t, ss.Metadata.Term, r.sm.metadataMu.term)
-		assert.Equal(t, Shard{ID: 1}, r.sm.metadataMu.shard)
+		assert.Equal(t, shard, r.sm.metadataMu.shard)
 
 		sms, err := r.sm.dataStorage.GetInitialStates()
 		assert.NoError(t, err)
diff --git a/raftstore/replica_event_raft_ready_test.go b/raftstore/replica_event_raft_ready_test.go
index 9e5a061..4ebc5c1 100644
--- a/raftstore/replica_event_raft_ready_test.go
+++ b/raftstore/replica_event_raft_ready_test.go
@@ -223,7 +223,7 @@ func TestIssue386(t *testing.T) {
 		HardState: raftpb.HardState{Commit: 1},
 	}, s.logdb.NewWorkerContext()))
 
-	rs, err := s.logdb.ReadRaftState(1, 2)
+	rs, err := s.logdb.ReadRaftState(1, 2, 0)
 	assert.NoError(t, err)
 	assert.Equal(t, uint64(1), rs.State.Commit)
 }
@@ -248,8 +248,8 @@ func TestApplyReceivedSnapshot(t *testing.T) {
 		base := kv.NewBaseStorage(dsMem, fs)
 		ds := kv.NewKVDataStorage(base, nil)
 		defer ds.Close()
-		shard := Shard{ID: 1}
-		replicaRec := Replica{ID: 1}
+		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)
 
 		rd := raft.Ready{Snapshot: ss}
@@ -257,7 +257,7 @@ func TestApplyReceivedSnapshot(t *testing.T) {
 		assert.NoError(t, r.processReady(rd, r.logdb.NewWorkerContext()))
 		assert.Equal(t, ss.Metadata.Index, r.sm.metadataMu.index)
 		assert.Equal(t, ss.Metadata.Term, r.sm.metadataMu.term)
-		assert.Equal(t, Shard{ID: 1}, r.sm.metadataMu.shard)
+		assert.Equal(t, shard, r.sm.metadataMu.shard)
 
 		sms, err := r.sm.dataStorage.GetInitialStates()
 		assert.NoError(t, err)
diff --git a/raftstore/replica_snapshot_test.go b/raftstore/replica_snapshot_test.go
index dcfbba9..48145d2 100644
--- a/raftstore/replica_snapshot_test.go
+++ b/raftstore/replica_snapshot_test.go
@@ -50,8 +50,8 @@ func runReplicaSnapshotTest(t *testing.T,
 		return fp
 	}
 	snapshotter := newSnapshotter(1, 1, logger, replicaSnapshotDir, ldb, fs)
-	shard := Shard{ID: 1}
-	replicaRec := Replica{ID: 1}
+	replicaRec := Replica{ID: 1, ContainerID: 100}
+	shard := Shard{ID: 1, Replicas: []Replica{replicaRec}}
 	dsMem := mem.NewStorage()
 	base := kv.NewBaseStorage(dsMem, fs)
 	ds := kv.NewKVDataStorage(base, nil)
@@ -65,6 +65,7 @@ func runReplicaSnapshotTest(t *testing.T,
 	sm := newStateMachine(logger, ds, ldb, shard, replicaRec, nil, nil)
 	sm.updateAppliedIndexTerm(100, 1)
 	r := &replica{
+		storeID:     100,
 		logger:      logger,
 		logdb:       ldb,
 		sm:          sm,
@@ -132,14 +133,16 @@ func TestReplicaSnapshotCanBeApplied(t *testing.T) {
 		base := kv.NewBaseStorage(dsMem, fs)
 		ds := kv.NewKVDataStorage(base, nil)
 		defer ds.Close()
-		shard := Shard{ID: 1}
-		replicaRec := Replica{ID: 1}
+		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)
 
+		r.replica = Replica{}
 		assert.NoError(t, r.applySnapshot(ss))
+		assert.Equal(t, r.replica, replicaRec)
 		assert.Equal(t, ss.Metadata.Index, r.sm.metadataMu.index)
 		assert.Equal(t, ss.Metadata.Term, r.sm.metadataMu.term)
-		assert.Equal(t, Shard{ID: 1}, r.sm.metadataMu.shard)
+		assert.Equal(t, shard, r.sm.metadataMu.shard)
 
 		sms, err := r.sm.dataStorage.GetInitialStates()
 		assert.NoError(t, err)
diff --git a/raftstore/replica_state_machine_exec_test.go b/raftstore/replica_state_machine_exec_test.go
index 5c8bb0f..f44f8b7 100644
--- a/raftstore/replica_state_machine_exec_test.go
+++ b/raftstore/replica_state_machine_exec_test.go
@@ -347,7 +347,7 @@ func TestDoExecCompactLog(t *testing.T) {
 	assert.NotNil(t, result.adminResult)
 	pr.lr.SetRange(0, 100)
 	pr.handleApplyResult(result)
-	state, err := pr.sm.logdb.ReadRaftState(pr.shardID, pr.replicaID)
+	state, err := pr.sm.logdb.ReadRaftState(pr.shardID, pr.replicaID, 0)
 	assert.NoError(t, err)
 	assert.Equal(t, uint64(3), state.FirstIndex)
 	assert.Equal(t, uint64(3), state.EntryCount)
diff --git a/raftstore/replica_test.go b/raftstore/replica_test.go
index eea923a..d9f63d7 100644
--- a/raftstore/replica_test.go
+++ b/raftstore/replica_test.go
@@ -17,9 +17,15 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"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/mem"
 	"github.com/matrixorigin/matrixcube/util/stop"
-	"github.com/stretchr/testify/assert"
 )
 
 type testReplicaGetter struct {
@@ -51,7 +57,26 @@ func TestInitAppliedIndex(t *testing.T) {
 	assert.NoError(t, pr.initAppliedIndex(ds))
 	assert.Equal(t, uint64(0), pr.appliedIndex)
 
-	err = ds.SaveShardMetadata([]meta.ShardMetadata{{ShardID: 2, LogIndex: 2, Metadata: meta.ShardLocalState{Shard: meta.Shard{ID: 2}}}})
+	err = ds.SaveShardMetadata([]meta.ShardMetadata{
+		{
+			ShardID:  2,
+			LogIndex: 2,
+			Metadata: meta.ShardLocalState{
+				Shard: meta.Shard{
+					ID: 2,
+				},
+			},
+		},
+	})
+	rd := raft.Ready{
+		Snapshot: raftpb.Snapshot{
+			Metadata: raftpb.SnapshotMetadata{
+				Index: 2,
+				Term:  100,
+			},
+		},
+	}
+	assert.NoError(t, pr.logdb.SaveRaftState(2, 2000, rd, pr.logdb.NewWorkerContext()))
 	assert.NoError(t, err)
 	ds.Sync([]uint64{2})
 	pr, err = newReplica(s, Shard{ID: 2}, Replica{ID: 2000}, "test")
@@ -66,3 +91,68 @@ func newTestReplica(shard Shard, peer Replica, s *store) *replica {
 	pr.setStarted()
 	return pr
 }
+
+func TestGetLogReaderMarkerState(t *testing.T) {
+	tests := []struct {
+		snapshotIndex      uint64
+		snapshotTerm       uint64
+		hasSnapshotRec     bool
+		persistentLogIndex uint64
+		persistentLogTerm  uint64
+		hasLogEntry        bool
+		resultIndex        uint64
+		resultTerm         uint64
+		err                error
+	}{
+		{0, 0, false, 0, 0, false, 0, 0, nil},
+		{100, 2, true, 50, 1, true, 100, 2, nil},
+		{100, 2, true, 0, 0, false, 100, 2, nil},
+		{100, 2, true, 100, 2, true, 100, 2, nil},
+		{100, 2, true, 200, 3, true, 200, 3, nil},
+		{0, 0, false, 50, 1, true, 50, 1, nil},
+		{0, 0, false, 50, 1, false, 0, 0, raft.ErrUnavailable},
+	}
+
+	shardID := uint64(1)
+	replicaID := uint64(1)
+	for _, tt := range tests {
+		func() {
+			logger := log.GetDefaultZapLogger()
+			memKV := mem.NewStorage()
+			defer memKV.Close()
+			logdb := logdb.NewKVLogDB(memKV, logger)
+			if tt.snapshotIndex > 0 && tt.hasSnapshotRec {
+				rd := raft.Ready{
+					Snapshot: raftpb.Snapshot{
+						Metadata: raftpb.SnapshotMetadata{
+							Index: tt.snapshotIndex,
+							Term:  tt.snapshotTerm,
+						},
+					},
+				}
+				assert.NoError(t, logdb.SaveRaftState(1, 1, rd, logdb.NewWorkerContext()))
+			}
+			if tt.persistentLogIndex > 0 && tt.hasLogEntry {
+				rd := raft.Ready{
+					Entries: []raftpb.Entry{{Index: tt.persistentLogIndex, Term: tt.persistentLogTerm}},
+				}
+				assert.NoError(t, logdb.SaveRaftState(1, 1, rd, logdb.NewWorkerContext()))
+			}
+			ds := &testDataStorage{
+				persistentLogIndex: tt.persistentLogIndex,
+			}
+			pr := &replica{
+				logdb:     logdb,
+				shardID:   shardID,
+				replicaID: replicaID,
+				sm: &stateMachine{
+					dataStorage: ds,
+				},
+			}
+			ss, err := pr.getLogReaderMarkerState()
+			assert.Equal(t, tt.err, err)
+			assert.Equal(t, tt.resultIndex, ss.Metadata.Index)
+			assert.Equal(t, tt.resultTerm, ss.Metadata.Term)
+		}()
+	}
+}