From f935290bbc178bc5a40a4458f15723260d5e3ed8 Mon Sep 17 00:00:00 2001
From: Xiang Li <xiangli.cs@gmail.com>
Date: Wed, 12 Oct 2016 10:42:55 -0700
Subject: [PATCH] mvcc: fix rev inconsistency

Try:

./etcdctl put foo bar
./etcdctl del foo
./etcdctl compact 3

restart etcd

./etcdctl get foo
mvcc: required revision has been compacted

The error is unexpected when range over the head revision.

Internally, we incorrectly set current revision smaller than the
compacted revision when we remove all keys around compacted revision.

This commit fixes the issue by recovering the current revision at least
to compacted revision.
---
 mvcc/kvstore.go                 |  7 ++++++
 mvcc/kvstore_compaction_test.go | 40 +++++++++++++++++++++++++++++++++
 2 files changed, 47 insertions(+)

diff --git a/mvcc/kvstore.go b/mvcc/kvstore.go
index adc1fb766..6819c69a1 100644
--- a/mvcc/kvstore.go
+++ b/mvcc/kvstore.go
@@ -408,6 +408,13 @@ func (s *store) restore() error {
 		s.currentRev = rev
 	}
 
+	// keys in the range [compacted revision -N, compaction] might all be deleted due to compaction.
+	// the correct revision should be set to compaction revision in the case, not the largest revision
+	// we have seen.
+	if s.currentRev.main < s.compactMainRev {
+		s.currentRev.main = s.compactMainRev
+	}
+
 	for key, lid := range keyToLease {
 		if s.le == nil {
 			panic("no lessor to attach lease")
diff --git a/mvcc/kvstore_compaction_test.go b/mvcc/kvstore_compaction_test.go
index 91c1d74c7..a631ed1ff 100644
--- a/mvcc/kvstore_compaction_test.go
+++ b/mvcc/kvstore_compaction_test.go
@@ -15,8 +15,10 @@
 package mvcc
 
 import (
+	"os"
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/coreos/etcd/lease"
 	"github.com/coreos/etcd/mvcc/backend"
@@ -93,3 +95,41 @@ func TestScheduleCompaction(t *testing.T) {
 		cleanup(s, b, tmpPath)
 	}
 }
+
+func TestCompactAllAndRestore(t *testing.T) {
+	b, tmpPath := backend.NewDefaultTmpBackend()
+	s0 := NewStore(b, &lease.FakeLessor{}, nil)
+	defer os.Remove(tmpPath)
+
+	s0.Put([]byte("foo"), []byte("bar"), lease.NoLease)
+	s0.Put([]byte("foo"), []byte("bar1"), lease.NoLease)
+	s0.Put([]byte("foo"), []byte("bar2"), lease.NoLease)
+	s0.DeleteRange([]byte("foo"), nil)
+
+	rev := s0.Rev()
+	// compact all keys
+	done, err := s0.Compact(rev)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout waiting for compaction to finish")
+	}
+
+	err = s0.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s1 := NewStore(b, &lease.FakeLessor{}, nil)
+	if s1.Rev() != rev {
+		t.Errorf("rev = %v, want %v", s1.Rev(), rev)
+	}
+	_, err = s1.Range([]byte("foo"), nil, RangeOptions{})
+	if err != nil {
+		t.Errorf("unexpect range error %v", err)
+	}
+}