Merge pull request #3359 from yichengq/storage-test

functional tests for storage package and some related fixes

storage/index_test.go

@@ -91,14 +91,10 @@ func TestIndexTombstone(t *testing.T) {
 	if err != nil {
 		t.Errorf("tombstone error = %v, want nil", err)
 	}
-	rev, _, _, err := index.Get([]byte("foo"), 7)
+	_, _, _, err = index.Get([]byte("foo"), 7)
-	if err != nil {
+	if err != ErrRevisionNotFound {
 		t.Errorf("get error = %v, want nil", err)
 	}
-	w := revision{main: 7}
-	if !reflect.DeepEqual(rev, w) {
-		t.Errorf("get revision = %+v, want %+v", rev, w)
-	}
 }
 
 func TestContinuousCompact(t *testing.T) {

storage/key_index.go

@@ -144,18 +144,14 @@ func (ki *keyIndex) compact(atRev int64, available map[revision]struct{}) {
 		return true
 	}
 
-	g := ki.findGeneration(atRev)
+	i, g := 0, &ki.generations[0]
-	if g == nil {
+	// find first generation includes atRev or created after atRev
-		return
+	for i < len(ki.generations)-1 {
-	}
+		if tomb := g.revs[len(g.revs)-1].main; tomb > atRev {
-
-	i := 0
-	for i <= len(ki.generations)-1 {
-		wg := &ki.generations[i]
-		if wg == g {
 			break
 		}
 		i++
+		g = &ki.generations[i]
 	}
 
 	if !g.isEmpty() {
@@ -180,9 +176,11 @@ func (ki *keyIndex) isEmpty() bool {
 }
 
 // findGeneartion finds out the generation of the keyIndex that the
-// given index belongs to.
+// given rev belongs to. If the given rev is at the gap of two generations,
+// which means that the key does not exist at the given rev, it returns nil.
 func (ki *keyIndex) findGeneration(rev int64) *generation {
-	cg := len(ki.generations) - 1
+	lastg := len(ki.generations) - 1
+	cg := lastg
 
 	for cg >= 0 {
 		if len(ki.generations[cg].revs) == 0 {
@@ -190,6 +188,11 @@ func (ki *keyIndex) findGeneration(rev int64) *generation {
 			continue
 		}
 		g := ki.generations[cg]
+		if cg != lastg {
+			if tomb := g.revs[len(g.revs)-1].main; tomb <= rev {
+				return nil
+			}
+		}
 		if g.revs[0].main <= rev {
 			return &ki.generations[cg]
 		}

storage/key_index_test.go

@@ -21,19 +21,19 @@ func TestKeyIndexGet(t *testing.T) {
 		wrev int64
 		werr error
 	}{
-		{13, 12, nil},
+		{13, 0, ErrRevisionNotFound},
-		{13, 12, nil},
+		{13, 0, ErrRevisionNotFound},
 
 		// get on generation 2
-		{12, 12, nil},
+		{12, 0, ErrRevisionNotFound},
 		{11, 10, nil},
 		{10, 10, nil},
 		{9, 8, nil},
 		{8, 8, nil},
-		{7, 6, nil},
+		{7, 0, ErrRevisionNotFound},
 
 		// get on generation 1
-		{6, 6, nil},
+		{6, 0, ErrRevisionNotFound},
 		{5, 4, nil},
 		{4, 4, nil},
 	}
@@ -291,6 +291,7 @@ func TestKeyIndexCompact(t *testing.T) {
 		}
 	}
 
+	ki = newTestKeyIndex()
 	// Jump Compaction
 	for i, tt := range tests {
 		if (i%2 == 0 && i < 6) || (i%2 == 1 && i > 6) {

storage/kv_test.go

@@ -1,65 +1,684 @@
 package storage
 
 import (
-	"fmt"
 	"os"
 	"reflect"
 	"testing"
+	"time"
+
+	"github.com/coreos/etcd/pkg/testutil"
+	"github.com/coreos/etcd/storage/storagepb"
 )
 
-type kv struct {
+// Functional tests for features implemented in v3 store. It treats v3 store
-	k, v []byte
+// as a black box, and tests it by feeding the input and validating the output.
+
+// TODO: add similar tests on operations in one txn/rev
+
+type (
+	rangeFunc       func(kv KV, key, end []byte, limit, rangeRev int64) ([]storagepb.KeyValue, int64, error)
+	putFunc         func(kv KV, key, value []byte) int64
+	deleteRangeFunc func(kv KV, key, end []byte) (n, rev int64)
+)
+
+var (
+	normalRangeFunc = func(kv KV, key, end []byte, limit, rangeRev int64) ([]storagepb.KeyValue, int64, error) {
+		return kv.Range(key, end, limit, rangeRev)
+	}
+	txnRangeFunc = func(kv KV, key, end []byte, limit, rangeRev int64) ([]storagepb.KeyValue, int64, error) {
+		id := kv.TxnBegin()
+		defer kv.TxnEnd(id)
+		return kv.TxnRange(id, key, end, limit, rangeRev)
+	}
+
+	normalPutFunc = func(kv KV, key, value []byte) int64 {
+		return kv.Put(key, value)
+	}
+	txnPutFunc = func(kv KV, key, value []byte) int64 {
+		id := kv.TxnBegin()
+		defer kv.TxnEnd(id)
+		rev, err := kv.TxnPut(id, key, value)
+		if err != nil {
+			panic("txn put error")
+		}
+		return rev
+	}
+
+	normalDeleteRangeFunc = func(kv KV, key, end []byte) (n, rev int64) {
+		return kv.DeleteRange(key, end)
+	}
+	txnDeleteRangeFunc = func(kv KV, key, end []byte) (n, rev int64) {
+		id := kv.TxnBegin()
+		defer kv.TxnEnd(id)
+		n, rev, err := kv.TxnDeleteRange(id, key, end)
+		if err != nil {
+			panic("txn delete error")
+		}
+		return n, rev
+	}
+)
+
+func TestKVRange(t *testing.T)    { testKVRange(t, normalRangeFunc) }
+func TestKVTxnRange(t *testing.T) { testKVRange(t, txnRangeFunc) }
+
+func testKVRange(t *testing.T, f rangeFunc) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar"))
+	s.Put([]byte("foo1"), []byte("bar1"))
+	s.Put([]byte("foo2"), []byte("bar2"))
+	kvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
+		{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
+		{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
+	}
+
+	wrev := int64(3)
+	tests := []struct {
+		key, end []byte
+		wkvs     []storagepb.KeyValue
+	}{
+		// get no keys
+		{
+			[]byte("doo"), []byte("foo"),
+			nil,
+		},
+		// get no keys when key == end
+		{
+			[]byte("foo"), []byte("foo"),
+			nil,
+		},
+		// get no keys when ranging single key
+		{
+			[]byte("doo"), nil,
+			nil,
+		},
+		// get all keys
+		{
+			[]byte("foo"), []byte("foo3"),
+			kvs,
+		},
+		// get partial keys
+		{
+			[]byte("foo"), []byte("foo1"),
+			kvs[:1],
+		},
+		// get single key
+		{
+			[]byte("foo"), nil,
+			kvs[:1],
+		},
+	}
+
+	for i, tt := range tests {
+		kvs, rev, err := f(s, tt.key, tt.end, 0, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if rev != wrev {
+			t.Errorf("#%d: rev = %d, want %d", i, rev, wrev)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
+		}
+	}
 }
 
-// TestWorkflow simulates the whole workflow that storage is used in normal
+func TestKVRangeRev(t *testing.T)    { testKVRangeRev(t, normalRangeFunc) }
-// etcd running, including key changes, compaction and restart.
+func TestKVTxnRangeRev(t *testing.T) { testKVRangeRev(t, normalRangeFunc) }
-func TestWorkflow(t *testing.T) {
+
-	s := newStore("test")
+func testKVRangeRev(t *testing.T, f rangeFunc) {
-	defer os.Remove("test")
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar"))
+	s.Put([]byte("foo1"), []byte("bar1"))
+	s.Put([]byte("foo2"), []byte("bar2"))
+	kvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
+		{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
+		{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
+	}
+
+	tests := []struct {
+		rev  int64
+		wrev int64
+		wkvs []storagepb.KeyValue
+	}{
+		{-1, 3, kvs},
+		{0, 3, kvs},
+		{1, 1, kvs[:1]},
+		{2, 2, kvs[:2]},
+		{3, 3, kvs},
+	}
+
+	for i, tt := range tests {
+		kvs, rev, err := f(s, []byte("foo"), []byte("foo3"), 0, tt.rev)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if rev != tt.wrev {
+			t.Errorf("#%d: rev = %d, want %d", i, rev, tt.wrev)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
+		}
+	}
+}
+
+func TestKVRangeBadRev(t *testing.T)    { testKVRangeBadRev(t, normalRangeFunc) }
+func TestKVTxnRangeBadRev(t *testing.T) { testKVRangeBadRev(t, normalRangeFunc) }
+
+func testKVRangeBadRev(t *testing.T, f rangeFunc) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar"))
+	s.Put([]byte("foo1"), []byte("bar1"))
+	s.Put([]byte("foo2"), []byte("bar2"))
+	if err := s.Compact(3); err != nil {
+		t.Fatalf("compact error (%v)", err)
+	}
+
+	tests := []struct {
+		rev  int64
+		werr error
+	}{
+		{-1, ErrCompacted},
+		{2, ErrCompacted},
+		{3, ErrCompacted},
+		{4, ErrFutureRev},
+		{100, ErrFutureRev},
+	}
+	for i, tt := range tests {
+		_, _, err := f(s, []byte("foo"), []byte("foo3"), 0, tt.rev)
+		if err != tt.werr {
+			t.Errorf("#%d: error = %v, want %v", i, err, tt.werr)
+		}
+	}
+}
+
+func TestKVRangeLimit(t *testing.T)    { testKVRangeLimit(t, normalRangeFunc) }
+func TestKVTxnRangeLimit(t *testing.T) { testKVRangeLimit(t, txnRangeFunc) }
+
+func testKVRangeLimit(t *testing.T, f rangeFunc) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar"))
+	s.Put([]byte("foo1"), []byte("bar1"))
+	s.Put([]byte("foo2"), []byte("bar2"))
+	kvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
+		{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
+		{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
+	}
+
+	wrev := int64(3)
+	tests := []struct {
+		limit int64
+		wkvs  []storagepb.KeyValue
+	}{
+		// no limit
+		{-1, kvs},
+		// no limit
+		{0, kvs},
+		{1, kvs[:1]},
+		{2, kvs[:2]},
+		{3, kvs},
+		{100, kvs},
+	}
+	for i, tt := range tests {
+		kvs, rev, err := f(s, []byte("foo"), []byte("foo3"), tt.limit, 0)
+		if err != nil {
+			t.Fatalf("#%d: range error (%v)", i, err)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
+		}
+		if rev != wrev {
+			t.Errorf("#%d: rev = %d, want %d", i, rev, wrev)
+		}
+	}
+}
+
+func TestKVPutMultipleTimes(t *testing.T)    { testKVPutMultipleTimes(t, normalPutFunc) }
+func TestKVTxnPutMultipleTimes(t *testing.T) { testKVPutMultipleTimes(t, txnPutFunc) }
+
+func testKVPutMultipleTimes(t *testing.T, f putFunc) {
+	s := New("test")
+	defer cleanup(s, "test")
 
-	var lastrev int64
-	var wkvs []kv
 	for i := 0; i < 10; i++ {
-		// regular compaction
+		base := int64(i + 1)
-		s.Compact(lastrev)
 
-		// put 100 keys into the store in each round
+		rev := f(s, []byte("foo"), []byte("bar"))
-		for k := 0; k < 100; k++ {
+		if wrev := base; rev != wrev {
-			key := fmt.Sprintf("bar_%03d_%03d", i, k)
+			t.Errorf("#%d: rev = %d, want %d", i, rev, base)
-			val := fmt.Sprintf("foo_%03d_%03d", i, k)
-			s.Put([]byte(key), []byte(val))
-			wkvs = append(wkvs, kv{k: []byte(key), v: []byte(val)})
 		}
 
-		// delete second-half keys in this round
+		kvs, _, err := s.Range([]byte("foo"), nil, 0, 0)
-		key := fmt.Sprintf("bar_%03d_050", i)
+		if err != nil {
-		end := fmt.Sprintf("bar_%03d_100", i)
+			t.Fatal(err)
-		if n, _ := s.DeleteRange([]byte(key), []byte(end)); n != 50 {
-			t.Errorf("#%d: delete number = %d, want 50", i, n)
 		}
-		wkvs = wkvs[:len(wkvs)-50]
+		wkvs := []storagepb.KeyValue{
+			{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: base, Version: base},
+		}
+		if !reflect.DeepEqual(kvs, wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
+		}
+	}
+}
 
-		// check existing keys
+func TestKVDeleteRange(t *testing.T)    { testKVDeleteRange(t, normalDeleteRangeFunc) }
-		kvs, rev, err := s.Range([]byte("bar"), []byte("bas"), 0, 0)
+func TestKVTxnDeleteRange(t *testing.T) { testKVDeleteRange(t, txnDeleteRangeFunc) }
+
+func testKVDeleteRange(t *testing.T, f deleteRangeFunc) {
+	tests := []struct {
+		key, end []byte
+
+		wrev int64
+		wN   int64
+	}{
+		{
+			[]byte("foo"), nil,
+			4, 1,
+		},
+		{
+			[]byte("foo"), []byte("foo1"),
+			4, 1,
+		},
+		{
+			[]byte("foo"), []byte("foo2"),
+			4, 2,
+		},
+		{
+			[]byte("foo"), []byte("foo3"),
+			4, 3,
+		},
+		{
+			[]byte("foo3"), []byte("foo8"),
+			3, 0,
+		},
+		{
+			[]byte("foo3"), nil,
+			3, 0,
+		},
+	}
+
+	for i, tt := range tests {
+		s := New("test")
+
+		s.Put([]byte("foo"), []byte("bar"))
+		s.Put([]byte("foo1"), []byte("bar1"))
+		s.Put([]byte("foo2"), []byte("bar2"))
+
+		n, rev := f(s, tt.key, tt.end)
+		if n != tt.wN || rev != tt.wrev {
+			t.Errorf("#%d: n = %d, rev = %d, want (%d, %d)", i, n, rev, tt.wN, tt.wrev)
+		}
+
+		cleanup(s, "test")
+	}
+}
+
+func TestKVDeleteMultipleTimes(t *testing.T)    { testKVDeleteMultipleTimes(t, normalDeleteRangeFunc) }
+func TestKVTxnDeleteMultipleTimes(t *testing.T) { testKVDeleteMultipleTimes(t, txnDeleteRangeFunc) }
+
+func testKVDeleteMultipleTimes(t *testing.T, f deleteRangeFunc) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar"))
+
+	n, rev := f(s, []byte("foo"), nil)
+	if n != 1 || rev != 2 {
+		t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 2)
+	}
+
+	for i := 0; i < 10; i++ {
+		n, rev := f(s, []byte("foo"), nil)
+		if n != 0 || rev != 2 {
+			t.Fatalf("#%d: n = %d, rev = %d, want (%d, %d)", i, n, rev, 0, 2)
+		}
+	}
+}
+
+// test that range, put, delete on single key in sequence repeatedly works correctly.
+func TestKVOperationInSequence(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	for i := 0; i < 10; i++ {
+		base := int64(i * 2)
+
+		// put foo
+		rev := s.Put([]byte("foo"), []byte("bar"))
+		if rev != base+1 {
+			t.Errorf("#%d: put rev = %d, want %d", i, rev, base+1)
+		}
+
+		kvs, rev, err := s.Range([]byte("foo"), nil, 0, base+1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		wkvs := []storagepb.KeyValue{
+			{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: base + 1, ModIndex: base + 1, Version: 1},
+		}
+		if !reflect.DeepEqual(kvs, wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
+		}
+		if rev != base+1 {
+			t.Errorf("#%d: range rev = %d, want %d", i, rev, base+1)
+		}
+
+		// delete foo
+		n, rev := s.DeleteRange([]byte("foo"), nil)
+		if n != 1 || rev != base+2 {
+			t.Errorf("#%d: n = %d, rev = %d, want (%d, %d)", i, n, rev, 1, base+2)
+		}
+
+		kvs, rev, err = s.Range([]byte("foo"), nil, 0, base+2)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if kvs != nil {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, nil)
+		}
+		if rev != base+2 {
+			t.Errorf("#%d: range rev = %d, want %d", i, rev, base+2)
+		}
+	}
+}
+
+func TestKVTxnBlockNonTnxOperations(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	tests := []func(){
+		func() { s.Range([]byte("foo"), nil, 0, 0) },
+		func() { s.Put([]byte("foo"), nil) },
+		func() { s.DeleteRange([]byte("foo"), nil) },
+	}
+	for i, tt := range tests {
+		id := s.TxnBegin()
+		done := make(chan struct{})
+		go func() {
+			tt()
+			done <- struct{}{}
+		}()
+		select {
+		case <-done:
+			t.Fatalf("#%d: operation failed to be blocked", i)
+		case <-time.After(10 * time.Millisecond):
+		}
+
+		s.TxnEnd(id)
+		select {
+		case <-done:
+		case <-time.After(10 * time.Millisecond):
+			t.Fatalf("#%d: operation failed to be unblocked", i)
+		}
+	}
+}
+
+func TestKVTxnWrongID(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	id := s.TxnBegin()
+	wrongid := id + 1
+
+	tests := []func() error{
+		func() error {
+			_, _, err := s.TxnRange(wrongid, []byte("foo"), nil, 0, 0)
+			return err
+		},
+		func() error {
+			_, err := s.TxnPut(wrongid, []byte("foo"), nil)
+			return err
+		},
+		func() error {
+			_, _, err := s.TxnDeleteRange(wrongid, []byte("foo"), nil)
+			return err
+		},
+		func() error { return s.TxnEnd(wrongid) },
+	}
+	for i, tt := range tests {
+		err := tt()
+		if err != ErrTxnIDMismatch {
+			t.Fatalf("#%d: err = %+v, want %+v", i, err, ErrTxnIDMismatch)
+		}
+	}
+
+	err := s.TxnEnd(id)
+	if err != nil {
+		t.Fatalf("end err = %+v, want %+v", err, nil)
+	}
+}
+
+// test that txn range, put, delete on single key in sequence repeatedly works correctly.
+func TestKVTnxOperationInSequence(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	for i := 0; i < 10; i++ {
+		id := s.TxnBegin()
+		base := int64(i)
+
+		// put foo
+		rev, err := s.TxnPut(id, []byte("foo"), []byte("bar"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if rev != base+1 {
+			t.Errorf("#%d: put rev = %d, want %d", i, rev, base+1)
+		}
+
+		kvs, rev, err := s.TxnRange(id, []byte("foo"), nil, 0, base+1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		wkvs := []storagepb.KeyValue{
+			{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: base + 1, ModIndex: base + 1, Version: 1},
+		}
+		if !reflect.DeepEqual(kvs, wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, wkvs)
+		}
+		if rev != base+1 {
+			t.Errorf("#%d: range rev = %d, want %d", i, rev, base+1)
+		}
+
+		// delete foo
+		n, rev, err := s.TxnDeleteRange(id, []byte("foo"), nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if n != 1 || rev != base+1 {
+			t.Errorf("#%d: n = %d, rev = %d, want (%d, %d)", i, n, rev, 1, base+1)
+		}
+
+		kvs, rev, err = s.TxnRange(id, []byte("foo"), nil, 0, base+1)
 		if err != nil {
 			t.Errorf("#%d: range error (%v)", i, err)
 		}
-		if len(kvs) != len(wkvs) {
+		if kvs != nil {
-			t.Fatalf("#%d: len(kvs) = %d, want %d", i, len(kvs), len(wkvs))
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, nil)
 		}
-		for j, kv := range kvs {
+		if rev != base+1 {
-			if !reflect.DeepEqual(kv.Key, wkvs[j].k) {
+			t.Errorf("#%d: range rev = %d, want %d", i, rev, base+1)
-				t.Errorf("#%d: keys[%d] = %s, want %s", i, j, kv.Key, wkvs[j].k)
-			}
-			if !reflect.DeepEqual(kv.Value, wkvs[j].v) {
-				t.Errorf("#%d: vals[%d] = %s, want %s", i, j, kv.Value, wkvs[j].v)
-			}
 		}
-		lastrev = rev
 
-		// the store is restarted and restored from the disk file
+		s.TxnEnd(id)
-		s.Close()
-		s = newStore("test")
-		s.Restore()
 	}
 }
+
+func TestKVCompactReserveLastValue(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar0"))
+	s.Put([]byte("foo"), []byte("bar1"))
+	s.DeleteRange([]byte("foo"), nil)
+	s.Put([]byte("foo"), []byte("bar2"))
+
+	// rev in tests will be called in Compact() one by one on the same store
+	tests := []struct {
+		rev int64
+		// wanted kvs right after the compacted rev
+		wkvs []storagepb.KeyValue
+	}{
+		{
+			0,
+			[]storagepb.KeyValue{
+				{Key: []byte("foo"), Value: []byte("bar0"), CreateIndex: 1, ModIndex: 1, Version: 1},
+			},
+		},
+		{
+			1,
+			[]storagepb.KeyValue{
+				{Key: []byte("foo"), Value: []byte("bar1"), CreateIndex: 1, ModIndex: 2, Version: 2},
+			},
+		},
+		{
+			2,
+			nil,
+		},
+		{
+			3,
+			[]storagepb.KeyValue{
+				{Key: []byte("foo"), Value: []byte("bar2"), CreateIndex: 4, ModIndex: 4, Version: 1},
+			},
+		},
+	}
+	for i, tt := range tests {
+		err := s.Compact(tt.rev)
+		if err != nil {
+			t.Errorf("#%d: unexpect compact error %v", i, err)
+		}
+		kvs, _, err := s.Range([]byte("foo"), nil, 0, tt.rev+1)
+		if err != nil {
+			t.Errorf("#%d: unexpect range error %v", i, err)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
+		}
+	}
+}
+
+func TestKVCompactBad(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar0"))
+	s.Put([]byte("foo"), []byte("bar1"))
+	s.Put([]byte("foo"), []byte("bar2"))
+
+	// rev in tests will be called in Compact() one by one on the same store
+	tests := []struct {
+		rev  int64
+		werr error
+	}{
+		{0, nil},
+		{1, nil},
+		{1, ErrCompacted},
+		{3, nil},
+		{4, ErrFutureRev},
+		{100, ErrFutureRev},
+	}
+	for i, tt := range tests {
+		err := s.Compact(tt.rev)
+		if err != tt.werr {
+			t.Errorf("#%d: compact error = %v, want %v", i, err, tt.werr)
+		}
+	}
+}
+
+func TestKVRestore(t *testing.T) {
+	tests := []func(kv KV){
+		func(kv KV) {
+			kv.Put([]byte("foo"), []byte("bar0"))
+			kv.Put([]byte("foo"), []byte("bar1"))
+			kv.Put([]byte("foo"), []byte("bar2"))
+		},
+		func(kv KV) {
+			kv.Put([]byte("foo"), []byte("bar0"))
+			kv.DeleteRange([]byte("foo"), nil)
+			kv.Put([]byte("foo"), []byte("bar1"))
+		},
+		func(kv KV) {
+			kv.Put([]byte("foo"), []byte("bar0"))
+			kv.Put([]byte("foo"), []byte("bar1"))
+			kv.Compact(1)
+		},
+	}
+	for i, tt := range tests {
+		s := New("test")
+		tt(s)
+		var kvss [][]storagepb.KeyValue
+		for k := int64(0); k < 10; k++ {
+			kvs, _, _ := s.Range([]byte("a"), []byte("z"), 0, k)
+			kvss = append(kvss, kvs)
+		}
+		s.Close()
+
+		ns := New("test")
+		ns.Restore()
+		// wait for possible compaction to finish
+		testutil.WaitSchedule()
+		var nkvss [][]storagepb.KeyValue
+		for k := int64(0); k < 10; k++ {
+			nkvs, _, _ := ns.Range([]byte("a"), []byte("z"), 0, k)
+			nkvss = append(nkvss, nkvs)
+		}
+		cleanup(ns, "test")
+
+		if !reflect.DeepEqual(nkvss, kvss) {
+			t.Errorf("#%d: kvs history = %+v, want %+v", i, nkvss, kvss)
+		}
+	}
+}
+
+func TestKVSnapshot(t *testing.T) {
+	s := New("test")
+	defer cleanup(s, "test")
+
+	s.Put([]byte("foo"), []byte("bar"))
+	s.Put([]byte("foo1"), []byte("bar1"))
+	s.Put([]byte("foo2"), []byte("bar2"))
+	wkvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
+		{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
+		{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
+	}
+
+	f, err := os.Create("new_test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = s.Snapshot(f)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.Close()
+
+	ns := New("new_test")
+	defer cleanup(ns, "new_test")
+	ns.Restore()
+	kvs, rev, err := ns.Range([]byte("a"), []byte("z"), 0, 0)
+	if err != nil {
+		t.Errorf("unexpect range error (%v)", err)
+	}
+	if !reflect.DeepEqual(kvs, wkvs) {
+		t.Errorf("kvs = %+v, want %+v", kvs, wkvs)
+	}
+	if rev != 3 {
+		t.Errorf("rev = %d, want %d", rev, 3)
+	}
+}
+
+func cleanup(s KV, path string) {
+	s.Close()
+	os.Remove(path)
+}

storage/kvstore.go

@@ -146,6 +146,8 @@ func (s *store) TxnDeleteRange(txnID int64, key, end []byte) (n, rev int64, err
 	n = s.deleteRange(key, end, s.currentRev.main+1)
 	if n != 0 || s.currentRev.sub != 0 {
 		rev = int64(s.currentRev.main + 1)
+	} else {
+		rev = int64(s.currentRev.main)
 	}
 	return n, rev, nil
 }
@@ -156,6 +158,9 @@ func (s *store) Compact(rev int64) error {
 	if rev <= s.compactMainRev {
 		return ErrCompacted
 	}
+	if rev > s.currentRev.main {
+		return ErrFutureRev
+	}
 
 	s.compactMainRev = rev
 
@@ -166,6 +171,8 @@ func (s *store) Compact(rev int64) error {
 	tx.Lock()
 	tx.UnsafePut(metaBucketName, scheduledCompactKeyName, rbytes)
 	tx.Unlock()
+	// ensure that desired compaction is persisted
+	s.b.ForceCommit()
 
 	keep := s.kvindex.Compact(rev)
 
@@ -252,14 +259,16 @@ func (a *store) Equal(b *store) bool {
 
 // range is a keyword in Go, add Keys suffix.
 func (s *store) rangeKeys(key, end []byte, limit, rangeRev int64) (kvs []storagepb.KeyValue, rev int64, err error) {
-	if rangeRev > s.currentRev.main {
+	curRev := int64(s.currentRev.main)
+	if s.currentRev.sub > 0 {
+		curRev += 1
+	}
+
+	if rangeRev > curRev {
 		return nil, s.currentRev.main, ErrFutureRev
 	}
 	if rangeRev <= 0 {
-		rev = int64(s.currentRev.main)
+		rev = curRev
-		if s.currentRev.sub > 0 {
-			rev += 1
-		}
 	} else {
 		rev = rangeRev
 	}

storage/kvstore_test.go

@@ -1,9 +1,7 @@
 package storage
 
 import (
-	"bytes"
 	"crypto/rand"
-	"math"
 	"os"
 	"reflect"
 	"testing"
@@ -12,86 +10,8 @@ import (
 	"github.com/coreos/etcd/storage/storagepb"
 )
 
-func TestRange(t *testing.T) {
+// TODO: improve to a unit test
-	s := newStore("test")
+func TestRangeLimitWhenKeyDeleted(t *testing.T) {
-	defer os.Remove("test")
-
-	s.Put([]byte("foo"), []byte("bar"))
-	s.Put([]byte("foo1"), []byte("bar1"))
-	s.Put([]byte("foo2"), []byte("bar2"))
-	kvs := []storagepb.KeyValue{
-		{Key: []byte("foo"), Value: []byte("bar"), CreateIndex: 1, ModIndex: 1, Version: 1},
-		{Key: []byte("foo1"), Value: []byte("bar1"), CreateIndex: 2, ModIndex: 2, Version: 1},
-		{Key: []byte("foo2"), Value: []byte("bar2"), CreateIndex: 3, ModIndex: 3, Version: 1},
-	}
-
-	tests := []struct {
-		key, end []byte
-		rev      int64
-
-		wrev int64
-		wkvs []storagepb.KeyValue
-	}{
-		{
-			[]byte("foo"), []byte("foo3"), 0,
-			3, kvs,
-		},
-		{
-			[]byte("foo"), []byte("foo1"), 0,
-			3, kvs[:1],
-		},
-		{
-			[]byte("foo"), []byte("foo3"), 1,
-			1, kvs[:1],
-		},
-		{
-			[]byte("foo"), []byte("foo3"), 2,
-			2, kvs[:2],
-		},
-	}
-
-	for i, tt := range tests {
-		kvs, rev, err := s.Range(tt.key, tt.end, 0, tt.rev)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if rev != tt.wrev {
-			t.Errorf("#%d: rev = %d, want %d", i, tt.rev, tt.wrev)
-		}
-		if !reflect.DeepEqual(kvs, tt.wkvs) {
-			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
-		}
-	}
-}
-
-func TestRangeBadRev(t *testing.T) {
-	s := newStore("test")
-	defer os.Remove("test")
-
-	s.Put([]byte("foo"), []byte("bar"))
-	s.Put([]byte("foo1"), []byte("bar1"))
-	s.Put([]byte("foo2"), []byte("bar2"))
-	if err := s.Compact(3); err != nil {
-		t.Fatalf("compact error (%v)", err)
-	}
-
-	tests := []struct {
-		rev  int64
-		werr error
-	}{
-		{2, ErrCompacted},
-		{3, ErrCompacted},
-		{4, ErrFutureRev},
-	}
-	for i, tt := range tests {
-		_, _, err := s.Range([]byte("foo"), []byte("foo3"), 0, tt.rev)
-		if err != tt.werr {
-			t.Errorf("#%d: error = %v, want %v", i, err, tt.werr)
-		}
-	}
-}
-
-func TestRangeLimit(t *testing.T) {
 	s := newStore("test")
 	defer os.Remove("test")
 
@@ -125,313 +45,6 @@ func TestRangeLimit(t *testing.T) {
 	}
 }
 
-func TestSimpleDeleteRange(t *testing.T) {
-	tests := []struct {
-		key, end []byte
-
-		wrev int64
-		wN   int64
-	}{
-		{
-			[]byte("foo"), []byte("foo1"),
-			4, 1,
-		},
-		{
-			[]byte("foo"), []byte("foo2"),
-			4, 2,
-		},
-		{
-			[]byte("foo"), []byte("foo3"),
-			4, 3,
-		},
-		{
-			[]byte("foo3"), []byte("foo8"),
-			3, 0,
-		},
-	}
-
-	for i, tt := range tests {
-		s := newStore("test")
-
-		s.Put([]byte("foo"), []byte("bar"))
-		s.Put([]byte("foo1"), []byte("bar1"))
-		s.Put([]byte("foo2"), []byte("bar2"))
-
-		n, rev := s.DeleteRange(tt.key, tt.end)
-		if n != tt.wN {
-			t.Errorf("#%d: n = %d, want %d", i, n, tt.wN)
-		}
-		if rev != tt.wrev {
-			t.Errorf("#%d: rev = %d, wang %d", i, rev, tt.wrev)
-		}
-
-		os.Remove("test")
-	}
-}
-
-func TestRangeInSequence(t *testing.T) {
-	s := newStore("test")
-	defer os.Remove("test")
-
-	s.Put([]byte("foo"), []byte("bar"))
-	s.Put([]byte("foo1"), []byte("bar1"))
-	s.Put([]byte("foo2"), []byte("bar2"))
-
-	// remove foo
-	n, rev := s.DeleteRange([]byte("foo"), nil)
-	if n != 1 || rev != 4 {
-		t.Fatalf("n = %d, index = %d, want (%d, %d)", n, rev, 1, 4)
-	}
-
-	// before removal foo
-	kvs, rev, err := s.Range([]byte("foo"), []byte("foo3"), 0, 3)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(kvs) != 3 {
-		t.Fatalf("len(kvs) = %d, want %d", len(kvs), 3)
-	}
-
-	// after removal foo
-	kvs, rev, err = s.Range([]byte("foo"), []byte("foo3"), 0, 4)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(kvs) != 2 {
-		t.Fatalf("len(kvs) = %d, want %d", len(kvs), 2)
-	}
-
-	// remove again -> expect nothing
-	n, rev = s.DeleteRange([]byte("foo"), nil)
-	if n != 0 || rev != 4 {
-		t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 0, 4)
-	}
-
-	// remove foo1
-	n, rev = s.DeleteRange([]byte("foo"), []byte("foo2"))
-	if n != 1 || rev != 5 {
-		t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 5)
-	}
-
-	// after removal foo1
-	kvs, rev, err = s.Range([]byte("foo"), []byte("foo3"), 0, 5)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(kvs) != 1 {
-		t.Fatalf("len(kvs) = %d, want %d", len(kvs), 1)
-	}
-
-	// remove foo2
-	n, rev = s.DeleteRange([]byte("foo2"), []byte("foo3"))
-	if n != 1 || rev != 6 {
-		t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 6)
-	}
-
-	// after removal foo2
-	kvs, rev, err = s.Range([]byte("foo"), []byte("foo3"), 0, 6)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(kvs) != 0 {
-		t.Fatalf("len(kvs) = %d, want %d", len(kvs), 0)
-	}
-}
-
-func TestOneTxn(t *testing.T) {
-	s := newStore("test")
-	defer os.Remove("test")
-
-	id := s.TxnBegin()
-	for i := 0; i < 3; i++ {
-		s.TxnPut(id, []byte("foo"), []byte("bar"))
-		s.TxnPut(id, []byte("foo1"), []byte("bar1"))
-		s.TxnPut(id, []byte("foo2"), []byte("bar2"))
-
-		// remove foo
-		n, rev, err := s.TxnDeleteRange(id, []byte("foo"), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if n != 1 || rev != 1 {
-			t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 1)
-		}
-
-		kvs, rev, err := s.TxnRange(id, []byte("foo"), []byte("foo3"), 0, 0)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if len(kvs) != 2 {
-			t.Fatalf("len(kvs) = %d, want %d", len(kvs), 2)
-		}
-
-		// remove again -> expect nothing
-		n, rev, err = s.TxnDeleteRange(id, []byte("foo"), nil)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if n != 0 || rev != 1 {
-			t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 0, 1)
-		}
-
-		// remove foo1
-		n, rev, err = s.TxnDeleteRange(id, []byte("foo"), []byte("foo2"))
-		if err != nil {
-			t.Fatal(err)
-		}
-		if n != 1 || rev != 1 {
-			t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 1)
-		}
-
-		// after removal foo1
-		kvs, rev, err = s.TxnRange(id, []byte("foo"), []byte("foo3"), 0, 0)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if len(kvs) != 1 {
-			t.Fatalf("len(kvs) = %d, want %d", len(kvs), 1)
-		}
-
-		// remove foo2
-		n, rev, err = s.TxnDeleteRange(id, []byte("foo2"), []byte("foo3"))
-		if err != nil {
-			t.Fatal(err)
-		}
-		if n != 1 || rev != 1 {
-			t.Fatalf("n = %d, rev = %d, want (%d, %d)", n, rev, 1, 1)
-		}
-
-		// after removal foo2
-		kvs, rev, err = s.TxnRange(id, []byte("foo"), []byte("foo3"), 0, 0)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if len(kvs) != 0 {
-			t.Fatalf("len(kvs) = %d, want %d", len(kvs), 0)
-		}
-	}
-	err := s.TxnEnd(id)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// After txn
-	kvs, rev, err := s.Range([]byte("foo"), []byte("foo3"), 0, 1)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if len(kvs) != 0 {
-		t.Fatalf("len(kvs) = %d, want %d", len(kvs), 0)
-	}
-	if rev != 1 {
-		t.Fatalf("rev = %d, want %d", rev, 1)
-	}
-}
-
-func TestCompaction(t *testing.T) {
-	s := newStore("test")
-	defer os.Remove("test")
-
-	s.Put([]byte("foo"), []byte("bar"))
-	s.Put([]byte("foo1"), []byte("bar1"))
-	s.Put([]byte("foo2"), []byte("bar2"))
-	s.Put([]byte("foo"), []byte("bar11"))
-	s.Put([]byte("foo1"), []byte("bar12"))
-	s.Put([]byte("foo2"), []byte("bar13"))
-	s.Put([]byte("foo1"), []byte("bar14"))
-	s.DeleteRange([]byte("foo"), []byte("foo200"))
-	s.Put([]byte("foo4"), []byte("bar4"))
-
-	err := s.Compact(4)
-	if err != nil {
-		t.Errorf("unexpect compact error %v", err)
-	}
-
-	err = s.Compact(4)
-	if err != ErrCompacted {
-		t.Errorf("err = %v, want %v", err, ErrCompacted)
-	}
-
-	_, _, err = s.Range([]byte("foo"), nil, 0, 4)
-	if err != ErrCompacted {
-		t.Errorf("err = %v, want %v", err, ErrCompacted)
-	}
-
-	// compact should not compact the last value of foo
-	kvs, rev, err := s.Range([]byte("foo"), nil, 0, 5)
-	if err != nil {
-		t.Errorf("unexpected range error %v", err)
-	}
-	if !bytes.Equal(kvs[0].Value, []byte("bar11")) {
-		t.Errorf("value = %s, want %s", string(kvs[0].Value), "bar11")
-	}
-	if rev != 5 {
-		t.Errorf("rev = %d, want %d", rev, 5)
-	}
-
-	// compact everything
-	err = s.Compact(8)
-	if err != nil {
-		t.Errorf("unexpect compact error %v", err)
-	}
-
-	kvs, rev, err = s.Range([]byte("foo"), []byte("fop"), 0, 0)
-	if err != nil {
-		t.Errorf("unexpected range error %v", err)
-	}
-	if len(kvs) != 1 {
-		t.Errorf("len(kvs) = %d, want %d", len(kvs), 1)
-	}
-	if !bytes.Equal(kvs[0].Value, []byte("bar4")) {
-		t.Errorf("value = %s, want %s", string(kvs[0].Value), "bar4")
-	}
-	if rev != 9 {
-		t.Errorf("rev = %d, want %d", rev, 9)
-	}
-}
-
-func TestRestore(t *testing.T) {
-	s0 := newStore("test")
-	defer os.Remove("test")
-
-	s0.Put([]byte("foo"), []byte("bar"))
-	s0.Put([]byte("foo1"), []byte("bar1"))
-	s0.Put([]byte("foo2"), []byte("bar2"))
-	s0.Put([]byte("foo"), []byte("bar11"))
-	s0.Put([]byte("foo1"), []byte("bar12"))
-	s0.Put([]byte("foo2"), []byte("bar13"))
-	s0.Put([]byte("foo1"), []byte("bar14"))
-	s0.Put([]byte("foo3"), []byte("bar3"))
-	s0.DeleteRange([]byte("foo3"), nil)
-	s0.Put([]byte("foo3"), []byte("bar31"))
-	s0.DeleteRange([]byte("foo3"), nil)
-
-	mink := newRevBytes()
-	revToBytes(revision{main: 0, sub: 0}, mink)
-	maxk := newRevBytes()
-	revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, maxk)
-	s0kvs, _, err := s0.rangeKeys(mink, maxk, 0, 0)
-	if err != nil {
-		t.Fatalf("rangeKeys on s0 error (%v)", err)
-	}
-
-	s0.Close()
-
-	s1 := newStore("test")
-	s1.Restore()
-
-	if !s0.Equal(s1) {
-		t.Errorf("not equal!")
-	}
-	s1kvs, _, err := s1.rangeKeys(mink, maxk, 0, 0)
-	if err != nil {
-		t.Fatalf("rangeKeys on s1 error (%v)", err)
-	}
-	if !reflect.DeepEqual(s1kvs, s0kvs) {
-		t.Errorf("s1kvs = %+v, want %+v", s1kvs, s0kvs)
-	}
-}
-
 func TestRestoreContinueUnfinishedCompaction(t *testing.T) {
 	s0 := newStore("test")
 	defer os.Remove("test")

test

@@ -16,7 +16,7 @@ COVER=${COVER:-"-cover"}
 source ./build
 
 # Hack: gofmt ./ will recursively check the .git directory. So use *.go for gofmt.
-TESTABLE_AND_FORMATTABLE="client discovery error etcdctl/command etcdmain etcdserver etcdserver/auth etcdserver/etcdhttp etcdserver/etcdhttp/httptypes pkg/fileutil pkg/flags pkg/idutil pkg/ioutil pkg/netutil pkg/osutil pkg/pbutil pkg/types pkg/transport pkg/wait proxy raft snap store version wal"
+TESTABLE_AND_FORMATTABLE="client discovery error etcdctl/command etcdmain etcdserver etcdserver/auth etcdserver/etcdhttp etcdserver/etcdhttp/httptypes pkg/fileutil pkg/flags pkg/idutil pkg/ioutil pkg/netutil pkg/osutil pkg/pbutil pkg/types pkg/transport pkg/wait proxy raft snap storage store version wal"
 # TODO: add it to race testing when the issue is resolved
 # https://github.com/golang/go/issues/9946
 NO_RACE_TESTABLE="rafthttp"