This is an automated email from the ASF dual-hosted git repository.
Alanxtl 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 106675bb5 fix: self-heal application-level discovery after transient
metadata failure (#3615) (#3625)
106675bb5 is described below
commit 106675bb54e85b7f0dfe90ab6ba48b72c0c76a31
Author: AsperforMias <[email protected]>
AuthorDate: Fri Aug 21 10:18:05 2026 +0800
fix: self-heal application-level discovery after transient metadata failure
(#3615) (#3625)
* fix: self-heal application-level discovery after transient metadata
failure (#3615)
Two stacked recovery gaps made a transient MetadataService fetch failure
permanent for application-level service discovery:
1. ServiceInstancesChangedListenerImpl skipped instances whose metadata
fetch failed and never retried; with no further registry event the
consumer directory stayed empty forever. Add a shared retry timer per
listener: unresolved revisions are re-resolved by replaying the latest
instance snapshot with capped exponential backoff (1s..30s + jitter,
unlimited attempts). Retries stop naturally when instances leave the
snapshot, subscribers are detached, or the registry is destroyed.
Metadata RPCs no longer run under the listener state mutex.
2. RegistryDirectory's closing tombstone vetoed the rebuild even after a
successful retry when the provider restarted with the same address,
because a stale pre-shutdown snapshot and a genuine restart were
indistinguishable. The tombstone now records the export timestamp and
only vetoes re-adds carrying the same timestamp; a different timestamp
proves a genuine restart and clears the tombstone.
* refactor: address review comments (#3615)
- Drop the caller-less hasActiveClosingTombstone wrapper; production code
uses activeClosingTombstone directly and the tests keep a local helper.
- Demote the per-event "received instance notification" log to debug:
it fires on every registry push and is too noisy at info level.
- refreshServiceURLs now reports whether every revision resolved, and
OnEvent surfaces unresolved revisions as an error so registry
dispatchers log the partial failure instead of it staying silent.
The retry loop is unchanged and still self-schedules.
* fix: close metadata retry lifecycle gaps found in review (#3615)
A retry timer could outlive its listener in three ways:
- scheduleMetadataRetry only checked unresolvedRevisions, so a refresh
that failed before any subscriber attached (the SubscribeURL order)
armed the timer, and a listener discarded after losing the install
race kept retrying — and was kept alive by the timer closure.
- RemoveListener racing with the timer callback still allowed the
callback to refresh and re-schedule with no subscribers left.
- Destroy raced with an in-flight refresh: stopMetadataRetry canceled
the pending timer, but the refresh's trailing scheduleMetadataRetry
armed a new one after the registry was gone.
The listener now has a closed flag; the scheduler and the timer
callback re-check closed and subscriber presence under the mutex, and
a refresh that fails with no subscribers is re-armed by the next
AddListenerAndNotify instead of running detached. SubscribeURL closes
the listener that loses the install race.
* fix: discard subscribes that outlive registry Destroy (#3615)
closed covered listeners already installed when Destroy runs, but a
SubscribeURL still in its initial GetInstances/metadata phase is
invisible to Destroy's listener sweep: when the call finished after
Destroy returned, it would install the listener, attach the subscriber
and arm a fresh unlimited retry timer on a dead registry (a blocking
fetch probe shows metadata fetches continuing to grow after Destroy).
The registry now carries a destroyed flag. SubscribeURL bails early on
destroyed registries and re-checks the flag under the write lock right
before installing: a late listener is closed and discarded instead of
installed, so no AddListener call and no retry timer can appear after
Destroy.
Adds the deterministic SubscribeURL x Destroy interleaving test
requested in review (blocking fetch probe; verified to fail without
the fix: listener installed, AddListener called, fetches continue)
plus a post-Destroy subscribe no-op test.
* fix: unsubscribe with the same protocol-qualified listener key (#3615)
SubscribeURL registers the notify listener under
url.ServiceKey()+":"+protocol,
but UnSubscribe removed it with the bare url.ServiceKey(), so the last
subscriber was never detached and the metadata retry timer kept probing
after
UnSubscribe. Derive both sides from a shared protocolSubscribeKey helper and
cover the public SubscribeURL -> UnSubscribe lifecycle with a regression
test.
---
registry/directory/directory.go | 35 +-
registry/directory/directory_test.go | 57 +++-
registry/servicediscovery/metadata_retry_test.go | 372 +++++++++++++++++++++
.../servicediscovery/service_discovery_registry.go | 61 +++-
.../service_instances_changed_listener_impl.go | 236 +++++++++++--
...service_instances_changed_listener_impl_test.go | 7 +-
.../subscribe_destroy_race_test.go | 164 +++++++++
7 files changed, 879 insertions(+), 53 deletions(-)
diff --git a/registry/directory/directory.go b/registry/directory/directory.go
index c0b7bd0dc..a2b5368f0 100644
--- a/registry/directory/directory.go
+++ b/registry/directory/directory.go
@@ -87,8 +87,12 @@ type closingTombstone struct {
InstanceKey string
ServiceKey string
Address string
- Source string
- ExpireAt time.Time
+ // Timestamp is the export timestamp of the closing instance's URL. A
re-add
+ // carrying a different timestamp is a genuine restart, not a stale
+ // pre-shutdown registry snapshot.
+ Timestamp string
+ Source string
+ ExpireAt time.Time
}
var defaultClosingTombstoneTTL = func() time.Duration {
@@ -645,24 +649,25 @@ func (dir *RegistryDirectory)
markClosingTombstone(instanceKey string, invoker p
if invoker != nil && invoker.GetURL() != nil {
tombstone.ServiceKey = invoker.GetURL().ServiceKey()
tombstone.Address = invoker.GetURL().Location
+ tombstone.Timestamp =
invoker.GetURL().GetParam(constant.TimestampKey, "")
}
dir.closingTombstones.Store(instanceKey, tombstone)
}
-func (dir *RegistryDirectory) hasActiveClosingTombstone(instanceKey string)
bool {
+func (dir *RegistryDirectory) activeClosingTombstone(instanceKey string)
(closingTombstone, bool) {
if instanceKey == "" {
- return false
+ return closingTombstone{}, false
}
tombstoneValue, ok := dir.closingTombstones.Load(instanceKey)
if !ok {
- return false
+ return closingTombstone{}, false
}
tombstone := tombstoneValue.(closingTombstone)
if time.Now().After(tombstone.ExpireAt) {
dir.closingTombstones.Delete(instanceKey)
- return false
+ return closingTombstone{}, false
}
- return true
+ return tombstone, true
}
func (dir *RegistryDirectory) clearClosingTombstone(instanceKey string) {
@@ -712,9 +717,19 @@ func (dir *RegistryDirectory) cacheInvoker(url
*common.URL, event *registry.Serv
func (dir *RegistryDirectory) doCacheInvoker(newUrl *common.URL, event
*registry.ServiceEvent) (protocolbase.Invoker, bool) {
key := event.Key()
dir.cleanupExpiredClosingTombstones()
- if dir.hasActiveClosingTombstone(key) {
- logger.Infof("[Registry][Directory] skip rebuilding closing
instance due to tombstone, instance key: %s", key)
- return nil, true
+ if tombstone, ok := dir.activeClosingTombstone(key); ok {
+ // A tombstone guards against re-adding an instance from a stale
+ // pre-shutdown registry snapshot. If the re-add carries a
different
+ // export timestamp, the instance has genuinely restarted with
the same
+ // address: vetoing it would keep the directory empty until the
next
+ // registry event, which may never come.
+ newTimestamp := newUrl.GetParam(constant.TimestampKey, "")
+ if tombstone.Timestamp == "" || newTimestamp == "" ||
newTimestamp == tombstone.Timestamp {
+ logger.Infof("[Registry][Directory] skip rebuilding
closing instance due to tombstone, instance key: %s", key)
+ return nil, true
+ }
+ logger.Infof("[Registry][Directory] instance %s restarted with
a new export timestamp, clearing closing tombstone", key)
+ dir.clearClosingTombstone(key)
}
cacheInvoker, ok := dir.cacheInvokersMap.Load(key)
var existingInvoker protocolbase.Invoker
diff --git a/registry/directory/directory_test.go
b/registry/directory/directory_test.go
index 1cfbeab6f..c6a5be7fc 100644
--- a/registry/directory/directory_test.go
+++ b/registry/directory/directory_test.go
@@ -351,6 +351,14 @@ func TestRemoveClosingInstanceReturnsFalseForUnknownKey(t
*testing.T) {
assert.Empty(t, registryDirectory.snapshotCacheInvokers())
}
+// hasActiveClosingTombstone reports whether an unexpired tombstone exists for
+// the instance key. Test helper; production code uses activeClosingTombstone
+// directly because it also needs the tombstone payload.
+func hasActiveClosingTombstone(dir *RegistryDirectory, instanceKey string)
bool {
+ _, ok := dir.activeClosingTombstone(instanceKey)
+ return ok
+}
+
func TestClosingTombstonePreventsRebuildUntilDeleteEvent(t *testing.T) {
registryDirectory, mockRegistry := normalRegistryDir(true)
@@ -368,7 +376,7 @@ func TestClosingTombstonePreventsRebuildUntilDeleteEvent(t
*testing.T) {
removed := registryDirectory.RemoveClosingInstance(key)
require.True(t, removed)
assert.Empty(t, registryDirectory.snapshotCacheInvokers())
- assert.True(t, registryDirectory.hasActiveClosingTombstone(key))
+ assert.True(t, hasActiveClosingTombstone(registryDirectory, key))
mockRegistry.MockEvent(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: providerURL})
time.Sleep(1e9)
@@ -376,7 +384,7 @@ func TestClosingTombstonePreventsRebuildUntilDeleteEvent(t
*testing.T) {
mockRegistry.MockEvent(®istry.ServiceEvent{Action:
remoting.EventTypeDel, Service: providerURL})
time.Sleep(1e9)
- assert.False(t, registryDirectory.hasActiveClosingTombstone(key))
+ assert.False(t, hasActiveClosingTombstone(registryDirectory, key))
mockRegistry.MockEvent(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: providerURL})
time.Sleep(1e9)
@@ -402,13 +410,56 @@ func TestExpiredClosingTombstoneAllowsRebuild(t
*testing.T) {
assert.Empty(t, registryDirectory.snapshotCacheInvokers())
time.Sleep(40 * time.Millisecond)
- assert.False(t, registryDirectory.hasActiveClosingTombstone(key))
+ assert.False(t, hasActiveClosingTombstone(registryDirectory, key))
mockRegistry.MockEvent(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: providerURL})
time.Sleep(1e9)
assert.Len(t, registryDirectory.snapshotCacheInvokers(), 1)
}
+// TestClosingTombstoneAllowsRebuildAfterGenuineRestart verifies the tombstone
+// only vetoes stale pre-shutdown snapshots: a re-add with a new export
+// timestamp is a genuine same-address restart and must rebuild immediately.
+func TestClosingTombstoneAllowsRebuildAfterGenuineRestart(t *testing.T) {
+ registryDirectory, mockRegistry := normalRegistryDir(true)
+
+ oldURL, _ :=
common.NewURL("dubbo://0.0.0.0:20000/org.apache.dubbo-go.mockService",
+ common.WithParamsValue(constant.ClusterKey, "mock1"),
+ common.WithParamsValue(constant.GroupKey, "group"),
+ common.WithParamsValue(constant.VersionKey, "1.0.0"),
+ common.WithParamsValue(constant.TimestampKey, "1000"))
+ newURL, _ :=
common.NewURL("dubbo://0.0.0.0:20000/org.apache.dubbo-go.mockService",
+ common.WithParamsValue(constant.ClusterKey, "mock1"),
+ common.WithParamsValue(constant.GroupKey, "group"),
+ common.WithParamsValue(constant.VersionKey, "1.0.0"),
+ common.WithParamsValue(constant.TimestampKey, "2000"))
+
+ oldEvent := ®istry.ServiceEvent{Action: remoting.EventTypeAdd,
Service: oldURL}
+ newEvent := ®istry.ServiceEvent{Action: remoting.EventTypeAdd,
Service: newURL}
+ key := registryDirectory.invokerCacheKey(oldEvent)
+ // The discrimination only matters when both URLs map to the same
instance key.
+ require.Equal(t, key, registryDirectory.invokerCacheKey(newEvent))
+
+ mockRegistry.MockEvent(oldEvent)
+ time.Sleep(1e9)
+ require.Len(t, registryDirectory.snapshotCacheInvokers(), 1)
+
+ require.True(t, registryDirectory.RemoveClosingInstance(key))
+ assert.Empty(t, registryDirectory.snapshotCacheInvokers())
+ require.True(t, hasActiveClosingTombstone(registryDirectory, key))
+
+ // Stale snapshot re-add (same export timestamp) stays vetoed.
+ mockRegistry.MockEvent(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: oldURL})
+ time.Sleep(1e9)
+ assert.Empty(t, registryDirectory.snapshotCacheInvokers())
+
+ // Genuine restart (new export timestamp) rebuilds despite the active
tombstone.
+ mockRegistry.MockEvent(newEvent)
+ time.Sleep(1e9)
+ assert.Len(t, registryDirectory.snapshotCacheInvokers(), 1)
+ assert.False(t, hasActiveClosingTombstone(registryDirectory, key))
+}
+
func TestRefreshConfiguratorsUseLatestBatch(t *testing.T) {
realConfigurator := extension.GetDefaultConfiguratorFunc()
diff --git a/registry/servicediscovery/metadata_retry_test.go
b/registry/servicediscovery/metadata_retry_test.go
new file mode 100644
index 000000000..1b6f51ef0
--- /dev/null
+++ b/registry/servicediscovery/metadata_retry_test.go
@@ -0,0 +1,372 @@
+/*
+ * 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"
+)
+
+// retryNotifyListener is a concurrency-safe notify listener: retry rebuilds
run
+// on the retry timer goroutine, so test listeners must synchronize access.
+type retryNotifyListener struct {
+ mu sync.Mutex
+ events []*registry.ServiceEvent
+}
+
+func (c *retryNotifyListener) Notify(event *registry.ServiceEvent) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.events = append(c.events, event)
+}
+
+func (c *retryNotifyListener) NotifyAll(events []*registry.ServiceEvent,
callback func()) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.events = append([]*registry.ServiceEvent(nil), events...)
+ if callback != nil {
+ callback()
+ }
+}
+
+func (c *retryNotifyListener) snapshot() []*registry.ServiceEvent {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return append([]*registry.ServiceEvent(nil), c.events...)
+}
+
+// stubMetadataFetch replaces the metadata fetcher and restores it on cleanup.
+func stubMetadataFetch(t *testing.T, fetch func(ctx context.Context, app
string, instance registry.ServiceInstance, revision, registryId string)
(*info.MetadataInfo, error)) {
+ t.Helper()
+ original := metadataInfoFetcher
+ metadataInfoFetcher = fetch
+ t.Cleanup(func() { metadataInfoFetcher = original })
+}
+
+// stubRetryDelays shrinks the backoff so retries fire within test time
budgets.
+func stubRetryDelays(t *testing.T, initial, max time.Duration) {
+ t.Helper()
+ origInitial, origMax := metadataRetryInitialDelay, metadataRetryMaxDelay
+ metadataRetryInitialDelay, metadataRetryMaxDelay = initial, max
+ t.Cleanup(func() { metadataRetryInitialDelay, metadataRetryMaxDelay =
origInitial, origMax })
+}
+
+// settleRetryListener stops any pending retry by flushing an empty snapshot:
+// with no instances left, every unresolved revision is dropped and the timer
+// is canceled. Register after stub cleanups so it runs before them.
+func settleRetryListener(t *testing.T, listener
*ServiceInstancesChangedListenerImpl) {
+ t.Helper()
+ t.Cleanup(func() {
+ _ =
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{}))
+ })
+}
+
+// TestMetadataRetryRecoversWithoutFurtherEvent is the core regression test for
+// the permanent empty-directory issue: the first metadata fetch fails, no
+// further registry event arrives, and the listener must still recover on its
own.
+func TestMetadataRetryRecoversWithoutFurtherEvent(t *testing.T) {
+ const revision = "rev-retry-recover"
+ const port = 22101
+ stubRetryDelays(t, 5*time.Millisecond, 20*time.Millisecond)
+
+ meta := newTestMetadataInfo(t, revision, port, "")
+ var calls atomic.Int32
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ if calls.Add(1) == 1 {
+ return nil, perrors.New("transient metadata failure")
+ }
+ return meta, nil
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ notify := &retryNotifyListener{}
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
+ t.Cleanup(func() { metaCache.Delete(testApp + ":" + constant.DefaultKey
+ ":" + revision) })
+
+ err :=
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revision),
+ }))
+ require.Error(t, err, "first fetch failed, OnEvent must surface the
unresolved revision")
+ require.Empty(t, notify.snapshot(), "first fetch failed, nothing should
be notified yet")
+
+ require.Eventually(t, func() bool {
+ return len(notify.snapshot()) > 0
+ }, 3*time.Second, 10*time.Millisecond, "retry must rebuild service URLs
without any further registry event")
+}
+
+// TestMetadataRetryStopsWhenInstanceRemoved verifies retries do not resurrect
or
+// keep probing instances that disappeared from the registry snapshot.
+func TestMetadataRetryStopsWhenInstanceRemoved(t *testing.T) {
+ const revision = "rev-retry-stop"
+ const port = 22102
+ stubRetryDelays(t, 5*time.Millisecond, 10*time.Millisecond)
+
+ var calls atomic.Int32
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ calls.Add(1)
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ notify := &retryNotifyListener{}
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
+
+ require.Error(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revision),
+ })), "fetch failed, OnEvent must surface the unresolved revision")
+ require.Eventually(t, func() bool { return calls.Load() >= 2 },
time.Second, 5*time.Millisecond,
+ "retry should have run at least once")
+
+ // The instance leaves the snapshot: retries must stop.
+ require.NoError(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{})))
+ time.Sleep(30 * time.Millisecond)
+ before := calls.Load()
+ time.Sleep(100 * time.Millisecond)
+ assert.Equal(t, before, calls.Load(), "no fetch should happen after the
instance is removed")
+ assert.Nil(t, listener.retryTimer, "retry timer must be canceled")
+}
+
+// TestMetadataRetryFollowsLatestRevision verifies a superseded revision is not
+// retried once the snapshot moved to a newer revision.
+func TestMetadataRetryFollowsLatestRevision(t *testing.T) {
+ const revOld = "rev-retry-old"
+ const revNew = "rev-retry-new"
+ const port = 22103
+ // Initial delay large enough to deliver the second event before any
retry fires.
+ stubRetryDelays(t, 200*time.Millisecond, 400*time.Millisecond)
+
+ metaNew := newTestMetadataInfo(t, revNew, port, "")
+ var oldCalls, newCalls atomic.Int32
+ stubMetadataFetch(t, func(_ context.Context, _ string, _
registry.ServiceInstance, revision string, _ string) (*info.MetadataInfo,
error) {
+ if revision == revOld {
+ oldCalls.Add(1)
+ return nil, perrors.New("transient metadata failure")
+ }
+ newCalls.Add(1)
+ return metaNew, nil
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ notify := &retryNotifyListener{}
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
+ t.Cleanup(func() { metaCache.Delete(testApp + ":" + constant.DefaultKey
+ ":" + revNew) })
+
+ require.Error(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revOld),
+ })), "revOld fetch failed, OnEvent must surface the unresolved
revision")
+ require.NoError(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revNew),
+ })))
+
+ require.Eventually(t, func() bool { return len(notify.snapshot()) > 0
}, time.Second, 10*time.Millisecond)
+ time.Sleep(500 * time.Millisecond)
+ assert.Equal(t, int32(1), oldCalls.Load(), "superseded revision must
not be retried")
+}
+
+// TestMetadataRetryUsesSingleTimer verifies repeated failing events share one
+// retry timer instead of stacking one per event.
+func TestMetadataRetryUsesSingleTimer(t *testing.T) {
+ const revision = "rev-retry-single-timer"
+ const port = 22104
+ stubRetryDelays(t, time.Second, 2*time.Second)
+
+ var calls atomic.Int32
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ calls.Add(1)
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ settleRetryListener(t, listener)
+ notify := &retryNotifyListener{}
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
+
+ for range 3 {
+ require.Error(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revision),
+ })), "fetch failed, OnEvent must surface the unresolved
revision")
+ }
+
+ require.NotNil(t, listener.retryTimer)
+ assert.Equal(t, 1, listener.retryAttempts, "repeated events must share
the pending retry timer")
+ assert.Equal(t, int32(3), calls.Load(), "each event triggers one fetch,
retries come only from the timer")
+}
+
+// TestMetadataRetryListenerDetach verifies removing the last subscriber
cancels
+// the retry timer, and re-attaching a subscriber resumes retries.
+func TestMetadataRetryListenerDetach(t *testing.T) {
+ const revision = "rev-retry-detach"
+ const port = 22105
+ stubRetryDelays(t, time.Second, 2*time.Second)
+
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ settleRetryListener(t, listener)
+ notify := &retryNotifyListener{}
+ key := common.MatchKey(testInterface, constant.TriProtocol)
+ listener.AddListenerAndNotify(key, notify)
+
+ require.Error(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revision),
+ })), "fetch failed, OnEvent must surface the unresolved revision")
+ require.NotNil(t, listener.retryTimer, "retry should be scheduled after
a failed fetch")
+
+ listener.RemoveListener(key)
+ assert.Nil(t, listener.retryTimer, "removing the last subscriber must
cancel the retry timer")
+
+ listener.AddListenerAndNotify(key, notify)
+ assert.NotNil(t, listener.retryTimer, "re-attaching a subscriber must
resume retries")
+}
+
+// TestMetadataRetryWaitsForSubscriber verifies a failed refresh with no
+// attached subscriber does not arm the retry timer: a listener that is never
+// installed (e.g. discarded after losing the subscribe install race) must not
+// keep probing metadata. Attaching a subscriber re-arms the retry.
+func TestMetadataRetryWaitsForSubscriber(t *testing.T) {
+ const revision = "rev-retry-no-subscriber"
+ const port = 22106
+ stubRetryDelays(t, time.Second, 2*time.Second)
+
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ settleRetryListener(t, listener)
+
+ // Snapshot arrives before any subscriber attaches (the SubscribeURL
order).
+ require.Error(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revision),
+ })))
+ assert.Nil(t, listener.retryTimer, "no subscriber attached: retry must
not be armed")
+
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), &retryNotifyListener{})
+ assert.NotNil(t, listener.retryTimer, "attaching a subscriber must arm
the pending retry")
+}
+
+// TestMetadataRetryStopsAfterClose verifies a closed listener cannot re-arm
the
+// retry timer — neither via a direct schedule nor via an in-flight refresh
+// that finishes after the owning registry was destroyed.
+func TestMetadataRetryStopsAfterClose(t *testing.T) {
+ const revision = "rev-retry-closed"
+ const port = 22107
+ stubRetryDelays(t, time.Second, 2*time.Second)
+
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), &retryNotifyListener{})
+
+ require.Error(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
+ newTestServiceInstanceOnly(port, "", revision),
+ })))
+ require.NotNil(t, listener.retryTimer, "retry should be scheduled after
a failed fetch")
+
+ listener.stopMetadataRetry()
+ assert.Nil(t, listener.retryTimer, "close must cancel the pending
retry")
+
+ listener.scheduleMetadataRetry()
+ assert.Nil(t, listener.retryTimer, "closed listener must not re-arm the
retry")
+
+ // Simulate a refresh that was in flight while the registry was
destroyed:
+ // its trailing schedule must not arm a new timer.
+ listener.refreshServiceURLs()
+ assert.Nil(t, listener.retryTimer, "in-flight refresh finishing after
close must not arm a retry")
+}
+
+// TestUnSubscribeStopsMetadataRetry is the regression test for the public
+// SubscribeURL -> UnSubscribe lifecycle: the listener is registered under the
+// protocol-qualified key, so UnSubscribe must remove it with the same key.
+// With an unresolved revision and a pending retry timer, UnSubscribe must
leave
+// the listener subscriber-less and stop the timer from probing metadata.
+func TestUnSubscribeStopsMetadataRetry(t *testing.T) {
+ const revision = "rev-unsubscribe-retry"
+ const port = 22108
+ stubRetryDelays(t, 5*time.Millisecond, 20*time.Millisecond)
+
+ var fetchCalls atomic.Int32
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ fetchCalls.Add(1)
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ setupEnvironment(t)
+ registryURL, _ := common.NewURL(testRegistryURL,
+ common.WithParamsValue(constant.RegistryKey, "mock"))
+ reg, err := newServiceDiscoveryRegistry(registryURL)
+ require.NoError(t, err)
+ sdReg, ok := reg.(*serviceDiscoveryRegistry)
+ require.True(t, ok)
+ sdReg.serviceDiscovery = &destroyRaceDiscovery{
+ instances:
[]registry.ServiceInstance{newTestServiceInstanceOnly(port, "", revision)},
+ }
+
+ consumerURL, err := common.NewURL("dubbo://127.0.0.1:20000/",
+ common.WithInterface(testInterface),
+ common.WithParamsValue(constant.SideKey, constant.SideConsumer),
+ common.WithParamsValue(constant.ProvidedBy, testApp))
+ require.NoError(t, err)
+
+ sdReg.SubscribeURL(consumerURL, &retryNotifyListener{},
gxset.NewSet(testApp))
+
+ listener, ok :=
sdReg.getServiceListener(testApp).(*ServiceInstancesChangedListenerImpl)
+ require.True(t, ok, "SubscribeURL must install the listener")
+ require.Eventually(t, func() bool {
+ listener.mutex.Lock()
+ defer listener.mutex.Unlock()
+ return listener.retryTimer != nil
+ }, 2*time.Second, time.Millisecond, "failed metadata fetch must arm the
retry timer")
+
+ require.NoError(t, sdReg.UnSubscribe(consumerURL,
&retryNotifyListener{}))
+
+ listener.mutex.Lock()
+ remaining := len(listener.listeners)
+ timer := listener.retryTimer
+ listener.mutex.Unlock()
+ assert.Zero(t, remaining, "UnSubscribe must remove the subscriber
registered by SubscribeURL")
+ assert.Nil(t, timer, "removing the last subscriber must cancel the
retry timer")
+
+ // No retry may fire after the last subscriber left.
+ before := fetchCalls.Load()
+ time.Sleep(150 * time.Millisecond)
+ assert.Equal(t, before, fetchCalls.Load(), "no metadata fetch may
happen after UnSubscribe")
+}
diff --git a/registry/servicediscovery/service_discovery_registry.go
b/registry/servicediscovery/service_discovery_registry.go
index 4676a25e7..d09203e8d 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -72,6 +72,11 @@ type serviceDiscoveryRegistry struct {
serviceListeners
map[string]registry.ServiceInstancesChangedListener
serviceMappingListeners map[string]mapping.MappingListener
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.
+ destroyed bool
}
func newServiceDiscoveryRegistry(url *common.URL) (registry.Registry, error) {
@@ -247,7 +252,7 @@ func (s *serviceDiscoveryRegistry) UnSubscribe(url
*common.URL, listener registr
}
serviceNamesKey := sortServices(services)
if l := s.getServiceListener(serviceNamesKey); l != nil {
- l.RemoveListener(url.ServiceKey())
+ l.RemoveListener(protocolSubscribeKey(url))
}
s.stopListen(url)
err := s.serviceNameMapping.Remove(url)
@@ -351,6 +356,16 @@ func (s *serviceDiscoveryRegistry) Destroy() {
s.cancel()
}
s.stopMetadataTimers()
+ s.lock.Lock()
+ s.destroyed = true
+ for _, l := range s.serviceListeners {
+ // Destroy drops listeners without RemoveListener; cancel any
pending
+ // metadata retry so its timer cannot leak.
+ if impl, ok := l.(*ServiceInstancesChangedListenerImpl); ok {
+ impl.stopMetadataRetry()
+ }
+ }
+ s.lock.Unlock()
err := s.serviceDiscovery.Destroy()
if err != nil {
logger.Errorf("[Registry][ServiceDiscovery] destroy
serviceDiscovery catch error, err=%s", err.Error())
@@ -562,11 +577,12 @@ func (s *serviceDiscoveryRegistry) Subscribe(url
*common.URL, notify registry.No
func (s *serviceDiscoveryRegistry) SubscribeURL(url *common.URL, notify
registry.NotifyListener, services *gxset.HashSet) {
serviceNamesKey := sortServices(services)
- protocol := constant.TriProtocol // consume "tri" protocol by default,
other protocols need to be specified on reference/consumer explicitly
- if url.Protocol != "" {
- protocol = url.Protocol
+ protocolServiceKey := protocolSubscribeKey(url)
+
+ // A destroyed registry accepts no new subscriptions.
+ if s.isDestroyed() {
+ return
}
- protocolServiceKey := url.ServiceKey() + ":" + protocol
// Fast path: reuse an already installed listener without touching
external calls.
if listener := s.getServiceListener(serviceNamesKey); listener != nil {
@@ -595,7 +611,23 @@ func (s *serviceDiscoveryRegistry) SubscribeURL(url
*common.URL, notify registry
// Install under a short write lock with a double-check so a concurrent
// subscriber for the same key does not install a duplicate listener.
s.lock.Lock()
+ if s.destroyed {
+ // Destroy ran while the initial GetInstances/metadata phase
above was
+ // in flight: the listener was invisible to it, so it is not
closed.
+ // Discard it here instead of installing into a dead registry.
+ s.lock.Unlock()
+ if impl, ok := listener.(*ServiceInstancesChangedListenerImpl);
ok {
+ impl.stopMetadataRetry()
+ }
+ logger.Warnf("[Registry][ServiceDiscovery] discard late
subscribe for applications=%s: registry already destroyed", serviceNamesKey)
+ return
+ }
if existing := s.serviceListeners[serviceNamesKey]; existing != nil {
+ // The loser of the install race is dropped without
subscribers; close
+ // it so it can never arm a metadata retry or be kept alive by
one.
+ if impl, ok := listener.(*ServiceInstancesChangedListenerImpl);
ok {
+ impl.stopMetadataRetry()
+ }
listener = existing
} else {
s.serviceListeners[serviceNamesKey] = listener
@@ -613,6 +645,13 @@ 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
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires
the
// listener into the service discovery so the caller does not block on it.
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL,
serviceNamesKey, protocolServiceKey string,
@@ -635,6 +674,18 @@ func (s *serviceDiscoveryRegistry) subscribeAndNotify(url
*common.URL, serviceNa
}()
}
+// 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
+// removed and the metadata retry keeps probing after UnSubscribe.
+func protocolSubscribeKey(url *common.URL) string {
+ protocol := constant.TriProtocol // consume "tri" protocol by default,
other protocols need to be specified on reference/consumer explicitly
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
func sortServices(services *gxset.HashSet) string {
list := make([]string, 0, services.Size())
for _, v := range services.Values() {
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl.go
b/registry/servicediscovery/service_instances_changed_listener_impl.go
index 479a0cd11..e535bb3b0 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl.go
@@ -20,6 +20,8 @@ package servicediscovery
import (
"context"
"encoding/gob"
+ "maps"
+ "math/rand/v2"
"reflect"
"sync"
"time"
@@ -69,6 +71,20 @@ type ServiceInstancesChangedListenerImpl struct {
revisionToMetadata map[string]*info.MetadataInfo
allInstances map[string][]registry.ServiceInstance
mutex sync.Mutex
+
+ // buildMu serializes service URL rebuilds so registry events and
metadata
+ // retries cannot interleave. Unlike mutex it may be held across
metadata RPCs.
+ buildMu sync.Mutex
+ // unresolvedRevisions tracks revision keys whose metadata fetch failed
and
+ // must be retried. Guarded by mutex.
+ unresolvedRevisions map[string]struct{}
+ retryTimer *time.Timer
+ retryAttempts int
+ lastFailureLog map[string]time.Time
+ // closed is set when the owning registry drops this listener for good
+ // (Destroy, or a duplicate listener discarded during subscribe). A
closed
+ // listener never arms a new retry timer. Guarded by mutex.
+ closed bool
}
func NewServiceInstancesChangedListener(app string, registryId string,
services *gxset.HashSet) registry.ServiceInstancesChangedListener {
@@ -85,14 +101,16 @@ func NewServiceInstancesChangedListenerWithContext(ctx
context.Context, app stri
initCache(app)
})
return &ServiceInstancesChangedListenerImpl{
- ctx: ctx,
- app: app,
- registryId: registryId,
- serviceNames: services,
- listeners: make(map[string]registry.NotifyListener),
- serviceUrls: make(map[string][]*common.URL),
- revisionToMetadata: make(map[string]*info.MetadataInfo),
- allInstances: make(map[string][]registry.ServiceInstance),
+ ctx: ctx,
+ app: app,
+ registryId: registryId,
+ serviceNames: services,
+ listeners: make(map[string]registry.NotifyListener),
+ serviceUrls: make(map[string][]*common.URL),
+ revisionToMetadata: make(map[string]*info.MetadataInfo),
+ allInstances:
make(map[string][]registry.ServiceInstance),
+ unresolvedRevisions: make(map[string]struct{}),
+ lastFailureLog: make(map[string]time.Time),
}
}
@@ -103,19 +121,43 @@ func (lstn *ServiceInstancesChangedListenerImpl)
OnEvent(e observer.Event) error
return nil
}
- lstn.mutex.Lock()
- defer lstn.mutex.Unlock()
+ logger.Debugf("[Registry][ServiceDiscovery] received instance
notification event, service=%s size=%d", ce.ServiceName, len(ce.Instances))
+ lstn.mutex.Lock()
lstn.allInstances[ce.ServiceName] = ce.Instances
- revisionToInstances := make(map[string][]registry.ServiceInstance,
len(lstn.revisionToMetadata))
- newRevisionToMetadata := make(map[string]*info.MetadataInfo,
len(lstn.revisionToMetadata))
+ lstn.mutex.Unlock()
+
+ if !lstn.refreshServiceURLs() {
+ return perrors.Errorf("metadata unresolved for some revisions
of service=%s, retry is scheduled", ce.ServiceName)
+ }
+ return nil
+}
+
+// refreshServiceURLs rebuilds service URLs from the latest instance snapshot
and
+// notifies subscribers. The build is serialized by buildMu, but lstn.mutex is
+// only held while reading or committing in-memory state: metadata RPCs run in
+// between without it, so a slow or unreachable provider cannot block event
+// processing or retry scheduling. It reports whether every revision resolved;
+// unresolved revisions are retried by the shared retry timer.
+func (lstn *ServiceInstancesChangedListenerImpl) refreshServiceURLs() bool {
+ lstn.buildMu.Lock()
+ defer lstn.buildMu.Unlock()
+
+ lstn.mutex.Lock()
+ allInstances := make(map[string][]registry.ServiceInstance,
len(lstn.allInstances))
+ maps.Copy(allInstances, lstn.allInstances)
+ cachedMetadata := make(map[string]*info.MetadataInfo,
len(lstn.revisionToMetadata))
+ maps.Copy(cachedMetadata, lstn.revisionToMetadata)
+ lstn.mutex.Unlock()
+
+ revisionToInstances := make(map[string][]registry.ServiceInstance,
len(cachedMetadata))
+ newRevisionToMetadata := make(map[string]*info.MetadataInfo,
len(cachedMetadata))
// The same service match key can be exported by several revisions.
// Keep each revision's ServiceInfo so provider-specific params are not
collapsed.
- serviceToRevisionServices :=
make(map[string]map[string]*info.ServiceInfo, len(lstn.serviceUrls))
-
- logger.Infof("[Registry][ServiceDiscovery] received instance
notification event, service=%s size=%d", ce.ServiceName, len(ce.Instances))
+ serviceToRevisionServices :=
make(map[string]map[string]*info.ServiceInfo, len(cachedMetadata))
+ unresolved := make(map[string]struct{})
- for _, instances := range lstn.allInstances {
+ for _, instances := range allInstances {
for _, instance := range instances {
if instance.GetMetadata() == nil {
logger.Warnf("[Registry][ServiceDiscovery]
instance metadata is nil, host=%s", instance.GetHost())
@@ -133,19 +175,19 @@ func (lstn *ServiceInstancesChangedListenerImpl)
OnEvent(e observer.Event) error
providerApp := instance.GetServiceName()
key := metadataCacheKey(providerApp, lstn.registryId,
revision)
- subInstances := revisionToInstances[key]
- if subInstances == nil {
- subInstances = make([]registry.ServiceInstance,
0, 8)
+ revisionToInstances[key] =
append(revisionToInstances[key], instance)
+ metadataInfo := newRevisionToMetadata[key]
+ if metadataInfo == nil {
+ metadataInfo = cachedMetadata[key]
}
- revisionToInstances[key] = append(subInstances,
instance)
- metadataInfo := lstn.revisionToMetadata[key]
if metadataInfo == nil {
- meta, err :=
GetMetadataInfoWithContext(lstn.ctx, providerApp, instance, revision,
lstn.registryId)
+ meta, err := metadataInfoFetcher(lstn.ctx,
providerApp, instance, revision, lstn.registryId)
if err != nil {
// Skip this instance if metadata fetch
fails (e.g., old Java Dubbo version)
- // Try next instance with same revision
-
logger.Warnf("[Registry][ServiceDiscovery] failed to get metadata from instance
%s (revision %s), err=%v, skipping this instance",
- instance.GetHost(), revision,
err)
+ // Try next instance with same
revision. The revision is recorded as
+ // unresolved so it is retried later
instead of being dropped silently.
+ lstn.logMetadataFetchFailure(key,
instance.GetHost(), revision, err)
+ unresolved[key] = struct{}{}
continue
}
metadataInfo = meta
@@ -153,6 +195,7 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
if metadataInfo == nil {
logger.Warnf("[Registry][ServiceDiscovery]
metadata info is nil for instance %s (revision %s), skipping this instance",
instance.GetHost(), revision)
+ unresolved[key] = struct{}{}
continue
}
instance.SetServiceMetadata(metadataInfo)
@@ -167,11 +210,6 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
newRevisionToMetadata[key] = metadataInfo
}
}
- lstn.revisionToMetadata = newRevisionToMetadata
- for key, metadataInfo := range newRevisionToMetadata {
- // key is already provider-app scoped and matches the disk
cache key format.
- metaCache.Set(key, metadataInfo)
- }
newServiceURLs := make(map[string][]*common.URL,
len(serviceToRevisionServices))
for serviceKey, revisionServices := range serviceToRevisionServices {
@@ -186,9 +224,27 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
newServiceURLs[serviceKey] = urls
}
+ lstn.mutex.Lock()
+ lstn.revisionToMetadata = newRevisionToMetadata
lstn.serviceUrls = newServiceURLs
- for key, notifyListener := range lstn.listeners {
- urls := lstn.serviceUrls[key]
+ lstn.unresolvedRevisions = unresolved
+ // Drop throttling state for revisions that resolved or disappeared.
+ for key := range lstn.lastFailureLog {
+ if _, ok := unresolved[key]; !ok {
+ delete(lstn.lastFailureLog, key)
+ }
+ }
+ listeners := make(map[string]registry.NotifyListener,
len(lstn.listeners))
+ maps.Copy(listeners, lstn.listeners)
+ lstn.mutex.Unlock()
+
+ for key, metadataInfo := range newRevisionToMetadata {
+ // key is already provider-app scoped and matches the disk
cache key format.
+ metaCache.Set(key, metadataInfo)
+ }
+
+ for key, notifyListener := range listeners {
+ urls := newServiceURLs[key]
events := make([]*registry.ServiceEvent, 0, len(urls))
for _, url := range urls {
events = append(events, ®istry.ServiceEvent{
@@ -199,7 +255,8 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
notifyListener.NotifyAll(events, func() {})
}
- return nil
+ lstn.scheduleMetadataRetry()
+ return len(unresolved) == 0
}
func toInstanceServiceURLs(instance registry.ServiceInstance, serviceInfo
*info.ServiceInfo) []*common.URL {
@@ -230,8 +287,15 @@ func (lstn *ServiceInstancesChangedListenerImpl)
AddListenerAndNotify(serviceKey
lstn.mutex.Lock()
lstn.listeners[serviceKey] = notify
urls := lstn.serviceUrls[serviceKey]
+ hasUnresolved := len(lstn.unresolvedRevisions) > 0
lstn.mutex.Unlock()
+ if hasUnresolved {
+ // A subscriber (re-)attached while metadata is still
unresolved; make
+ // sure the retry loop is running for it.
+ lstn.scheduleMetadataRetry()
+ }
+
for _, url := range urls {
notify.Notify(®istry.ServiceEvent{
Action: remoting.EventTypeAdd,
@@ -245,6 +309,12 @@ func (lstn *ServiceInstancesChangedListenerImpl)
RemoveListener(serviceKey strin
lstn.mutex.Lock()
defer lstn.mutex.Unlock()
delete(lstn.listeners, serviceKey)
+ if len(lstn.listeners) == 0 && lstn.retryTimer != nil {
+ // No subscriber left: stop retrying so the timer does not keep
the
+ // listener alive after it is dropped.
+ lstn.retryTimer.Stop()
+ lstn.retryTimer = nil
+ }
}
// GetServiceNames return all listener service names
@@ -316,6 +386,106 @@ func GetMetadataInfoWithContext(ctx context.Context, app
string, instance regist
return metadataInfo, nil
}
+var (
+ // metadataRetryInitialDelay is the first backoff delay before retrying
a failed
+ // metadata fetch. Package-level so tests can shrink it.
+ metadataRetryInitialDelay = time.Second
+ // metadataRetryMaxDelay caps the backoff. The retry count itself is
+ // intentionally unlimited: retries only target instances the registry
still
+ // reports as alive, and a capped count would re-introduce the permanent
+ // empty-directory failure this mechanism fixes.
+ metadataRetryMaxDelay = 30 * time.Second
+ // metadataFetchFailureLogInterval throttles repeated fetch-failure
warnings
+ // for the same revision key.
+ 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
+// in-flight refresh can arm a new one afterwards.
+func (lstn *ServiceInstancesChangedListenerImpl) stopMetadataRetry() {
+ lstn.mutex.Lock()
+ defer lstn.mutex.Unlock()
+ lstn.closed = true
+ if lstn.retryTimer != nil {
+ lstn.retryTimer.Stop()
+ lstn.retryTimer = nil
+ }
+}
+
+// scheduleMetadataRetry arms the shared retry timer while unresolved revisions
+// remain. Retries replay the latest instance snapshot, so revisions whose
+// instances disappeared from the registry are dropped naturally on the next
+// run. No timer is armed once the listener is closed or while it has no
+// subscribers: AddListenerAndNotify re-arms the retry when a subscriber
+// attaches, and a subscriber-less listener must not keep probing metadata.
+func (lstn *ServiceInstancesChangedListenerImpl) scheduleMetadataRetry() {
+ lstn.mutex.Lock()
+ defer lstn.mutex.Unlock()
+ if len(lstn.unresolvedRevisions) == 0 || lstn.closed ||
len(lstn.listeners) == 0 {
+ if lstn.retryTimer != nil {
+ lstn.retryTimer.Stop()
+ lstn.retryTimer = nil
+ }
+ lstn.retryAttempts = 0
+ return
+ }
+ if lstn.retryTimer != nil {
+ // One shared timer per listener: repeated events must not
multiply retries.
+ return
+ }
+ delay := metadataRetryDelay(lstn.retryAttempts)
+ lstn.retryAttempts++
+ lstn.retryTimer = time.AfterFunc(delay, func() {
+ lstn.mutex.Lock()
+ lstn.retryTimer = nil
+ // Re-check under the lock: the last subscriber may have been
removed
+ // or the listener closed while the timer was pending.
+ run := !lstn.closed && len(lstn.listeners) > 0 &&
len(lstn.unresolvedRevisions) > 0
+ lstn.mutex.Unlock()
+ if run {
+ lstn.refreshServiceURLs()
+ }
+ })
+}
+
+// metadataRetryDelay returns exponential backoff (initial << attempt) capped
at
+// metadataRetryMaxDelay, plus up to 25% jitter to desynchronize retries across
+// consumers after a correlated provider restart.
+func metadataRetryDelay(attempt int) time.Duration {
+ delay := metadataRetryMaxDelay
+ if attempt >= 0 && attempt < 30 { // guard against shift overflow
+ delay = metadataRetryInitialDelay << attempt
+ if delay <= 0 || delay > metadataRetryMaxDelay {
+ delay = metadataRetryMaxDelay
+ }
+ }
+ return delay + time.Duration(rand.Int64N(int64(delay/4)+1))
+}
+
+// logMetadataFetchFailure logs a throttled warning for a failed metadata
fetch.
+func (lstn *ServiceInstancesChangedListenerImpl) logMetadataFetchFailure(key,
host, revision string, err error) {
+ lstn.mutex.Lock()
+ last, logged := lstn.lastFailureLog[key]
+ now := time.Now()
+ shouldLog := !logged || now.Sub(last) >= metadataFetchFailureLogInterval
+ if shouldLog {
+ lstn.lastFailureLog[key] = now
+ }
+ lstn.mutex.Unlock()
+ if shouldLog {
+ logger.Warnf("[Registry][ServiceDiscovery] failed to get
metadata from instance %s (revision %s), err=%v, skipping this instance",
+ host, revision, err)
+ }
+}
+
func getMetadataStorageType(instance registry.ServiceInstance) string {
instanceMetadata := instance.GetMetadata()
if instanceMetadata == nil {
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl_test.go
b/registry/servicediscovery/service_instances_changed_listener_impl_test.go
index 071627eac..232f6b4dc 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl_test.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl_test.go
@@ -128,7 +128,10 @@ func
TestServiceInstancesChangedListenerRefreshesAndClearsEnvironmentWhenRevisio
}
func TestServiceInstancesChangedListenerSkipsNilMetadataWithoutPanic(t
*testing.T) {
- listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey, gxset.NewSet(testApp))
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey,
gxset.NewSet(testApp)).(*ServiceInstancesChangedListenerImpl)
+ // Nil metadata leaves the revision unresolved and arms the retry timer;
+ // settle it on cleanup so no retry outlives the test.
+ settleRetryListener(t, listener)
notify := &capturingNotifyListener{}
listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
@@ -147,7 +150,7 @@ func
TestServiceInstancesChangedListenerSkipsNilMetadataWithoutPanic(t *testing.
instance,
}))
})
- require.NoError(t, err)
+ require.Error(t, err, "nil metadata leaves the revision unresolved,
OnEvent must surface it")
assert.Empty(t, notify.events)
}
diff --git a/registry/servicediscovery/subscribe_destroy_race_test.go
b/registry/servicediscovery/subscribe_destroy_race_test.go
new file mode 100644
index 000000000..545a84af7
--- /dev/null
+++ b/registry/servicediscovery/subscribe_destroy_race_test.go
@@ -0,0 +1,164 @@
+/*
+ * 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"
+)
+
+// destroyRaceDiscovery is a concurrency-safe ServiceDiscovery stub that counts
+// the calls a late subscribe must not make after Destroy.
+type destroyRaceDiscovery struct {
+ mockServiceDiscovery
+ mu sync.Mutex
+ instances []registry.ServiceInstance
+ getInstancesCalls int
+ addListenerCalls int
+}
+
+func (m *destroyRaceDiscovery) GetInstances(string) []registry.ServiceInstance
{
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.getInstancesCalls++
+ return append([]registry.ServiceInstance(nil), m.instances...)
+}
+
+func (m *destroyRaceDiscovery)
AddListener(registry.ServiceInstancesChangedListener) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.addListenerCalls++
+ return nil
+}
+
+func (m *destroyRaceDiscovery) counts() (getInstances, addListener int) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return m.getInstancesCalls, m.addListenerCalls
+}
+
+func newDestroyRaceRegistry(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),
+ }
+}
+
+// TestSubscribeURLDiscardedWhenDestroyRacesInitialLoad is the deterministic
+// interleaving test for the review finding: Destroy runs while SubscribeURL is
+// still in its initial GetInstances/metadata phase (invisible to Destroy's
+// listener sweep). The late subscribe must be discarded at install time: no
+// listener installed, no AddListener, and no retry timer probing metadata
+// after Destroy.
+func TestSubscribeURLDiscardedWhenDestroyRacesInitialLoad(t *testing.T) {
+ const revision = "rev-destroy-race"
+ const port = 22401
+ stubRetryDelays(t, 5*time.Millisecond, 20*time.Millisecond)
+
+ entered := make(chan struct{})
+ release := make(chan struct{})
+ var enteredOnce sync.Once
+ var fetchCalls atomic.Int32
+ stubMetadataFetch(t, func(context.Context, string,
registry.ServiceInstance, string, string) (*info.MetadataInfo, error) {
+ fetchCalls.Add(1)
+ enteredOnce.Do(func() { close(entered) })
+ <-release
+ return nil, perrors.New("metadata unreachable")
+ })
+
+ sd := &destroyRaceDiscovery{
+ instances:
[]registry.ServiceInstance{newTestServiceInstanceOnly(port, "", revision)},
+ }
+ reg := newDestroyRaceRegistry(sd)
+
+ consumerURL, err := common.NewURL("tri://127.0.0.1:20000/",
+ common.WithInterface(testInterface),
+ common.WithParamsValue(constant.SideKey, constant.SideConsumer))
+ require.NoError(t, err)
+
+ subscribeDone := make(chan struct{})
+ go func() {
+ defer close(subscribeDone)
+ reg.SubscribeURL(consumerURL, &retryNotifyListener{},
gxset.NewSet(testApp))
+ }()
+
+ // SubscribeURL is now blocked inside the metadata fetch of its initial
+ // load phase, invisible to Destroy's listener sweep.
+ select {
+ case <-entered:
+ case <-time.After(2 * time.Second):
+ t.Fatal("SubscribeURL never reached the metadata fetch")
+ }
+ reg.Destroy()
+ close(release)
+ select {
+ case <-subscribeDone:
+ case <-time.After(2 * time.Second):
+ t.Fatal("SubscribeURL did not return after the blocked fetch
was released")
+ }
+
+ assert.Nil(t, reg.getServiceListener(testApp), "late listener must not
be installed after Destroy")
+ _, addListenerCalls := sd.counts()
+ assert.Equal(t, 0, addListenerCalls, "AddListener must not be called
after Destroy")
+
+ // The in-flight fetch may complete, but no retry timer may probe
further.
+ before := fetchCalls.Load()
+ time.Sleep(150 * time.Millisecond)
+ assert.Equal(t, before, fetchCalls.Load(), "no metadata fetch may
happen after Destroy")
+}
+
+// TestSubscribeURLAfterDestroyIsNoop verifies a subscribe issued after Destroy
+// does not touch the service discovery at all.
+func TestSubscribeURLAfterDestroyIsNoop(t *testing.T) {
+ sd := &destroyRaceDiscovery{}
+ reg := newDestroyRaceRegistry(sd)
+ reg.Destroy()
+
+ consumerURL, err := common.NewURL("tri://127.0.0.1:20000/",
+ common.WithInterface(testInterface),
+ common.WithParamsValue(constant.SideKey, constant.SideConsumer))
+ require.NoError(t, err)
+ reg.SubscribeURL(consumerURL, &retryNotifyListener{},
gxset.NewSet(testApp))
+
+ getInstancesCalls, addListenerCalls := sd.counts()
+ assert.Equal(t, 0, getInstancesCalls, "destroyed registry must not
query instances")
+ assert.Equal(t, 0, addListenerCalls, "destroyed registry must not add
listeners")
+ assert.Nil(t, reg.getServiceListener(testApp), "destroyed registry must
not install listeners")
+}
