Revert "*: handle auth invalid token and old revision errors in watch"

This reverts commit 0c6e466.
etcd-io · Mar 21, 2023 · 614dcca · 614dcca
1 parent 46ae7eb
commit 614dcca
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 47 deletions.
diff --git a/clientv3/watch.go b/clientv3/watch.go
@@ -18,7 +18,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
 	"sync"
 	"time"
 
@@ -586,26 +585,6 @@ func (w *watchGrpcStream) run() {
 
 			switch {
 			case pbresp.Created:
-				cancelReasonError := v3rpc.Error(errors.New(pbresp.CancelReason))
-				if shouldRetryWatch(cancelReasonError) {
-					var newErr error
-					if wc, newErr = w.newWatchClient(); newErr != nil {
-						w.lg.Error("failed to create a new watch client", zap.Error(newErr))
-						return
-					}
-
-					if len(w.resuming) != 0 {
-						if ws := w.resuming[0]; ws != nil {
-							if err := wc.Send(ws.initReq.toPB()); err != nil {
-								w.lg.Debug("error when sending request", zap.Error(err))
-							}
-						}
-					}
-
-					cur = nil
-					continue
-				}
-
 				// response to head of queue creation
 				if ws := w.resuming[0]; ws != nil {
 					w.addSubstream(pbresp, ws)
@@ -725,11 +704,6 @@ func (w *watchGrpcStream) run() {
 	}
 }
 
-func shouldRetryWatch(cancelReasonError error) bool {
-	return (strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCInvalidAuthToken.Error()) == 0) ||
-		(strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCAuthOldRevision.Error()) == 0)
-}
-
 // nextResume chooses the next resuming to register with the grpc stream. Abandoned
 // streams are marked as nil in the queue since the head must wait for its inflight registration.
 func (w *watchGrpcStream) nextResume() *watcherStream {

diff --git a/etcdserver/api/v3rpc/watch.go b/etcdserver/api/v3rpc/watch.go
@@ -234,16 +234,16 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
 	return err
 }
 
-func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) error {
+func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) bool {
 	authInfo, err := sws.ag.AuthInfoFromCtx(sws.gRPCStream.Context())
 	if err != nil {
-		return err
+		return false
 	}
 	if authInfo == nil {
 		// if auth is enabled, IsRangePermitted() can cause an error
 		authInfo = &auth.AuthInfo{}
 	}
-	return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd)
+	return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd) == nil
 }
 
 func (sws *serverWatchStream) recvLoop() error {
@@ -277,29 +277,13 @@ func (sws *serverWatchStream) recvLoop() error {
 				creq.RangeEnd = []byte{}
 			}
 
-			err := sws.isWatchPermitted(creq)
-			if err != nil {
-				var cancelReason string
-				switch err {
-				case auth.ErrInvalidAuthToken:
-					cancelReason = rpctypes.ErrGRPCInvalidAuthToken.Error()
-				case auth.ErrAuthOldRevision:
-					cancelReason = rpctypes.ErrGRPCAuthOldRevision.Error()
-				case auth.ErrUserEmpty:
-					cancelReason = rpctypes.ErrGRPCUserEmpty.Error()
-				default:
-					if err != auth.ErrPermissionDenied {
-						sws.lg.Error("unexpected error code", zap.Error(err))
-					}
-					cancelReason = rpctypes.ErrGRPCPermissionDenied.Error()
-				}
-
+			if !sws.isWatchPermitted(creq) {
 				wr := &pb.WatchResponse{
 					Header:       sws.newResponseHeader(sws.watchStream.Rev()),
 					WatchId:      clientv3.InvalidWatchID,
 					Canceled:     true,
 					Created:      true,
-					CancelReason: cancelReason,
+					CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
 				}
 
 				select {