*: avoid closing a watch with ID 0 incorrectly

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

server/etcdserver/api/v3rpc/watch.go

@@ -24,6 +24,7 @@ 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"
 	"go.etcd.io/etcd/server/v3/auth"
 	"go.etcd.io/etcd/server/v3/etcdserver"
 	"go.etcd.io/etcd/server/v3/etcdserver/apply"
@@ -270,7 +271,7 @@ func (sws *serverWatchStream) recvLoop() error {
 			if !sws.isWatchPermitted(creq) {
 				wr := &pb.WatchResponse{
 					Header:       sws.newResponseHeader(sws.watchStream.Rev()),
-					WatchId:      creq.WatchId,
+					WatchId:      int64(mvcc.InvalidWatchID),
 					Canceled:     true,
 					Created:      true,
 					CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
@@ -465,7 +466,10 @@ func (sws *serverWatchStream) sendLoop() {
 
 			// track id creation
 			wid := mvcc.WatchID(c.WatchId)
-			if c.Canceled {
+
+			verify.Assert((c.Canceled && c.Created) && wid != mvcc.InvalidWatchID, "the watch response invalid")
+
+			if c.Canceled && wid != mvcc.InvalidWatchID {
 				delete(ids, wid)
 				continue
 			}

server/storage/mvcc/watcher.go

@@ -22,9 +22,14 @@ import (
 	"go.etcd.io/etcd/api/v3/mvccpb"
 )
 
-// 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
+const (
+	// InvalidWatchID represents an invalid watch ID and prevents duplication with an existing watch.
+	InvalidWatchID WatchID = -1
+
+	// 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 WatchID = 0
+)
 
 var (
 	ErrWatcherNotExist    = errors.New("mvcc: watcher does not exist")