Merge pull request #20245 from serathius/skip-future-progress-notification-release-3.5

[release-3.5] Skip sending progress notification for watch with starting revision in the future

Author: fuweid

server/etcdserver/api/v3rpc/watch.go

@@ -303,12 +303,7 @@ func (sws *serverWatchStream) recvLoop() error {
 
 			filters := FiltersFromRequest(creq)
 
-			wsrev := sws.watchStream.Rev()
-			rev := creq.StartRevision
-			if rev == 0 {
-				rev = wsrev + 1
-			}
-			id, err := sws.watchStream.Watch(mvcc.WatchID(creq.WatchId), creq.Key, creq.RangeEnd, rev, filters...)
+			id, err := sws.watchStream.Watch(mvcc.WatchID(creq.WatchId), creq.Key, creq.RangeEnd, creq.StartRevision, filters...)
 			if err == nil {
 				sws.mu.Lock()
 				if creq.ProgressNotify {
@@ -326,7 +321,7 @@ func (sws *serverWatchStream) recvLoop() error {
 			}
 
 			wr := &pb.WatchResponse{
-				Header:   sws.newResponseHeader(wsrev),
+				Header:   sws.newResponseHeader(sws.watchStream.Rev()),
 				WatchId:  int64(id),
 				Created:  true,
 				Canceled: err != nil,

server/mvcc/watchable_store.go

@@ -117,12 +117,13 @@ func (s *watchableStore) NewWatchStream() WatchStream {
 
 func (s *watchableStore) watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse, fcs ...FilterFunc) (*watcher, cancelFunc) {
 	wa := &watcher{
-		key:    key,
-		end:    end,
-		minRev: startRev,
-		id:     id,
-		ch:     ch,
-		fcs:    fcs,
+		key:      key,
+		end:      end,
+		startRev: startRev,
+		minRev:   startRev,
+		id:       id,
+		ch:       ch,
+		fcs:      fcs,
 	}
 
 	s.mu.Lock()
@@ -495,11 +496,15 @@ func (s *watchableStore) progressIfSync(watchers map[WatchID]*watcher, responseW
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
+	rev := s.rev()
 	// Any watcher unsynced?
 	for _, w := range watchers {
 		if _, ok := s.synced.watchers[w]; !ok {
 			return false
 		}
+		if rev < w.startRev {
+			return false
+		}
 	}
 
 	// If all watchers are synchronised, send out progress
@@ -508,7 +513,7 @@ func (s *watchableStore) progressIfSync(watchers map[WatchID]*watcher, responseW
 	// notification will be broadcasted client-side if required
 	// (see dispatchEvent in client/v3/watch.go)
 	for _, w := range watchers {
-		w.send(WatchResponse{WatchID: responseWatchID, Revision: s.rev()})
+		w.send(WatchResponse{WatchID: responseWatchID, Revision: rev})
 		return true
 	}
 	return true
@@ -535,6 +540,7 @@ type watcher struct {
 	// except when the watcher were to be moved from "synced" watcher group
 	restore bool
 
+	startRev int64
 	// minRev is the minimum revision update the watcher will accept
 	minRev int64
 	id     WatchID

server/mvcc/watcher_test.go

@@ -288,9 +288,7 @@ func TestWatchStreamCancelWatcherByID(t *testing.T) {
 	}
 }
 
-// TestWatcherRequestProgress ensures synced watcher can correctly
-// report its correct progress.
-func TestWatcherRequestProgress(t *testing.T) {
+func TestWatcherRequestProgressBadId(t *testing.T) {
 	b, tmpPath := betesting.NewDefaultTmpBackend(t)
 
 	// manually create watchableStore instead of newWatchableStore
@@ -302,14 +300,12 @@ func TestWatcherRequestProgress(t *testing.T) {
 		unsynced: newWatcherGroup(),
 		synced:   newWatcherGroup(),
 	}
-
 	defer func() {
 		s.store.Close()
 		os.Remove(tmpPath)
 	}()
 
 	testKey := []byte("foo")
-	notTestKey := []byte("bad")
 	testValue := []byte("bar")
 	s.Put(testKey, testValue, lease.NoLease)
 
@@ -322,26 +318,91 @@ func TestWatcherRequestProgress(t *testing.T) {
 		t.Fatalf("unexpected %+v", resp)
 	default:
 	}
+}
 
-	id, _ := w.Watch(0, notTestKey, nil, 1)
-	w.RequestProgress(id)
-	select {
-	case resp := <-w.Chan():
-		t.Fatalf("unexpected %+v", resp)
-	default:
+func TestWatcherRequestProgress(t *testing.T) {
+	testKey := []byte("foo")
+	notTestKey := []byte("bad")
+	testValue := []byte("bar")
+	tcs := []struct {
+		name                     string
+		startRev                 int64
+		expectProgressBeforeSync bool
+		expectProgressAfterSync  bool
+	}{
+		{
+			name:                     "Zero revision",
+			startRev:                 0,
+			expectProgressBeforeSync: true,
+			expectProgressAfterSync:  true,
+		},
+		{
+			name:                    "Old revision",
+			startRev:                1,
+			expectProgressAfterSync: true,
+		},
+		{
+			name:                    "Current revision",
+			startRev:                2,
+			expectProgressAfterSync: true,
+		},
+		{
+			name:     "Current revision plus one",
+			startRev: 3,
+		},
+		{
+			name:     "Current revision plus two",
+			startRev: 4,
+		},
 	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			b, tmpPath := betesting.NewDefaultTmpBackend(t)
+
+			// manually create watchableStore instead of newWatchableStore
+			// because newWatchableStore automatically calls syncWatchers
+			// method to sync watchers in unsynced map. We want to keep watchers
+			// in unsynced to test if syncWatchers works as expected.
+			s := &watchableStore{
+				store:    NewStore(zap.NewExample(), b, &lease.FakeLessor{}, StoreConfig{}),
+				unsynced: newWatcherGroup(),
+				synced:   newWatcherGroup(),
+			}
 
-	s.syncWatchers()
+			defer func() {
+				s.store.Close()
+				os.Remove(tmpPath)
+			}()
+
+			s.Put(testKey, testValue, lease.NoLease)
 
-	w.RequestProgress(id)
-	wrs := WatchResponse{WatchID: id, Revision: 2}
+			w := s.NewWatchStream()
+
+			id, _ := w.Watch(0, notTestKey, nil, tc.startRev)
+			w.RequestProgress(id)
+			asssertProgressSent(t, w, id, tc.expectProgressBeforeSync)
+			s.syncWatchers()
+			w.RequestProgress(id)
+			asssertProgressSent(t, w, id, tc.expectProgressAfterSync)
+		})
+	}
+}
+
+func asssertProgressSent(t *testing.T, stream WatchStream, id WatchID, expectProgress bool) {
 	select {
-	case resp := <-w.Chan():
-		if !reflect.DeepEqual(resp, wrs) {
-			t.Fatalf("got %+v, expect %+v", resp, wrs)
+	case resp := <-stream.Chan():
+		if expectProgress {
+			wrs := WatchResponse{WatchID: id, Revision: 2}
+			if !reflect.DeepEqual(resp, wrs) {
+				t.Fatalf("got %+v, expect %+v", resp, wrs)
+			}
+		} else {
+			t.Fatalf("unexpected response %+v", resp)
+		}
+	default:
+		if expectProgress {
+			t.Fatalf("failed to receive progress")
 		}
-	case <-time.After(time.Second):
-		t.Fatal("failed to receive progress")
 	}
 }