fix(versioning): close owner review correctness gaps

Author: mochengqian

pkg/core/discovery/subscriber/zk_config.go

@@ -170,7 +170,7 @@ func processConfigUpsert[T coremodel.Resource](
 			return err
 		}
 		emitter.Send(events.NewResourceChangedEventWithContext(cache.Added, nil, newRuleRes, map[string]string{
-			"source-registry": sourceRegistryZookeeper,
+			events.SourceRegistryContextKey: sourceRegistryZookeeper,
 		}))
 		return nil
 	}
@@ -189,7 +189,7 @@ func processConfigUpsert[T coremodel.Resource](
 	}
 
 	emitter.Send(events.NewResourceChangedEventWithContext(cache.Updated, oldMetadataRes, newRuleRes, map[string]string{
-		"source-registry": sourceRegistryZookeeper,
+		events.SourceRegistryContextKey: sourceRegistryZookeeper,
 	}))
 	return nil
 }
@@ -228,7 +228,7 @@ func processConfigDelete[T coremodel.Resource](
 		return err
 	}
 	emitter.Send(events.NewResourceChangedEventWithContext(cache.Deleted, oldRuleRes, nil, map[string]string{
-		"source-registry": sourceRegistryZookeeper,
+		events.SourceRegistryContextKey: sourceRegistryZookeeper,
 	}))
 	return nil
 }

pkg/core/events/eventbus.go

@@ -26,14 +26,16 @@ import (
 	"github.com/apache/dubbo-admin/pkg/core/resource/model"
 )
 
