restore dupe detection

Signed-off-by: Tim Vaillancourt <[email protected]>

go/vt/vtorc/discovery/queue.go

@@ -26,6 +26,7 @@ push() operation never blocks while pop() blocks on an empty queue.
 package discovery
 
 import (
+	"sync"
 	"time"
 
 	"vitess.io/vitess/go/vt/log"
@@ -40,7 +41,8 @@ type queueItem struct {
 
 // Queue is an implementation of discovery.Queue.
 type Queue struct {
-	queue chan queueItem
+	enqueued sync.Map
+	queue    chan queueItem
 }
 
 // NewQueue creates a new queue.
@@ -58,10 +60,14 @@ func (q *Queue) QueueLen() int {
 // Push enqueues a key if it is not on a queue and is not being
 // processed; silently returns otherwise.
 func (q *Queue) Push(key string) {
+	if _, found := q.enqueued.Load(key); found {
+		return
+	}
 	q.queue <- queueItem{
 		PushedAt: time.Now(),
 		Key:      key,
 	}
+	q.enqueued.Store(key, struct{}{})
 }
 
 // Consume fetches a key to process; blocks if queue is empty.
@@ -76,3 +82,9 @@ func (q *Queue) Consume() string {
 
 	return item.Key
 }
+
+// Release removes a key from a list of being processed keys
+// which allows that key to be pushed into the queue again.
+func (q *Queue) Release(key string) {
+	q.enqueued.Delete(key)
+}

go/vt/vtorc/discovery/queue_test.go

@@ -29,8 +29,22 @@ func TestQueue(t *testing.T) {
 	// Push
 	q.Push(t.Name())
 	require.Equal(t, 1, q.QueueLen())
+	_, found := q.enqueued.Load(t.Name())
+	require.True(t, found)
+
+	// Push duplicate
+	q.Push(t.Name())
+	require.Equal(t, 1, q.QueueLen())
 
 	// Consume
 	require.Equal(t, t.Name(), q.Consume())
 	require.Zero(t, q.QueueLen())
+	_, found = q.enqueued.Load(t.Name())
+	require.True(t, found)
+
+	// Release
+	q.Release(t.Name())
+	require.Zero(t, q.QueueLen())
+	_, found = q.enqueued.Load(t.Name())
+	require.False(t, found)
 }

go/vt/vtorc/logic/vtorc.go

@@ -115,6 +115,7 @@ func handleDiscoveryRequests() {
 			for {
 				tabletAlias := discoveryQueue.Consume()
 				DiscoverInstance(tabletAlias, false /* forceDiscovery */)
+				discoveryQueue.Release(tabletAlias)
 			}
 		}()
 	}