This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new 5683612b1 fix: remove stale providers after initial subscribe load
(#3479)
5683612b1 is described below
commit 5683612b1dfbbf755e5bb10bfcc6faf56d7d1d19
Author: AsperforMias <[email protected]>
AuthorDate: Wed Jul 8 19:27:28 2026 +0800
fix: remove stale providers after initial subscribe load (#3479)
* fix(registry/nacos): seed listener with initial instances
---
registry/nacos/listener.go | 27 ++++++-
registry/nacos/registry.go | 40 +++++++++--
registry/nacos/registry_test.go | 155 ++++++++++++++++++++++++++++++++++++++--
3 files changed, 210 insertions(+), 12 deletions(-)
diff --git a/registry/nacos/listener.go b/registry/nacos/listener.go
index d9b0bd01c..8a5971642 100644
--- a/registry/nacos/listener.go
+++ b/registry/nacos/listener.go
@@ -74,6 +74,29 @@ func NewNacosListenerWithServiceName(serviceName string,
regURL *common.URL, nam
}
}
+func (nl *nacosListener) setInstanceSnapshot(instances []model.Instance) {
+ nl.cacheLock.Lock()
+ defer nl.cacheLock.Unlock()
+
+ nl.instanceMap = buildInstanceMap(instances)
+}
+
+func buildInstanceMap(instances []model.Instance) map[string]model.Instance {
+ instanceMap := make(map[string]model.Instance, len(instances))
+ for i := range instances {
+ if !instances[i].Enable {
+ continue
+ }
+ host := instanceHost(instances[i])
+ instanceMap[host] = instances[i]
+ }
+ return instanceMap
+}
+
+func instanceHost(instance model.Instance) string {
+ return instance.Ip + ":" + strconv.Itoa(int(instance.Port))
+}
+
func generateUrl(instance model.Instance) *common.URL {
if instance.Metadata == nil {
logger.Errorf("[Registry][Nacos] nacos instance metadata is
empty, instance=%+v", instance)
@@ -125,7 +148,7 @@ func (nl *nacosListener) Callback(services
[]model.Instance, err error) {
// instance is not available,so ignore it
continue
}
- host := services[i].Ip + ":" +
strconv.Itoa(int(services[i].Port))
+ host := instanceHost(services[i])
instance := services[i]
newInstanceMap[host] = instance
if old, ok := nl.instanceMap[host]; !ok && instance.Healthy {
@@ -184,7 +207,7 @@ func (nl *nacosListener) listenService(serviceName string)
error {
}
err := nl.namingClient.Client().Subscribe(nl.subscribeParam)
if err == nil {
-
listenerCache.Store(nl.subscribeParam.ServiceName+nl.subscribeParam.GroupName,
nl)
+
listenerCache.Store(subscribeCacheKey(nl.subscribeParam.ServiceName,
nl.subscribeParam.GroupName), nl)
}
return nil
}
diff --git a/registry/nacos/registry.go b/registry/nacos/registry.go
index 745f1db53..7b47366ff 100644
--- a/registry/nacos/registry.go
+++ b/registry/nacos/registry.go
@@ -34,6 +34,7 @@ import (
nacosClient "github.com/dubbogo/gost/database/kv/nacos"
"github.com/dubbogo/gost/log/logger"
+ "github.com/nacos-group/nacos-sdk-go/v2/model"
"github.com/nacos-group/nacos-sdk-go/v2/vo"
perrors "github.com/pkg/errors"
@@ -61,11 +62,14 @@ func init() {
type nacosRegistry struct {
*common.URL
+ // namingClient is the pooled Nacos client shared by registry
operations and listeners.
namingClient *nacosClient.NacosNamingClient
registryUrls []*common.URL
- done chan struct{}
- availability availabilityCache
- wg sync.WaitGroup
+ // initialSubscribeInstances keeps the synchronous
LoadSubscribeInstances result as the diff baseline for the first Nacos push.
+ initialSubscribeInstances sync.Map
+ done chan struct{}
+ availability availabilityCache
+ wg sync.WaitGroup
}
type availabilityCache struct {
@@ -256,7 +260,7 @@ func (nr *nacosRegistry) subscribeAll(url *common.URL,
notifyListener registry.N
return
}
for _, name := range serviceNames {
- if _, ok := listenerCache.Load(name + groupName); ok {
+ if _, ok := listenerCache.Load(subscribeCacheKey(name,
groupName)); ok {
// has subscribed ,ignore
continue
}
@@ -279,6 +283,13 @@ func (nr *nacosRegistry) subscribe(serviceName string,
notifyListener registry.N
return perrors.New("nacosRegistry is not available.")
}
listener := NewNacosListenerWithServiceName(serviceName, nr.URL,
nr.namingClient)
+ groupName := nr.GetParam(constant.RegistryGroupKey, defaultGroup)
+ cacheKey := subscribeCacheKey(serviceName, groupName)
+ if value, ok := nr.initialSubscribeInstances.Load(cacheKey); ok {
+ if instances, ok := value.([]model.Instance); ok {
+ listener.setInstanceSnapshot(instances)
+ }
+ }
// will add to listenerCache when subscribe success
err := listener.listenService(serviceName)
metrics.Publish(metricsRegistry.NewSubscribeEvent(err == nil))
@@ -286,6 +297,7 @@ func (nr *nacosRegistry) subscribe(serviceName string,
notifyListener registry.N
logger.Warnf("[Registry][Nacos] subscribe service %s err=%v",
serviceName, perrors.WithStack(err))
return err
}
+ nr.initialSubscribeInstances.Delete(cacheKey)
// handleServiceEvents will block to wait notify event and exit when
error occur
nr.wg.Add(1)
go func() {
@@ -343,7 +355,9 @@ func (nr *nacosRegistry) handleServiceEvents(listener
registry.Listener, notifyL
// UnSubscribe :
func (nr *nacosRegistry) UnSubscribe(url *common.URL, _
registry.NotifyListener) error {
- param := createSubscribeParam(getSubscribeName(url),
nr.GetParam(constant.RegistryGroupKey, defaultGroup), nil)
+ serviceName := getSubscribeName(url)
+ groupName := nr.GetParam(constant.RegistryGroupKey, defaultGroup)
+ param := createSubscribeParam(serviceName, groupName, nil)
if param == nil {
return nil
}
@@ -351,6 +365,7 @@ func (nr *nacosRegistry) UnSubscribe(url *common.URL, _
registry.NotifyListener)
if err != nil {
return perrors.New("UnSubscribe [" + param.ServiceName + "] to
nacos failed")
}
+ nr.initialSubscribeInstances.Delete(subscribeCacheKey(serviceName,
groupName))
return nil
}
@@ -366,6 +381,11 @@ func (nr *nacosRegistry) LoadSubscribeInstances(url
*common.URL, notify registry
return perrors.New(fmt.Sprintf("could not query the instances
for serviceName=%s,groupName=%s,error=%v",
serviceName, groupName, err))
}
+ if len(instances) > 0 {
+ copied := make([]model.Instance, len(instances))
+ copy(copied, instances)
+
nr.initialSubscribeInstances.Store(subscribeCacheKey(serviceName, groupName),
copied)
+ }
for i := range instances {
if newUrl := generateUrl(instances[i]); newUrl != nil {
@@ -375,9 +395,13 @@ func (nr *nacosRegistry) LoadSubscribeInstances(url
*common.URL, notify registry
return nil
}
+func subscribeCacheKey(serviceName string, groupName string) string {
+ return fmt.Sprintf("%d:%s:%s", len(serviceName), serviceName, groupName)
+}
+
func createSubscribeParam(serviceName string, groupName string, cb callback)
*vo.SubscribeParam {
if cb == nil {
- v, ok := listenerCache.Load(serviceName + groupName)
+ v, ok := listenerCache.Load(subscribeCacheKey(serviceName,
groupName))
if !ok {
return nil
}
@@ -453,6 +477,10 @@ func (nr *nacosRegistry) Destroy() {
}
nr.registryUrls = nil
+ nr.initialSubscribeInstances.Range(func(key any, _ any) bool {
+ nr.initialSubscribeInstances.Delete(key)
+ return true
+ })
nr.CloseAndNilClient()
}
diff --git a/registry/nacos/registry_test.go b/registry/nacos/registry_test.go
index 3b157b4d2..d7ae17533 100644
--- a/registry/nacos/registry_test.go
+++ b/registry/nacos/registry_test.go
@@ -40,6 +40,7 @@ import (
"dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/registry"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
)
// MockINamingClient is a mock of INamingClient interface
@@ -426,7 +427,7 @@ func TestNacosRegistryDestroy(t *testing.T) {
SubscribeCallback: nl.Callback,
}
nl.subscribeParam = subscribeParam
- listenerCache.Store(serviceName+"testgroup", nl)
+ listenerCache.Store(subscribeCacheKey(serviceName, "testgroup"), nl)
// Simulate unsubscribe and unregister instances
mockNamingClient.EXPECT().Unsubscribe(subscribeParam).Return(nil)
@@ -451,7 +452,7 @@ func TestNacosRegistryDestroy(t *testing.T) {
t.Errorf("namingClient was not set to nil")
}
- if _, ok := listenerCache.Load(serviceName + "testgroup"); ok {
+ if _, ok := listenerCache.Load(subscribeCacheKey(serviceName,
"testgroup")); ok {
t.Errorf("listenerCache was not cleared")
}
}
@@ -601,7 +602,7 @@ func TestNacosRegistryCloseListener(t *testing.T) {
SubscribeCallback: nl.Callback,
}
nl.subscribeParam = subscribeParam
- listenerCache.Store(serviceName+"testgroup", nl)
+ listenerCache.Store(subscribeCacheKey(serviceName, "testgroup"), nl)
mockNamingClient.EXPECT().Unsubscribe(subscribeParam).Return(nil)
@@ -610,7 +611,7 @@ func TestNacosRegistryCloseListener(t *testing.T) {
nr.CloseListener()
// Verify whether to clear the entries in the listenerCache
- if _, ok := listenerCache.Load(serviceName + "testgroup"); ok {
+ if _, ok := listenerCache.Load(subscribeCacheKey(serviceName,
"testgroup")); ok {
t.Errorf("listenerCache was not cleared")
}
@@ -685,3 +686,149 @@ func
TestNacosRegistryScheduledLookUpStopsWhenDoneClosed(t *testing.T) {
t.Fatalf("scheduledLookUp should exit quickly after done is
closed")
}
}
+
+func TestNacosRegistryInitialSubscribeSnapshotDiff(t *testing.T) {
+ clearListenerCacheForTest()
+
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ mockNamingClient := NewMockINamingClient(ctrl)
+ nc := &nacosClient.NacosNamingClient{}
+ nc.SetClient(mockNamingClient)
+
+ regURL, _ :=
common.NewURL("registry://127.0.0.1:8848?registry.group=testgroup")
+ nr := &nacosRegistry{
+ URL: regURL,
+ namingClient: nc,
+ done: make(chan struct{}),
+ registryUrls: []*common.URL{},
+ }
+
+ urlMap := url.Values{}
+ urlMap.Set(constant.RegistryRoleKey, strconv.Itoa(common.CONSUMER))
+ urlMap.Set(constant.InterfaceKey, "com.test.InitialSnapshotService")
+ testURL, _ :=
common.NewURL("consumer://127.0.0.1/com.test.InitialSnapshotService",
common.WithParams(urlMap))
+
+ instanceA := newNacosTestInstance("10.0.0.1", 20001,
"com.test.InitialSnapshotService")
+ instanceB := newNacosTestInstance("10.0.0.2", 20002,
"com.test.InitialSnapshotService")
+
+ var subscribeParam *vo.SubscribeParam
+
mockNamingClient.EXPECT().SelectAllInstances(gomock.Any()).Return([]model.Instance{instanceA,
instanceB}, nil)
+
mockNamingClient.EXPECT().GetAllServicesInfo(gomock.Any()).Return(model.ServiceList{},
nil).AnyTimes()
+
mockNamingClient.EXPECT().Subscribe(gomock.Any()).DoAndReturn(func(param
*vo.SubscribeParam) error {
+ subscribeParam = param
+ return nil
+ })
+ mockNamingClient.EXPECT().Unsubscribe(gomock.Any()).AnyTimes()
+
+ notify := newRecordingNotifyListener()
+ if err := nr.LoadSubscribeInstances(testURL, notify); err != nil {
+ t.Fatalf("LoadSubscribeInstances() error = %v", err)
+ }
+ assertRecordedEvent(t, notify.nextEvent(t), remoting.EventTypeAdd,
"10.0.0.1", "20001")
+ assertRecordedEvent(t, notify.nextEvent(t), remoting.EventTypeAdd,
"10.0.0.2", "20002")
+
+ if err := nr.subscribe(getSubscribeName(testURL), notify); err != nil {
+ t.Fatalf("subscribe() error = %v", err)
+ }
+ if subscribeParam == nil || subscribeParam.SubscribeCallback == nil {
+ t.Fatalf("SubscribeCallback was not captured")
+ }
+
+ subscribeParam.SubscribeCallback([]model.Instance{instanceA}, nil)
+ assertRecordedEvent(t, notify.nextEvent(t), remoting.EventTypeDel,
"10.0.0.2", "20002")
+ notify.assertNoEvent(t)
+
+ nr.CloseListener()
+ close(nr.done)
+ nr.wg.Wait()
+}
+
+func clearListenerCacheForTest() {
+ listenerCache.Range(func(key any, _ any) bool {
+ listenerCache.Delete(key)
+ return true
+ })
+}
+
+func newNacosTestInstance(ip string, port uint64, interfaceName string)
model.Instance {
+ return model.Instance{
+ Ip: ip,
+ Port: port,
+ Enable: true,
+ Healthy: true,
+ Ephemeral: true,
+ Metadata: map[string]string{
+ constant.InterfaceKey: interfaceName,
+ constant.NacosPathKey: "/" + interfaceName,
+ constant.NacosProtocolKey: "tri",
+ constant.NacosCategoryKey:
common.DubboNodes[common.PROVIDER],
+ constant.RegistryRoleKey:
strconv.Itoa(common.PROVIDER),
+ constant.RegistryGroupKey: "testgroup",
+ constant.SerializationKey: "protobuf",
+ constant.ReleaseKey: "test",
+ constant.ApplicationKey: "test-provider",
+ constant.MethodsKey: "",
+ constant.TimestampKey: "1",
+ constant.AnyhostKey: "true",
+ constant.TokenKey: "",
+ constant.WarmupKey: "",
+ constant.SideKey: "provider",
+ constant.NacosGroupKey: "testgroup",
+ constant.NacosNamespaceID: "",
+ constant.NacosNotLoadLocalCache: "true",
+ },
+ }
+}
+
+type recordingNotifyListener struct {
+ events chan *registry.ServiceEvent
+}
+
+func newRecordingNotifyListener() *recordingNotifyListener {
+ return &recordingNotifyListener{events: make(chan
*registry.ServiceEvent, 8)}
+}
+
+func (r *recordingNotifyListener) Notify(event *registry.ServiceEvent) {
+ r.events <- event
+}
+
+func (r *recordingNotifyListener) NotifyAll(events []*registry.ServiceEvent,
callback func()) {
+ for _, event := range events {
+ r.Notify(event)
+ }
+ if callback != nil {
+ callback()
+ }
+}
+
+func (r *recordingNotifyListener) nextEvent(t *testing.T)
*registry.ServiceEvent {
+ t.Helper()
+ select {
+ case event := <-r.events:
+ return event
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for notify event")
+ return nil
+ }
+}
+
+func (r *recordingNotifyListener) assertNoEvent(t *testing.T) {
+ t.Helper()
+ select {
+ case event := <-r.events:
+ t.Fatalf("unexpected notify event: action=%s service=%s",
event.Action, event.Service)
+ case <-time.After(100 * time.Millisecond):
+ }
+}
+
+func assertRecordedEvent(t *testing.T, event *registry.ServiceEvent, action
remoting.EventType, ip string, port string) {
+ t.Helper()
+ if event.Action != action {
+ t.Fatalf("event action = %s, want %s", event.Action, action)
+ }
+ if event.Service.Ip != ip || event.Service.Port != port {
+ t.Fatalf("event service = %s:%s, want %s:%s", event.Service.Ip,
event.Service.Port, ip, port)
+ }
+}