+const SourceRegistryContextKey = "source-registry"
+
 type Event interface {
 	// Type returns the type of the event, see definitions in cache.DeltaType
 	Type() cache.DeltaType
 	// OldObj returns the old object, nil if old object doesn't exist in store
 	OldObj() model.Resource
 	// NewObj returns the new object, nil if event type is in [cache.Deleted]
 	NewObj() model.Resource
-	// Context returns the context of the event, if event provider want to pass extra info to the consumer, just use context
+	// Context returns read-only event metadata. Subscribers must not mutate the returned map.
 	Context() map[string]string
 	// String returns the string representation of the event
 	String() string

pkg/core/versioning/component.go

@@ -116,7 +116,7 @@ func (c *component) Init(ctx runtime.BuilderContext) error {
 	return nil
 }
 
-func (c *component) Start(rt runtime.Runtime, _ <-chan struct{}) error {
+func (c *component) Start(rt runtime.Runtime, stop <-chan struct{}) error {
 	cfg := rt.Config().Versioning
 	if cfg == nil {
 		cfg = versioningcfg.Default()
@@ -140,6 +140,14 @@ func (c *component) Start(rt runtime.Runtime, _ <-chan struct{}) error {
 			}
 		}
 	}
+	if stop != nil {
+		go func() {
+			<-stop
+			for _, sub := range c.subscribers {
+				sub.FlushAll()
+			}
+		}()
+	}
 	return nil
 }
 

pkg/core/versioning/e2e_rollback_drill_test.go

@@ -60,7 +60,7 @@ func TestE2ERollbackDrill(t *testing.T) {
 
 	upstreamPush := newE2EConditionRoute(3)
 	bus.Send(events.NewResourceChangedEventWithContext(cache.Updated, adminEdit, upstreamPush, map[string]string{
-		"source-registry": "zookeeper",
+		events.SourceRegistryContextKey: "zookeeper",
 	}))
 	items = requireVersions(t, store, original.ResourceKey(), 3)
 	require.Equal(t, SourceUpstream, items[0].Source)

pkg/core/versioning/hint.go

@@ -64,6 +64,7 @@ func (r *AdminHintRegistry) Put(kind coremodel.ResourceKind, resourceKey, conten
 	}
 	r.mu.Lock()
 	defer r.mu.Unlock()
+	r.pruneExpiredLocked()
 	r.hints[hintKey{kind: kind, resourceKey: resourceKey, hash: contentHash}] = hint
 }
 
@@ -73,6 +74,7 @@ func (r *AdminHintRegistry) Take(kind coremodel.ResourceKind, resourceKey, conte
 	}
 	r.mu.Lock()
 	defer r.mu.Unlock()
+	r.pruneExpiredLocked()
 	key := hintKey{kind: kind, resourceKey: resourceKey, hash: contentHash}
 	hint, ok := r.hints[key]
 	if !ok {

pkg/core/versioning/store.go

@@ -67,7 +67,7 @@ func (s *MemoryStore) InsertVersion(req InsertRequest, maxVersions int64) (*Vers
 	}
 	if ids := s.byRule[key]; len(ids) > 0 {
 		latest := s.versions[ids[len(ids)-1]]
-		if latest != nil && latest.ContentHash == req.ContentHash && latest.Operation == req.Operation {
+		if shouldDedupVersion(latest, req) {
 			cp := *latest
 			if meta.CurrentVersion != nil && *meta.CurrentVersion == cp.ID {
 				cp.IsCurrent = true
@@ -110,6 +110,16 @@ func (s *MemoryStore) InsertVersion(req InsertRequest, maxVersions int64) (*Vers
 	return &cp, nil
 }
 
+func shouldDedupVersion(latest *Version, req InsertRequest) bool {
+	if latest == nil || latest.ContentHash != req.ContentHash {
+		return false
+	}
+	if latest.Operation == OperationDelete || req.Operation == OperationDelete {
+		return latest.Operation == req.Operation
+	}
+	return true
+}
+
 func (s *MemoryStore) ListVersions(kind coremodel.ResourceKind, resourceKey string) ([]Version, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

pkg/core/versioning/store_gorm.go

@@ -61,7 +61,7 @@ func (s *GormStore) InsertVersion(req InsertRequest, maxVersions int64) (*Versio
 		err = tx.Where("rule_kind = ? AND resource_key = ?", req.RuleKind, req.ResourceKey).
 			Order("version_no DESC").
 			First(&latest).Error
-		if err == nil && latest.ContentHash == req.ContentHash && latest.Operation == req.Operation {
+		if err == nil && shouldDedupVersion(&latest, req) {
 			inserted = latest
 			return nil
 		}

pkg/core/versioning/subscriber.go

@@ -44,8 +44,6 @@ type pendingEvent struct {
 	timer *time.Timer
 }
 
-const sourceRegistryContextKey = "source-registry"
-
 func NewSubscriber(kind coremodel.ResourceKind, store Store, hints *AdminHintRegistry, maxVersions int64, coalesceWindow time.Duration) *Subscriber {
 	return &Subscriber{
 		kind:           kind,
@@ -160,14 +158,12 @@ func (s *Subscriber) record(event events.Event) error {
 	author := "system:upstream"
 	reason := ""
 	var rolledBackFromID *int64
-	hinted := false
 	if ctx := event.Context(); ctx != nil {
-		if registry := ctx[sourceRegistryContextKey]; registry != "" {
+		if registry := ctx[events.SourceRegistryContextKey]; registry != "" {
 			author = "system:" + registry
 		}
 	}
 	if hint, ok := s.hints.Take(res.ResourceKind(), res.ResourceKey(), hash); ok {
-		hinted = true
 		source = hint.Source
 		if source == "" {
 			source = SourceAdmin
@@ -196,15 +192,6 @@ func (s *Subscriber) record(event events.Event) error {
 	if author == "" {
 		author = "system:unknown"
 	}
-	if !hinted && source == SourceUpstream && op != OperationDelete {
-		latest, err := s.store.LatestVersion(ruleKind, resourceKey)
-		if err != nil {
-			return err
-		}
-		if latest != nil && latest.ContentHash == hash {
-			return nil
-		}
-	}
 	_, err = s.store.InsertVersion(InsertRequest{
 		RuleKind:         ruleKind,
 		Mesh:             mesh,

pkg/core/versioning/versioning_test.go

@@ -29,7 +29,10 @@ import (
 	meshproto "github.com/apache/dubbo-admin/api/mesh/v1alpha1"
 	appconfig "github.com/apache/dubbo-admin/pkg/config/app"
 	eventbusconfig "github.com/apache/dubbo-admin/pkg/config/eventbus"
+	"github.com/apache/dubbo-admin/pkg/config/mode"
+	versioningcfg "github.com/apache/dubbo-admin/pkg/config/versioning"
 	"github.com/apache/dubbo-admin/pkg/core/events"
+	"github.com/apache/dubbo-admin/pkg/core/manager"
 	meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1"
 	"github.com/apache/dubbo-admin/pkg/core/resource/model"
 	coreruntime "github.com/apache/dubbo-admin/pkg/core/runtime"
@@ -74,6 +77,26 @@ func TestAdminHintTTLHitAndMiss(t *testing.T) {
 	require.False(t, ok)
 }
 
+func TestAdminHintPrunesExpiredOnPutAndTake(t *testing.T) {
+	now := time.Now()
+	reg := NewAdminHintRegistry()
+	reg.now = func() time.Time { return now }
+
+	reg.Put(meshresource.ConditionRouteKind, "m/stale", "stale", AdminHint{ExpiresAt: now.Add(-time.Second)})
+	require.Len(t, reg.hints, 1)
+
+	reg.Put(meshresource.ConditionRouteKind, "m/fresh", "fresh", AdminHint{
+		ExpiresAt: now.Add(time.Second),
+	})
+	require.Len(t, reg.hints, 1)
+
+	reg.Put(meshresource.ConditionRouteKind, "m/stale", "stale", AdminHint{ExpiresAt: now.Add(-time.Second)})
+	hint, ok := reg.Take(meshresource.ConditionRouteKind, "m/fresh", "fresh")
+	require.True(t, ok)
+	require.Empty(t, hint.Author)
+	require.Empty(t, reg.hints)
+}
+
 func TestMemoryStoreRetentionCurrentPointerAndDelete(t *testing.T) {
 	store := NewMemoryStore()
 	key := "mesh/demo.condition-router"
@@ -219,7 +242,7 @@ func TestSubscriberRespectsRegistrySourceContext(t *testing.T) {
 		cache.Updated,
 		nil,
 		upstreamRes,
-		map[string]string{"source-registry": "zookeeper"},
+		map[string]string{events.SourceRegistryContextKey: "zookeeper"},
 	)))
 
 	items, err := store.ListVersions(meshresource.ConditionRouteKind, upstreamRes.ResourceKey())
@@ -241,7 +264,7 @@ func TestSubscriberSkipsNoopUpstreamEchoAfterBootstrap(t *testing.T) {
 		cache.Updated,
 		nil,
 		original,
-		map[string]string{"source-registry": "zookeeper"},
+		map[string]string{events.SourceRegistryContextKey: "zookeeper"},
 	)))
 
 	items, err := store.ListVersions(meshresource.ConditionRouteKind, original.ResourceKey())
@@ -256,7 +279,7 @@ func TestSubscriberSkipsNoopUpstreamEchoAfterBootstrap(t *testing.T) {
 		cache.Updated,
 		original,
 		changed,
-		map[string]string{"source-registry": "zookeeper"},
+		map[string]string{events.SourceRegistryContextKey: "zookeeper"},
 	)))
 
 	items, err = store.ListVersions(meshresource.ConditionRouteKind, original.ResourceKey())
@@ -267,6 +290,26 @@ func TestSubscriberSkipsNoopUpstreamEchoAfterBootstrap(t *testing.T) {
 	require.True(t, items[0].IsCurrent)
 }
 
+func TestSubscriberRecordsEmptyCreateAfterDelete(t *testing.T) {
+	store := NewMemoryStore()
+	hints := NewAdminHintRegistry()
+	sub := NewSubscriber(meshresource.ConditionRouteKind, store, hints, 5, 0)
+	res := meshresource.NewConditionRouteResourceWithAttributes("demo.condition-router", "mesh")
+	res.Spec = &meshproto.ConditionRoute{}
+
+	require.NoError(t, sub.ProcessEvent(events.NewResourceChangedEvent(cache.Added, nil, res)))
+	require.NoError(t, sub.ProcessEvent(events.NewResourceChangedEvent(cache.Deleted, res, nil)))
+	require.NoError(t, sub.ProcessEvent(events.NewResourceChangedEvent(cache.Added, nil, res)))
+
+	items, err := store.ListVersions(meshresource.ConditionRouteKind, res.ResourceKey())
+	require.NoError(t, err)
+	require.Len(t, items, 3)
+	require.Equal(t, OperationCreate, items[0].Operation)
+	require.True(t, items[0].IsCurrent)
+	require.Equal(t, OperationDelete, items[1].Operation)
+	require.False(t, items[1].IsCurrent)
+}
+
 func TestSubscriberRecordsDeleteWithAdminHintSnapshot(t *testing.T) {
 	store := NewMemoryStore()
 	hints := NewAdminHintRegistry()
@@ -410,6 +453,37 @@ func TestDisabledServiceHistoryReturnsFeatureDisabled(t *testing.T) {
 	require.ErrorIs(t, err, ErrFeatureDisabled)
 }
 
+func TestComponentFlushesPendingVersionsOnStop(t *testing.T) {
+	store := NewMemoryStore()
+	sub := NewSubscriber(meshresource.ConditionRouteKind, store, NewAdminHintRegistry(), 5, time.Hour)
+	comp := &component{
+		store:       store,
+		subscribers: []*Subscriber{sub},
+	}
+	res := meshresource.NewConditionRouteResourceWithAttributes("demo.condition-router", "mesh")
+	res.Spec = &meshproto.ConditionRoute{Key: "demo", Priority: 1}
+	require.NoError(t, sub.ProcessEvent(events.NewResourceChangedEvent(cache.Added, nil, res)))
+
+	stop := make(chan struct{})
+	require.NoError(t, comp.Start(testRuntime{
+		cfg: appconfig.AdminConfig{
+			Versioning: &versioningcfg.Config{
+				Enabled:            true,
+				MaxVersionsPerRule: 5,
+			},
+		},
+		components: map[coreruntime.ComponentType]coreruntime.Component{
+			coreruntime.ResourceManager: testRMComponent{rm: fakeNoopResourceManager{}},
+		},
+	}, stop))
+	close(stop)
+
+	require.Eventually(t, func() bool {
+		items, err := store.ListVersions(meshresource.ConditionRouteKind, res.ResourceKey())
+		return err == nil && len(items) == 1
+	}, time.Second, 10*time.Millisecond)
+}
+
 type fakeVersionResourceManager struct {
 	subscriber *Subscriber
 }
@@ -564,3 +638,70 @@ func (c testBuilderContext) GetActivatedComponent(coreruntime.ComponentType) (co
 func (c testBuilderContext) ActivateComponent(coreruntime.Component) error {
 	return nil
 }
+
+type testRuntime struct {
+	cfg        appconfig.AdminConfig
+	components map[coreruntime.ComponentType]coreruntime.Component
+}
+
+func (r testRuntime) GetInstanceId() string {
+	return "test-instance"
+}
+
+func (r testRuntime) GetClusterId() string {
+	return "test-cluster"
+}
+
+func (r testRuntime) GetStartTime() time.Time {
+	return time.Now()
+}
+
+func (r testRuntime) GetMode() mode.Mode {
+	return mode.Test
+}
+
+func (r testRuntime) Config() appconfig.AdminConfig {
+	return r.cfg
+}
+
+func (r testRuntime) GetComponent(typ coreruntime.ComponentType) (coreruntime.Component, error) {
+	return r.components[typ], nil
+}
+
+func (r testRuntime) AppContext() context.Context {
+	return context.Background()
+}
+
+func (r testRuntime) Add(...coreruntime.Component) {}
+
+func (r testRuntime) Start(<-chan struct{}) error {
+	return nil
+}
+
+type testRMComponent struct {
+	rm manager.ResourceManager
+}
+
+func (c testRMComponent) Type() coreruntime.ComponentType {
+	return coreruntime.ResourceManager
+}
+
+func (c testRMComponent) Order() int {
+	return 0
+}
+
+func (c testRMComponent) RequiredDependencies() []coreruntime.ComponentType {
+	return nil
+}
+
+func (c testRMComponent) Init(coreruntime.BuilderContext) error {
+	return nil
+}
+
+func (c testRMComponent) Start(coreruntime.Runtime, <-chan struct{}) error {
+	return nil
+}
+
+func (c testRMComponent) ResourceManager() manager.ResourceManager {
+	return c.rm
+}