storage: remove the event concept from key-value layer

The point is to decouple the key-value storage layer and the
event notification layer clearly. It gives the watchableKV the
flexibility to define whatever event structure it wants without
breaking the ondisk format at key-value storage layer.

Changes:

1. change the format of key and value stored in backend

Store KeyValue struct instead of Event struct in backend value for
better abstraction as xiang suggests. And record the corresponded
action in the backend key.

2. Remove word 'event' from functions

storage/index.go

@@ -28,7 +28,7 @@ type index interface {
 	Put(key []byte, rev revision)
 	Restore(key []byte, created, modified revision, ver int64)
 	Tombstone(key []byte, rev revision) error
-	RangeEvents(key, end []byte, rev int64) []revision
+	RangeSince(key, end []byte, rev int64) []revision
 	Compact(rev int64) map[revision]struct{}
 	Equal(b index) bool
 }
@@ -134,10 +134,10 @@ func (ti *treeIndex) Tombstone(key []byte, rev revision) error {
 	return ki.tombstone(rev.main, rev.sub)
 }
 
-// RangeEvents returns all revisions from key(including) to end(excluding)
+// RangeSince returns all revisions from key(including) to end(excluding)
 // at or after the given rev. The returned slice is sorted in the order
 // of revision.
-func (ti *treeIndex) RangeEvents(key, end []byte, rev int64) []revision {
+func (ti *treeIndex) RangeSince(key, end []byte, rev int64) []revision {
 	ti.RLock()
 	defer ti.RUnlock()
 

storage/index_test.go

@@ -136,7 +136,7 @@ func TestIndexTombstone(t *testing.T) {
 	}
 }
 
-func TestIndexRangeEvents(t *testing.T) {
+func TestIndexRangeSince(t *testing.T) {
 	allKeys := [][]byte{[]byte("foo"), []byte("foo1"), []byte("foo2"), []byte("foo2"), []byte("foo1"), []byte("foo")}
 	allRevs := []revision{{main: 1}, {main: 2}, {main: 3}, {main: 4}, {main: 5}, {main: 6}}
 
@@ -184,7 +184,7 @@ func TestIndexRangeEvents(t *testing.T) {
 		},
 	}
 	for i, tt := range tests {
-		revs := index.RangeEvents(tt.key, tt.end, atRev)
+		revs := index.RangeSince(tt.key, tt.end, atRev)
 		if !reflect.DeepEqual(revs, tt.wrevs) {
 			t.Errorf("#%d: revs = %+v, want %+v", i, revs, tt.wrevs)
 		}

storage/kvstore.go

@@ -32,6 +32,13 @@ var (
 	keyBucketName  = []byte("key")
 	metaBucketName = []byte("meta")
 
+	// markedRevBytesLen is the byte length of marked revision.
+	// The first `revBytesLen` bytes represents a normal revision. The last
+	// one byte is the mark.
+	markedRevBytesLen      = revBytesLen + 1
+	markBytePosition       = markedRevBytesLen - 1
+	markTombstone     byte = 't'
+
 	scheduledCompactKeyName = []byte("scheduledCompactRev")
 	finishedCompactKeyName  = []byte("finishedCompactRev")
 
@@ -193,7 +200,7 @@ func (s *store) TxnDeleteRange(txnID int64, key, end []byte) (n, rev int64, err
 	return n, rev, nil
 }
 
-// RangeEvents gets the events from key to end starting from startRev.
+// RangeHistory ranges the history from key to end starting from startRev.
 // If `end` is nil, the request only observes the events on key.
 // If `end` is not nil, it observes the events on key range [key, range_end).
 // Limit limits the number of events returned.
@@ -202,28 +209,29 @@ func (s *store) TxnDeleteRange(txnID int64, key, end []byte) (n, rev int64, err
 // If the required start rev is compacted, ErrCompacted will be returned.
 // If the required start rev has not happened, ErrFutureRev will be returned.
 //
-// RangeEvents returns events that satisfy the requirement (0 <= n <= limit).
-// If events in the revision range have not all happened, it returns immeidately
+// RangeHistory returns revision bytes slice and key-values that satisfy the requirement (0 <= n <= limit).
+// If history in the revision range has not all happened, it returns immeidately
 // what is available.
 // It also returns nextRev which indicates the start revision used for the following
 // RangeEvents call. The nextRev could be smaller than the given endRev if the store
 // has not progressed so far or it hits the event limit.
 //
-// TODO: return byte slices instead of events to avoid meaningless encode and decode.
-func (s *store) RangeEvents(key, end []byte, limit, startRev int64) (evs []storagepb.Event, nextRev int64, err error) {
+// TODO: return byte slices instead of keyValues to avoid meaningless encode and decode.
+// This also helps to return raw (key, val) pair directly to make API consistent.
+func (s *store) RangeHistory(key, end []byte, limit, startRev int64) (revbs [][]byte, kvs []storagepb.KeyValue, nextRev int64, err error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	if startRev > 0 && startRev <= s.compactMainRev {
-		return nil, 0, ErrCompacted
+		return nil, nil, 0, ErrCompacted
 	}
 	if startRev > s.currentRev.main {
-		return nil, 0, ErrFutureRev
+		return nil, nil, 0, ErrFutureRev
 	}
 
-	revs := s.kvindex.RangeEvents(key, end, startRev)
+	revs := s.kvindex.RangeSince(key, end, startRev)
 	if len(revs) == 0 {
-		return nil, s.currentRev.main + 1, nil
+		return nil, nil, s.currentRev.main + 1, nil
 	}
 
 	tx := s.b.BatchTx()
@@ -231,24 +239,24 @@ func (s *store) RangeEvents(key, end []byte, limit, startRev int64) (evs []stora
 	defer tx.Unlock()
 	// fetch events from the backend using revisions
 	for _, rev := range revs {
-		revbytes := newRevBytes()
-		revToBytes(rev, revbytes)
+		start, end := revBytesRange(rev)
 
-		_, vs := tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
+		ks, vs := tx.UnsafeRange(keyBucketName, start, end, 0)
 		if len(vs) != 1 {
 			log.Fatalf("storage: range cannot find rev (%d,%d)", rev.main, rev.sub)
 		}
 
-		e := storagepb.Event{}
-		if err := e.Unmarshal(vs[0]); err != nil {
+		var kv storagepb.KeyValue
+		if err := kv.Unmarshal(vs[0]); err != nil {
 			log.Fatalf("storage: cannot unmarshal event: %v", err)
 		}
-		evs = append(evs, e)
-		if limit > 0 && len(evs) >= int(limit) {
-			return evs, rev.main + 1, nil
+		revbs = append(revbs, ks[0])
+		kvs = append(kvs, kv)
+		if limit > 0 && len(kvs) >= int(limit) {
+			return revbs, kvs, rev.main + 1, nil
 		}
 	}
-	return evs, s.currentRev.main + 1, nil
+	return revbs, kvs, s.currentRev.main + 1, nil
 }
 
 func (s *store) Compact(rev int64) error {
@@ -316,21 +324,19 @@ func (s *store) Restore() error {
 	// TODO: limit N to reduce max memory usage
 	keys, vals := tx.UnsafeRange(keyBucketName, min, max, 0)
 	for i, key := range keys {
-		e := &storagepb.Event{}
-		if err := e.Unmarshal(vals[i]); err != nil {
+		var kv storagepb.KeyValue
+		if err := kv.Unmarshal(vals[i]); err != nil {
 			log.Fatalf("storage: cannot unmarshal event: %v", err)
 		}
 
-		rev := bytesToRev(key)
+		rev := bytesToRev(key[:revBytesLen])
 
 		// restore index
-		switch e.Type {
-		case storagepb.PUT:
-			s.kvindex.Restore(e.Kv.Key, revision{e.Kv.CreateRevision, 0}, rev, e.Kv.Version)
-		case storagepb.DELETE:
-			s.kvindex.Tombstone(e.Kv.Key, rev)
+		switch {
+		case isTombstone(key):
+			s.kvindex.Tombstone(kv.Key, rev)
 		default:
-			log.Panicf("storage: unexpected event type %s", e.Type)
+			s.kvindex.Restore(kv.Key, revision{kv.CreateRevision, 0}, rev, kv.Version)
 		}
 
 		// update revision
@@ -392,19 +398,18 @@ func (s *store) rangeKeys(key, end []byte, limit, rangeRev int64) (kvs []storage
 	}
 
 	for _, revpair := range revpairs {
-		revbytes := newRevBytes()
-		revToBytes(revpair, revbytes)
+		start, end := revBytesRange(revpair)
 
-		_, vs := s.tx.UnsafeRange(keyBucketName, revbytes, nil, 0)
+		_, vs := s.tx.UnsafeRange(keyBucketName, start, end, 0)
 		if len(vs) != 1 {
 			log.Fatalf("storage: range cannot find rev (%d,%d)", revpair.main, revpair.sub)
 		}
 
-		e := &storagepb.Event{}
-		if err := e.Unmarshal(vs[0]); err != nil {
+		var kv storagepb.KeyValue
+		if err := kv.Unmarshal(vs[0]); err != nil {
 			log.Fatalf("storage: cannot unmarshal event: %v", err)
 		}
-		kvs = append(kvs, *e.Kv)
+		kvs = append(kvs, kv)
 		if limit > 0 && len(kvs) >= int(limit) {
 			break
 		}
@@ -426,18 +431,15 @@ func (s *store) put(key, value []byte) {
 	revToBytes(revision{main: rev, sub: s.currentRev.sub}, ibytes)
 
 	ver = ver + 1
-	event := storagepb.Event{
-		Type: storagepb.PUT,
-		Kv: &storagepb.KeyValue{
-			Key:            key,
-			Value:          value,
-			CreateRevision: c,
-			ModRevision:    rev,
-			Version:        ver,
-		},
-	}
-
-	d, err := event.Marshal()
+	kv := storagepb.KeyValue{
+		Key:            key,
+		Value:          value,
+		CreateRevision: c,
+		ModRevision:    rev,
+		Version:        ver,
+	}
+
+	d, err := kv.Marshal()
 	if err != nil {
 		log.Fatalf("storage: cannot marshal event: %v", err)
 	}
@@ -469,15 +471,13 @@ func (s *store) delete(key []byte) {
 
 	ibytes := newRevBytes()
 	revToBytes(revision{main: mainrev, sub: s.currentRev.sub}, ibytes)
+	ibytes = appendMarkTombstone(ibytes)
 
-	event := storagepb.Event{
-		Type: storagepb.DELETE,
-		Kv: &storagepb.KeyValue{
-			Key: key,
-		},
+	kv := storagepb.KeyValue{
+		Key: key,
 	}
 
-	d, err := event.Marshal()
+	d, err := kv.Marshal()
 	if err != nil {
 		log.Fatalf("storage: cannot marshal event: %v", err)
 	}
@@ -489,3 +489,29 @@ func (s *store) delete(key []byte) {
 	}
 	s.currentRev.sub += 1
 }
+
+// appendMarkTombstone appends tombstone mark to normal revision bytes.
+func appendMarkTombstone(b []byte) []byte {
+	if len(b) != revBytesLen {
+		log.Panicf("cannot append mark to non normal revision bytes")
+	}
+	return append(b, markTombstone)
+}
+
+// isTombstone checks whether the revision bytes is a tombstone.
+func isTombstone(b []byte) bool {
+	return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone
+}
+
+// revBytesRange returns the range of revision bytes at
+// the given revision.
+func revBytesRange(rev revision) (start, end []byte) {
+	start = newRevBytes()
+	revToBytes(rev, start)
+
+	end = newRevBytes()
+	endRev := revision{main: rev.main, sub: rev.sub + 1}
+	revToBytes(endRev, end)
+
+	return start, end
+}

storage/kvstore_test.go

@@ -46,22 +46,21 @@ func TestStorePut(t *testing.T) {
 		r   indexGetResp
 
 		wrev    revision
-		wev     storagepb.Event
+		wkey    []byte
+		wkv     storagepb.KeyValue
 		wputrev revision
 	}{
 		{
 			revision{1, 0},
 			indexGetResp{revision{}, revision{}, 0, ErrRevisionNotFound},
 			revision{1, 1},
-			storagepb.Event{
-				Type: storagepb.PUT,
-				Kv: &storagepb.KeyValue{
-					Key:            []byte("foo"),
-					Value:          []byte("bar"),
-					CreateRevision: 2,
-					ModRevision:    2,
-					Version:        1,
-				},
+			newTestKeyBytes(revision{2, 0}, false),
+			storagepb.KeyValue{
+				Key:            []byte("foo"),
+				Value:          []byte("bar"),
+				CreateRevision: 2,
+				ModRevision:    2,
+				Version:        1,
 			},
 			revision{2, 0},
 		},
@@ -69,15 +68,13 @@ func TestStorePut(t *testing.T) {
 			revision{1, 1},
 			indexGetResp{revision{2, 0}, revision{2, 0}, 1, nil},
 			revision{1, 2},
-			storagepb.Event{
-				Type: storagepb.PUT,
-				Kv: &storagepb.KeyValue{
-					Key:            []byte("foo"),
-					Value:          []byte("bar"),
-					CreateRevision: 2,
-					ModRevision:    2,
-					Version:        2,
-				},
+			newTestKeyBytes(revision{2, 1}, false),
+			storagepb.KeyValue{
+				Key:            []byte("foo"),
+				Value:          []byte("bar"),
+				CreateRevision: 2,
+				ModRevision:    2,
+				Version:        2,
 			},
 			revision{2, 1},
 		},
@@ -85,15 +82,13 @@ func TestStorePut(t *testing.T) {
 			revision{2, 0},
 			indexGetResp{revision{2, 1}, revision{2, 0}, 2, nil},
 			revision{2, 1},
-			storagepb.Event{
-				Type: storagepb.PUT,
-				Kv: &storagepb.KeyValue{
-					Key:            []byte("foo"),
-					Value:          []byte("bar"),
-					CreateRevision: 2,
-					ModRevision:    3,
-					Version:        3,
-				},
+			newTestKeyBytes(revision{3, 0}, false),
+			storagepb.KeyValue{
+				Key:            []byte("foo"),
+				Value:          []byte("bar"),
+				CreateRevision: 2,
+				ModRevision:    3,
+				Version:        3,
 			},
 			revision{3, 0},
 		},
@@ -106,12 +101,12 @@ func TestStorePut(t *testing.T) {
 
 		s.put([]byte("foo"), []byte("bar"))
 
-		data, err := tt.wev.Marshal()
+		data, err := tt.wkv.Marshal()
 		if err != nil {
 			t.Errorf("#%d: marshal err = %v, want nil", i, err)
 		}
 		wact := []testutil.Action{
-			{"put", []interface{}{keyBucketName, newTestBytes(tt.wputrev), data}},
+			{"put", []interface{}{keyBucketName, tt.wkey, data}},
 		}
 		if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
 			t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
@@ -130,17 +125,15 @@ func TestStorePut(t *testing.T) {
 }
 
 func TestStoreRange(t *testing.T) {
-	ev := storagepb.Event{
-		Type: storagepb.PUT,
-		Kv: &storagepb.KeyValue{
-			Key:            []byte("foo"),
-			Value:          []byte("bar"),
-			CreateRevision: 1,
-			ModRevision:    2,
-			Version:        1,
-		},
-	}
-	evb, err := ev.Marshal()
+	key := newTestKeyBytes(revision{2, 0}, false)
+	kv := storagepb.KeyValue{
+		Key:            []byte("foo"),
+		Value:          []byte("bar"),
+		CreateRevision: 1,
+		ModRevision:    2,
+		Version:        1,
+	}
+	kvb, err := kv.Marshal()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -153,11 +146,11 @@ func TestStoreRange(t *testing.T) {
 	}{
 		{
 			indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
-			rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
+			rangeResp{[][]byte{key}, [][]byte{kvb}},
 		},
 		{
 			indexRangeResp{[][]byte{[]byte("foo"), []byte("foo1")}, []revision{{2, 0}, {3, 0}}},
-			rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
+			rangeResp{[][]byte{key}, [][]byte{kvb}},
 		},
 	}
 	for i, tt := range tests {
@@ -171,15 +164,16 @@ func TestStoreRange(t *testing.T) {
 		if err != nil {
 			t.Errorf("#%d: err = %v, want nil", i, err)
 		}
-		if w := []storagepb.KeyValue{*ev.Kv}; !reflect.DeepEqual(kvs, w) {
+		if w := []storagepb.KeyValue{kv}; !reflect.DeepEqual(kvs, w) {
 			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, w)
 		}
 		if rev != wrev {
 			t.Errorf("#%d: rev = %d, want %d", i, rev, wrev)
 		}
 
+		wstart, wend := revBytesRange(tt.idxr.revs[0])
 		wact := []testutil.Action{
-			{"range", []interface{}{keyBucketName, newTestBytes(tt.idxr.revs[0]), []byte(nil), int64(0)}},
+			{"range", []interface{}{keyBucketName, wstart, wend, int64(0)}},
 		}
 		if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
 			t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
@@ -201,6 +195,7 @@ func TestStoreDeleteRange(t *testing.T) {
 		rev revision
 		r   indexRangeResp
 
+		wkey    []byte
 		wrev    revision
 		wrrev   int64
 		wdelrev revision
@@ -208,6 +203,7 @@ func TestStoreDeleteRange(t *testing.T) {
 		{
 			revision{2, 0},
 			indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
+			newTestKeyBytes(revision{3, 0}, true),
 			revision{2, 1},
 			2,
 			revision{3, 0},
@@ -215,6 +211,7 @@ func TestStoreDeleteRange(t *testing.T) {
 		{
 			revision{2, 1},
 			indexRangeResp{[][]byte{[]byte("foo")}, []revision{{2, 0}}},
+			newTestKeyBytes(revision{3, 1}, true),
 			revision{2, 2},
 			3,
 			revision{3, 1},
@@ -231,17 +228,14 @@ func TestStoreDeleteRange(t *testing.T) {
 			t.Errorf("#%d: n = %d, want 1", i, n)
 		}
 
-		data, err := (&storagepb.Event{
-			Type: storagepb.DELETE,
-			Kv: &storagepb.KeyValue{
-				Key: []byte("foo"),
-			},
+		data, err := (&storagepb.KeyValue{
+			Key: []byte("foo"),
 		}).Marshal()
 		if err != nil {
 			t.Errorf("#%d: marshal err = %v, want nil", i, err)
 		}
 		wact := []testutil.Action{
-			{"put", []interface{}{keyBucketName, newTestBytes(tt.wdelrev), data}},
+			{"put", []interface{}{keyBucketName, tt.wkey, data}},
 		}
 		if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
 			t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
@@ -259,18 +253,16 @@ func TestStoreDeleteRange(t *testing.T) {
 	}
 }
 
-func TestStoreRangeEvents(t *testing.T) {
-	ev := storagepb.Event{
-		Type: storagepb.PUT,
-		Kv: &storagepb.KeyValue{
-			Key:            []byte("foo"),
-			Value:          []byte("bar"),
-			CreateRevision: 1,
-			ModRevision:    2,
-			Version:        1,
-		},
+func TestStoreRangeHistory(t *testing.T) {
+	key := newTestKeyBytes(revision{2, 0}, false)
+	kv := storagepb.KeyValue{
+		Key:            []byte("foo"),
+		Value:          []byte("bar"),
+		CreateRevision: 1,
+		ModRevision:    2,
+		Version:        1,
 	}
-	evb, err := ev.Marshal()
+	kvb, err := kv.Marshal()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -282,11 +274,11 @@ func TestStoreRangeEvents(t *testing.T) {
 	}{
 		{
 			indexRangeEventsResp{[]revision{{2, 0}}},
-			rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
+			rangeResp{[][]byte{key}, [][]byte{kvb}},
 		},
 		{
 			indexRangeEventsResp{[]revision{{2, 0}, {3, 0}}},
-			rangeResp{[][]byte{newTestBytes(revision{2, 0})}, [][]byte{evb}},
+			rangeResp{[][]byte{key}, [][]byte{kvb}},
 		},
 	}
 	for i, tt := range tests {
@@ -295,12 +287,15 @@ func TestStoreRangeEvents(t *testing.T) {
 		index.indexRangeEventsRespc <- tt.idxr
 		b.tx.rangeRespc <- tt.r
 
-		evs, _, err := s.RangeEvents([]byte("foo"), []byte("goo"), 1, 1)
+		keys, kvs, _, err := s.RangeHistory([]byte("foo"), []byte("goo"), 1, 1)
 		if err != nil {
 			t.Errorf("#%d: err = %v, want nil", i, err)
 		}
-		if w := []storagepb.Event{ev}; !reflect.DeepEqual(evs, w) {
-			t.Errorf("#%d: evs = %+v, want %+v", i, evs, w)
+		if w := [][]byte{key}; !reflect.DeepEqual(keys, w) {
+			t.Errorf("#%d: keys = %+v, want %+v", i, keys, w)
+		}
+		if w := []storagepb.KeyValue{kv}; !reflect.DeepEqual(kvs, w) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, w)
 		}
 
 		wact := []testutil.Action{
@@ -309,8 +304,9 @@ func TestStoreRangeEvents(t *testing.T) {
 		if g := index.Action(); !reflect.DeepEqual(g, wact) {
 			t.Errorf("#%d: index action = %+v, want %+v", i, g, wact)
 		}
+		wstart, wend := revBytesRange(tt.idxr.revs[0])
 		wact = []testutil.Action{
-			{"range", []interface{}{keyBucketName, newTestBytes(tt.idxr.revs[0]), []byte(nil), int64(0)}},
+			{"range", []interface{}{keyBucketName, wstart, wend, int64(0)}},
 		}
 		if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
 			t.Errorf("#%d: tx action = %+v, want %+v", i, g, wact)
@@ -325,7 +321,9 @@ func TestStoreCompact(t *testing.T) {
 	s, b, index := newFakeStore()
 	s.currentRev = revision{3, 0}
 	index.indexCompactRespc <- map[revision]struct{}{revision{1, 0}: {}}
-	b.tx.rangeRespc <- rangeResp{[][]byte{newTestBytes(revision{1, 0}), newTestBytes(revision{2, 0})}, nil}
+	key1 := newTestKeyBytes(revision{1, 0}, false)
+	key2 := newTestKeyBytes(revision{2, 0}, false)
+	b.tx.rangeRespc <- rangeResp{[][]byte{key1, key2}, nil}
 
 	s.Compact(3)
 	s.wg.Wait()
@@ -336,10 +334,10 @@ func TestStoreCompact(t *testing.T) {
 	end := make([]byte, 8)
 	binary.BigEndian.PutUint64(end, uint64(4))
 	wact := []testutil.Action{
-		{"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestBytes(revision{3, 0})}},
+		{"put", []interface{}{metaBucketName, scheduledCompactKeyName, newTestRevBytes(revision{3, 0})}},
 		{"range", []interface{}{keyBucketName, make([]byte, 17), end, int64(10000)}},
-		{"delete", []interface{}{keyBucketName, newTestBytes(revision{2, 0})}},
-		{"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestBytes(revision{3, 0})}},
+		{"delete", []interface{}{keyBucketName, key2}},
+		{"put", []interface{}{metaBucketName, finishedCompactKeyName, newTestRevBytes(revision{3, 0})}},
 	}
 	if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
 		t.Errorf("tx actions = %+v, want %+v", g, wact)
@@ -355,33 +353,29 @@ func TestStoreCompact(t *testing.T) {
 func TestStoreRestore(t *testing.T) {
 	s, b, index := newFakeStore()
 
-	putev := storagepb.Event{
-		Type: storagepb.PUT,
-		Kv: &storagepb.KeyValue{
-			Key:            []byte("foo"),
-			Value:          []byte("bar"),
-			CreateRevision: 3,
-			ModRevision:    3,
-			Version:        1,
-		},
+	putkey := newTestKeyBytes(revision{3, 0}, false)
+	putkv := storagepb.KeyValue{
+		Key:            []byte("foo"),
+		Value:          []byte("bar"),
+		CreateRevision: 3,
+		ModRevision:    3,
+		Version:        1,
 	}
-	putevb, err := putev.Marshal()
+	putkvb, err := putkv.Marshal()
 	if err != nil {
 		t.Fatal(err)
 	}
-	delev := storagepb.Event{
-		Type: storagepb.DELETE,
-		Kv: &storagepb.KeyValue{
-			Key: []byte("foo"),
-		},
+	delkey := newTestKeyBytes(revision{4, 0}, true)
+	delkv := storagepb.KeyValue{
+		Key: []byte("foo"),
 	}
-	delevb, err := delev.Marshal()
+	delkvb, err := delkv.Marshal()
 	if err != nil {
 		t.Fatal(err)
 	}
-	b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestBytes(revision{2, 0})}}
-	b.tx.rangeRespc <- rangeResp{[][]byte{newTestBytes(revision{3, 0}), newTestBytes(revision{4, 0})}, [][]byte{putevb, delevb}}
-	b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestBytes(revision{2, 0})}}
+	b.tx.rangeRespc <- rangeResp{[][]byte{finishedCompactKeyName}, [][]byte{newTestRevBytes(revision{2, 0})}}
+	b.tx.rangeRespc <- rangeResp{[][]byte{putkey, delkey}, [][]byte{putkvb, delkvb}}
+	b.tx.rangeRespc <- rangeResp{[][]byte{scheduledCompactKeyName}, [][]byte{newTestRevBytes(revision{2, 0})}}
 
 	s.Restore()
 
@@ -394,7 +388,7 @@ func TestStoreRestore(t *testing.T) {
 	}
 	wact := []testutil.Action{
 		{"range", []interface{}{metaBucketName, finishedCompactKeyName, []byte(nil), int64(0)}},
-		{"range", []interface{}{keyBucketName, newTestBytes(revision{}), newTestBytes(revision{math.MaxInt64, math.MaxInt64}), int64(0)}},
+		{"range", []interface{}{keyBucketName, newTestRevBytes(revision{}), newTestRevBytes(revision{math.MaxInt64, math.MaxInt64}), int64(0)}},
 		{"range", []interface{}{metaBucketName, scheduledCompactKeyName, []byte(nil), int64(0)}},
 	}
 	if g := b.tx.Action(); !reflect.DeepEqual(g, wact) {
@@ -410,79 +404,79 @@ func TestStoreRestore(t *testing.T) {
 }
 
 // tests end parameter works well
-func TestStoreRangeEventsEnd(t *testing.T) {
+func TestStoreRangeHistoryEnd(t *testing.T) {
 	s := newStore(tmpPath)
 	defer cleanup(s, tmpPath)
 
 	s.Put([]byte("foo"), []byte("bar"))
 	s.Put([]byte("foo1"), []byte("bar1"))
 	s.Put([]byte("foo2"), []byte("bar2"))
-	evs := []storagepb.Event{
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
-		},
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
-		},
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
-		},
+	keys := [][]byte{
+		newTestKeyBytes(revision{1, 0}, false),
+		newTestKeyBytes(revision{2, 0}, false),
+		newTestKeyBytes(revision{3, 0}, false),
+	}
+	kvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
+		{Key: []byte("foo1"), Value: []byte("bar1"), CreateRevision: 2, ModRevision: 2, Version: 1},
+		{Key: []byte("foo2"), Value: []byte("bar2"), CreateRevision: 3, ModRevision: 3, Version: 1},
 	}
 
 	tests := []struct {
 		key, end []byte
-		wevs     []storagepb.Event
+		wkeys    [][]byte
+		wkvs     []storagepb.KeyValue
 	}{
 		// get no keys
 		{
 			[]byte("doo"), []byte("foo"),
-			nil,
+			nil, nil,
 		},
 		// get no keys when key == end
 		{
 			[]byte("foo"), []byte("foo"),
-			nil,
+			nil, nil,
 		},
 		// get no keys when ranging single key
 		{
 			[]byte("doo"), nil,
-			nil,
+			nil, nil,
 		},
 		// get all keys
 		{
 			[]byte("foo"), []byte("foo3"),
-			evs,
+			keys, kvs,
 		},
 		// get partial keys
 		{
 			[]byte("foo"), []byte("foo1"),
-			evs[:1],
+			keys[:1], kvs[:1],
 		},
 		// get single key
 		{
 			[]byte("foo"), nil,
-			evs[:1],
+			keys[:1], kvs[:1],
 		},
 	}
 
 	for i, tt := range tests {
-		evs, rev, err := s.RangeEvents(tt.key, tt.end, 0, 1)
+		keys, kvs, rev, err := s.RangeHistory(tt.key, tt.end, 0, 1)
 		if err != nil {
 			t.Fatal(err)
 		}
 		if rev != 4 {
 			t.Errorf("#%d: rev = %d, want %d", i, rev, 4)
 		}
-		if !reflect.DeepEqual(evs, tt.wevs) {
-			t.Errorf("#%d: evs = %+v, want %+v", i, evs, tt.wevs)
+		if !reflect.DeepEqual(keys, tt.wkeys) {
+			t.Errorf("#%d: actions = %+v, want %+v", i, keys, tt.wkeys)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
 		}
 	}
 }
 
-func TestStoreRangeEventsRev(t *testing.T) {
+func TestStoreRangeHistoryRev(t *testing.T) {
 	s := newStore(tmpPath)
 	defer cleanup(s, tmpPath)
 
@@ -490,39 +484,39 @@ func TestStoreRangeEventsRev(t *testing.T) {
 	s.DeleteRange([]byte("foo"), nil)
 	s.Put([]byte("foo"), []byte("bar"))
 	s.Put([]byte("unrelated"), []byte("unrelated"))
-	evs := []storagepb.Event{
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
-		},
-		{
-			Type: storagepb.DELETE,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo")},
-		},
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
-		},
+	keys := [][]byte{
+		newTestKeyBytes(revision{1, 0}, false),
+		newTestKeyBytes(revision{2, 0}, true),
+		newTestKeyBytes(revision{3, 0}, false),
+	}
+	kvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
+		{Key: []byte("foo")},
+		{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
 	}
 
 	tests := []struct {
 		start int64
 
-		wevs  []storagepb.Event
+		wkeys [][]byte
+		wkvs  []storagepb.KeyValue
 		wnext int64
 	}{
-		{0, evs, 5},
-		{1, evs, 5},
-		{3, evs[2:], 5},
+		{0, keys, kvs, 5},
+		{1, keys, kvs, 5},
+		{3, keys[2:], kvs[2:], 5},
 	}
 
 	for i, tt := range tests {
-		evs, next, err := s.RangeEvents([]byte("foo"), nil, 0, tt.start)
+		keys, kvs, next, err := s.RangeHistory([]byte("foo"), nil, 0, tt.start)
 		if err != nil {
 			t.Fatal(err)
 		}
-		if !reflect.DeepEqual(evs, tt.wevs) {
-			t.Errorf("#%d: evs = %+v, want %+v", i, evs, tt.wevs)
+		if !reflect.DeepEqual(keys, tt.wkeys) {
+			t.Errorf("#%d: acts = %+v, want %+v", i, keys, tt.wkeys)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
 		}
 		if next != tt.wnext {
 			t.Errorf("#%d: next = %d, want %d", i, next, tt.wnext)
@@ -530,7 +524,7 @@ func TestStoreRangeEventsRev(t *testing.T) {
 	}
 }
 
-func TestStoreRangeEventsBad(t *testing.T) {
+func TestStoreRangeHistoryBad(t *testing.T) {
 	s := newStore(tmpPath)
 	defer cleanup(s, tmpPath)
 
@@ -552,55 +546,55 @@ func TestStoreRangeEventsBad(t *testing.T) {
 		{10, ErrFutureRev},
 	}
 	for i, tt := range tests {
-		_, _, err := s.RangeEvents([]byte("foo"), nil, 0, tt.rev)
+		_, _, _, err := s.RangeHistory([]byte("foo"), nil, 0, tt.rev)
 		if err != tt.werr {
 			t.Errorf("#%d: error = %v, want %v", i, err, tt.werr)
 		}
 	}
 }
 
-func TestStoreRangeEventsLimit(t *testing.T) {
+func TestStoreRangeHistoryLimit(t *testing.T) {
 	s := newStore(tmpPath)
 	defer cleanup(s, tmpPath)
 
 	s.Put([]byte("foo"), []byte("bar"))
 	s.DeleteRange([]byte("foo"), nil)
 	s.Put([]byte("foo"), []byte("bar"))
-	evs := []storagepb.Event{
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
-		},
-		{
-			Type: storagepb.DELETE,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo")},
-		},
-		{
-			Type: storagepb.PUT,
-			Kv:   &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
-		},
+	keys := [][]byte{
+		newTestKeyBytes(revision{1, 0}, false),
+		newTestKeyBytes(revision{2, 0}, true),
+		newTestKeyBytes(revision{3, 0}, false),
+	}
+	kvs := []storagepb.KeyValue{
+		{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
+		{Key: []byte("foo")},
+		{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 3, ModRevision: 3, Version: 1},
 	}
 
 	tests := []struct {
 		limit int64
-		wevs  []storagepb.Event
+		wkeys [][]byte
+		wkvs  []storagepb.KeyValue
 	}{
 		// no limit
-		{-1, evs},
+		{-1, keys, kvs},
 		// no limit
-		{0, evs},
-		{1, evs[:1]},
-		{2, evs[:2]},
-		{3, evs},
-		{100, evs},
+		{0, keys, kvs},
+		{1, keys[:1], kvs[:1]},
+		{2, keys[:2], kvs[:2]},
+		{3, keys, kvs},
+		{100, keys, kvs},
 	}
 	for i, tt := range tests {
-		evs, _, err := s.RangeEvents([]byte("foo"), nil, tt.limit, 1)
+		keys, kvs, _, err := s.RangeHistory([]byte("foo"), nil, tt.limit, 1)
 		if err != nil {
 			t.Fatalf("#%d: range error (%v)", i, err)
 		}
-		if !reflect.DeepEqual(evs, tt.wevs) {
-			t.Errorf("#%d: evs = %+v, want %+v", i, evs, tt.wevs)
+		if !reflect.DeepEqual(keys, tt.wkeys) {
+			t.Errorf("#%d: acts = %+v, want %+v", i, keys, tt.wkeys)
+		}
+		if !reflect.DeepEqual(kvs, tt.wkvs) {
+			t.Errorf("#%d: kvs = %+v, want %+v", i, kvs, tt.wkvs)
 		}
 	}
 }
@@ -688,9 +682,18 @@ func BenchmarkStorePut(b *testing.B) {
 	}
 }
 
-func newTestBytes(rev revision) []byte {
+func newTestRevBytes(rev revision) []byte {
+	bytes := newRevBytes()
+	revToBytes(rev, bytes)
+	return bytes
+}
+
+func newTestKeyBytes(rev revision, tombstone bool) []byte {
 	bytes := newRevBytes()
 	revToBytes(rev, bytes)
+	if tombstone {
+		bytes = appendMarkTombstone(bytes)
+	}
 	return bytes
 }
 
@@ -792,7 +795,7 @@ func (i *fakeIndex) Tombstone(key []byte, rev revision) error {
 	i.Recorder.Record(testutil.Action{Name: "tombstone", Params: []interface{}{key, rev}})
 	return nil
 }
-func (i *fakeIndex) RangeEvents(key, end []byte, rev int64) []revision {
+func (i *fakeIndex) RangeSince(key, end []byte, rev int64) []revision {
 	i.Recorder.Record(testutil.Action{Name: "rangeEvents", Params: []interface{}{key, end, rev}})
 	r := <-i.indexRangeEventsRespc
 	return r.revs

storage/revision.go

@@ -16,6 +16,11 @@ package storage
 
 import "encoding/binary"
 
+// revBytesLen is the byte length of a normal revision.
+// First 8 bytes is the revision.main in big-endian format. The 9th byte
+// is a '_'. The last 8 bytes is the revision.sub in big-endian format.
+const revBytesLen = 8 + 1 + 8
+
 type revision struct {
 	main int64
 	sub  int64
@@ -32,7 +37,7 @@ func (a revision) GreaterThan(b revision) bool {
 }
 
 func newRevBytes() []byte {
-	return make([]byte, 8+1+8)
+	return make([]byte, revBytesLen)
 }
 
 func revToBytes(rev revision, bytes []byte) {

storage/watchable_store.go

@@ -247,7 +247,7 @@ func (s *watchableStore) syncWatchings() {
 		if limit == 0 {
 			continue
 		}
-		evs, nextRev, err := s.store.RangeEvents(w.key, end, int64(limit), w.cur)
+		revbs, kvs, nextRev, err := s.store.RangeHistory(w.key, end, int64(limit), w.cur)
 		if err != nil {
 			// TODO: send error event to watching
 			delete(s.unsynced, w)
@@ -255,8 +255,19 @@ func (s *watchableStore) syncWatchings() {
 		}
 
 		// push events to the channel
-		for _, ev := range evs {
-			w.ch <- ev
+		for i, kv := range kvs {
+			var evt storagepb.Event_EventType
+			switch {
+			case isTombstone(revbs[i]):
+				evt = storagepb.DELETE
+			default:
+				evt = storagepb.PUT
+			}
+
+			w.ch <- storagepb.Event{
+				Type: evt,
+				Kv:   &kv,
+			}
 			pendingEventsGauge.Inc()
 		}
 		// switch to tracking future events if needed