From 6103504d4b4650f5d65fcf75c83883a862172498 Mon Sep 17 00:00:00 2001
From: Marek Siarkowicz <siarkowicz@google.com>
Date: Fri, 8 Mar 2024 11:11:40 +0100
Subject: [PATCH] Fix progress notification for watch that doesn't get any
 events

When implementing the fix for progress notifications
(https://github.com/etcd-io/etcd/pull/15237) we made a incorrect
assumption that that unsynched watches will always get at least one event.

Unsynched watches include not only slow watchers, but also newly created
watches that requested current or older revision. In case that non of the events
match watch filter, those newly created watches might become synched
without any event going through.

Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>
---
 server/etcdserver/api/v3rpc/watch.go | 20 +----------
 tests/integration/v3_watch_test.go   | 53 +++++++++++++++++++++++++---
 2 files changed, 50 insertions(+), 23 deletions(-)

diff --git a/server/etcdserver/api/v3rpc/watch.go b/server/etcdserver/api/v3rpc/watch.go
index 3b82fe8d848..9b28319806b 100644
--- a/server/etcdserver/api/v3rpc/watch.go
+++ b/server/etcdserver/api/v3rpc/watch.go
@@ -145,10 +145,6 @@ type serverWatchStream struct {
 	// records fragmented watch IDs
 	fragment map[mvcc.WatchID]bool
 
-	// indicates whether we have an outstanding global progress
-	// notification to send
-	deferredProgress bool
-
 	// closec indicates the stream is closed.
 	closec chan struct{}
 
@@ -178,8 +174,6 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
 		prevKV:   make(map[mvcc.WatchID]bool),
 		fragment: make(map[mvcc.WatchID]bool),
 
-		deferredProgress: false,
-
 		closec: make(chan struct{}),
 	}
 
@@ -367,14 +361,7 @@ func (sws *serverWatchStream) recvLoop() error {
 		case *pb.WatchRequest_ProgressRequest:
 			if uv.ProgressRequest != nil {
 				sws.mu.Lock()
-				// Ignore if deferred progress notification is already in progress
-				if !sws.deferredProgress {
-					// Request progress for all watchers,
-					// force generation of a response
-					if !sws.watchStream.RequestProgressAll() {
-						sws.deferredProgress = true
-					}
-				}
+				sws.watchStream.RequestProgressAll()
 				sws.mu.Unlock()
 			}
 		default:
@@ -483,11 +470,6 @@ func (sws *serverWatchStream) sendLoop() {
 				// elide next progress update if sent a key update
 				sws.progress[wresp.WatchID] = false
 			}
-			if sws.deferredProgress {
-				if sws.watchStream.RequestProgressAll() {
-					sws.deferredProgress = false
-				}
-			}
 			sws.mu.Unlock()
 
 		case c, ok := <-sws.ctrlStream:
diff --git a/tests/integration/v3_watch_test.go b/tests/integration/v3_watch_test.go
index 3d094c57281..40b07fc5f31 100644
--- a/tests/integration/v3_watch_test.go
+++ b/tests/integration/v3_watch_test.go
@@ -1433,8 +1433,8 @@ func TestV3WatchProgressWaitsForSync(t *testing.T) {
 	wch := client.Watch(ctx, "foo", clientv3.WithRev(1))
 
 	// Immediately request a progress notification. As the client
-	// is unsynchronised, the server will have to defer the
-	// notification internally.
+	// is unsynchronised, the server will not sent any notification,
+	//as client can infer progress from events.
 	err := client.RequestProgress(ctx)
 	require.NoError(t, err)
 
@@ -1454,8 +1454,9 @@ func TestV3WatchProgressWaitsForSync(t *testing.T) {
 		}
 		event_count += len(wr.Events)
 	}
-
-	// ... followed by the requested progress notification
+	// client needs to request progress notification again
+	err = client.RequestProgress(ctx)
+	require.NoError(t, err)
 	wr2 := <-wch
 	if wr2.Err() != nil {
 		t.Fatal(fmt.Errorf("watch error: %w", wr2.Err()))
@@ -1467,3 +1468,47 @@ func TestV3WatchProgressWaitsForSync(t *testing.T) {
 		t.Fatal("Wrong revision in progress notification!")
 	}
 }
+
+func TestV3WatchProgressWaitsForSyncNoEvents(t *testing.T) {
+	if integration.ThroughProxy {
+		t.Skip("grpc proxy currently does not support requesting progress notifications")
+	}
+	integration.BeforeTest(t)
+
+	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	client := clus.RandClient()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	resp, err := client.Put(ctx, "bar", "1")
+	require.NoError(t, err)
+
+	wch := client.Watch(ctx, "foo", clientv3.WithRev(resp.Header.Revision))
+	// Request the progress notification on newly created watch that was not yet synced.
+	err = client.RequestProgress(ctx)
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+
+	require.NoError(t, err)
+	gotProgressNotification := false
+	for {
+		select {
+		case <-ticker.C:
+			err := client.RequestProgress(ctx)
+			require.NoError(t, err)
+		case resp := <-wch:
+			if resp.Err() != nil {
+				t.Fatal(fmt.Errorf("watch error: %w", resp.Err()))
+			}
+			if resp.IsProgressNotify() {
+				gotProgressNotification = true
+			}
+		}
+		if gotProgressNotification {
+			break
+		}
+	}
+	require.True(t, gotProgressNotification, "Expected to get progress notification")
+}