feat(metadata): add concurrency safety for application-level metadata state (#3353)

metadata/info/metadata_info.go

@@ -21,6 +21,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 )
 
 import (
@@ -63,6 +64,7 @@ type MetadataInfo struct {
 	Services              map[string]*ServiceInfo  `json:"services,omitempty" hessian:"services"`
 	exportedServiceURLs   map[string][]*common.URL `hessian:"-"` // server exported service urls
 	subscribedServiceURLs map[string][]*common.URL `hessian:"-"` // client subscribed service urls
+	mu                    sync.RWMutex             `json:"-" hessian:"-"`
 }
 
 func NewAppMetadataInfo(app string) *MetadataInfo {
@@ -95,6 +97,15 @@ func (info *MetadataInfo) JavaClassName() string {
 
 // AddService add provider service info to MetadataInfo
 func (info *MetadataInfo) AddService(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
+	info.addServiceWithoutLock(url)
+}
+
+// addServiceWithoutLock adds a service URL without acquiring the lock.
+// The caller must hold info.mu.Lock() before calling this method.
+func (info *MetadataInfo) addServiceWithoutLock(url *common.URL) {
 	service := NewServiceInfoWithURL(url)
 	info.Services[service.GetMatchKey()] = service
 	addUrl(info.exportedServiceURLs, url)
@@ -131,6 +142,9 @@ func deleteItem(slice []*common.URL, index int) []*common.URL {
 }
 
 func (info *MetadataInfo) RemoveService(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	service := NewServiceInfoWithURL(url)
 	removeUrl(info.exportedServiceURLs, url)
 	if replacement := info.findExportedServiceURL(service.GetMatchKey()); replacement != nil {
@@ -142,15 +156,24 @@ func (info *MetadataInfo) RemoveService(url *common.URL) {
 
 // AddSubscribeURL client subscribe a service url
 func (info *MetadataInfo) AddSubscribeURL(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	addUrl(info.subscribedServiceURLs, url)
 }
 
 // RemoveSubscribeURL client unsubscribe a service url
 func (info *MetadataInfo) RemoveSubscribeURL(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	removeUrl(info.subscribedServiceURLs, url)
 }
 
 func (info *MetadataInfo) GetExportedServiceURLs() []*common.URL {
+	info.mu.RLock()
+	defer info.mu.RUnlock()
+
 	res := make([]*common.URL, 0)
 	for _, urls := range info.exportedServiceURLs {
 		res = append(res, urls...)
@@ -159,18 +182,37 @@ func (info *MetadataInfo) GetExportedServiceURLs() []*common.URL {
 }
 
 func (info *MetadataInfo) GetSubscribedURLs() []*common.URL {
+	info.mu.RLock()
+	defer info.mu.RUnlock()
+
 	res := make([]*common.URL, 0)
 	for _, urls := range info.subscribedServiceURLs {
 		res = append(res, urls...)
 	}
 	return res
 }
 
+// GetServices returns a deep copy of the Services map for safe iteration by external callers.
+// Each ServiceInfo is fully copied with lazy fields eagerly populated to prevent write-on-read races.
+func (info *MetadataInfo) GetServices() map[string]*ServiceInfo {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
+	cp := make(map[string]*ServiceInfo, len(info.Services))
+	for k, v := range info.Services {
+		cp[k] = v.DeepCopy()
+	}
+	return cp
+}
+
 func (info *MetadataInfo) ReplaceExportedServices(urls []*common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	info.Services = make(map[string]*ServiceInfo)
 	info.exportedServiceURLs = make(map[string][]*common.URL)
 	for _, serviceURL := range urls {
-		info.AddService(serviceURL)
+		info.addServiceWithoutLock(serviceURL)
 	}
 }
 
@@ -283,3 +325,23 @@ func (si *ServiceInfo) GetServiceKey() string {
 	si.ServiceKey = common.ServiceKey(si.Name, si.Group, si.Version)
 	return si.ServiceKey
 }
+
+// DeepCopy returns a fully independent copy of ServiceInfo with lazy fields eagerly populated.
+func (si *ServiceInfo) DeepCopy() *ServiceInfo {
+	params := make(map[string]string, len(si.Params))
+	for k, v := range si.Params {
+		params[k] = v
+	}
+	return &ServiceInfo{
+		Name:       si.Name,
+		Group:      si.Group,
+		Version:    si.Version,
+		Protocol:   si.Protocol,
+		Port:       si.Port,
+		Path:       si.Path,
+		Params:     params,
+		ServiceKey: si.GetServiceKey(),
+		MatchKey:   si.GetMatchKey(),
+		URL:        si.URL,
+	}
+}

metadata/info/metadata_info_test.go

@@ -188,6 +188,27 @@ func TestServiceInfoGetMatchKey(t *testing.T) {
 	assert.NotEmpty(t, si.GetMatchKey())
 }
 
+func TestMetadataInfoGetServices(t *testing.T) {
+	metadataInfo := &MetadataInfo{
+		Services:              make(map[string]*ServiceInfo),
+		exportedServiceURLs:   make(map[string][]*common.URL),
+		subscribedServiceURLs: make(map[string][]*common.URL),
+	}
+	url, _ := common.NewURL("dubbo://127.0.0.1:20000?application=foo&category=providers&check=false&dubbo=dubbo-go+v1.5.0&interface=com.foo.Bar&methods=GetPetByID%2CGetPetTypes&organization=Apache&owner=foo&revision=1.0.0&side=provider&version=1.0.0")
+	metadataInfo.AddService(url)
+
+	services := metadataInfo.GetServices()
+	require.Len(t, services, 1)
+	assert.NotEmpty(t, services)
+
+	// GetServices returns a copy: modifying the original does not affect the snapshot
+	metadataInfo.RemoveService(url)
+	assert.Len(t, services, 1)
+
+	// A fresh call reflects the removal
+	assert.Empty(t, metadataInfo.GetServices())
+}
+
 func TestServiceInfoJavaClassName(t *testing.T) {
 	assert.Equalf(t, "org.apache.dubbo.metadata.MetadataInfo", NewAppMetadataInfo("dubbo").JavaClassName(), "JavaClassName()")
 }

metadata/metadata.go

@@ -18,14 +18,19 @@
 // Package metadata collects and exposes information of all services for service discovery purpose.
 package metadata
 
+import (
+	"sync"
+)
+
 import (
 	"dubbo.apache.org/dubbo-go/v3/common"
 	"dubbo.apache.org/dubbo-go/v3/common/constant"
 	"dubbo.apache.org/dubbo-go/v3/metadata/info"
 )
 
 var (
-	registryMetadataInfo                 = make(map[string]*info.MetadataInfo)
+	registryMetadataInfo = make(map[string]*info.MetadataInfo)
+	registryMetadataLock sync.RWMutex
 	metadataService      MetadataService = &DefaultMetadataService{metadataMap: registryMetadataInfo}
 )
 
@@ -34,27 +39,37 @@ func GetMetadataService() MetadataService {
 }
 
 func GetMetadataInfo(registryId string) *info.MetadataInfo {
+	registryMetadataLock.RLock()
+	defer registryMetadataLock.RUnlock()
 	return registryMetadataInfo[registryId]
 }
 
 func AddService(registryId string, url *common.URL) {
+	registryMetadataLock.Lock()
 	if _, exist := registryMetadataInfo[registryId]; !exist {
 		registryMetadataInfo[registryId] = info.NewMetadataInfo(
 			url.GetParam(constant.ApplicationKey, ""),
 			url.GetParam(constant.ApplicationTagKey, ""),
 		)
 	}
-	registryMetadataInfo[registryId].AddService(url)
+	metaInfo := registryMetadataInfo[registryId]
+	registryMetadataLock.Unlock()
+
+	metaInfo.AddService(url)
 }
 
 func AddSubscribeURL(registryId string, url *common.URL) {
+	registryMetadataLock.Lock()
 	if _, exist := registryMetadataInfo[registryId]; !exist {
 		registryMetadataInfo[registryId] = info.NewMetadataInfo(
 			url.GetParam(constant.ApplicationKey, ""),
 			url.GetParam(constant.ApplicationTagKey, ""),
 		)
 	}
-	registryMetadataInfo[registryId].AddSubscribeURL(url)
+	metaInfo := registryMetadataInfo[registryId]
+	registryMetadataLock.Unlock()
+
+	metaInfo.AddSubscribeURL(url)
 }
 
 func RemoveService(registryId string, url *common.URL) {

metadata/metadata_service.go

@@ -95,6 +95,8 @@ func (mts *DefaultMetadataService) GetMetadataInfo(revision string) (*info.Metad
 	if revision == "" {
 		return nil, nil
 	}
+	registryMetadataLock.RLock()
+	defer registryMetadataLock.RUnlock()
 	for _, metadataInfo := range mts.metadataMap {
 		if metadataInfo.Revision == revision {
 			return metadataInfo, nil
@@ -106,6 +108,8 @@ func (mts *DefaultMetadataService) GetMetadataInfo(revision string) (*info.Metad
 
 // GetExportedServiceURLs get exported service urls
 func (mts *DefaultMetadataService) GetExportedServiceURLs() ([]*common.URL, error) {
+	registryMetadataLock.RLock()
+	defer registryMetadataLock.RUnlock()
 	urls := make([]*common.URL, 0)
 	for _, metadataInfo := range mts.metadataMap {
 		urls = append(urls, metadataInfo.GetExportedServiceURLs()...)
@@ -124,6 +128,8 @@ func (mts *DefaultMetadataService) GetMetadataServiceURL() (*common.URL, error)
 }
 
 func (mts *DefaultMetadataService) GetSubscribedURLs() ([]*common.URL, error) {
+	registryMetadataLock.RLock()
+	defer registryMetadataLock.RUnlock()
 	urls := make([]*common.URL, 0)
 	for _, metadataInfo := range mts.metadataMap {
 		urls = append(urls, metadataInfo.GetSubscribedURLs()...)
@@ -257,7 +263,7 @@ func (mtsV2 *MetadataServiceV2) GetMetadataInfo(ctx context.Context, req *triple
 	return &tripleapi.MetadataInfoV2{
 		App:      metadataInfo.App,
 		Version:  metadataInfo.Revision,
-		Services: convertV2(metadataInfo.Services),
+		Services: convertV2(metadataInfo.GetServices()),
 	}, err
 }
 

metadata/metadata_service_test.go

@@ -19,6 +19,7 @@ package metadata
 
 import (
 	"strconv"
+	"sync"
 	"testing"
 )
 
@@ -362,3 +363,35 @@ func Test_serviceExporterExport(t *testing.T) {
 		require.NoError(t, err)
 	})
 }
+
+func TestDefaultMetadataServiceConcurrentReadAccess(t *testing.T) {
+	mts := &DefaultMetadataService{
+		metadataMap: newMetadataMap(),
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			urls, err := mts.GetExportedServiceURLs()
+			assert.NoError(t, err)
+			assert.NotEmpty(t, urls)
+		}()
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			urls, err := mts.GetSubscribedURLs()
+			assert.NoError(t, err)
+			assert.NotEmpty(t, urls)
+		}()
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			info, err := mts.GetMetadataInfo("1")
+			assert.NoError(t, err)
+			assert.NotNil(t, info)
+		}()
+	}
+	wg.Wait()
+}

metadata/metadata_test.go

@@ -18,6 +18,7 @@
 package metadata
 
 import (
+	"sync"
 	"testing"
 )
 
@@ -179,3 +180,36 @@ func TestGetMetadataService(t *testing.T) {
 		})
 	}
 }
+
+func TestAddServiceConcurrent(t *testing.T) {
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		registryId := "concurrent-reg"
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			url := common.NewURLWithOptions(
+				common.WithProtocol("dubbo"),
+				common.WithParamsValue(constant.ApplicationKey, "dubbo"),
+				common.WithParamsValue(constant.ApplicationTagKey, "v1"),
+			)
+			AddService(registryId, url)
+		}(i)
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			url := common.NewURLWithOptions(
+				common.WithProtocol("dubbo"),
+				common.WithParamsValue(constant.ApplicationKey, "dubbo"),
+				common.WithParamsValue(constant.ApplicationTagKey, "v1"),
+			)
+			AddSubscribeURL(registryId, url)
+		}(i)
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			_ = GetMetadataInfo(registryId)
+		}(i)
+	}
+	wg.Wait()
+}

metadata/report/nacos/report_test.go

@@ -171,7 +171,7 @@ func newNacosMetadataReport(f fields) *nacosMetadataReport {
 }
 
 func Test_nacosMetadataReport_GetAppMetadata(t *testing.T) {
-	mi := info.MetadataInfo{
+	mi := &info.MetadataInfo{
 		App: "GetAppMetadata",
 	}
 	data, _ := json.Marshal(mi)
@@ -198,7 +198,7 @@ func Test_nacosMetadataReport_GetAppMetadata(t *testing.T) {
 				application: "dubbo",
 				revision:    "revision",
 			},
-			want:    &mi,
+			want:    mi,
 			wantErr: false,
 		},
 	}