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 6647d6a12 fix: retry failed application-level AddListener with capped
backoff (#3624) (#3634)
6647d6a12 is described below
commit 6647d6a1237e9aaef5db0926dc7f6c947d94684d
Author: AsperforMias <[email protected]>
AuthorDate: Fri Aug 21 18:57:53 2026 +0800
fix: retry failed application-level AddListener with capped backoff (#3624)
(#3634)
* fix: retry failed application-level AddListener with capped backoff
(#3624)
subscribeAndNotify ran serviceDiscovery.AddListener fire-and-forget: a
transient registry error (reconnect storm, leader election) was logged
once and never retried, so the consumer silently missed every instance
push until restart.
Add a retry loop mirroring the interface-level subscribe backoff from
registry/nacos (#3178): one pending retry per serviceNamesKey,
exponential backoff (1s..30s) with 25% jitter, unlimited attempts
while subscribers remain. A successful retry re-syncs the latest
instance snapshot so changes missed while the subscription was down
are picked up. Pending retries are canceled on Destroy and are not
armed once the registry is destroyed or no subscriber remains.
Also fix the inverted event.Succ flag and a %s/%d log mismatch in the
touched subscribe path.
* fix: remove notify listener by the same key subscribe used (#3624)
UnSubscribe called RemoveListener with url.ServiceKey() while
SubscribeURL had registered the notify listener under
url.ServiceKey()+":"+protocol, so the entry was never removed and
subscriber-count-based cleanup could never fire.
RemoveListener now uses the same protocolServiceKeyOf key, and the
last unsubscribe cancels any pending AddListener retry for the key.
* test: regression tests for application-level subscribe retry (#3624)
Cover the issue scenario and the retry contract:
- first AddListener fails, retry establishes the subscription and
re-syncs the latest snapshot so the consumer receives instance
events without a restart;
- backoff grows exponentially, is capped, and stays inside the
[delay, delay+25%] jitter band;
- retries stop after UnSubscribe and after Destroy, and no new timer
is armed once the registry is destroyed;
- repeated failing subscribes share a single pending retry timer.
* fix: resolve pending subscribe retry on success and preserve in-flight
state (#3624)
Two review findings on the retry lifecycle:
- A subscribe that succeeded while a retry was still pending did not
cancel the timer, so it later fired an extra AddListener. That is not
safe for service discoveries whose AddListener is not idempotent
(Polaris AddSubscriber, ZooKeeper ListenServiceEvent). A successful
AddListener now resolves the pending retry for the same listener.
- retryAddListener deleted the state before the external call, so a
concurrent failing subscribe could seed a fresh attempts=0 state and
the in-flight attempt's grown backoff was dropped on the next dedup.
The state now stays in the map while in flight (marked inFlight), the
failure path re-arms it in place under the lock, and a canceled flag
stops a superseded in-flight attempt from rescheduling.
Regression tests (verified to fail without the fix):
- first subscribe fails, a concurrent one succeeds: no third
AddListener call after the backoff window;
- a failing subscribe during an in-flight retry dedups against the
in-flight state instead of replacing it.
---
.../servicediscovery/service_discovery_registry.go | 270 +++++++++++++--
.../service_instances_changed_listener_impl.go | 21 +-
registry/servicediscovery/subscribe_retry_test.go | 370 +++++++++++++++++++++
3 files changed, 628 insertions(+), 33 deletions(-)
diff --git a/registry/servicediscovery/service_discovery_registry.go
b/registry/servicediscovery/service_discovery_registry.go
index d09203e8d..d19e5ef89 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -71,11 +71,15 @@ type serviceDiscoveryRegistry struct {
metadataReport report.MetadataReport
serviceListeners
map[string]registry.ServiceInstancesChangedListener
serviceMappingListeners map[string]mapping.MappingListener
- renewAppMetadataTimer *time.Timer
+ // subscribeRetries holds at most one pending AddListener retry per
+ // serviceNamesKey. Guarded by lock.
+ subscribeRetries map[string]*subscribeRetry
+ renewAppMetadataTimer *time.Timer
// destroyed is set by Destroy. SubscribeURL re-checks it under lock
right
// before installing a listener, so a subscribe whose initial
GetInstances /
// metadata phase outlives Destroy is discarded instead of being
installed
- // into a dead registry. Guarded by lock.
+ // into a dead registry, and late AddListener failures cannot arm new
retry
+ // timers. Guarded by lock.
destroyed bool
}
@@ -94,6 +98,7 @@ func newServiceDiscoveryRegistry(url *common.URL)
(registry.Registry, error) {
serviceNameMapping: extension.GetGlobalServiceNameMapping(),
metadataReport:
metadata.GetMetadataReportByRegistry(url.GetParam(constant.RegistryIdKey, "")),
serviceListeners:
make(map[string]registry.ServiceInstancesChangedListener),
+ subscribeRetries: make(map[string]*subscribeRetry),
// cache for mapping listener
serviceMappingListeners:
make(map[string]mapping.MappingListener),
}, nil
@@ -252,7 +257,13 @@ func (s *serviceDiscoveryRegistry) UnSubscribe(url
*common.URL, listener registr
}
serviceNamesKey := sortServices(services)
if l := s.getServiceListener(serviceNamesKey); l != nil {
+ // Must match the key SubscribeURL used for
AddListenerAndNotify,
+ // otherwise the entry leaks and subscriber tracking breaks.
l.RemoveListener(protocolSubscribeKey(url))
+ if !listenerHasSubscribers(l) {
+ // Last subscriber left: stop retrying AddListener for
this key.
+ s.cancelSubscribeRetry(serviceNamesKey)
+ }
}
s.stopListen(url)
err := s.serviceNameMapping.Remove(url)
@@ -358,6 +369,10 @@ func (s *serviceDiscoveryRegistry) Destroy() {
s.stopMetadataTimers()
s.lock.Lock()
s.destroyed = true
+ for key := range s.subscribeRetries {
+ // Cancel pending AddListener retries so their timers cannot
leak.
+ s.cancelSubscribeRetryLocked(key)
+ }
for _, l := range s.serviceListeners {
// Destroy drops listeners without RemoveListener; cancel any
pending
// metadata retry so its timer cannot leak.
@@ -596,17 +611,7 @@ func (s *serviceDiscoveryRegistry) SubscribeURL(url
*common.URL, notify registry
// other subscribe/unsubscribe on this registry. The lock below only
guards
// the serviceListeners check/install, never the external work.
listener := NewServiceInstancesChangedListenerWithContext(s.ctx,
url.GetParam(constant.ApplicationKey, ""),
s.url.GetParam(constant.RegistryIdKey, constant.DefaultKey), services)
- for _, serviceNameTmp := range services.Values() {
- serviceName := serviceNameTmp.(string)
- instances := s.serviceDiscovery.GetInstances(serviceName)
- logger.Infof("[Registry][ServiceDiscovery] synchronized
instance notification on application %s subscription, instance list size %s",
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)
- }
- }
+ s.loadLatestInstances(listener)
// Install under a short write lock with a double-check so a concurrent
// subscriber for the same key does not install a duplicate listener.
@@ -645,35 +650,246 @@ func (s *serviceDiscoveryRegistry)
getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
-// isDestroyed reports whether Destroy has run.
-func (s *serviceDiscoveryRegistry) isDestroyed() bool {
- s.lock.RLock()
- defer s.lock.RUnlock()
- return s.destroyed
+// 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(serviceNamesKey, 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. A success resolves any pending retry for
+// the key so its timer cannot fire an extra (possibly non-idempotent)
+// AddListener.
+func (s *serviceDiscoveryRegistry) addInstanceListener(serviceNamesKey string,
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))
+ if err == nil {
+ s.resolveSubscribeRetry(serviceNamesKey, listener)
+ }
+ 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 or in-flight AddListener retry for one
+// serviceNamesKey. The entry stays in the map while the attempt is in flight
+// (inFlight) so a concurrent failure cannot seed a fresh attempts=0 state and
+// reset the backoff. canceled is set when the state is resolved by a
+// successful subscribe or canceled by unsubscribe/Destroy; an in-flight
+// attempt checks it before rescheduling.
+type subscribeRetry struct {
+ listener registry.ServiceInstancesChangedListener
+ url *common.URL
+ timer *time.Timer
+ attempts int
+ inFlight bool
+ canceled bool
+}
+
+// scheduleSubscribeRetry arms the retry timer for serviceNamesKey after a
+// failed AddListener call. One pending retry per key: repeated failures share
+// the same state 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 || state.canceled {
+ 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
+ }
+ s.subscribeRetries[serviceNamesKey] = state
+ s.armSubscribeRetryLocked(serviceNamesKey, state)
+}
+
+// armSubscribeRetryLocked computes the next backoff delay and arms the retry
+// timer; caller must hold s.lock and the state must be in the map.
+func (s *serviceDiscoveryRegistry) armSubscribeRetryLocked(serviceNamesKey
string, state *subscribeRetry) {
+ delay := subscribeRetryDelay(state.attempts)
+ state.attempts++
+ state.timer = time.AfterFunc(delay, func() {
+ s.retryAddListener(serviceNamesKey)
+ })
+ 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. An in-flight attempt is not interrupted but will not
+// reschedule.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetryLocked(serviceNamesKey
string) {
+ if state, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ state.canceled = true
+ if state.timer != nil {
+ state.timer.Stop()
+ }
+ delete(s.subscribeRetries, serviceNamesKey)
+ }
+}
+
+// resolveSubscribeRetry drops a pending retry after a successful AddListener
+// for the same listener: the subscription is established, so the timer must
+// not fire an extra (possibly non-idempotent) AddListener. An in-flight
+// attempt that later fails sees canceled and does not reschedule.
+func (s *serviceDiscoveryRegistry) resolveSubscribeRetry(serviceNamesKey
string, listener registry.ServiceInstancesChangedListener) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ state, ok := s.subscribeRetries[serviceNamesKey]
+ if !ok || state.listener != listener {
+ return
+ }
+ state.canceled = true
+ if state.timer != nil {
+ state.timer.Stop()
+ }
+ delete(s.subscribeRetries, serviceNamesKey)
+}
+
+// retryAddListener re-runs AddListener after the backoff delay. The state
+// stays in the map while the attempt is in flight so concurrent failures
+// dedup against it and the backoff attempts keep growing. 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
+ }
+ state.timer = nil
+ state.inFlight = true
+ active := !s.destroyed && !state.canceled &&
+ s.serviceListeners[serviceNamesKey] == state.listener &&
+ listenerHasSubscribers(state.listener)
+ s.lock.Unlock()
+
+ if !active {
+ // Registry destroyed, listener replaced, or no subscriber left:
+ // drop the state so the loop cannot leak.
+ s.lock.Lock()
+ if s.subscribeRetries[serviceNamesKey] == state {
+ delete(s.subscribeRetries, serviceNamesKey)
+ }
+ s.lock.Unlock()
+ return
+ }
+ if err := s.addInstanceListener(serviceNamesKey, state.url,
state.listener); err != nil {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ if state.canceled || s.destroyed ||
!listenerHasSubscribers(state.listener) {
+ // A concurrent successful subscribe resolved this
state, or the
+ // listener is gone: drop it instead of rescheduling.
+ if s.subscribeRetries[serviceNamesKey] == state {
+ delete(s.subscribeRetries, serviceNamesKey)
+ }
+ return
+ }
+ // Re-arm in place: the entry never left the map, so the backoff
+ // attempts keep growing and no duplicate state can appear.
+ state.inFlight = false
+ s.armSubscribeRetryLocked(serviceNamesKey, state)
+ logger.Warnf("[Registry][ServiceDiscovery] retry add instance
listener failed, applications=%s attempt=%d err=%s",
+ serviceNamesKey, state.attempts, err.Error())
+ return
+ }
+ // Success: addInstanceListener already resolved the state.
+ if s.isDestroyed() {
+ return
+ }
+ logger.Infof("[Registry][ServiceDiscovery] instance listener for
applications=%s established after %d retries, re-syncing latest instances",
+ serviceNamesKey, state.attempts)
+ s.loadLatestInstances(state.listener)
+}
+
+// subscribeRetryDelay returns exponential backoff (initial << attempt) capped
+// at subscribeRetryMaxDelay, plus up to 25% jitter to desynchronize retries
+// across consumers after a correlated registry failure.
+func subscribeRetryDelay(attempt int) time.Duration {
+ delay := subscribeRetryMaxDelay
+ if attempt >= 0 && attempt < 30 { // guard against shift overflow
+ delay = subscribeRetryInitialDelay << attempt
+ if delay <= 0 || delay > subscribeRetryMaxDelay {
+ delay = subscribeRetryMaxDelay
+ }
+ }
+ return delay + time.Duration(rand.Int64N(int64(delay/4)+1))
+}
+
+// isDestroyed reports whether Destroy has run.
+func (s *serviceDiscoveryRegistry) isDestroyed() bool {
+ s.lock.RLock()
+ defer s.lock.RUnlock()
+ return s.destroyed
+}
+
+// listenerHasSubscribers reports whether the listener still has subscribers.
+// Unknown implementations are assumed active so retries are never dropped
+// silently.
+func listenerHasSubscribers(listener registry.ServiceInstancesChangedListener)
bool {
+ if impl, ok := listener.(*ServiceInstancesChangedListenerImpl); ok {
+ return impl.hasSubscribers()
+ }
+ return true
+}
+
// protocolSubscribeKey builds the key under which a subscription's notify
// listener is registered on the ServiceInstancesChangedListener. SubscribeURL
// and UnSubscribe must derive it the same way, or the last subscriber is never
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl.go
b/registry/servicediscovery/service_instances_changed_listener_impl.go
index e535bb3b0..ae5916f16 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl.go
@@ -317,6 +317,15 @@ func (lstn *ServiceInstancesChangedListenerImpl)
RemoveListener(serviceKey strin
}
}
+// hasSubscribers reports whether any notify listener is still attached. The
+// owning registry uses it to stop pending subscribe retries once the last
+// subscriber unsubscribes.
+func (lstn *ServiceInstancesChangedListenerImpl) hasSubscribers() bool {
+ lstn.mutex.Lock()
+ defer lstn.mutex.Unlock()
+ return len(lstn.listeners) > 0
+}
+
// GetServiceNames return all listener service names
func (lstn *ServiceInstancesChangedListenerImpl) GetServiceNames()
*gxset.HashSet {
return lstn.serviceNames
@@ -349,6 +358,12 @@ func metadataCacheKey(app, registryId, revision string)
string {
return app + ":" + registryId + ":" + revision
}
+// metadataInfoFetcher resolves MetadataInfo for a revision; a package-level
+// indirection so tests can inject transient failures. It follows
+// GetMetadataInfoWithContext so listener refreshes are canceled with the
+// listener's lifecycle context.
+var metadataInfoFetcher = GetMetadataInfoWithContext
+
// GetMetadataInfo retrieves the MetadataInfo for a service instance by
revision.
// Results are cached by app+registryId+revision, where app must be the
provider
// application name. For "remote" storage type, it fetches from the metadata
report
@@ -400,12 +415,6 @@ var (
metadataFetchFailureLogInterval = 5 * time.Minute
)
-// metadataInfoFetcher resolves MetadataInfo for a revision; a package-level
-// indirection so tests can inject transient failures. It follows
-// GetMetadataInfoWithContext so listener refreshes are canceled with the
-// listener's lifecycle context.
-var metadataInfoFetcher = GetMetadataInfoWithContext
-
// stopMetadataRetry marks the listener closed and cancels any pending metadata
// retry. It is called when the owning registry is destroyed and drops this
// listener without RemoveListener, so neither the pending timer nor an
diff --git a/registry/servicediscovery/subscribe_retry_test.go
b/registry/servicediscovery/subscribe_retry_test.go
new file mode 100644
index 000000000..d67183a97
--- /dev/null
+++ b/registry/servicediscovery/subscribe_retry_test.go
@@ -0,0 +1,370 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package servicediscovery
+
+import (
+ "context"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+import (
+ gxset "github.com/dubbogo/gost/container/set"
+
+ perrors "github.com/pkg/errors"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/metadata/info"
+ "dubbo.apache.org/dubbo-go/v3/registry"
+)
+
+// stubSubscribeRetryDelays shrinks the subscribe backoff so retries fire
within
+// test time budgets.
+func stubSubscribeRetryDelays(t *testing.T, initial, max time.Duration) {
+ t.Helper()
+ origInitial, origMax := subscribeRetryInitialDelay,
subscribeRetryMaxDelay
+ subscribeRetryInitialDelay, subscribeRetryMaxDelay = initial, max
+ t.Cleanup(func() { subscribeRetryInitialDelay, subscribeRetryMaxDelay =
origInitial, origMax })
+}
+
+// retrySubscribeDiscovery is a concurrency-safe ServiceDiscovery stub whose
+// AddListener keeps failing while fail is set, so tests can control exactly
+// when the subscription is allowed to be established. When blockOnCall > 0,
+// the Nth AddListener call blocks until blockRelease is closed, letting tests
+// hold an attempt in flight deterministically.
+type retrySubscribeDiscovery struct {
+ mockServiceDiscovery
+ mu sync.Mutex
+ fail atomic.Bool
+ addCalls int
+ instances []registry.ServiceInstance
+ blockOnCall int
+ blockEntered chan struct{}
+ blockRelease chan struct{}
+}
+
+func (m *retrySubscribeDiscovery)
AddListener(registry.ServiceInstancesChangedListener) error {
+ m.mu.Lock()
+ m.addCalls++
+ call := m.addCalls
+ m.mu.Unlock()
+ if call == m.blockOnCall && m.blockRelease != nil {
+ close(m.blockEntered)
+ <-m.blockRelease
+ }
+ if m.fail.Load() {
+ return perrors.New("transient subscribe failure")
+ }
+ return nil
+}
+
+func (m *retrySubscribeDiscovery) GetInstances(string)
[]registry.ServiceInstance {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return append([]registry.ServiceInstance(nil), m.instances...)
+}
+
+func (m *retrySubscribeDiscovery) setInstances(instances
[]registry.ServiceInstance) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.instances = instances
+}
+
+func (m *retrySubscribeDiscovery) addListenerCalls() int {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return m.addCalls
+}
+
+func newRetryTestRegistry(sd registry.ServiceDiscovery)
*serviceDiscoveryRegistry {
+ registryURL, _ := common.NewURL(testRegistryURL,
+ common.WithParamsValue(constant.RegistryKey, "mock"))
+ return &serviceDiscoveryRegistry{
+ url: registryURL,
+ serviceDiscovery: sd,
+ serviceListeners:
make(map[string]registry.ServiceInstancesChangedListener),
+ subscribeRetries: make(map[string]*subscribeRetry),
+ serviceNameMapping: &mockServiceNameMapping{data:
map[string]*gxset.HashSet{testInterface: gxset.NewSet(testApp)}},
+ }
+}
+
+func newRetryConsumerURL(t *testing.T) *common.URL {
+ t.Helper()
+ consumerURL, err := common.NewURL("tri://127.0.0.1:20000/",
+ common.WithInterface(testInterface),
+ common.WithParamsValue(constant.SideKey, constant.SideConsumer))
+ require.NoError(t, err)
+ return consumerURL
+}
+
+// TestSubscribeRetryRecoversAfterTransientFailure is the core regression test
+// for issue #3624: the first AddListener fails, no further registry event
+// arrives, and the subscription must still be established by the retry loop —
+// including a re-sync of the latest instance snapshot so the consumer receives
+// instance events without a restart.
+func TestSubscribeRetryRecoversAfterTransientFailure(t *testing.T) {
+ const revision = "rev-subscribe-retry"
+ const port = 22301
+ stubSubscribeRetryDelays(t, 5*time.Millisecond, 20*time.Millisecond)
+
+ meta := newTestMetadataInfo(t, revision, port, "")
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ return meta, nil
+ })
+ t.Cleanup(func() { metaCache.Delete(testApp + ":" + constant.DefaultKey
+ ":" + revision) })
+
+ sd := &retrySubscribeDiscovery{}
+ sd.fail.Store(true)
+ reg := newRetryTestRegistry(sd)
+ t.Cleanup(reg.Destroy)
+
+ notify := &retryNotifyListener{}
+ reg.SubscribeURL(newRetryConsumerURL(t), notify, gxset.NewSet(testApp))
+
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 1
}, time.Second, 5*time.Millisecond,
+ "initial AddListener attempt should have run")
+ require.Empty(t, notify.snapshot(), "no instance is visible while the
subscription is down")
+
+ // The provider changes while the subscription is not established.
+
sd.setInstances([]registry.ServiceInstance{newTestServiceInstanceOnly(port, "",
revision)})
+ sd.fail.Store(false)
+
+ require.Eventually(t, func() bool { return len(notify.snapshot()) > 0
}, 3*time.Second, 10*time.Millisecond,
+ "retry must establish the subscription and re-sync the latest
snapshot")
+ assert.Empty(t, reg.subscribeRetries, "retry state must be cleared once
the subscription is established")
+}
+
+// TestSubscribeRetryBackoffDelay verifies the backoff grows exponentially, is
+// capped, and stays within the [delay, delay+25%] jitter band.
+func TestSubscribeRetryBackoffDelay(t *testing.T) {
+ initial, max := time.Second, 30*time.Second
+ stubSubscribeRetryDelays(t, initial, max)
+ for _, attempt := range []int{0, 1, 2, 3} {
+ want := initial << attempt
+ for range 50 {
+ got := subscribeRetryDelay(attempt)
+ assert.GreaterOrEqual(t, got, want)
+ assert.LessOrEqual(t, got, want+want/4)
+ }
+ }
+ // Beyond the cap every attempt lands in the capped jitter band.
+ for _, attempt := range []int{5, 10, 29, 30, 100} {
+ for range 50 {
+ got := subscribeRetryDelay(attempt)
+ assert.GreaterOrEqual(t, got, max)
+ assert.LessOrEqual(t, got, max+max/4)
+ }
+ }
+}
+
+// TestSubscribeRetryStopsAfterUnSubscribe verifies retries stop once the last
+// subscriber unsubscribes.
+func TestSubscribeRetryStopsAfterUnSubscribe(t *testing.T) {
+ stubSubscribeRetryDelays(t, 5*time.Millisecond, 10*time.Millisecond)
+
+ sd := &retrySubscribeDiscovery{}
+ sd.fail.Store(true)
+ reg := newRetryTestRegistry(sd)
+ t.Cleanup(reg.Destroy)
+
+ consumerURL := newRetryConsumerURL(t)
+ reg.SubscribeURL(consumerURL, &retryNotifyListener{},
gxset.NewSet(testApp))
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 2
}, time.Second, 5*time.Millisecond,
+ "retry loop should be running")
+
+ require.NoError(t, reg.UnSubscribe(consumerURL, &retryNotifyListener{}))
+ require.Eventually(t, func() bool {
+ reg.lock.RLock()
+ defer reg.lock.RUnlock()
+ return len(reg.subscribeRetries) == 0
+ }, time.Second, 5*time.Millisecond, "pending retry must be canceled on
unsubscribe")
+
+ time.Sleep(30 * time.Millisecond)
+ before := sd.addListenerCalls()
+ time.Sleep(100 * time.Millisecond)
+ assert.Equal(t, before, sd.addListenerCalls(), "no AddListener attempt
should happen after unsubscribe")
+}
+
+// TestSubscribeRetryStopsAfterDestroy verifies Destroy cancels pending
retries.
+func TestSubscribeRetryStopsAfterDestroy(t *testing.T) {
+ stubSubscribeRetryDelays(t, 5*time.Millisecond, 10*time.Millisecond)
+
+ sd := &retrySubscribeDiscovery{}
+ sd.fail.Store(true)
+ reg := newRetryTestRegistry(sd)
+
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
gxset.NewSet(testApp))
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 2
}, time.Second, 5*time.Millisecond,
+ "retry loop should be running")
+
+ reg.Destroy()
+ assert.Empty(t, reg.subscribeRetries, "Destroy must cancel pending
retries")
+
+ time.Sleep(30 * time.Millisecond)
+ before := sd.addListenerCalls()
+ time.Sleep(100 * time.Millisecond)
+ assert.Equal(t, before, sd.addListenerCalls(), "no AddListener attempt
should happen after Destroy")
+
+ // A late failure after Destroy must not arm a new retry timer.
+ reg.scheduleSubscribeRetry(testApp, &subscribeRetry{})
+ assert.Empty(t, reg.subscribeRetries, "retries must not be scheduled
after Destroy")
+}
+
+// TestSubscribeRetryUsesSingleTimer verifies repeated failing subscribes for
+// the same applications share one pending retry instead of stacking timers.
+func TestSubscribeRetryUsesSingleTimer(t *testing.T) {
+ stubSubscribeRetryDelays(t, time.Second, 2*time.Second)
+
+ sd := &retrySubscribeDiscovery{}
+ sd.fail.Store(true)
+ reg := newRetryTestRegistry(sd)
+ t.Cleanup(reg.Destroy)
+
+ services := gxset.NewSet(testApp)
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
services)
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 1
}, time.Second, 5*time.Millisecond)
+ // A second subscribe hits the fast path and fails again; it must reuse
the
+ // pending retry, not arm a second timer.
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
services)
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 2
}, time.Second, 5*time.Millisecond)
+ // addCalls is incremented before the failure path schedules; give both
+ // schedule attempts time to settle before asserting on the map.
+ time.Sleep(100 * time.Millisecond)
+
+ reg.lock.RLock()
+ defer reg.lock.RUnlock()
+ assert.Len(t, reg.subscribeRetries, 1, "repeated failures must share
the pending retry timer")
+ assert.Equal(t, 1, reg.subscribeRetries[testApp].attempts)
+}
+
+// TestSubscribeRetryResolvedByConcurrentSuccess covers the review finding: a
+// subscribe that succeeds while a retry is pending must cancel the pending
+// timer — otherwise it fires an extra AddListener, which is not safe for
+// service discoveries whose AddListener is not idempotent (e.g. Polaris
+// AddSubscriber, ZooKeeper ListenServiceEvent).
+func TestSubscribeRetryResolvedByConcurrentSuccess(t *testing.T) {
+ stubSubscribeRetryDelays(t, 500*time.Millisecond, time.Second)
+
+ sd := &retrySubscribeDiscovery{}
+ sd.fail.Store(true)
+ reg := newRetryTestRegistry(sd)
+ t.Cleanup(reg.Destroy)
+
+ services := gxset.NewSet(testApp)
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
services)
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 1
}, time.Second, 5*time.Millisecond,
+ "initial AddListener attempt should have run")
+ require.Eventually(t, func() bool {
+ reg.lock.RLock()
+ defer reg.lock.RUnlock()
+ return len(reg.subscribeRetries) == 1
+ }, time.Second, 5*time.Millisecond, "retry should be pending after the
failed subscribe")
+
+ // A later subscribe succeeds while the retry is still pending.
+ sd.fail.Store(false)
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
services)
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 2
}, time.Second, 5*time.Millisecond)
+
+ // Wait well beyond the backoff: the pending retry must have been
resolved.
+ time.Sleep(700 * time.Millisecond)
+ assert.Equal(t, 2, sd.addListenerCalls(), "successful subscribe must
cancel the pending retry")
+ reg.lock.RLock()
+ defer reg.lock.RUnlock()
+ assert.Empty(t, reg.subscribeRetries, "retry state must be resolved
after success")
+}
+
+// TestSubscribeRetryInFlightStatePreserved covers the review finding: the
+// retry state must stay in the map while its attempt is in flight, so a
+// concurrent failing subscribe dedups against it instead of seeding a fresh
+// attempts=0 state that resets the backoff.
+func TestSubscribeRetryInFlightStatePreserved(t *testing.T) {
+ // Large delays: no timer fires during the test; the retry is driven
+ // synchronously instead.
+ stubSubscribeRetryDelays(t, time.Hour, 2*time.Hour)
+
+ sd := &retrySubscribeDiscovery{
+ blockOnCall: 2,
+ blockEntered: make(chan struct{}),
+ blockRelease: make(chan struct{}),
+ }
+ sd.fail.Store(true)
+ reg := newRetryTestRegistry(sd)
+ t.Cleanup(reg.Destroy)
+
+ services := gxset.NewSet(testApp)
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
services)
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 1
}, time.Second, 5*time.Millisecond)
+
+ var state *subscribeRetry
+ require.Eventually(t, func() bool {
+ reg.lock.RLock()
+ defer reg.lock.RUnlock()
+ state = reg.subscribeRetries[testApp]
+ return state != nil
+ }, time.Second, 5*time.Millisecond, "retry should be pending after the
failed subscribe")
+ reg.lock.Lock()
+ if state.timer != nil {
+ state.timer.Stop()
+ }
+ reg.lock.Unlock()
+
+ // Drive the retry attempt synchronously; it blocks inside AddListener
+ // call #2, i.e. in flight.
+ retryDone := make(chan struct{})
+ go func() {
+ defer close(retryDone)
+ reg.retryAddListener(testApp)
+ }()
+ select {
+ case <-sd.blockEntered:
+ case <-time.After(2 * time.Second):
+ t.Fatal("retry attempt never reached AddListener")
+ }
+
+ // A concurrent subscribe fails while the retry is in flight.
+ reg.SubscribeURL(newRetryConsumerURL(t), &retryNotifyListener{},
services)
+ require.Eventually(t, func() bool { return sd.addListenerCalls() >= 3
}, time.Second, 5*time.Millisecond)
+ // Let its schedule attempt finish before the in-flight attempt
resolves,
+ // so the dedup check runs while the in-flight state is still present.
+ time.Sleep(100 * time.Millisecond)
+
+ reg.lock.RLock()
+ current := reg.subscribeRetries[testApp]
+ reg.lock.RUnlock()
+ assert.Same(t, state, current,
+ "in-flight retry state must be preserved, not replaced by a
fresh attempts=0 state")
+
+ // Let the in-flight attempt succeed: the state is resolved.
+ sd.fail.Store(false)
+ close(sd.blockRelease)
+ <-retryDone
+ require.Eventually(t, func() bool {
+ reg.lock.RLock()
+ defer reg.lock.RUnlock()
+ return len(reg.subscribeRetries) == 0
+ }, time.Second, 5*time.Millisecond, "successful in-flight retry must
resolve the state")
+ assert.Equal(t, 3, sd.addListenerCalls(), "no extra AddListener beyond
the three driven attempts")
+}
