receipt: dequeue not returning watch err

the error from watch is not returned, if the client token expired, the `ev` return by watch will be nil, therefore the Deque method will panic in such case.

Signed-off-by: guokairui <wxsms@foxmail.com>

client/v3/experimental/recipes/client.go

@@ -27,6 +27,7 @@ var (
 	ErrWaitMismatch   = errors.New("unexpected wait result")
 	ErrTooManyClients = errors.New("too many clients")
 	ErrNoWatcher      = errors.New("no watcher channel")
+	ErrWatchFailed    = errors.New("watch failed")
 )
 
 // deleteRevKey deletes a key by revision, returning false if key is missing

client/v3/experimental/recipes/watch.go

@@ -29,7 +29,7 @@ func WaitEvents(c *clientv3.Client, key string, rev int64, evs []mvccpb.Event_Ev
 	if wc == nil {
 		return nil, ErrNoWatcher
 	}
-	return waitEvents(wc, evs), nil
+	return waitEvents(wc, evs)
 }
 
 func WaitPrefixEvents(c *clientv3.Client, prefix string, rev int64, evs []mvccpb.Event_EventType) (*clientv3.Event, error) {
@@ -39,20 +39,23 @@ func WaitPrefixEvents(c *clientv3.Client, prefix string, rev int64, evs []mvccpb
 	if wc == nil {
 		return nil, ErrNoWatcher
 	}
-	return waitEvents(wc, evs), nil
+	return waitEvents(wc, evs)
 }
 
-func waitEvents(wc clientv3.WatchChan, evs []mvccpb.Event_EventType) *clientv3.Event {
+func waitEvents(wc clientv3.WatchChan, evs []mvccpb.Event_EventType) (*clientv3.Event, error) {
 	i := 0
 	for wresp := range wc {
+		if wresp.Canceled {
+			return nil, wresp.Err()
+		}
 		for _, ev := range wresp.Events {
 			if ev.Type == evs[i] {
 				i++
 				if i == len(evs) {
-					return ev
+					return ev, nil
 				}
 			}
 		}
 	}
-	return nil
+	return nil, ErrWatchFailed
 }

tests/integration/clientv3/experimental/recipes/v3_queue_test.go

@@ -17,8 +17,10 @@ package recipes_test
 import (
 	"fmt"
 	"math/rand"
+	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	recipe "go.etcd.io/etcd/client/v3/experimental/recipes"
 	integration2 "go.etcd.io/etcd/tests/v3/framework/integration"
@@ -78,6 +80,27 @@ func TestQueueManyReaderManyWriter(t *testing.T) {
 	testQueueNReaderMWriter(t, manyQueueClients, manyQueueClients)
 }
 
+// TestQueueAuthTokenExpired queue should return err on watch failed
+func TestQueueWatchError(t *testing.T) {
+	integration2.BeforeTest(t)
+	clus := integration2.NewCluster(t, &integration2.ClusterConfig{Size: 1})
+	etcdc := clus.RandClient()
+	queue := recipe.NewQueue(etcdc, "testq")
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		_, err := queue.Dequeue()
+		if err != recipe.ErrWatchFailed {
+			t.Errorf("expect ErrWatchFailed, got %v", err)
+		}
+		wg.Done()
+	}()
+	time.Sleep(1 * time.Second)
+	// queue.Enqueue("1")
+	clus.Terminate(t)
+	wg.Wait()
+}
+
 // BenchmarkQueue benchmarks Queues using many/many readers/writers
 func BenchmarkQueue(b *testing.B) {
 	integration2.BeforeTest(b)