From 34a468de36d63eb95c3c5e9366cbbeb28690346f Mon Sep 17 00:00:00 2001
From: Yicheng Qin <qycqycqycqycqyc@gmail.com>
Date: Tue, 2 Dec 2014 23:57:02 -0800
Subject: [PATCH] raft: return 0 for term of compacted index

It is necessary to make this check because of the following case:

1. memory storage contains ents from index 0 to 50, and unstable has
ents from index 50 to 60.
2. raft receives an incoming snapshot with index 100.
3. raft restores its unstable to 100, but has not applied snapshot on memory storage.
4. raft receives an out-dated MsgApp from index 60.
5. raft finds the term of index 60 to check the match.
6. raft asks memory storage about the term of index 60 after it failed to get
it from unstable.
7. memory storage panics because it knows nothing about index 60.
---
 raft/log.go      |  4 +++-
 raft/log_test.go | 34 ++++++++++++++++++++++++++++++++--
 2 files changed, 35 insertions(+), 3 deletions(-)

diff --git a/raft/log.go b/raft/log.go
index 6cb1d851d..5d2e63596 100644
--- a/raft/log.go
+++ b/raft/log.go
@@ -199,7 +199,9 @@ func (l *raftLog) stableSnapTo(i uint64) { l.unstable.stableSnapTo(i) }
 func (l *raftLog) lastTerm() uint64 { return l.term(l.lastIndex()) }
 
 func (l *raftLog) term(i uint64) uint64 {
-	if i > l.lastIndex() {
+	// the valid term range is [index of dummy entry, last index]
+	dummyIndex := l.firstIndex() - 1
+	if i < dummyIndex || i > l.lastIndex() {
 		return 0
 	}
 
diff --git a/raft/log_test.go b/raft/log_test.go
index 609654c60..f85a4f9d5 100644
--- a/raft/log_test.go
+++ b/raft/log_test.go
@@ -634,7 +634,7 @@ func TestTerm(t *testing.T) {
 	num := uint64(100)
 
 	storage := NewMemoryStorage()
-	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: offset}})
+	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: offset, Term: 1}})
 	l := newLog(storage)
 	for i = 1; i < num; i++ {
 		l.append(pb.Entry{Index: offset + i, Term: i})
@@ -645,7 +645,7 @@ func TestTerm(t *testing.T) {
 		w     uint64
 	}{
 		{offset - 1, 0},
-		{offset, 0},
+		{offset, 1},
 		{offset + num/2, num / 2},
 		{offset + num - 1, num - 1},
 		{offset + num, 0},
@@ -659,6 +659,36 @@ func TestTerm(t *testing.T) {
 	}
 }
 
+func TestTermWithUnstableSnapshot(t *testing.T) {
+	storagesnapi := uint64(100)
+	unstablesnapi := storagesnapi + 5
+
+	storage := NewMemoryStorage()
+	storage.ApplySnapshot(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: storagesnapi, Term: 1}})
+	l := newLog(storage)
+	l.restore(pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: unstablesnapi, Term: 1}})
+
+	tests := []struct {
+		index uint64
+		w     uint64
+	}{
+		// cannot get term from storage
+		{storagesnapi, 0},
+		// cannot get term from the gap between storage ents and unstable snapshot
+		{storagesnapi + 1, 0},
+		{unstablesnapi - 1, 0},
+		// get term from unstable snapshot index
+		{unstablesnapi, 1},
+	}
+
+	for i, tt := range tests {
+		term := l.term(tt.index)
+		if !reflect.DeepEqual(term, tt.w) {
+			t.Errorf("#%d: at = %d, want %d", i, term, tt.w)
+		}
+	}
+}
+
 func TestSlice(t *testing.T) {
 	var i uint64
 	offset := uint64(100)