fix: self-heal application-level discovery after transient metadata failure (#3615)

registry/directory/directory.go

@@ -87,8 +87,12 @@
 	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 @@
 	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) {
@@ -709,12 +714,22 @@
 	return nil
 }
 
 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

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,15 +376,15 @@ 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(&registry.ServiceEvent{Action: remoting.EventTypeAdd, Service: providerURL})
 	time.Sleep(1e9)
 	assert.Empty(t, registryDirectory.snapshotCacheInvokers())
 
 	mockRegistry.MockEvent(&registry.ServiceEvent{Action: remoting.EventTypeDel, Service: providerURL})
 	time.Sleep(1e9)
-	assert.False(t, registryDirectory.hasActiveClosingTombstone(key))
+	assert.False(t, hasActiveClosingTombstone(registryDirectory, key))
 
 	mockRegistry.MockEvent(&registry.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(&registry.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 := &registry.ServiceEvent{Action: remoting.EventTypeAdd, Service: oldURL}
+	newEvent := &registry.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(&registry.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()