You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by "AsperforMias (via GitHub)" <gi...@apache.org> on 2026/08/10 02:39:46 UTC
[PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
AsperforMias opened a new pull request, #3634:
URL: https://github.com/apache/dubbo-go/pull/3634
### Description
Fixes #3624
**Root cause.** In application-level service discovery, `serviceDiscoveryRegistry.subscribeAndNotify` runs `serviceDiscovery.AddListener(listener)` fire-and-forget in a goroutine. A transient registry error (reconnect storm, leader election, brief unavailability) was logged once and never retried. Since the initial `GetInstances` snapshot has already been processed, the consumer looks healthy but silently misses every subsequent instance-change push until restart — a single pod can go permanently stale while others stay fine.
**Fix.** Mirror the retry discipline of the interface-level nacos subscribe backoff (#3178), at the subscription-channel layer:
- One pending retry per `serviceNamesKey` (i.e. per listener): repeated failing subscribes share a single timer instead of stacking.
- Exponential backoff 1s → 30s cap + up to 25% jitter; attempts are intentionally unlimited while subscribers remain (a capped count would re-introduce the permanent staleness this fixes).
- A successful retry **re-syncs the latest instance snapshot** via `GetInstances`/`OnEvent`, so instance changes missed while the subscription was down are picked up instead of relying on a future push.
- Pending retries are canceled when the last subscriber unsubscribes and on `Destroy`; no new timer is armed once the registry is destroyed or no subscriber remains.
- Observability: initial failure keeps the existing error log; each failed retry logs one warning with the attempt count and error; a successful recovery logs once; every attempt publishes the existing `NewServerSubscribeEvent` metric. A dedicated "subscription not established" gauge is left to #3356.
Two latent bugs in the touched paths are fixed along the way:
- `UnSubscribe` removed the notify listener by `url.ServiceKey()` while `SubscribeURL` had registered it under `ServiceKey()+":"+protocol`, so the entry was never removed and subscriber-count-based cleanup could never fire. Both sides now use the same `protocolServiceKeyOf` key.
- The `event.Succ = err != nil` flag in the subscribe path was inverted (and a `%s`/`%d` log mismatch next to it).
**Verification.**
- Unit/regression tests (`registry/servicediscovery/subscribe_retry_test.go`, run with `-race`): first `AddListener` fails → retry establishes the subscription → the consumer receives instance events without a restart; backoff stays in the `[delay, delay+25%]` band and is capped; retries stop after `UnSubscribe` and after `Destroy`, and no timer is armed post-`Destroy`; repeated failing subscribes share one timer.
- Real-Nacos e2e (nacos 2.5.1, docker compose, application-level discovery): the consumer wraps the real nacos `ServiceDiscovery` so the first `AddListener` fails (the deterministic variant suggested in the issue). On `upstream/main`: baseline calls work from the initial snapshot, the provider is then replaced, and the consumer stays at `No provider available` for 60s+ until restarted — bug reproduced. On this branch: `established after 1 retries, re-syncing latest instances`, the consumer follows the provider replacement and calls keep succeeding **without a restart**.
Note: this adds a small `metadataInfoFetcher` indirection for test fault injection; #3625 introduces the same variable, so whichever PR lands second keeps one copy (trivial conflict).
### Checklist
- [x] I confirm the target branch is `develop`
- [x] Code has passed local testing
- [x] I have added tests that prove my fix is effective or that my feature works
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "AsperforMias (via GitHub)" <gi...@apache.org>.
AsperforMias commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5354419943
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5351827912
pls fix conflict
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5254964064
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5235291331
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5354294816
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl commented on code in PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#discussion_r3758015465
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry) getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized instance notification on application %s subscription, instance list size %d", serviceName, len(instances))
+ if err := listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery] ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL, serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event := metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
logger.Errorf("[Registry][ServiceDiscovery] add instance listener catch error, url=%s err=%s", url.String(), err.Error())
+ s.scheduleSubscribeRetry(serviceNamesKey, &subscribeRetry{listener: listener, url: url})
}
}()
}
+// addInstanceListener installs the listener into the service discovery and
+// publishes the subscribe metrics.
+func (s *serviceDiscoveryRegistry) addInstanceListener(url *common.URL, listener registry.ServiceInstancesChangedListener) error {
+ event := metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
+ err := s.serviceDiscovery.AddListener(listener)
+ event.Succ = err == nil
+ event.End = time.Now()
+ event.Attachment[constant.InterfaceKey] = url.Interface()
+ metrics.Publish(event)
+ metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
+ return err
+}
+
+var (
+ // subscribeRetryInitialDelay is the first backoff delay before retrying a
+ // failed AddListener call. Package-level so tests can shrink it.
+ subscribeRetryInitialDelay = time.Second
+ // subscribeRetryMaxDelay caps the backoff. The retry count itself is
+ // intentionally unlimited: a capped count would re-introduce the
+ // permanently stale consumer this mechanism fixes.
+ subscribeRetryMaxDelay = 30 * time.Second
+)
+
+// subscribeRetry is a pending AddListener retry for one serviceNamesKey.
+type subscribeRetry struct {
+ listener registry.ServiceInstancesChangedListener
+ url *common.URL
+ timer *time.Timer
+ attempts int
+}
+
+// scheduleSubscribeRetry arms the retry timer for serviceNamesKey after a
+// failed AddListener call. One pending retry per key: repeated failures share
+// the same timer instead of stacking new ones. Retries continue with capped
+// exponential backoff and jitter until the subscription is established, the
+// last subscriber unsubscribes, or the registry is destroyed.
+func (s *serviceDiscoveryRegistry) scheduleSubscribeRetry(serviceNamesKey string, state *subscribeRetry) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ if s.destroyed {
+ return
+ }
+ if !listenerHasSubscribers(state.listener) {
+ // No subscriber left (e.g. unsubscribe raced with a failing retry):
+ // do not arm a timer nobody waits for.
+ return
+ }
+ if _, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ return
+ }
+ delay := subscribeRetryDelay(state.attempts)
+ state.attempts++
+ state.timer = time.AfterFunc(delay, func() {
+ s.retryAddListener(serviceNamesKey)
+ })
+ s.subscribeRetries[serviceNamesKey] = state
+ logger.Debugf("[Registry][ServiceDiscovery] instance listener for applications=%s not established, retry in %s", serviceNamesKey, delay)
+}
+
+// cancelSubscribeRetry stops a pending AddListener retry, if any.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetry(serviceNamesKey string) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ s.cancelSubscribeRetryLocked(serviceNamesKey)
+}
+
+// cancelSubscribeRetryLocked stops a pending AddListener retry; caller must
+// hold s.lock.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetryLocked(serviceNamesKey string) {
+ if state, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ state.timer.Stop()
+ delete(s.subscribeRetries, serviceNamesKey)
+ }
+}
+
+// retryAddListener re-runs AddListener after the backoff delay. On success it
+// re-syncs the latest instance snapshot so instance changes missed while the
+// subscription was down are picked up instead of leaving the consumer on a
+// stale view.
+func (s *serviceDiscoveryRegistry) retryAddListener(serviceNamesKey string) {
+ s.lock.Lock()
+ state, ok := s.subscribeRetries[serviceNamesKey]
+ if !ok {
+ s.lock.Unlock()
+ return
+ }
+ delete(s.subscribeRetries, serviceNamesKey)
+ active := !s.destroyed &&
+ s.serviceListeners[serviceNamesKey] == state.listener &&
+ listenerHasSubscribers(state.listener)
+ s.lock.Unlock()
+
+ if !active {
+ // Registry destroyed, listener replaced, or no subscriber left:
+ // stop retrying so the loop cannot leak.
+ return
+ }
Review Comment:
retry state is deleted while the external retry attempt is still in flight
deletes `subscribeRetries[serviceNamesKey]` before calling `AddListener` at line 729. During that call, another `SubscribeURL` for the same key can fail and schedule a fresh `subscribeRetry{attempts: 0}` because the map is temporarily empty. If the in-flight retry then also fails, its higher-attempt state is dropped by `scheduleSubscribeRetry` because the new low-attempt timer already exists. That breaks the PR’s “one pending retry per key” and sustained exponential backoff guarantees, and can cause repeated low-delay retry bursts during reconnect storms. Keeping the retry state present with an “in flight” marker, or replacing only when preserving/maxing attempts, would close this window.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "Alanxtl (via GitHub)" <gi...@apache.org>.
Alanxtl commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5364395030
pls fix conflict after merging #3625
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "AlexStocks (via GitHub)" <gi...@apache.org>.
AlexStocks commented on code in PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#discussion_r3749849916
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry) getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized instance notification on application %s subscription, instance list size %d", serviceName, len(instances))
+ if err := listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery] ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL, serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event := metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
Review Comment:
[P1] 后续成功的 `AddListener` 不会撤销同一 `serviceNamesKey` 已挂起的重试。首个订阅失败后，如果 timer 触发前另一个 `SubscribeURL` 成功，这个 goroutine 的成功分支什么都不做，旧 timer 仍会再次调用 `AddListener`；当前 Head 的公开入口探针得到 2 次必要调用后仍出现第 3 次。Polaris 的 `AddSubscriber` 和 ZooKeeper 的 `ListenServiceEvent` 会重复注册回调/监听，不能依赖幂等。成功时应在确认 listener/state 匹配后取消 pending retry，并补“先失败、后续订阅成功、等待超过 backoff 仍只有两次调用”的测试。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "AlexStocks (via GitHub)" <gi...@apache.org>.
AlexStocks merged PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "AsperforMias (via GitHub)" <gi...@apache.org>.
AsperforMias commented on code in PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#discussion_r3760893926
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry) getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized instance notification on application %s subscription, instance list size %d", serviceName, len(instances))
+ if err := listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery] ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL, serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event := metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
logger.Errorf("[Registry][ServiceDiscovery] add instance listener catch error, url=%s err=%s", url.String(), err.Error())
+ s.scheduleSubscribeRetry(serviceNamesKey, &subscribeRetry{listener: listener, url: url})
}
}()
}
+// addInstanceListener installs the listener into the service discovery and
+// publishes the subscribe metrics.
+func (s *serviceDiscoveryRegistry) addInstanceListener(url *common.URL, listener registry.ServiceInstancesChangedListener) error {
+ event := metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
+ err := s.serviceDiscovery.AddListener(listener)
+ event.Succ = err == nil
+ event.End = time.Now()
+ event.Attachment[constant.InterfaceKey] = url.Interface()
+ metrics.Publish(event)
+ metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
+ return err
+}
+
+var (
+ // subscribeRetryInitialDelay is the first backoff delay before retrying a
+ // failed AddListener call. Package-level so tests can shrink it.
+ subscribeRetryInitialDelay = time.Second
+ // subscribeRetryMaxDelay caps the backoff. The retry count itself is
+ // intentionally unlimited: a capped count would re-introduce the
+ // permanently stale consumer this mechanism fixes.
+ subscribeRetryMaxDelay = 30 * time.Second
+)
+
+// subscribeRetry is a pending AddListener retry for one serviceNamesKey.
+type subscribeRetry struct {
+ listener registry.ServiceInstancesChangedListener
+ url *common.URL
+ timer *time.Timer
+ attempts int
+}
+
+// scheduleSubscribeRetry arms the retry timer for serviceNamesKey after a
+// failed AddListener call. One pending retry per key: repeated failures share
+// the same timer instead of stacking new ones. Retries continue with capped
+// exponential backoff and jitter until the subscription is established, the
+// last subscriber unsubscribes, or the registry is destroyed.
+func (s *serviceDiscoveryRegistry) scheduleSubscribeRetry(serviceNamesKey string, state *subscribeRetry) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ if s.destroyed {
+ return
+ }
+ if !listenerHasSubscribers(state.listener) {
+ // No subscriber left (e.g. unsubscribe raced with a failing retry):
+ // do not arm a timer nobody waits for.
+ return
+ }
+ if _, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ return
+ }
+ delay := subscribeRetryDelay(state.attempts)
+ state.attempts++
+ state.timer = time.AfterFunc(delay, func() {
+ s.retryAddListener(serviceNamesKey)
+ })
+ s.subscribeRetries[serviceNamesKey] = state
+ logger.Debugf("[Registry][ServiceDiscovery] instance listener for applications=%s not established, retry in %s", serviceNamesKey, delay)
+}
+
+// cancelSubscribeRetry stops a pending AddListener retry, if any.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetry(serviceNamesKey string) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ s.cancelSubscribeRetryLocked(serviceNamesKey)
+}
+
+// cancelSubscribeRetryLocked stops a pending AddListener retry; caller must
+// hold s.lock.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetryLocked(serviceNamesKey string) {
+ if state, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ state.timer.Stop()
+ delete(s.subscribeRetries, serviceNamesKey)
+ }
+}
+
+// retryAddListener re-runs AddListener after the backoff delay. On success it
+// re-syncs the latest instance snapshot so instance changes missed while the
+// subscription was down are picked up instead of leaving the consumer on a
+// stale view.
+func (s *serviceDiscoveryRegistry) retryAddListener(serviceNamesKey string) {
+ s.lock.Lock()
+ state, ok := s.subscribeRetries[serviceNamesKey]
+ if !ok {
+ s.lock.Unlock()
+ return
+ }
+ delete(s.subscribeRetries, serviceNamesKey)
+ active := !s.destroyed &&
+ s.serviceListeners[serviceNamesKey] == state.listener &&
+ listenerHasSubscribers(state.listener)
+ s.lock.Unlock()
+
+ if !active {
+ // Registry destroyed, listener replaced, or no subscriber left:
+ // stop retrying so the loop cannot leak.
+ return
+ }
Review Comment:
done
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry) getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized instance notification on application %s subscription, instance list size %d", serviceName, len(instances))
+ if err := listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery] ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL, serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event := metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5235315403
## [Codecov](https://app.codecov.io/gh/apache/dubbo-go/pull/3634?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `89.79592%` with `10 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 54.04%. Comparing base ([`60d1c2a`](https://app.codecov.io/gh/apache/dubbo-go/commit/60d1c2a949f0ee0da4be3fe09fb79d295491e040?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`0dfed19`](https://app.codecov.io/gh/apache/dubbo-go/commit/0dfed19034f022f70f5fce558e917e549e72a9f3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 901 commits behind head on develop.
| [Files with missing lines](https://app.codecov.io/gh/apache/dubbo-go/pull/3634?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...try/servicediscovery/service\_discovery\_registry.go](https://app.codecov.io/gh/apache/dubbo-go/pull/3634?src=pr&el=tree&filepath=registry%2Fservicediscovery%2Fservice_discovery_registry.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cmVnaXN0cnkvc2VydmljZWRpc2NvdmVyeS9zZXJ2aWNlX2Rpc2NvdmVyeV9yZWdpc3RyeS5nbw==) | 89.24% | [6 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/dubbo-go/pull/3634?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #3634 +/- ##
===========================================
+ Coverage 46.76% 54.04% +7.28%
===========================================
Files 295 474 +179
Lines 17172 36469 +19297
===========================================
+ Hits 8031 19711 +11680
- Misses 8287 15177 +6890
- Partials 854 1581 +727
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/dubbo-go/pull/3634?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] fix: retry failed application-level AddListener with capped backoff (#3624) [dubbo-go]
Posted by "sonarqubecloud[bot] (via GitHub)" <gi...@apache.org>.
sonarqubecloud[bot] commented on PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#issuecomment-5365863170
## [](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634) **Quality Gate passed**
Issues
 [0 New issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0 Accepted issues](https://sonarcloud.io/project/issues?id=apache_dubbo-go&pullRequest=3634&issueStatuses=ACCEPTED)
Measures
 [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_dubbo-go&pullRequest=3634&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_coverage&view=list)
 [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=apache_dubbo-go&pullRequest=3634&metric=new_duplicated_lines_density&view=list)
<!-- sqra-placement-anchor -->
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=apache_dubbo-go&pullRequest=3634)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org