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 62476d510 fix(filter, protocol, registry, config_center): guard unsafe
type assertions (#3563)
62476d510 is described below
commit 62476d5109e41eb2fb47afc5868372dab35f6e9f
Author: Nene7ko_ <[email protected]>
AuthorDate: Thu Aug 6 15:58:52 2026 +0800
fix(filter, protocol, registry, config_center): guard unsafe type
assertions (#3563)
* fix: guard unsafe type assertions
* fix: address PR 3563 review findings
Signed-off-by: Nene7ko_ <[email protected]>
---------
Signed-off-by: Nene7ko_ <[email protected]>
---
config_center/apollo/impl.go | 6 +-
config_center/apollo/impl_test.go | 18 ++++-
filter/accesslog/filter.go | 51 ++++++++------
filter/accesslog/filter_test.go | 27 +++++++
protocol/dubbo/dubbo_codec.go | 24 ++++++-
protocol/dubbo/dubbo_codec_test.go | 28 ++++++++
registry/directory/directory.go | 132 +++++++++++++++++++++++++----------
registry/directory/directory_test.go | 45 ++++++++++++
registry/protocol/protocol.go | 79 +++++++++++++++++----
registry/protocol/protocol_test.go | 42 +++++++++++
10 files changed, 376 insertions(+), 76 deletions(-)
diff --git a/config_center/apollo/impl.go b/config_center/apollo/impl.go
index 9ee978454..bde5e56f6 100644
--- a/config_center/apollo/impl.go
+++ b/config_center/apollo/impl.go
@@ -118,7 +118,11 @@ func (c *apolloConfiguration) GetInternalProperty(key
string, opts ...cc.Option)
if err != nil {
return "", perrors.Wrap(err, "get config value failed")
}
- return value.(string), nil
+ stringValue, ok := value.(string)
+ if !ok {
+ return "", perrors.Errorf("apollo config value for key %q is
not a string: %T", key, value)
+ }
+ return stringValue, nil
}
func (c *apolloConfiguration) GetRule(key string, opts ...cc.Option) (string,
error) {
diff --git a/config_center/apollo/impl_test.go
b/config_center/apollo/impl_test.go
index 9c72644ba..88e4606de 100644
--- a/config_center/apollo/impl_test.go
+++ b/config_center/apollo/impl_test.go
@@ -68,7 +68,7 @@ const (
"homepageUrl": "http://localhost:8080";
}]`
- mockConfigCacheRes = `{"content":"dubbo:\n application:\n name:
\"demo-server\"\n version: \"2.0\"\n"}`
+ mockConfigCacheRes = `{"content":"dubbo:\n application:\n name:
\"demo-server\"\n version: \"2.0\"\n retries: 3\n"}`
mockConfigCacheJsonRes = `{
"content": "{\n \"dubbo\": {\n \"application\": {\n
\"name\": \"demo-server\",\n \"version\": \"2.0\"\n },\n
\"otel\": {\n \"trace\": {\n \"enable\": true,\n
\"sample-ratio\": 1.123\n }\n }\n }\n}"
@@ -80,7 +80,7 @@ var mockConfigRes = `{
"cluster": "default",
"namespaceName": "mockDubbogo.yaml",
"configurations":{
- "content":"dubbo:\n application:\n name: \"demo-server\"\n
version: \"2.0\"\n"
+ "content":"dubbo:\n application:\n name: \"demo-server\"\n
version: \"2.0\"\n retries: 3\n"
},
"releaseKey": "20191104105242-0f13805d89f834a4"
}`
@@ -181,6 +181,20 @@ func TestGetConfigItem(t *testing.T) {
assert.Equal(t, "demo-server", appName)
}
+func TestGetConfigItemReturnsErrorForNonStringValue(t *testing.T) {
+ configuration := initMockApollo(t)
+
+ var (
+ value string
+ err error
+ )
+ require.NotPanics(t, func() {
+ value, err =
configuration.GetInternalProperty(constant.ApplicationConfigPrefix + ".retries")
+ })
+ require.Error(t, err)
+ assert.Empty(t, value)
+}
+
func initMockApollo(t *testing.T) *apolloConfiguration {
// Register the YAML format parser with concurrent safety.
extension.AddFormatParser(apolloconstant.YAML, &Parser{})
diff --git a/filter/accesslog/filter.go b/filter/accesslog/filter.go
index 8330ba4b2..cdb5279e5 100644
--- a/filter/accesslog/filter.go
+++ b/filter/accesslog/filter.go
@@ -136,30 +136,24 @@ func (f *Filter) logIntoChannel(accessLogData Data) {
func (f *Filter) buildAccessLogData(_ base.Invoker, invocation
base.Invocation) map[string]string {
dataMap := make(map[string]string, 16)
attachments := invocation.Attachments()
- itf := attachments[constant.InterfaceKey]
- if itf == nil || len(itf.(string)) == 0 {
- itf = attachments[constant.PathKey]
+ itf, ok := stringAttachment(attachments, constant.InterfaceKey)
+ if !ok || len(itf) == 0 {
+ itf, _ = stringAttachment(attachments, constant.PathKey)
}
- if itf != nil {
- dataMap[constant.InterfaceKey] = itf.(string)
+ if itf != "" {
+ dataMap[constant.InterfaceKey] = itf
}
- if v, ok := attachments[constant.MethodKey]; ok && v != nil {
- dataMap[constant.MethodKey] = v.(string)
- }
- if v, ok := attachments[constant.VersionKey]; ok && v != nil {
- dataMap[constant.VersionKey] = v.(string)
- }
- if v, ok := attachments[constant.GroupKey]; ok && v != nil {
- dataMap[constant.GroupKey] = v.(string)
- }
- if v, ok := attachments[constant.TimestampKey]; ok && v != nil {
- dataMap[constant.TimestampKey] = v.(string)
- }
- if v, ok := attachments[constant.LocalAddr]; ok && v != nil {
- dataMap[constant.LocalAddr] = v.(string)
- }
- if v, ok := attachments[constant.RemoteAddr]; ok && v != nil {
- dataMap[constant.RemoteAddr] = v.(string)
+ for _, key := range []string{
+ constant.MethodKey,
+ constant.VersionKey,
+ constant.GroupKey,
+ constant.TimestampKey,
+ constant.LocalAddr,
+ constant.RemoteAddr,
+ } {
+ if value, ok := stringAttachment(attachments, key); ok {
+ dataMap[key] = value
+ }
}
if len(invocation.Arguments()) > 0 {
@@ -184,6 +178,19 @@ func (f *Filter) buildAccessLogData(_ base.Invoker,
invocation base.Invocation)
return dataMap
}
+func stringAttachment(attachments map[string]any, key string) (string, bool) {
+ value, exists := attachments[key]
+ if !exists || value == nil {
+ return "", false
+ }
+ stringValue, ok := value.(string)
+ if !ok {
+ logger.Debugf("[Filter][AccessLog] attachment %q has unexpected
type %T and will be omitted", key, value)
+ return "", false
+ }
+ return stringValue, true
+}
+
// OnResponse do nothing
func (f *Filter) OnResponse(_ context.Context, result result.Result, _
base.Invoker, _ base.Invocation) result.Result {
return result
diff --git a/filter/accesslog/filter_test.go b/filter/accesslog/filter_test.go
index 96a4ccd9f..6e38a04dd 100644
--- a/filter/accesslog/filter_test.go
+++ b/filter/accesslog/filter_test.go
@@ -26,6 +26,7 @@ import (
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
)
import (
@@ -82,3 +83,29 @@ func TestFilterOnResponse(t *testing.T) {
response := filter.OnResponse(context.TODO(), rpcResult, nil, nil)
assert.Equal(t, rpcResult, response)
}
+
+func TestBuildAccessLogDataSkipsNonStringAttachments(t *testing.T) {
+ attachments := map[string]any{
+ constant.InterfaceKey: 42,
+ constant.PathKey: "fallback.Service",
+ constant.MethodKey: true,
+ constant.VersionKey: 1,
+ constant.GroupKey: []string{"group"},
+ constant.TimestampKey: 1234567890,
+ constant.LocalAddr: struct{}{},
+ constant.RemoteAddr: nil,
+ }
+ inv := invocation.NewRPCInvocation("MethodName", nil, attachments)
+
+ var data map[string]string
+ require.NotPanics(t, func() {
+ data = (&Filter{}).buildAccessLogData(nil, inv)
+ })
+ assert.Equal(t, "fallback.Service", data[constant.InterfaceKey])
+ assert.NotContains(t, data, constant.MethodKey)
+ assert.NotContains(t, data, constant.VersionKey)
+ assert.NotContains(t, data, constant.GroupKey)
+ assert.NotContains(t, data, constant.TimestampKey)
+ assert.NotContains(t, data, constant.LocalAddr)
+ assert.NotContains(t, data, constant.RemoteAddr)
+}
diff --git a/protocol/dubbo/dubbo_codec.go b/protocol/dubbo/dubbo_codec.go
index 00ca3ae29..56a6175bc 100644
--- a/protocol/dubbo/dubbo_codec.go
+++ b/protocol/dubbo/dubbo_codec.go
@@ -144,10 +144,14 @@ func (c *DubboCodec) EncodeResponse(response
*remoting.Response) (*bytes.Buffer,
},
}
if !response.IsHeartbeat() {
+ rpcResult, err := rpcResultFromResponse(response.Result)
+ if err != nil {
+ return nil, err
+ }
resp.Body = &impl.ResponsePayload{
- RspObj: response.Result.(result.RPCResult).Rest,
- Exception: response.Result.(result.RPCResult).Err,
- Attachments: response.Result.(result.RPCResult).Attrs,
+ RspObj: rpcResult.Rest,
+ Exception: rpcResult.Err,
+ Attachments: rpcResult.Attrs,
}
}
@@ -161,6 +165,20 @@ func (c *DubboCodec) EncodeResponse(response
*remoting.Response) (*bytes.Buffer,
return bytes.NewBuffer(pkg), nil
}
+func rpcResultFromResponse(value any) (result.RPCResult, error) {
+ switch rpcResult := value.(type) {
+ case result.RPCResult:
+ return rpcResult, nil
+ case *result.RPCResult:
+ if rpcResult == nil {
+ return result.RPCResult{}, perrors.New("dubbo response
result is a nil *result.RPCResult")
+ }
+ return *rpcResult, nil
+ default:
+ return result.RPCResult{}, perrors.Errorf("dubbo response
result has unexpected type %T", value)
+ }
+}
+
// Decode data, including request and response.
func (c *DubboCodec) Decode(data []byte) (*remoting.DecodeResult, int, error) {
dataLen := len(data)
diff --git a/protocol/dubbo/dubbo_codec_test.go
b/protocol/dubbo/dubbo_codec_test.go
index 2a81641c6..139f92036 100644
--- a/protocol/dubbo/dubbo_codec_test.go
+++ b/protocol/dubbo/dubbo_codec_test.go
@@ -26,6 +26,12 @@ import (
"github.com/stretchr/testify/require"
)
+import (
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/protocol/result"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+)
+
// TestIsRequest tests isRequest with various bit patterns
func TestIsRequest(t *testing.T) {
codec := &DubboCodec{}
@@ -101,6 +107,28 @@ func TestCodecType(t *testing.T) {
assert.NotNil(t, codec.encodeHeartbeatRequest)
}
+func TestEncodeResponseAcceptsRPCResultPointer(t *testing.T) {
+ response := remoting.NewResponse(1, "2.0.2")
+ response.SerialID = constant.SHessian2
+ response.Result = &result.RPCResult{}
+
+ require.NotPanics(t, func() {
+ _, err := (&DubboCodec{}).EncodeResponse(response)
+ require.NoError(t, err)
+ })
+}
+
+func TestEncodeResponseRejectsUnexpectedResultType(t *testing.T) {
+ response := remoting.NewResponse(1, "2.0.2")
+ response.SerialID = constant.SHessian2
+ response.Result = "not-an-rpc-result"
+
+ require.NotPanics(t, func() {
+ _, err := (&DubboCodec{}).EncodeResponse(response)
+ require.Error(t, err)
+ })
+}
+
// TestIsRequestEdgeCases tests isRequest with edge case bit patterns
func TestIsRequestEdgeCases(t *testing.T) {
codec := &DubboCodec{}
diff --git a/registry/directory/directory.go b/registry/directory/directory.go
index f5dbdf835..c0b7bd0dc 100644
--- a/registry/directory/directory.go
+++ b/registry/directory/directory.go
@@ -21,6 +21,7 @@ import (
"fmt"
"net/url"
"os"
+ "reflect"
"sync"
"time"
)
@@ -346,9 +347,15 @@ func (dir *RegistryDirectory) refreshAllInvokers(events
[]*registry.ServiceEvent
defer dir.registerLock.Unlock()
// get need clear invokers from original invoker list
dir.cacheInvokersMap.Range(func(k, v any) bool {
- if !dir.eventMatched(k.(string), providerEvents) {
+ key, ok := k.(string)
+ if !ok {
+ logger.Errorf("[Registry][Directory] cached
invoker has unexpected key type %T", k)
+ dir.cacheInvokersMap.Delete(k)
+ return true
+ }
+ if !dir.eventMatched(key, providerEvents) {
// delete unused invoker from cache
- if invoker :=
dir.uncacheInvokerWithKey(k.(string)); invoker != nil {
+ if invoker := dir.uncacheInvokerWithKey(key);
invoker != nil {
oldInvokers = append(oldInvokers,
invoker)
}
}
@@ -496,44 +503,52 @@ func cloneServiceEvents(events []*registry.ServiceEvent)
[]*registry.ServiceEven
}
func (dir *RegistryDirectory) toGroupInvokers() []protocolbase.Invoker {
+ groupInvokersMap := dir.groupInvokersFromCache()
+ if len(groupInvokersMap) == 1 {
+ for _, invokers := range groupInvokersMap {
+ return invokers
+ }
+ }
+ return dir.joinGroupInvokers(groupInvokersMap)
+}
+
+func (dir *RegistryDirectory) groupInvokersFromCache()
map[string][]protocolbase.Invoker {
groupInvokersMap := make(map[string][]protocolbase.Invoker)
dir.cacheInvokersMap.Range(func(key, value any) bool {
- invoker := value.(protocolbase.Invoker)
+ invoker, ok := cachedInvoker(key, value)
+ if !ok {
+ dir.cacheInvokersMap.Delete(key)
+ return true
+ }
group := invoker.GetURL().GetParam(constant.GroupKey, "")
groupInvokersMap[group] = append(groupInvokersMap[group],
invoker)
return true
})
+ return groupInvokersMap
+}
+func (dir *RegistryDirectory) joinGroupInvokers(groupInvokersMap
map[string][]protocolbase.Invoker) []protocolbase.Invoker {
groupInvokersList := make([]protocolbase.Invoker, 0,
len(groupInvokersMap))
- if len(groupInvokersMap) == 1 {
- // len is 1 it means no group setting ,so do not need cluster
again
- for _, invokers := range groupInvokersMap {
- groupInvokersList = invokers
+ for _, invokers := range groupInvokersMap {
+ staticDir := static.NewDirectory(invokers)
+ clusterKey := dir.GetURL().SubURL.GetParam(constant.ClusterKey,
constant.DefaultCluster)
+ cluster, err := extension.GetCluster(clusterKey)
+ if err != nil {
+ logger.Errorf("[Registry][Directory] directory get
cluster %s error, err=%w, will skip this group",
+ clusterKey, err)
+ continue
}
- } else {
- for _, invokers := range groupInvokersMap {
- staticDir := static.NewDirectory(invokers)
- clusterKey :=
dir.GetURL().SubURL.GetParam(constant.ClusterKey, constant.DefaultCluster)
- cluster, err := extension.GetCluster(clusterKey)
- if err != nil {
- logger.Errorf("[Registry][Directory] directory
get cluster %s error, err=%w, will skip this group",
- clusterKey, err)
- continue
- }
- if cluster == nil {
- logger.Errorf("[Registry][Directory] directory
cluster is nil for key %s, will skip this group", clusterKey)
- continue
- }
- err = staticDir.BuildRouterChain(invokers, dir.GetURL())
- if err != nil {
- logger.Errorf("[Registry][Directory]
buildRouterChain error, err=%v", err)
- continue
- }
- groupInvokersList = append(groupInvokersList,
cluster.Join(staticDir))
+ if cluster == nil {
+ logger.Errorf("[Registry][Directory] directory cluster
is nil for key %s, will skip this group", clusterKey)
+ continue
}
+ if err = staticDir.BuildRouterChain(invokers, dir.GetURL());
err != nil {
+ logger.Errorf("[Registry][Directory] buildRouterChain
error, err=%v", err)
+ continue
+ }
+ groupInvokersList = append(groupInvokersList,
cluster.Join(staticDir))
}
-
return groupInvokersList
}
@@ -541,8 +556,19 @@ func (dir *RegistryDirectory)
uncacheInvokerWithClusterID(clusterID string) []pr
logger.Debugf("[Registry][Directory] all service will be deleted in
cache invokers with clusterID=%s", clusterID)
invokerKeys := make([]string, 0)
dir.cacheInvokersMap.Range(func(key, cacheInvoker any) bool {
- if
cacheInvoker.(protocolbase.Invoker).GetURL().GetParam(constant.MeshClusterIDKey,
"") == clusterID {
- invokerKeys = append(invokerKeys, key.(string))
+ invoker, ok := cachedInvoker(key, cacheInvoker)
+ if !ok {
+ dir.cacheInvokersMap.Delete(key)
+ return true
+ }
+ keyString, ok := key.(string)
+ if !ok {
+ logger.Errorf("[Registry][Directory] cached invoker has
unexpected key type %T", key)
+ dir.cacheInvokersMap.Delete(key)
+ return true
+ }
+ if invoker.GetURL().GetParam(constant.MeshClusterIDKey, "") ==
clusterID {
+ invokerKeys = append(invokerKeys, keyString)
}
return true
})
@@ -568,7 +594,11 @@ func (dir *RegistryDirectory) uncacheInvokerWithKey(key
string) protocolbase.Inv
protocolbase.RemoveUrlKeyUnhealthyStatus(key)
if cacheInvoker, ok := dir.cacheInvokersMap.Load(key); ok {
dir.cacheInvokersMap.Delete(key)
- return cacheInvoker.(protocolbase.Invoker)
+ invoker, valid := cachedInvoker(key, cacheInvoker)
+ if !valid {
+ return nil
+ }
+ return invoker
}
return nil
}
@@ -586,7 +616,7 @@ func (dir *RegistryDirectory)
RemoveClosingInstance(instanceKey string) bool {
defer dir.registerLock.Unlock()
if cacheInvoker, ok := dir.cacheInvokersMap.Load(instanceKey);
ok {
- removed = cacheInvoker.(protocolbase.Invoker)
+ removed, _ = cachedInvoker(instanceKey, cacheInvoker)
}
dir.markClosingTombstone(instanceKey, removed, "closing-event")
removed = dir.uncacheInvokerWithKey(instanceKey)
@@ -686,7 +716,15 @@ func (dir *RegistryDirectory) doCacheInvoker(newUrl
*common.URL, event *registry
logger.Infof("[Registry][Directory] skip rebuilding closing
instance due to tombstone, instance key: %s", key)
return nil, true
}
- if cacheInvoker, ok := dir.cacheInvokersMap.Load(key); !ok {
+ cacheInvoker, ok := dir.cacheInvokersMap.Load(key)
+ var existingInvoker protocolbase.Invoker
+ if ok {
+ existingInvoker, ok = cachedInvoker(key, cacheInvoker)
+ if !ok {
+ dir.cacheInvokersMap.Delete(key)
+ }
+ }
+ if !ok {
logger.Debugf("[Registry][Directory] service will be added in
cache invokers, url=%s", newUrl)
newInvoker :=
extension.GetProtocol(protocolwrapper.FILTER).Refer(newUrl)
if newInvoker != nil {
@@ -698,15 +736,15 @@ func (dir *RegistryDirectory) doCacheInvoker(newUrl
*common.URL, event *registry
metrics.Publish(metricsRegistry.NewDirectoryEvent(metricsRegistry.NumValidTotal))
// if cached invoker has the same URL with the new URL, then no
need to re-refer, and no need to destroy
// the old invoker.
- if common.GetCompareURLEqualFunc()(newUrl,
cacheInvoker.(protocolbase.Invoker).GetURL()) {
+ if common.GetCompareURLEqualFunc()(newUrl,
existingInvoker.GetURL()) {
return nil, true
}
- logger.Debugf("[Registry][Directory] service will be updated in
cache invokers, newUrl=%s oldUrl=%s", newUrl,
cacheInvoker.(protocolbase.Invoker).GetURL())
+ logger.Debugf("[Registry][Directory] service will be updated in
cache invokers, newUrl=%s oldUrl=%s", newUrl, existingInvoker.GetURL())
newInvoker :=
extension.GetProtocol(protocolwrapper.FILTER).Refer(newUrl)
if newInvoker != nil {
dir.cacheInvokersMap.Store(key, newInvoker)
- return cacheInvoker.(protocolbase.Invoker), true
+ return existingInvoker, true
} else {
logger.Warnf("[Registry][Directory] service will be
updated in cache invokers fail, result is null, url=%s", newUrl.String())
}
@@ -714,6 +752,28 @@ func (dir *RegistryDirectory) doCacheInvoker(newUrl
*common.URL, event *registry
return nil, false
}
+func cachedInvoker(key, value any) (protocolbase.Invoker, bool) {
+ invoker, ok := value.(protocolbase.Invoker)
+ if !ok || isNilInvoker(invoker) {
+ logger.Errorf("[Registry][Directory] cached invoker has
unexpected type %T for key %v", value, key)
+ return nil, false
+ }
+ return invoker, true
+}
+
+func isNilInvoker(invoker protocolbase.Invoker) bool {
+ if invoker == nil {
+ return true
+ }
+ value := reflect.ValueOf(invoker)
+ switch value.Kind() {
+ case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map,
reflect.Pointer, reflect.Slice, reflect.UnsafePointer:
+ return value.IsNil()
+ default:
+ return false
+ }
+}
+
// List selected protocol invokers from the directory
func (dir *RegistryDirectory) List(invocation protocolbase.Invocation)
[]protocolbase.Invoker {
routerChain := dir.RouterChain()
diff --git a/registry/directory/directory_test.go
b/registry/directory/directory_test.go
index 5254a5f71..1cfbeab6f 100644
--- a/registry/directory/directory_test.go
+++ b/registry/directory/directory_test.go
@@ -66,6 +66,38 @@ func TestSubscribe_InvalidUrl(t *testing.T) {
require.Error(t, err)
}
+func TestUnexpectedCachedInvokerTypeDoesNotPanic(t *testing.T) {
+ registryDirectory, _ := normalRegistryDir()
+ const key = "unexpected-invoker-type"
+
+ t.Run("grouping", func(t *testing.T) {
+ registryDirectory.cacheInvokersMap.Store(key, "not-an-invoker")
+ require.NotPanics(t, func() {
+ registryDirectory.toGroupInvokers()
+ })
+ _, exists := registryDirectory.cacheInvokersMap.Load(key)
+ assert.False(t, exists)
+ })
+
+ t.Run("uncache", func(t *testing.T) {
+ registryDirectory.cacheInvokersMap.Store(key, "not-an-invoker")
+ var invoker protocolbase.Invoker
+ require.NotPanics(t, func() {
+ invoker = registryDirectory.uncacheInvokerWithKey(key)
+ })
+ assert.Nil(t, invoker)
+ })
+
+ t.Run("remove closing instance", func(t *testing.T) {
+ registryDirectory.cacheInvokersMap.Store(key, "not-an-invoker")
+ var removed bool
+ require.NotPanics(t, func() {
+ removed = registryDirectory.RemoveClosingInstance(key)
+ })
+ assert.False(t, removed)
+ })
+}
+
func TestNewRegistryDirectoryResolvesApplicationAttribute(t *testing.T) {
tests := []struct {
name string
@@ -116,6 +148,19 @@ func
TestNewRegistryDirectoryResolvesApplicationAttribute(t *testing.T) {
}
}
+func TestTypedNilCachedInvokerDoesNotPanic(t *testing.T) {
+ registryDirectory, _ := normalRegistryDir()
+ const key = "typed-nil-invoker"
+ var typedNil *protocolbase.BaseInvoker
+ registryDirectory.cacheInvokersMap.Store(key, typedNil)
+
+ require.NotPanics(t, func() {
+ registryDirectory.toGroupInvokers()
+ })
+ _, exists := registryDirectory.cacheInvokersMap.Load(key)
+ assert.False(t, exists)
+}
+
func TestNewRegistryDirectoryCopiesRegistriesAttributeFromSubURL(t *testing.T)
{
registryURL, subURL := newRegistryDirectoryAttributeTestURL(t)
registries := map[string]*global.RegistryConfig{
diff --git a/registry/protocol/protocol.go b/registry/protocol/protocol.go
index ccd9491fb..d00ee2351 100644
--- a/registry/protocol/protocol.go
+++ b/registry/protocol/protocol.go
@@ -19,6 +19,7 @@ package protocol
import (
"context"
+ "reflect"
"strings"
"sync"
"time"
@@ -86,15 +87,51 @@ func (proto *registryProtocol) getRegistry(registryUrl
*common.URL) registry.Reg
if namespace != "" {
cacheKey = cacheKey + "?" + constant.NacosNamespaceID + "=" +
namespace
}
- actualReg, _ := proto.registries.LoadOrStore(cacheKey, func() any {
- reg, err := extension.GetRegistry(registryUrl.Protocol,
registryUrl)
- if err != nil {
- logger.Errorf("[Registry] registry cannot connect
successfully, err=%s", err.Error())
- panic(err)
+ if actualReg, loaded := proto.registries.Load(cacheKey); loaded {
+ reg, ok := cachedRegistry(actualReg, cacheKey)
+ if !ok {
+ proto.registries.Delete(cacheKey)
}
return reg
- }())
- return actualReg.(registry.Registry)
+ }
+
+ reg, err := extension.GetRegistry(registryUrl.Protocol, registryUrl)
+ if err != nil {
+ logger.Errorf("[Registry] registry cannot connect successfully,
err=%s", err.Error())
+ panic(err)
+ }
+ reg, ok := cachedRegistry(reg, cacheKey)
+ if !ok {
+ return nil
+ }
+ actualReg, _ := proto.registries.LoadOrStore(cacheKey, reg)
+ cachedReg, ok := cachedRegistry(actualReg, cacheKey)
+ if !ok {
+ proto.registries.Delete(cacheKey)
+ }
+ return cachedReg
+}
+
+func cachedRegistry(value any, cacheKey string) (registry.Registry, bool) {
+ reg, ok := value.(registry.Registry)
+ if !ok || isNilRegistry(reg) {
+ logger.Errorf("[Registry] cached registry has unexpected type
%T for key %s", value, cacheKey)
+ return nil, false
+ }
+ return reg, true
+}
+
+func isNilRegistry(reg registry.Registry) bool {
+ if reg == nil {
+ return true
+ }
+ value := reflect.ValueOf(reg)
+ switch value.Kind() {
+ case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map,
reflect.Pointer, reflect.Slice, reflect.UnsafePointer:
+ return value.IsNil()
+ default:
+ return false
+ }
}
func getCacheKey(invoker base.Invoker) string {
@@ -174,9 +211,11 @@ func shutdownFromAttribute(url *common.URL)
(*global.ShutdownConfig, bool) {
// GetRegistries returns all underlying registry instances.
func (proto *registryProtocol) GetRegistries() []registry.Registry {
var rs []registry.Registry
- proto.registries.Range(func(_, v any) bool {
- if r, ok := v.(registry.Registry); ok {
+ proto.registries.Range(func(key, v any) bool {
+ if r, ok := v.(registry.Registry); ok && !isNilRegistry(r) {
rs = append(rs, r)
+ } else {
+ proto.registries.Delete(key)
}
return true
})
@@ -192,6 +231,10 @@ func (proto *registryProtocol) Refer(url *common.URL)
base.Invoker {
}
reg := proto.getRegistry(url)
+ if reg == nil {
+ logger.Errorf("[Registry] consumer service %v has no valid
registry, and will return nil invoker", serviceUrl.String())
+ return nil
+ }
// new registry directory for store service url from registry
dic, err := extension.GetDirectoryInstance(registryUrl, reg)
@@ -255,6 +298,10 @@ func (proto *registryProtocol) Export(originInvoker
base.Invoker) base.Exporter
if len(registryUrl.Protocol) > 0 {
// url to registry
reg := proto.getRegistry(registryUrl)
+ if reg == nil {
+ logger.Errorf("[Registry] provider service %v has no
valid registry, and will return nil exporter", providerUrl.String())
+ return nil
+ }
registeredProviderUrl := getUrlToRegistry(providerUrl,
registryUrl)
err := reg.Register(registeredProviderUrl)
@@ -496,10 +543,14 @@ func (proto *registryProtocol) Destroy() {
// the work for unexport should be finished in
protocol.UnExport(), see also config.destroyProviderProtocols().
exporter := value.(*exporterChangeableWrapper)
reg := proto.getRegistry(getRegistryUrl(exporter.originInvoker))
- if err := reg.UnRegister(exporter.registerUrl); err != nil {
- logger.Warnf("[Registry] unRegister consumer url
failed, url=%s err=%v", exporter.registerUrl.String(), err)
+ if reg == nil {
+ logger.Warnf("[Registry] skip unregister because no
valid registry was found, url=%s", exporter.registerUrl.String())
+ } else {
+ if err := reg.UnRegister(exporter.registerUrl); err !=
nil {
+ logger.Warnf("[Registry] unRegister consumer
url failed, url=%s err=%v", exporter.registerUrl.String(), err)
+ }
+ proto.unsubscribeOverrideListener(reg,
exporter.subscribeUrl)
}
- proto.unsubscribeOverrideListener(reg, exporter.subscribeUrl)
proto.serviceConfigurationListeners.Delete(getProviderUrl(exporter.originInvoker).ServiceKey())
// close all protocol server after consumerUpdateWait +
stepTimeout(max time wait during
@@ -559,6 +610,10 @@ func (proto *registryProtocol) UnregisterRegistries() {
proto.bounds.Range(func(_, value any) bool {
exporter := value.(*exporterChangeableWrapper)
reg := proto.getRegistry(getRegistryUrl(exporter.originInvoker))
+ if reg == nil {
+ logger.Warnf("[Registry] skip unregister because no
valid registry was found, url=%s", exporter.registerUrl.String())
+ return true
+ }
if err := reg.UnRegister(exporter.registerUrl); err != nil {
logger.Warnf("[Registry] unRegister consumer url
failed, url=%s err=%v", exporter.registerUrl.String(), err)
}
diff --git a/registry/protocol/protocol_test.go
b/registry/protocol/protocol_test.go
index 6da1720cd..0dfbeb274 100644
--- a/registry/protocol/protocol_test.go
+++ b/registry/protocol/protocol_test.go
@@ -79,6 +79,48 @@ func referNormal(t *testing.T, regProtocol
*registryProtocol) {
assert.Equal(t, invoker.GetURL().String(), url.String())
}
+func TestGetRegistryRejectsInvalidCachedValues(t *testing.T) {
+ tests := []struct {
+ name string
+ value any
+ }{
+ {name: "unexpected type", value: "not-a-registry"},
+ {name: "typed nil", value: (*registry.MockRegistry)(nil)},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ regProtocol := newRegistryProtocol()
+ registryURL, err :=
common.NewURL("mock://127.0.0.1:1111")
+ require.NoError(t, err)
+ extension.SetRegistry("mock", registry.NewMockRegistry)
+ regProtocol.registries.Store(registryURL.PrimitiveURL,
tt.value)
+
+ var actual registry.Registry
+ require.NotPanics(t, func() {
+ actual = regProtocol.getRegistry(registryURL)
+ })
+ assert.Nil(t, actual)
+ _, exists :=
regProtocol.registries.Load(registryURL.PrimitiveURL)
+ assert.False(t, exists)
+ assert.Empty(t, regProtocol.GetRegistries())
+
+ actual = regProtocol.getRegistry(registryURL)
+ assert.NotNil(t, actual)
+ })
+ }
+}
+
+func TestGetRegistriesSkipsTypedNil(t *testing.T) {
+ regProtocol := newRegistryProtocol()
+ var typedNil *registry.MockRegistry
+ regProtocol.registries.Store("typed-nil", typedNil)
+
+ assert.Empty(t, regProtocol.GetRegistries())
+ _, exists := regProtocol.registries.Load("typed-nil")
+ assert.False(t, exists)
+}
+
func TestRefer(t *testing.T) {
regProtocol := newRegistryProtocol()
referNormal(t, regProtocol)
