From adf61b30e356896ba13cc73a8ba3df2406629379 Mon Sep 17 00:00:00 2001 From: yang Date: Tue, 12 May 2026 12:32:59 +0800 Subject: [PATCH 1/2] fix discov duplicate watch adds --- core/discov/internal/registry.go | 53 +++++++++--- core/discov/internal/registry_test.go | 112 +++++++++++++++++++++++++- core/discov/subscriber.go | 11 +++ core/discov/subscriber_test.go | 26 ++++++ 4 files changed, 190 insertions(+), 12 deletions(-) diff --git a/core/discov/internal/registry.go b/core/discov/internal/registry.go index c4ab411707b2..1e5644c0addc 100644 --- a/core/discov/internal/registry.go +++ b/core/discov/internal/registry.go @@ -263,24 +263,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) @@ -433,7 +464,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 diff --git a/core/discov/internal/registry_test.go b/core/discov/internal/registry_test.go index 21f7e079e92a..7cc15ad74a3e 100644 --- a/core/discov/internal/registry_test.go +++ b/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) { @@ -517,7 +627,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{}) diff --git a/core/discov/subscriber.go b/core/discov/subscriber.go index 3b7b43164451..d00de8dfb38f 100644 --- a/core/discov/subscriber.go +++ b/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 diff --git a/core/discov/subscriber_test.go b/core/discov/subscriber_test.go index 4c5c355eb921..5b47eb3b19e3 100644 --- a/core/discov/subscriber_test.go +++ b/core/discov/subscriber_test.go @@ -201,6 +201,32 @@ func TestContainer(t *testing.T) { } } +func TestContainer_AddKvRepeatedKey(t *testing.T) { + t.Run("same key and value only kept once", func(t *testing.T) { + c := newContainer(false) + + c.OnAdd(internal.KV{Key: "first", Val: "a"}) + c.OnAdd(internal.KV{Key: "first", Val: "a"}) + c.OnAdd(internal.KV{Key: "first", Val: "a"}) + + assert.Equal(t, []string{"first"}, c.values["a"]) + assert.Equal(t, "a", c.mapping["first"]) + assert.ElementsMatch(t, []string{"a"}, c.GetValues()) + }) + + t.Run("same key moved to new value", func(t *testing.T) { + c := newContainer(false) + + c.OnAdd(internal.KV{Key: "first", Val: "a"}) + c.OnAdd(internal.KV{Key: "first", Val: "b"}) + + assert.NotContains(t, c.values, "a") + assert.Equal(t, []string{"first"}, c.values["b"]) + assert.Equal(t, "b", c.mapping["first"]) + assert.ElementsMatch(t, []string{"b"}, c.GetValues()) + }) +} + func TestSubscriber(t *testing.T) { sub := new(Subscriber) Exclusive()(sub) From 26f6a8c44c923cac259b77263291899ccfe41b66 Mon Sep 17 00:00:00 2001 From: yang Date: Tue, 12 May 2026 13:02:27 +0800 Subject: [PATCH 2/2] fix discov duplicate watch monitors --- core/discov/internal/registry.go | 75 +++++++++++++++++---------- core/discov/internal/registry_test.go | 37 +++++++++++++ 2 files changed, 85 insertions(+), 27 deletions(-) diff --git a/core/discov/internal/registry.go b/core/discov/internal/registry.go index 1e5644c0addc..8b5a9748face 100644 --- a/core/discov/internal/registry.go +++ b/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) { @@ -352,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) { diff --git a/core/discov/internal/registry_test.go b/core/discov/internal/registry_test.go index 7cc15ad74a3e..a4fd3c83fc12 100644 --- a/core/discov/internal/registry_test.go +++ b/core/discov/internal/registry_test.go @@ -548,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)