Merge pull request #20241 from serathius/skip-future-progress-notification-release-3.6

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

Author: serathius

server/etcdserver/api/v3rpc/watch.go

@@ -305,12 +305,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 {
@@ -328,7 +323,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/storage/mvcc/watchable_store.go

@@ -122,12 +122,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()
@@ -532,11 +533,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
@@ -545,7 +550,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
@@ -572,6 +577,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/storage/mvcc/watcher_test.go

@@ -289,48 +289,92 @@ 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, _ := betesting.NewDefaultTmpBackend(t)
 	s := newWatchableStore(zaptest.NewLogger(t), b, &lease.FakeLessor{}, StoreConfig{})
 
 	defer cleanup(s, b)
-
-	testKey := []byte("foo")
-	notTestKey := []byte("bad")
-	testValue := []byte("bar")
-	s.Put(testKey, testValue, lease.NoLease)
-
 	w := s.NewWatchStream()
-
 	badID := WatchID(1000)
 	w.RequestProgress(badID)
 	select {
 	case resp := <-w.Chan():
 		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, _ := betesting.NewDefaultTmpBackend(t)
+			s := newWatchableStore(zaptest.NewLogger(t), b, &lease.FakeLessor{}, StoreConfig{})
 
-	s.syncWatchers([]mvccpb.Event{})
+			defer cleanup(s, b)
+
+			s.Put(testKey, testValue, lease.NoLease)
+
+			w := s.NewWatchStream()
+
+			id, _ := w.Watch(0, notTestKey, nil, tc.startRev)
+			w.RequestProgress(id)
+			asssertProgressSent(t, w, id, tc.expectProgressBeforeSync)
+			s.syncWatchers([]mvccpb.Event{})
+			w.RequestProgress(id)
+			asssertProgressSent(t, w, id, tc.expectProgressAfterSync)
+		})
+	}
+}
 
-	w.RequestProgress(id)
-	wrs := WatchResponse{WatchID: id, Revision: 2}
+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")
 	}
 }