*: avoid closing a watch with ID 0 incorrectly

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

add test

1

1

1

1

1

1

server/etcdserver/api/v3rpc/watch.go

@@ -16,6 +16,7 @@ package v3rpc
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"math/rand"
 	"sync"
@@ -24,6 +25,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"
+	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/mvcc"
@@ -285,7 +287,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,
@@ -319,7 +321,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),
@@ -356,7 +361,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:
@@ -479,7 +484,12 @@ func (sws *serverWatchStream) sendLoop() {
 
 			// track id creation
 			wid := mvcc.WatchID(c.WatchId)
-			if c.Canceled {
+
+			if !(!(c.Canceled && c.Created) || wid == clientv3.InvalidWatchID) {
+				panic(fmt.Sprintf("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
 			}