mvcc: allow clients to assign watcher IDs

This allows for watchers to be created concurrently without needing potentially complex and latency-adding queuing on the client

Documentation/dev-guide/apispec/swagger/rpc.swagger.json

@@ -2283,6 +2283,11 @@
           "description": "start_revision is an optional revision to watch from (inclusive). No start_revision is \"now\".",
           "type": "string",
           "format": "int64"
+        },
+        "watch_id": {
+          "description": "If watch_id is provided, it will be used to assign to this watcher. Since\ncreating a watcher in etcd is not a synchronous operation, this can be\nused ensure that ordering is correct when creating multiple watchers on\nthe same stream. Creating a watcher with an ID already in use on the\nstream will cause an error to be returned.",
+          "type": "string",
+          "format": "int64"
         }
       }
     },

etcdserver/api/v3rpc/rpctypes/error.go

@@ -57,6 +57,7 @@ var (
 	ErrGRPCAuthNotEnabled       = status.New(codes.FailedPrecondition, "etcdserver: authentication is not enabled").Err()
 	ErrGRPCInvalidAuthToken     = status.New(codes.Unauthenticated, "etcdserver: invalid auth token").Err()
 	ErrGRPCInvalidAuthMgmt      = status.New(codes.InvalidArgument, "etcdserver: invalid auth management").Err()
+	ErrGRPCDuplicateWatcherID   = status.New(codes.InvalidArgument, "etcdserver: duplicate watcher ID").Err()
 
 	ErrGRPCNoLeader                   = status.New(codes.Unavailable, "etcdserver: no leader").Err()
 	ErrGRPCNotLeader                  = status.New(codes.FailedPrecondition, "etcdserver: not leader").Err()

etcdserver/api/v3rpc/watch.go

@@ -170,6 +170,59 @@ func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) bool
 	return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd) == nil
 }
 
+func (sws *serverWatchStream) recvWatchRequest(creq *pb.WatchCreateRequest) (wr *pb.WatchResponse, id mvcc.WatchID, err error) {
+	if len(creq.Key) == 0 {
+		// \x00 is the smallest key
+		creq.Key = []byte{0}
+	}
+	if len(creq.RangeEnd) == 0 {
+		// force nil since watchstream.Watch distinguishes
+		// between nil and []byte{} for single key / >=
+		creq.RangeEnd = nil
+	}
+	if len(creq.RangeEnd) == 1 && creq.RangeEnd[0] == 0 {
+		// support  >= key queries
+		creq.RangeEnd = []byte{}
+	}
+
+	if !sws.isWatchPermitted(creq) {
+		return nil, mvcc.WatchID(creq.WatchId), rpctypes.ErrGRPCPermissionDenied
+	}
+
+	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...)
+	switch {
+	case err == mvcc.ErrEmptyWatcherRange:
+		// fall through, still send the response, but don't bother setting up
+		// the watcher record since nothing can be sent on it.
+	case err != nil:
+		return nil, id, err
+	default:
+		sws.mu.Lock()
+		if creq.ProgressNotify {
+			sws.progress[id] = true
+		}
+		if creq.PrevKv {
+			sws.prevKV[id] = true
+		}
+		sws.mu.Unlock()
+	}
+
+	return &pb.WatchResponse{
+		Header:   sws.newResponseHeader(wsrev),
+		WatchId:  int64(id),
+		Created:  true,
+		Canceled: id == -1,
+	}, id, nil
+}
+
 func (sws *serverWatchStream) recvLoop() error {
 	for {
 		req, err := sws.gRPCStream.Recv()
@@ -185,66 +238,20 @@ func (sws *serverWatchStream) recvLoop() error {
 			if uv.CreateRequest == nil {
 				break
 			}
-
-			creq := uv.CreateRequest
-			if len(creq.Key) == 0 {
-				// \x00 is the smallest key
-				creq.Key = []byte{0}
-			}
-			if len(creq.RangeEnd) == 0 {
-				// force nil since watchstream.Watch distinguishes
-				// between nil and []byte{} for single key / >=
-				creq.RangeEnd = nil
-			}
-			if len(creq.RangeEnd) == 1 && creq.RangeEnd[0] == 0 {
-				// support  >= key queries
-				creq.RangeEnd = []byte{}
-			}
-
-			if !sws.isWatchPermitted(creq) {
-				wr := &pb.WatchResponse{
+			wr, id, err := sws.recvWatchRequest(uv.CreateRequest)
+			if err != nil {
+				wr = &pb.WatchResponse{
 					Header:       sws.newResponseHeader(sws.watchStream.Rev()),
-					WatchId:      -1,
+					WatchId:      int64(id),
 					Canceled:     true,
 					Created:      true,
-					CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
-				}
-
-				select {
-				case sws.ctrlStream <- wr:
-				case <-sws.closec:
+					CancelReason: err.Error(),
 				}
-				return nil
 			}
 
-			filters := FiltersFromRequest(creq)
-
-			wsrev := sws.watchStream.Rev()
-			rev := creq.StartRevision
-			if rev == 0 {
-				rev = wsrev + 1
-			}
-			id := sws.watchStream.Watch(creq.Key, creq.RangeEnd, rev, filters...)
-			if id != -1 {
-				sws.mu.Lock()
-				if creq.ProgressNotify {
-					sws.progress[id] = true
-				}
-				if creq.PrevKv {
-					sws.prevKV[id] = true
-				}
-				sws.mu.Unlock()
-			}
-			wr := &pb.WatchResponse{
-				Header:   sws.newResponseHeader(wsrev),
-				WatchId:  int64(id),
-				Created:  true,
-				Canceled: id == -1,
-			}
 			select {
 			case sws.ctrlStream <- wr:
 			case <-sws.closec:
-				return nil
 			}
 		case *pb.WatchRequest_CancelRequest:
 			if uv.CancelRequest != nil {