From 2dcfa83094fa0f396c10467090d31208a574f920 Mon Sep 17 00:00:00 2001
From: Hitoshi Mitake <h.mitake@gmail.com>
Date: Sun, 31 Jul 2022 23:42:38 +0900
Subject: [PATCH] *: handle auth invalid token and old revision errors in watch

Signed-off-by: Hitoshi Mitake <h.mitake@gmail.com>
---
 client/v3/watch.go                   | 26 ++++++++++++++++++++++++++
 server/etcdserver/api/v3rpc/watch.go | 26 +++++++++++++++++++++-----
 2 files changed, 47 insertions(+), 5 deletions(-)

diff --git a/client/v3/watch.go b/client/v3/watch.go
index b73925ba1..5bd2d4cd0 100644
--- a/client/v3/watch.go
+++ b/client/v3/watch.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -580,6 +581,26 @@ 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 len(w.resuming) != 0 {
 					if ws := w.resuming[0]; ws != nil {
@@ -688,6 +709,11 @@ 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/server/etcdserver/api/v3rpc/watch.go b/server/etcdserver/api/v3rpc/watch.go
index 1a3cff539..c206f7002 100644
--- a/server/etcdserver/api/v3rpc/watch.go
+++ b/server/etcdserver/api/v3rpc/watch.go
@@ -224,16 +224,16 @@ func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
 	return err
 }
 
-func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) bool {
+func (sws *serverWatchStream) isWatchPermitted(wcr *pb.WatchCreateRequest) error {
 	authInfo, err := sws.ag.AuthInfoFromCtx(sws.gRPCStream.Context())
 	if err != nil {
-		return false
+		return err
 	}
 	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) == nil
+	return sws.ag.AuthStore().IsRangePermitted(authInfo, wcr.Key, wcr.RangeEnd)
 }
 
 func (sws *serverWatchStream) recvLoop() error {
@@ -267,13 +267,29 @@ func (sws *serverWatchStream) recvLoop() error {
 				creq.RangeEnd = []byte{}
 			}
 
-			if !sws.isWatchPermitted(creq) {
+			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()
+				}
+
 				wr := &pb.WatchResponse{
 					Header:       sws.newResponseHeader(sws.watchStream.Rev()),
 					WatchId:      creq.WatchId,
 					Canceled:     true,
 					Created:      true,
-					CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
+					CancelReason: cancelReason,
 				}
 
 				select {