mvcc: sending events after restore

Fixes: etcd-io#8411

mvcc/watchable_store.go

@@ -180,6 +180,21 @@ func (s *watchableStore) cancelWatcher(wa *watcher) {
 	s.mu.Unlock()
 }
 
+func (s *watchableStore) Restore(b backend.Backend) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	err := s.store.Restore(b)
+	if err != nil {
+		return err
+	}
+
+	for wa := range s.synced.watchers {
+		s.unsynced.watchers.add(wa)
+	}
+	s.synced = newWatcherGroup()
+	return nil
+}
+
 // syncWatchersLoop syncs the watcher in the unsynced map every 100ms.
 func (s *watchableStore) syncWatchersLoop() {
 	defer s.wg.Done()

mvcc/watchable_store_test.go

@@ -17,7 +17,9 @@ package mvcc
 import (
 	"bytes"
 	"fmt"
+	"io/ioutil"
 	"os"
+	"path/filepath"
 	"reflect"
 	"sync"
 	"testing"
@@ -296,6 +298,46 @@ func TestWatchFutureRev(t *testing.T) {
 	}
 }
 
+func TestWatchRestore(t *testing.T) {
+	b, tmpPath := backend.NewDefaultTmpBackend()
+	s := newWatchableStore(b, &lease.FakeLessor{}, nil)
+	defer cleanup(s, b, tmpPath)
+
+	testKey := []byte("foo")
+	testValue := []byte("bar")
+	rev := s.Put(testKey, testValue, lease.NoLease)
+
+	dir, err := ioutil.TempDir(os.TempDir(), "etcd_backend_test")
+	if err != nil {
+		plog.Fatal(err)
+	}
+	newPath := filepath.Join(dir, "database_new")
+
+	newBackend := backend.NewDefaultBackend(newPath)
+	newStore := newWatchableStore(newBackend, &lease.FakeLessor{}, nil)
+	defer cleanup(newStore, newBackend, newPath)
+
+	w := newStore.NewWatchStream()
+	w.Watch(testKey, nil, rev-1)
+
+	newStore.Restore(b)
+
+	select {
+	case resp := <-w.Chan():
+		if resp.Revision != rev {
+			t.Fatalf("rev = %d, want %d", resp.Revision, rev)
+		}
+		if len(resp.Events) != 1 {
+			t.Fatalf("failed to get events from the response")
+		}
+		if resp.Events[0].Kv.ModRevision != rev {
+			t.Fatalf("kv.rev = %d, want %d", resp.Events[0].Kv.ModRevision, rev)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("failed to receive event in 1 second.")
+	}
+}
+
 // TestWatchBatchUnsynced tests batching on unsynced watchers
 func TestWatchBatchUnsynced(t *testing.T) {
 	b, tmpPath := backend.NewDefaultTmpBackend()