*: avoid closing a watch with ID 0 incorrectly

Signed-off-by: Kafuu Chino <KafuuChinoQ@gmail.com>

add test

client/v3/watch.go

@@ -25,7 +25,6 @@ import (
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	v3rpc "go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
-
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -38,6 +37,13 @@ const (
 	EventTypePut    = mvccpb.PUT
 
 	closeSendErrTimeout = 250 * time.Millisecond
+
+	// AutoWatchID is the watcher ID passed in WatchStream.Watch when no
+	// user-provided ID is available. If pass, an ID will automatically be assigned.
+	AutoWatchID = 0
+
+	// InvalidWatchID represents an invalid watch ID and prevents duplication with an existing watch.
+	InvalidWatchID = -1
 )
 
 type Event mvccpb.Event
@@ -451,7 +457,7 @@ func (w *watcher) closeStream(wgs *watchGrpcStream) {
 
 func (w *watchGrpcStream) addSubstream(resp *pb.WatchResponse, ws *watcherStream) {
 	// check watch ID for backward compatibility (<= v3.3)
-	if resp.WatchId == -1 || (resp.Canceled && resp.CancelReason != "") {
+	if resp.WatchId == InvalidWatchID || (resp.Canceled && resp.CancelReason != "") {
 		w.closeErr = v3rpc.Error(errors.New(resp.CancelReason))
 		// failed; no channel
 		close(ws.recvc)
@@ -482,7 +488,7 @@ func (w *watchGrpcStream) closeSubstream(ws *watcherStream) {
 	} else if ws.outc != nil {
 		close(ws.outc)
 	}
-	if ws.id != -1 {
+	if ws.id != InvalidWatchID {
 		delete(w.substreams, ws.id)
 		return
 	}
@@ -544,7 +550,7 @@ func (w *watchGrpcStream) run() {
 				// TODO: pass custom watch ID?
 				ws := &watcherStream{
 					initReq: *wreq,
-					id:      -1,
+					id:      InvalidWatchID,
 					outc:    outc,
 					// unbuffered so resumes won't cause repeat events
 					recvc: make(chan *WatchResponse),
@@ -690,7 +696,7 @@ func (w *watchGrpcStream) run() {
 			if len(w.substreams)+len(w.resuming) == 0 {
 				return
 			}
-			if ws.id != -1 {
+			if ws.id != InvalidWatchID {
 				// client is closing an established watch; close it on the server proactively instead of waiting
 				// to close when the next message arrives
 				cancelSet[ws.id] = struct{}{}
@@ -742,9 +748,9 @@ func (w *watchGrpcStream) dispatchEvent(pbresp *pb.WatchResponse) bool {
 		cancelReason:    pbresp.CancelReason,
 	}
 
-	// watch IDs are zero indexed, so request notify watch responses are assigned a watch ID of -1 to
+	// watch IDs are zero indexed, so request notify watch responses are assigned a watch ID of InvalidWatchID to
 	// indicate they should be broadcast.
-	if wr.IsProgressNotify() && pbresp.WatchId == -1 {
+	if wr.IsProgressNotify() && pbresp.WatchId == InvalidWatchID {
 		return w.broadcastResponse(wr)
 	}
 
@@ -899,7 +905,7 @@ func (w *watchGrpcStream) newWatchClient() (pb.Watch_WatchClient, error) {
 	w.resumec = make(chan struct{})
 	w.joinSubstreams()
 	for _, ws := range w.substreams {
-		ws.id = -1
+		ws.id = InvalidWatchID
 		w.resuming = append(w.resuming, ws)
 	}
 	// strip out nils, if any

server/etcdserver/api/v3rpc/watch.go

@@ -24,6 +24,8 @@ import (
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
+	"go.etcd.io/etcd/client/pkg/v3/verify"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/auth"
 	"go.etcd.io/etcd/server/v3/etcdserver"
 	"go.etcd.io/etcd/server/v3/etcdserver/apply"
@@ -286,7 +288,7 @@ func (sws *serverWatchStream) recvLoop() error {
 
 				wr := &pb.WatchResponse{
 					Header:       sws.newResponseHeader(sws.watchStream.Rev()),
-					WatchId:      creq.WatchId,
+					WatchId:      clientv3.InvalidWatchID,
 					Canceled:     true,
 					Created:      true,
 					CancelReason: cancelReason,
@@ -320,7 +322,10 @@ func (sws *serverWatchStream) recvLoop() error {
 					sws.fragment[id] = true
 				}
 				sws.mu.Unlock()
+			} else {
+				id = clientv3.InvalidWatchID
 			}
+
 			wr := &pb.WatchResponse{
 				Header:   sws.newResponseHeader(wsrev),
 				WatchId:  int64(id),
@@ -357,7 +362,7 @@ func (sws *serverWatchStream) recvLoop() error {
 			if uv.ProgressRequest != nil {
 				sws.ctrlStream <- &pb.WatchResponse{
 					Header:  sws.newResponseHeader(sws.watchStream.Rev()),
-					WatchId: -1, // response is not associated with any WatchId and will be broadcast to all watch channels
+					WatchId: clientv3.InvalidWatchID, // response is not associated with any WatchId and will be broadcast to all watch channels
 				}
 			}
 		default:
@@ -481,7 +486,10 @@ func (sws *serverWatchStream) sendLoop() {
 
 			// track id creation
 			wid := mvcc.WatchID(c.WatchId)
-			if c.Canceled {
+
+			verify.Assert(!(c.Canceled && c.Created) || wid == clientv3.InvalidWatchID, "unexpected watchId: %d, wanted: %d, since both 'Canceled' and 'Created' are true", wid, clientv3.InvalidWatchID)
+
+			if c.Canceled && wid != clientv3.InvalidWatchID {
 				delete(ids, wid)
 				continue
 			}

server/proxy/grpcproxy/watch.go

@@ -22,7 +22,6 @@ import (
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	"go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v3rpc"
-
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
@@ -238,7 +237,7 @@ func (wps *watchProxyStream) recvLoop() error {
 			if err := wps.checkPermissionForWatch(cr.Key, cr.RangeEnd); err != nil {
 				wps.watchCh <- &pb.WatchResponse{
 					Header:       &pb.ResponseHeader{},
-					WatchId:      -1,
+					WatchId:      clientv3.InvalidWatchID,
 					Created:      true,
 					Canceled:     true,
 					CancelReason: err.Error(),
@@ -258,7 +257,7 @@ func (wps *watchProxyStream) recvLoop() error {
 				filters:  v3rpc.FiltersFromRequest(cr),
 			}
 			if !w.wr.valid() {
-				w.post(&pb.WatchResponse{WatchId: -1, Created: true, Canceled: true})
+				w.post(&pb.WatchResponse{WatchId: clientv3.InvalidWatchID, Created: true, Canceled: true})
 				wps.mu.Unlock()
 				continue
 			}

server/storage/mvcc/watcher.go

@@ -20,12 +20,9 @@ import (
 	"sync"
 
 	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
-// AutoWatchID is the watcher ID passed in WatchStream.Watch when no
-// user-provided ID is available. If pass, an ID will automatically be assigned.
-const AutoWatchID WatchID = 0
-
 var (
 	ErrWatcherNotExist    = errors.New("mvcc: watcher does not exist")
 	ErrEmptyWatcherRange  = errors.New("mvcc: watcher range is empty")
@@ -118,7 +115,7 @@ func (ws *watchStream) Watch(id WatchID, key, end []byte, startRev int64, fcs ..
 		return -1, ErrEmptyWatcherRange
 	}
 
-	if id == AutoWatchID {
+	if id == clientv3.AutoWatchID {
 		for ws.watchers[ws.nextID] != nil {
 			ws.nextID++
 		}

tests/integration/v3_auth_test.go

@@ -531,3 +531,57 @@ func TestV3AuthWatchAndTokenExpire(t *testing.T) {
 	watchResponse = <-wChan
 	testutil.AssertNil(t, watchResponse.Err())
 }
+
+func TestV3AuthWatchErrorAndWatchId0(t *testing.T) {
+	integration.BeforeTest(t)
+	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
+	defer cancel()
+
+	users := []user{
+		{
+			name:     "user1",
+			password: "user1-123",
+			role:     "role1",
+			key:      "k1",
+			end:      "k2",
+		},
+	}
+	authSetupUsers(t, integration.ToGRPC(clus.Client(0)).Auth, users)
+
+	authSetupRoot(t, integration.ToGRPC(clus.Client(0)).Auth)
+
+	c, cerr := integration.NewClient(t, clientv3.Config{Endpoints: clus.Client(0).Endpoints(), Username: "user1", Password: "user1-123"})
+	if cerr != nil {
+		t.Fatal(cerr)
+	}
+	defer c.Close()
+
+	watchStartCh, watchEndCh := make(chan interface{}), make(chan interface{})
+
+	go func() {
+		wChan := c.Watch(ctx, "k1", clientv3.WithRev(1))
+		watchStartCh <- struct{}{}
+		watchResponse := <-wChan
+		t.Logf("watch response from k1: %v", watchResponse)
+		testutil.AssertTrue(t, len(watchResponse.Events) != 0)
+		watchEndCh <- struct{}{}
+	}()
+
+	// Chan for making sure that the above goroutine invokes Watch()
+	// So the above Watch() can get watch ID = 0
+	<-watchStartCh
+
+	wChan := c.Watch(ctx, "non-allowed-key", clientv3.WithRev(1))
+	watchResponse := <-wChan
+	testutil.AssertNotNil(t, watchResponse.Err()) // permission denied
+
+	_, err := c.Put(ctx, "k1", "val")
+	if err != nil {
+		t.Fatalf("Unexpected error from Put: %v", err)
+	}
+
+	<-watchEndCh
+}

tests/integration/v3_watch_test.go

@@ -26,6 +26,7 @@ import (
 
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v3rpc"
 	"go.etcd.io/etcd/tests/v3/framework/integration"
 )
@@ -390,8 +391,8 @@ func TestV3WatchWrongRange(t *testing.T) {
 		if cresp.Canceled != tt.canceled {
 			t.Fatalf("#%d: canceled %v, want %v", i, tt.canceled, cresp.Canceled)
 		}
-		if tt.canceled && cresp.WatchId != -1 {
-			t.Fatalf("#%d: canceled watch ID %d, want -1", i, cresp.WatchId)
+		if tt.canceled && cresp.WatchId != clientv3.InvalidWatchID {
+			t.Fatalf("#%d: canceled watch ID %d, want %d", i, cresp.WatchId, clientv3.InvalidWatchID)
 		}
 	}
 }