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

Signed-off-by: Hitoshi Mitake <h.mitake@gmail.com>

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 {