Skip to content

fix: close #3247 nacos/directory lifecycle race risks #3270

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Alanxtl merged 11 commits into from
Apr 17, 2026
Merged

fix: close #3247 nacos/directory lifecycle race risks #3270

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
9cd31ff
fix(registry/nacos): make scheduled lookup stoppable
xxs588 Mar 19, 2026
99fc172
fix(directory): harden subscribe timeout lifecycle
xxs588 Mar 19, 2026
86e191d
style: rename subscribedURLLock to subscribedUrlLock
xxs588 Mar 21, 2026
bd9badc
fix(directory): align subscribedUrlLock references
xxs588 Mar 21, 2026
beda059
test(directory): stabilize subscribe timeout case and clarify lifecyc…
xxs588 Mar 23, 2026
2a7102a
test(directory): avoid unsynchronized cacheInvokers reads in tests
xxs588 Mar 23, 2026
860efc6
fix(routerchain): snapshot invokers before routing
xxs588 Mar 25, 2026
a041ff6
fix(directory): rollback late register success after timeout
xxs588 Mar 25, 2026
3314048
fix(directory): decouple subscribe start from register timeout
xxs588 Apr 3, 2026
a121054
Merge branch 'develop' into split/3247-registry-lifecycle
xxs588 Apr 16, 2026
1a9c618
fix(routerchain): repair merge-conflict break in snapshotInvokers
xxs588 Apr 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions cluster/router/chain/chain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,10 +53,7 @@ type RouterChain struct {

// Route Loop routers in RouterChain and call Route method to determine the target invokers list.
func (c *RouterChain) Route(url *common.URL, invocation base.Invocation) []base.Invoker {
c.mutex.RLock()
invokers := c.invokers
c.mutex.RUnlock()

invokers := c.snapshotInvokers()
finalInvokers := make([]base.Invoker, 0, len(invokers))
// multiple invoker may include different methods, find correct invoker otherwise
// will return the invoker without methods
Expand Down Expand Up @@ -110,6 +107,15 @@ func (c *RouterChain) copyRouters() []router.PriorityRouter {
return ret
}

// snapshotInvokers returns a copy of current invokers under lock.
func (c *RouterChain) snapshotInvokers() []base.Invoker {
c.mutex.RLock()
defer c.mutex.RUnlock()
ret := make([]base.Invoker, len(c.invokers))
copy(ret, c.invokers)
return ret
}

// NewRouterChain init router chain
// Loop routerFactories and call NewRouter method
func NewRouterChain(url *common.URL) (*RouterChain, error) {
Expand Down
138 changes: 104 additions & 34 deletions registry/directory/directory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,9 +70,11 @@ type RegistryDirectory struct {
consumerURL *common.URL
cacheOriginUrl *common.URL
configurators []config_center.Configurator
configuratorsLock sync.RWMutex
consumerConfigurationListener *consumerConfigurationListener
referenceConfigurationListener *referenceConfigurationListener
registerLock sync.Mutex // this lock if for register
subscribedUrlLock sync.RWMutex
SubscribedUrl *common.URL
RegisteredUrl *common.URL
closingTombstones *sync.Map // map[string]closingTombstone
Expand Down Expand Up @@ -182,14 +184,7 @@ func NewRegistryDirectory(url *common.URL, registry registry.Registry) (director
// subscribe from registry
func (dir *RegistryDirectory) Subscribe(url *common.URL) error {
logger.Infof("Start subscribing for service :%s with a new go routine.", url.Key())

go func() {
dir.SubscribedUrl = url
if err := dir.registry.Subscribe(url, dir); err != nil {
logger.Error("registry.Subscribe(url:%v, dir:%v) = error:%v", url, dir, err)
}

}()
dir.setSubscribedURL(url)

// Get the timeout time from the registration center configuration (default time 5s)
registerUrl := dir.registry.GetURL()
Expand All @@ -208,26 +203,58 @@ func (dir *RegistryDirectory) Subscribe(url *common.URL) error {
timeout, _ = time.ParseDuration(constant.DefaultRegTimeout)
}

done := make(chan struct{})
serviceKey := url.Key()
// Registration is bounded by registry timeout (default 5s).
// On timeout, we return immediately and skip starting subscribe goroutine.
if err := dir.registerConsumerWithTimeout(url, timeout, serviceKey); err != nil {

@AlexStocks AlexStocks Apr 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P0] 这里把订阅启动放到了 Register 成功之后，改变了原有容错语义。registry/protocol/protocol.go 里 dic.Subscribe(...) 返回错误时只记日志，后面仍会继续 cluster.Join(dic) 并返回 invoker；也就是说一旦注册超时或失败，消费者已经被创建出来，但永远不会启动订阅协程，目录会一直拿不到 provider 列表。建议保持订阅与注册解耦，先启动 registry.Subscribe，再单独处理注册超时/回滚，至少不要让注册失败把服务发现直接短路掉。

@xxs588 xxs588 Apr 3, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

收到，我现在去修改

@xxs588 xxs588 Apr 3, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这次改为先启动 registry.Subscribe，再执行带超时的 Register，保持订阅与注册解耦。
同时在 -race 验证中发现了测试侧存在新竞态（测试直接读 cacheInvokers、以及 normalRegistryDir 异步启动 Subscribe 带来的并发噪音），已同步修成并发安全读取和同步初始化。

return err
}

go func() {
urlToReg := getConsumerUrlToRegistry(url)
err := dir.registry.Register(urlToReg)
if err != nil {
logger.Errorf("consumer service %v register registry %v error, error message is %v",
url.String(), dir.registry.GetURL().String(), err)
if err := dir.registry.Subscribe(url, dir); err != nil {
logger.Error("registry.Subscribe(url:%v, dir:%v) = error:%v", url, dir, err)
}
}()

logger.Infof("register completed successfully for service: %s", serviceKey)
return nil
}

close(done)
func (dir *RegistryDirectory) registerConsumerWithTimeout(url *common.URL, timeout time.Duration, serviceKey string) error {
registerErrCh := make(chan error, 1)
urlToReg := getConsumerUrlToRegistry(url.Clone())
go func() {
registerErrCh <- dir.registry.Register(urlToReg)
}()

timer := time.NewTimer(timeout)
defer timer.Stop()

select {
case <-done:
logger.Infof("register completed successfully for service: %s", url.Key())
case err := <-registerErrCh:
if err != nil {
registryURL := dir.registry.GetURL()
registryURLString := ""
if registryURL != nil {
registryURLString = registryURL.String()
}
logger.Errorf("consumer service %v register registry %v error, error message is %v",
url.String(), registryURLString, err)
return err
}
return nil
case <-time.After(timeout):
logger.Errorf("register timed out for service: %s", url.Key())
return fmt.Errorf("register timed out for service: %s", url.Key())
case <-timer.C:

@AlexStocks AlexStocks Mar 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里的超时只是让调用方提前返回，后台 Register goroutine 还会继续跑。如果它在超时之后成功，consumer 实际上已经注册了，但 Subscribe 不会启动，后续也没有补偿 UnRegister，状态会变成“返回失败但部分初始化成功”。建议把取消能力下推到 registry.Register，或者在超时后处理 late success 并做回滚。

@xxs588 xxs588 Mar 25, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

感谢 ，已修改成 registerConsumerWithTimeout() 超时返回后，如果 Register 迟到成功，则执行 UnRegister 回滚
逻辑详见图片
图片

图片 如果`Subscribe`超时返回错误了但是后面goroutine可能没那么快结束，所以再起一个goroutine等待结果，如果`Register`失败了就不用补偿直接return ，如果成功了说明出现 调用方收到超时失败，但实际上已经注册成功的状态，然后就再补偿 回滚把迟到成功的注册注销掉，如果回滚失败也补上日志方便定位

logger.Errorf("register timed out for service: %s", serviceKey)
go func() {
err := <-registerErrCh
if err != nil {
return
}
if unRegErr := dir.registry.UnRegister(urlToReg.Clone()); unRegErr != nil {
logger.Warnf("register timed out for service %s, but late unregister failed: %v", serviceKey, unRegErr)
}
}()
return fmt.Errorf("register timed out for service: %s", serviceKey)
}
}

Expand Down Expand Up @@ -397,7 +424,7 @@ func (dir *RegistryDirectory) convertUrl(res *registry.ServiceEvent) *common.URL
ret := res.Service
if ret.Protocol == constant.OverrideProtocol || // 1.for override url in 2.6.x
ret.GetParam(constant.CategoryKey, constant.DefaultCategory) == constant.ConfiguratorsCategory {
dir.configurators = append(dir.configurators, extension.GetDefaultConfigurator(ret))
dir.appendConfigurator(extension.GetDefaultConfigurator(ret))
ret = nil
} else if ret.Protocol == constant.RouterProtocol || // 2.for router
ret.GetParam(constant.CategoryKey, constant.DefaultCategory) == constant.RouterCategory {
Expand Down Expand Up @@ -630,9 +657,7 @@ func (dir *RegistryDirectory) List(invocation protocolbase.Invocation) []protoco
routerChain := dir.RouterChain()

if routerChain == nil {
dir.invokersLock.RLock()
defer dir.invokersLock.RUnlock()
return dir.cacheInvokers
return dir.snapshotCacheInvokers()
}
return routerChain.Route(dir.consumerURL, invocation)

@AlexStocks AlexStocks Mar 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里虽然把 routerChain == nil 的分支换成了快照，但 routerChain.Route() 仍然会无锁读 RouterChain.invokers，同时 setNewInvokers() 又会在另一个 goroutine 里通过 SetInvokers() 写这份切片。当前 HEAD 直接跑 go test -race ./registry/directory ./registry/nacos 仍会在 Test_List 报 race。建议在 RouterChain 内部先做带锁 snapshot 再路由，只修 cacheInvokers 读取还不够。

@xxs588 xxs588 Mar 25, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

感谢建议，已修复，Route() 不再直接读 c.invokers，改为先走 snapshotInvokers() 拿快照再路由，对cluster/router全包进行了测试，通过
这是修复的地方
图片

}
Expand All @@ -643,7 +668,7 @@ func (dir *RegistryDirectory) IsAvailable() bool {
return false
}

for _, ivk := range dir.cacheInvokers {
for _, ivk := range dir.snapshotCacheInvokers() {
if ivk.IsAvailable() {
return true
}
Expand All @@ -657,22 +682,23 @@ func (dir *RegistryDirectory) Destroy() {
// TODO:unregister & unsubscribe
dir.DoDestroy(func() {
graceful_shutdown.DefaultClosingDirectoryRegistry().Unregister(dir.closingServiceKey(), dir)
if dir.RegisteredUrl != nil {
err := dir.registry.UnRegister(dir.RegisteredUrl)
registeredURL, subscribedURL := dir.snapshotRegistryURLs()

if registeredURL != nil {
err := dir.registry.UnRegister(registeredURL)
if err != nil {
logger.Warnf("Unregister consumer url failed, %s, error: %v", dir.RegisteredUrl.String(), err)
logger.Warnf("Unregister consumer url failed, %s, error: %v", registeredURL.String(), err)
}
}

if dir.SubscribedUrl != nil {
err := dir.registry.UnSubscribe(dir.SubscribedUrl, dir)
if subscribedURL != nil {
err := dir.registry.UnSubscribe(subscribedURL, dir)
if err != nil {
logger.Warnf("Unsubscribe consumer url failed, %s, error: %v", dir.RegisteredUrl.String(), err)
logger.Warnf("Unsubscribe consumer url failed, %s, error: %v", subscribedURL.String(), err)
}
}

invokers := dir.cacheInvokers
dir.cacheInvokers = []protocolbase.Invoker{}
invokers := dir.swapCacheInvokers()
for _, ivk := range invokers {
ivk.Destroy()
}
Expand All @@ -692,11 +718,55 @@ func (dir *RegistryDirectory) closingServiceKey() string {
}

func (dir *RegistryDirectory) overrideUrl(targetUrl *common.URL) {
doOverrideUrl(dir.configurators, targetUrl)
// Use a read-only snapshot to avoid sharing mutable configurator slice during overrides.
doOverrideUrl(dir.snapshotConfigurators(), targetUrl)
doOverrideUrl(dir.consumerConfigurationListener.Configurators(), targetUrl)
doOverrideUrl(dir.referenceConfigurationListener.Configurators(), targetUrl)
}

func (dir *RegistryDirectory) snapshotCacheInvokers() []protocolbase.Invoker {
dir.invokersLock.RLock()
defer dir.invokersLock.RUnlock()
invokers := make([]protocolbase.Invoker, len(dir.cacheInvokers))
copy(invokers, dir.cacheInvokers)
return invokers
}

func (dir *RegistryDirectory) swapCacheInvokers() []protocolbase.Invoker {
dir.invokersLock.Lock()
defer dir.invokersLock.Unlock()
invokers := make([]protocolbase.Invoker, len(dir.cacheInvokers))
copy(invokers, dir.cacheInvokers)
dir.cacheInvokers = []protocolbase.Invoker{}
return invokers
}

func (dir *RegistryDirectory) appendConfigurator(configurator config_center.Configurator) {
dir.configuratorsLock.Lock()
defer dir.configuratorsLock.Unlock()
dir.configurators = append(dir.configurators, configurator)
}

func (dir *RegistryDirectory) snapshotConfigurators() []config_center.Configurator {
dir.configuratorsLock.RLock()
defer dir.configuratorsLock.RUnlock()
configurators := make([]config_center.Configurator, len(dir.configurators))
copy(configurators, dir.configurators)
return configurators
}

func (dir *RegistryDirectory) setSubscribedURL(url *common.URL) {
dir.subscribedUrlLock.Lock()
defer dir.subscribedUrlLock.Unlock()
dir.SubscribedUrl = url
}

func (dir *RegistryDirectory) snapshotRegistryURLs() (registeredURL *common.URL, subscribedURL *common.URL) {
dir.subscribedUrlLock.RLock()
defer dir.subscribedUrlLock.RUnlock()
return dir.RegisteredUrl, dir.SubscribedUrl
}

func (dir *RegistryDirectory) getConsumerUrl(c *common.URL) *common.URL {
processID := fmt.Sprintf("%d", os.Getpid())
localIP := common.GetLocalIp()
Expand Down
Loading
Loading