fix(discov): avoid duplicate watch events and monitors,duplicate addKv

core/discov/internal/registry.go

@@ -54,27 +54,24 @@ func (r *Registry) Monitor(endpoints []string, key string, exactMatch bool, l Up
 		exactMatch: exactMatch,
 	}
 
-	c, exists := r.getOrCreateCluster(endpoints)
-	// if exists, the existing values should be updated to the listener.
-	if exists {
-		c.lock.Lock()
-		watcher, ok := c.watchers[wkey]
-		if ok {
-			watcher.listeners = append(watcher.listeners, l)
+	c, _ := r.getOrCreateCluster(endpoints)
+	kvs, created := c.addListener(wkey, l)
+	if !created {
+		for _, kv := range kvs {
+			l.OnAdd(kv)
 		}
-		c.lock.Unlock()
 
-		if ok {
-			kvs := c.getCurrent(wkey)
-			for _, kv := range kvs {
-				l.OnAdd(kv)
-			}
+		return nil
+	}
 
-			return nil
-		}
+	cli, err := c.getClient()
+	if err != nil {
+		c.removeListener(wkey, l)
+		return err
 	}
 
-	return c.monitor(wkey, l)
+	c.monitor(cli, wkey)
+	return nil
 }
 
 func (r *Registry) Unmonitor(endpoints []string, key string, exactMatch bool, l UpdateListener) {
@@ -172,19 +169,51 @@ func newCluster(endpoints []string) *cluster {
 	}
 }
 
-func (c *cluster) addListener(key watchKey, l UpdateListener) {
+func (c *cluster) addListener(key watchKey, l UpdateListener) (current []KV, created bool) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
 	watcher, ok := c.watchers[key]
 	if ok {
 		watcher.listeners = append(watcher.listeners, l)
-		return
+		current = make([]KV, 0, len(watcher.values))
+		for k, v := range watcher.values {
+			current = append(current, KV{
+				Key: k,
+				Val: v,
+			})
+		}
+		return current, false
 	}
 
 	val := newWatchValue()
 	val.listeners = []UpdateListener{l}
 	c.watchers[key] = val
+	return nil, true
+}
+
+func (c *cluster) removeListener(key watchKey, l UpdateListener) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	watcher, ok := c.watchers[key]
+	if !ok {
+		return
+	}
+
+	for i, listener := range watcher.listeners {
+		if listener == l {
+			watcher.listeners = append(watcher.listeners[:i], watcher.listeners[i+1:]...)
+			break
+		}
+	}
+
+	if len(watcher.listeners) == 0 {
+		if watcher.cancel != nil {
+			watcher.cancel()
+		}
+		delete(c.watchers, key)
+	}
 }
 
 func (c *cluster) getClient() (EtcdClient, error) {
@@ -263,24 +292,55 @@ func (c *cluster) handleWatchEvents(ctx context.Context, key watchKey, events []
 	for _, ev := range events {
 		switch ev.Type {
 		case clientv3.EventTypePut:
+			var remove, add *KV
+			eventKey := string(ev.Kv.Key)
+			eventVal := string(ev.Kv.Value)
 			c.lock.Lock()
-			watcher.values[string(ev.Kv.Key)] = string(ev.Kv.Value)
+			oldVal, ok := watcher.values[eventKey]
+			switch {
+			case ok && oldVal == eventVal:
+				c.lock.Unlock()
+				continue
+			case ok:
+				remove = &KV{
+					Key: eventKey,
+					Val: oldVal,
+				}
+			}
+			watcher.values[eventKey] = eventVal
+			add = &KV{
+				Key: eventKey,
+				Val: eventVal,
+			}
 			c.lock.Unlock()
 			for _, l := range listeners {
-				l.OnAdd(KV{
-					Key: string(ev.Kv.Key),
-					Val: string(ev.Kv.Value),
-				})
+				if remove != nil {
+					l.OnDelete(*remove)
+				}
+				l.OnAdd(*add)
 			}
 		case clientv3.EventTypeDelete:
+			var remove *KV
+			eventKey := string(ev.Kv.Key)
 			c.lock.Lock()
-			delete(watcher.values, string(ev.Kv.Key))
+			if oldVal, ok := watcher.values[eventKey]; ok {
+				remove = &KV{
+					Key: eventKey,
+					Val: oldVal,
+				}
+				delete(watcher.values, eventKey)
+			} else if len(ev.Kv.Value) > 0 {
+				remove = &KV{
+					Key: eventKey,
+					Val: string(ev.Kv.Value),
+				}
+			}
 			c.lock.Unlock()
+			if remove == nil {
+				continue
+			}
 			for _, l := range listeners {
-				l.OnDelete(KV{
-					Key: string(ev.Kv.Key),
-					Val: string(ev.Kv.Value),
-				})
+				l.OnDelete(*remove)
 			}
 		default:
 			logc.Errorf(ctx, "Unknown event type: %v", ev.Type)
@@ -321,19 +381,11 @@ func (c *cluster) load(cli EtcdClient, key watchKey) int64 {
 	return resp.Header.Revision
 }
 
-func (c *cluster) monitor(key watchKey, l UpdateListener) error {
-	cli, err := c.getClient()
-	if err != nil {
-		return err
-	}
-
-	c.addListener(key, l)
+func (c *cluster) monitor(cli EtcdClient, key watchKey) {
 	rev := c.load(cli, key)
 	c.watchGroup.Run(func() {
 		c.watch(cli, key, rev)
 	})
-
-	return nil
 }
 
 func (c *cluster) newClient() (EtcdClient, error) {
@@ -433,7 +485,7 @@ func (c *cluster) setupWatch(cli EtcdClient, key watchKey, rev int64) (context.C
 	}
 
 	ctx, cancel := context.WithCancel(cli.Ctx())
-	
+
 	c.lock.Lock()
 	if watcher, ok := c.watchers[key]; ok {
 		watcher.cancel = cancel

core/discov/internal/registry_test.go

@@ -305,6 +305,116 @@ func TestCluster_handleWatchEvents(t *testing.T) {
 			}, nil)
 		})
 	})
+
+	t.Run("ignore duplicate put", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		l := NewMockUpdateListener(ctrl)
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					listeners: []UpdateListener{l},
+					values: map[string]string{
+						"any/1": "localhost:8080",
+					},
+				},
+			},
+		}
+
+		c.handleWatchEvents(context.Background(), watchKey{key: "any"}, []*clientv3.Event{
+			{
+				Type: clientv3.EventTypePut,
+				Kv: &mvccpb.KeyValue{
+					Key:   []byte("any/1"),
+					Value: []byte("localhost:8080"),
+				},
+			},
+		})
+
+		assert.Equal(t, map[string]string{
+			"any/1": "localhost:8080",
+		}, c.watchers[watchKey{key: "any"}].values)
+	})
+
+	t.Run("put changed value removes old before add", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		l := NewMockUpdateListener(ctrl)
+		gomock.InOrder(
+			l.EXPECT().OnDelete(KV{
+				Key: "any/1",
+				Val: "localhost:8080",
+			}),
+			l.EXPECT().OnAdd(KV{
+				Key: "any/1",
+				Val: "localhost:8081",
+			}),
+		)
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					listeners: []UpdateListener{l},
+					values: map[string]string{
+						"any/1": "localhost:8080",
+					},
+				},
+			},
+		}
+
+		c.handleWatchEvents(context.Background(), watchKey{key: "any"}, []*clientv3.Event{
+			{
+				Type: clientv3.EventTypePut,
+				Kv: &mvccpb.KeyValue{
+					Key:   []byte("any/1"),
+					Value: []byte("localhost:8081"),
+				},
+			},
+		})
+
+		assert.Equal(t, map[string]string{
+			"any/1": "localhost:8081",
+		}, c.watchers[watchKey{key: "any"}].values)
+	})
+
+	t.Run("delete uses previous value", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		l := NewMockUpdateListener(ctrl)
+		l.EXPECT().OnDelete(KV{
+			Key: "any/1",
+			Val: "localhost:8080",
+		})
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					listeners: []UpdateListener{l},
+					values: map[string]string{
+						"any/1": "localhost:8080",
+					},
+				},
+			},
+		}
+
+		c.handleWatchEvents(context.Background(), watchKey{key: "any"}, []*clientv3.Event{
+			{
+				Type: clientv3.EventTypeDelete,
+				Kv: &mvccpb.KeyValue{
+					Key: []byte("any/1"),
+				},
+			},
+		})
+
+		assert.Empty(t, c.watchers[watchKey{key: "any"}].values)
+	})
 }
 
 func TestCluster_addListener(t *testing.T) {
@@ -438,6 +548,43 @@ func TestRegistry_Monitor(t *testing.T) {
 	assert.Error(t, GetRegistry().Monitor(endpoints, "foo", false, new(mockListener)))
 }
 
+func TestRegistry_MonitorDuplicateKeyStartsOneWatch(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	cli := NewMockEtcdClient(ctrl)
+	ch := make(chan clientv3.WatchResponse)
+	watched := make(chan struct{})
+	cli.EXPECT().Ctx().Return(context.Background()).AnyTimes()
+	cli.EXPECT().Get(gomock.Any(), "foo/", gomock.Any()).Return(&clientv3.GetResponse{
+		Header: &etcdserverpb.ResponseHeader{
+			Revision: 1,
+		},
+	}, nil).Times(1)
+	cli.EXPECT().Watch(gomock.Any(), "foo/", gomock.Any()).DoAndReturn(
+		func(context.Context, string, ...clientv3.OpOption) clientv3.WatchChan {
+			close(watched)
+			return ch
+		}).Times(1)
+
+	endpoints := []string{stringx.Rand()}
+	connManager.Inject(getClusterKey(endpoints), cli)
+	reg := &Registry{
+		clusters: make(map[string]*cluster),
+	}
+	assert.NoError(t, reg.Monitor(endpoints, "foo", false, new(mockListener)))
+	assert.NoError(t, reg.Monitor(endpoints, "foo", false, new(mockListener)))
+	<-watched
+
+	c, ok := reg.getCluster(endpoints)
+	assert.True(t, ok)
+	wkey := watchKey{key: "foo"}
+	c.lock.RLock()
+	assert.Len(t, c.watchers[wkey].listeners, 2)
+	c.lock.RUnlock()
+	close(c.done)
+}
+
 func TestRegistry_Unmonitor(t *testing.T) {
 	svr, err := mockserver.StartMockServers(1)
 	assert.NoError(t, err)
@@ -517,7 +664,7 @@ func TestCluster_ConcurrentMonitor(t *testing.T) {
 		go func() {
 			defer wg.Done()
 			key := keys[idx%len(keys)]
-			
+
 			if idx%2 == 0 {
 				// Half the goroutines add listeners (write operation)
 				c.addListener(key, &mockListener{})

core/discov/subscriber.go

@@ -142,6 +142,17 @@ func (c *container) addKv(key, value string) ([]string, bool) {
 	defer c.lock.Unlock()
 
 	c.dirty.Set(true)
+	if old, ok := c.mapping[key]; ok {
+		if old == value {
+			for _, each := range c.values[value] {
+				if each == key {
+					return nil, false
+				}
+			}
+		}
+		c.doRemoveKey(key)
+	}
+
 	keys := c.values[value]
 	previous := append([]string(nil), keys...)
 	early := len(keys) > 0