AlexStocks commented on code in PR #3482: URL: https://github.com/apache/dubbo-go/pull/3482#discussion_r3651197564
########## registry/polaris/registry_test.go: ########## @@ -0,0 +1,1985 @@ +/* + * 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 polaris + +import ( + "context" + "errors" + "os" + "os/exec" + "reflect" + "strconv" + "strings" + "testing" + "time" +) + +import ( + "github.com/golang/protobuf/ptypes/wrappers" + + api "github.com/polarismesh/polaris-go" + "github.com/polarismesh/polaris-go/pkg/model" + "github.com/polarismesh/polaris-go/pkg/model/local" + "github.com/polarismesh/polaris-go/pkg/model/pb" + namingpb "github.com/polarismesh/polaris-go/pkg/model/pb/v1" +) + +import ( + "dubbo.apache.org/dubbo-go/v3/common" + "dubbo.apache.org/dubbo-go/v3/common/constant" + "dubbo.apache.org/dubbo-go/v3/registry" + "dubbo.apache.org/dubbo-go/v3/remoting" +) + +const ( + testPolarisNamespace = "test" + testPolarisRevisionKey = "revision" +) + +type fakePolarisConsumer struct { + api.ConsumerAPI + instanceResponses [][]model.Instance + getCalls int + watchInstances []model.Instance + watchEvents []model.SubScribeEvent + watchErr error + watchCalls int + watchRequest *api.WatchServiceRequest +} + +type gatedPolarisWatchResult struct { + response *model.WatchServiceResponse + err error +} + +type gatedPolarisConsumer struct { + *fakePolarisConsumer + watchRequests chan *api.WatchServiceRequest + watchResults chan gatedPolarisWatchResult +} + +func newGatedPolarisConsumer(instanceResponses ...[]model.Instance) *gatedPolarisConsumer { + return &gatedPolarisConsumer{ + fakePolarisConsumer: &fakePolarisConsumer{instanceResponses: instanceResponses}, + watchRequests: make(chan *api.WatchServiceRequest), + watchResults: make(chan gatedPolarisWatchResult), + } +} + +func (f *gatedPolarisConsumer) WatchService(request *api.WatchServiceRequest) (*model.WatchServiceResponse, error) { + f.watchRequests <- request + result := <-f.watchResults + return result.response, result.err +} + +func (f *fakePolarisConsumer) GetInstances(_ *api.GetInstancesRequest) (*model.InstancesResponse, error) { + var instances []model.Instance + if f.getCalls < len(f.instanceResponses) { + instances = f.instanceResponses[f.getCalls] + } + f.getCalls++ + return &model.InstancesResponse{Instances: copyInstances(instances)}, nil +} + +func (f *fakePolarisConsumer) WatchService(request *api.WatchServiceRequest) (*model.WatchServiceResponse, error) { + f.watchCalls++ + f.watchRequest = request + if f.watchErr != nil { + return nil, f.watchErr + } + events := make(chan model.SubScribeEvent, len(f.watchEvents)) + for _, event := range f.watchEvents { + events <- event + } + close(events) + return &model.WatchServiceResponse{ + EventChannel: events, + GetAllInstancesResp: &model.InstancesResponse{ + Instances: copyInstances(f.watchInstances), + }, + }, nil +} + +type recordedPolarisEvent struct { + action remoting.EventType + host string + port string + id string + revision string +} + +type recordedNotification struct { + kind polarisNotificationKind + events []recordedPolarisEvent + callbackNil bool +} + +type recordingPolarisNotifyListener struct { + notifications []recordedNotification + onNotify func(*registry.ServiceEvent) +} + +type nonComparablePolarisNotifyListener []*recordingPolarisNotifyListener + +type nonComparableMapPolarisNotifyListener map[string]*recordingPolarisNotifyListener + +type nonComparableFuncPolarisNotifyListener func(*registry.ServiceEvent, []*registry.ServiceEvent, func()) + +type nonComparableStructPolarisNotifyListener struct { + recorder *recordingPolarisNotifyListener + _ []recordedPolarisEvent +} + +type runtimeUnhashablePolarisNotifyListener struct { + state any + recorder *recordingPolarisNotifyListener +} + +type comparableValuePolarisNotifyListener struct { + id int + recorder *recordingPolarisNotifyListener +} + +type nilablePolarisNotifyListener struct{} + +type nilableChannelPolarisNotifyListener chan struct{} + +type stopSubscribePump struct{} + +func (l nonComparablePolarisNotifyListener) Notify(event *registry.ServiceEvent) { + if len(l) > 0 && l[0] != nil { + l[0].Notify(event) + } +} + +func (l nonComparablePolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + if len(l) > 0 && l[0] != nil { + l[0].NotifyAll(events, callback) + } +} + +func (l nonComparableMapPolarisNotifyListener) Notify(event *registry.ServiceEvent) { + if recorder := l["recorder"]; recorder != nil { + recorder.Notify(event) + } +} + +func (l nonComparableMapPolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + if recorder := l["recorder"]; recorder != nil { + recorder.NotifyAll(events, callback) + } +} + +func (l nonComparableFuncPolarisNotifyListener) Notify(event *registry.ServiceEvent) { + if l != nil { + l(event, nil, nil) + } +} + +func (l nonComparableFuncPolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + if l != nil { + l(nil, events, callback) + } +} + +func (l nonComparableStructPolarisNotifyListener) Notify(event *registry.ServiceEvent) { + if l.recorder != nil { + l.recorder.Notify(event) + } +} + +func (l nonComparableStructPolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + if l.recorder != nil { + l.recorder.NotifyAll(events, callback) + } +} + +func (l runtimeUnhashablePolarisNotifyListener) Notify(event *registry.ServiceEvent) { + if l.recorder != nil { + l.recorder.Notify(event) + } +} + +func (l runtimeUnhashablePolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + if l.recorder != nil { + l.recorder.NotifyAll(events, callback) + } +} + +func (l comparableValuePolarisNotifyListener) Notify(event *registry.ServiceEvent) { + l.recorder.Notify(event) +} + +func (l comparableValuePolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + l.recorder.NotifyAll(events, callback) +} + +func (*nilablePolarisNotifyListener) Notify(*registry.ServiceEvent) {} + +func (*nilablePolarisNotifyListener) NotifyAll([]*registry.ServiceEvent, func()) {} + +func (nilableChannelPolarisNotifyListener) Notify(*registry.ServiceEvent) {} + +func (nilableChannelPolarisNotifyListener) NotifyAll([]*registry.ServiceEvent, func()) {} + +func (r *recordingPolarisNotifyListener) Notify(event *registry.ServiceEvent) { + r.notifications = append(r.notifications, recordedNotification{ + kind: incrementalNotification, + events: []recordedPolarisEvent{recordedPolarisEventFromServiceEvent(event)}, + }) + if r.onNotify != nil { + r.onNotify(event) + } +} + +func (r *recordingPolarisNotifyListener) NotifyAll(events []*registry.ServiceEvent, callback func()) { + r.notifications = append(r.notifications, recordedNotification{ + kind: fullSnapshotNotification, + events: recordedPolarisEventsFromServiceEvents(events), + callbackNil: callback == nil, + }) + if callback != nil { + callback() + } +} + +func (r *recordingPolarisNotifyListener) recordInstances(action remoting.EventType, instances []model.Instance) { + if len(instances) == 0 { + return + } + events := make([]recordedPolarisEvent, 0, len(instances)) + for _, instance := range instances { + events = append(events, recordedPolarisEventFromInstance(action, instance)) + } + r.notifications = append(r.notifications, recordedNotification{kind: incrementalNotification, events: events}) +} + +func (r *recordingPolarisNotifyListener) recordedEvents() []recordedPolarisEvent { + var events []recordedPolarisEvent + for _, notification := range r.notifications { + events = append(events, notification.events...) + } + return events +} + +func TestPolarisInitialSnapshotReconciliation(t *testing.T) { + serviceName := "com.test.InitialSnapshotService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + sameKeyA := newPolarisTestInstance("new-polaris-id", "10.0.0.1", 20001, serviceName, true) + instanceA.GetMetadata()[testPolarisRevisionKey] = "before" + sameKeyA.GetMetadata()[testPolarisRevisionKey] = "after" + invalid := newPolarisTestInstance("invalid", "10.0.0.9", 20009, serviceName, false) + consumer := &fakePolarisConsumer{ + instanceResponses: [][]model.Instance{{instanceA, invalid}}, + watchInstances: []model.Instance{sameKeyA}, + } + pr := newTestPolarisRegistry(consumer) + notify := &recordingPolarisNotifyListener{} + if err := pr.LoadSubscribeInstances(newPolarisConsumerURL(serviceName), notify); err != nil { + t.Fatalf("LoadSubscribeInstances() error = %v", err) + } + watcher := mustStoppedRegistryWatcher(t, pr, serviceName) + listener, err := pr.createPolarisListener(serviceName, notify) + if err != nil { + t.Fatalf("createPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + if err := watcher.watchOnce(); err != nil { + t.Fatalf("watchOnce() error = %v", err) + } + + assertPolarisEventsAddress(t, notify.recordedEvents(), []recordedPolarisEvent{{ + action: remoting.EventTypeAdd, host: "10.0.0.1", port: "20001", + }}) + events := assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{{ + action: remoting.EventTypeAdd, host: "10.0.0.1", port: "20001", + }}) + assertPolarisEventInstance(t, events[0], remoting.EventTypeAdd, sameKeyA) + assertPolarisWatchRequest(t, consumer.watchRequest, watcher.subscribeParam, serviceName) + + t.Run("WatchService error is returned", func(t *testing.T) { + wantErr := errors.New("watch failed") + watcher := newStoppedPolarisWatcher(t, &fakePolarisConsumer{watchErr: wantErr}) + if err := watcher.watchOnce(); !errors.Is(err, wantErr) { + t.Fatalf("watchOnce() error = %v, want %v", err, wantErr) + } + if watcher.snapshotReady { + t.Fatal("watcher snapshot is ready after WatchService error") + } + }) +} + +func TestPolarisWatchServiceReconnectLifecycle(t *testing.T) { + serviceName := "com.test.ReconnectSnapshotService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + instanceB := newPolarisTestInstance("instance-b", "10.0.0.2", 20002, serviceName, true) + consumer := newGatedPolarisConsumer() + pr := newTestPolarisRegistry(consumer) + watcher := mustStoppedRegistryWatcher(t, pr, serviceName) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + firstRequest := runGatedPolarisWatchOnce(t, watcher, consumer, []model.Instance{instanceA, instanceB}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeAdd, host: "10.0.0.1", port: "20001"}, + {action: remoting.EventTypeAdd, host: "10.0.0.2", port: "20002"}, + }) + secondRequest := runGatedPolarisWatchOnce(t, watcher, consumer, []model.Instance{instanceB}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + {action: remoting.EventTypeAdd, host: "10.0.0.2", port: "20002"}, + }) + assertPolarisWatchRequest(t, firstRequest, watcher.subscribeParam, serviceName) + assertPolarisWatchRequest(t, secondRequest, watcher.subscribeParam, serviceName) + + t.Run("single instance to empty snapshot", func(t *testing.T) { + pr := newTestPolarisRegistry(nil) + watcher := mustStoppedRegistryWatcher(t, pr, serviceName) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + watcher.handleWatchSnapshot([]model.Instance{instanceA}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{{ + action: remoting.EventTypeAdd, host: "10.0.0.1", port: "20001", + }}) + + watcher.handleWatchSnapshot(nil) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{{ + action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001", + }}) + if len(watcher.currentInstances) != 0 { + t.Fatalf("current instance count = %d, want 0", len(watcher.currentInstances)) + } + }) +} + +func TestPolarisWatchSnapshotIsolatedFromInputMutation(t *testing.T) { + serviceName := "com.test.InputMutationSnapshotService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + wantKey := instanceA.GetInstanceKey() + wantHost := instanceA.GetHost() + wantPort := instanceA.GetPort() + wantInterface := instanceA.GetMetadata()["interface"] + wantPath := instanceA.GetMetadata()["path"] + watcher := newStoppedPolarisWatcher(t, nil) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + watcher.handleWatchSnapshot([]model.Instance{instanceA}) + assertPolarisListenerEvent(t, listener, remoting.EventTypeAdd, instanceA) + if len(watcher.currentInstances) != 1 { + t.Fatalf("current instance count = %d, want 1", len(watcher.currentInstances)) + } + current := watcher.currentInstances[0] + if current == instanceA { + t.Fatal("watcher current instance shares the input instance") + } + + delete(instanceA.GetMetadata(), "interface") + delete(instanceA.GetMetadata(), "path") + if current.GetMetadata()["interface"] != wantInterface || current.GetMetadata()["path"] != wantPath { + t.Fatalf("watcher metadata = %v, want interface=%q path=%q", current.GetMetadata(), wantInterface, wantPath) + } + + watcher.handleWatchSnapshot(nil) + select { + case value := <-listener.events.Out(): + notification, ok := value.(*polarisNotification) + if !ok || notification == nil { + t.Fatalf("listener notification type = %T, want *polarisNotification", value) + } + if notification.eventType != remoting.EventTypeDel || len(notification.instances) != 1 { + t.Fatalf("listener notification = %#v, want one DEL instance", notification) + } + deleted := notification.instances[0] + if deleted.GetInstanceKey() != wantKey || deleted.GetHost() != wantHost || deleted.GetPort() != wantPort { + t.Fatalf( + "deleted instance = (key=%v host=%s port=%d), want (key=%v host=%s port=%d)", + deleted.GetInstanceKey(), + deleted.GetHost(), + deleted.GetPort(), + wantKey, + wantHost, + wantPort, + ) + } + if deleted.GetMetadata()["interface"] != wantInterface || deleted.GetMetadata()["path"] != wantPath { + t.Fatalf("deleted instance metadata = %v, want interface=%q path=%q", deleted.GetMetadata(), wantInterface, wantPath) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for DEL after input mutation") + } + assertNoPolarisListenerEvent(t, listener) + if len(watcher.currentInstances) != 0 { + t.Fatalf("current instance count = %d, want 0", len(watcher.currentInstances)) + } +} + +func TestPolarisSubscriberMutationDoesNotAffectWatcherSnapshot(t *testing.T) { + serviceName := "com.test.SubscriberMutationSnapshotService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + wantInterface := instanceA.GetMetadata()["interface"] + wantPath := instanceA.GetMetadata()["path"] + watcher := newStoppedPolarisWatcher(t, nil) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + watcher.AddSubscriber(func(_ remoting.EventType, instances []model.Instance) { + if len(instances) == 0 { + return + } + delete(instances[0].GetMetadata(), "interface") + delete(instances[0].GetMetadata(), "path") + }) + var secondInterface string + var secondPath string + watcher.AddSubscriber(func(_ remoting.EventType, instances []model.Instance) { + if len(instances) == 0 { + return + } + secondInterface = instances[0].GetMetadata()["interface"] + secondPath = instances[0].GetMetadata()["path"] + }) + + watcher.handleWatchSnapshot([]model.Instance{instanceA}) + assertPolarisListenerEvent(t, listener, remoting.EventTypeAdd, instanceA) + if secondInterface != wantInterface || secondPath != wantPath { + t.Fatalf("second subscriber metadata = (interface=%q path=%q), want (interface=%q path=%q)", secondInterface, secondPath, wantInterface, wantPath) + } + if len(watcher.currentInstances) != 1 { + t.Fatalf("current instance count = %d, want 1", len(watcher.currentInstances)) + } + currentMetadata := watcher.currentInstances[0].GetMetadata() + if currentMetadata["interface"] != wantInterface || currentMetadata["path"] != wantPath { + t.Fatalf("watcher metadata = %v, want interface=%q path=%q", currentMetadata, wantInterface, wantPath) + } + + watcher.handleWatchSnapshot(nil) + assertPolarisListenerEvent(t, listener, remoting.EventTypeDel, instanceA) +} + +func TestPolarisComparableSubscriberDeletesInstanceThatBecomesInvalid(t *testing.T) { + serviceName := "com.test.ValidToInvalidSnapshotService" + validA := newPolarisTestInstance("valid-a", "10.0.0.1", 20001, serviceName, true) + invalidA := newPolarisTestInstance("invalid-a", "10.0.0.1", 20001, serviceName, false) + + t.Run("first reconciliation compares baseline with notifiable current set", func(t *testing.T) { + pr := newTestPolarisRegistry(&fakePolarisConsumer{ + instanceResponses: [][]model.Instance{{validA}}, + }) + notify := &recordingPolarisNotifyListener{} + if err := pr.LoadSubscribeInstances(newPolarisConsumerURL(serviceName), notify); err != nil { + t.Fatalf("LoadSubscribeInstances() error = %v", err) + } + watcher := mustStoppedRegistryWatcher(t, pr, serviceName) + listener, err := pr.createPolarisListener(serviceName, notify) + if err != nil { + t.Fatalf("createPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + watcher.handleWatchSnapshot([]model.Instance{invalidA}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + }) + }) + + t.Run("reconnect deletes previously valid instance before skipping invalid current", func(t *testing.T) { + watcher := newStoppedPolarisWatcher(t, nil) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + watcher.handleWatchSnapshot([]model.Instance{validA}) + assertPolarisListenerEvent(t, listener, remoting.EventTypeAdd, validA) + + watcher.handleWatchSnapshot([]model.Instance{invalidA}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + }) + }) +} + +func TestPolarisRegistrySubscriberUpdateNotificationMatrix(t *testing.T) { + serviceName := "com.test.UpdateNotificationMatrixService" + validA := newPolarisTestInstance("valid-a", "10.0.0.1", 20001, serviceName, true) + updatedA := newPolarisTestInstance("updated-a", "10.0.0.1", 20001, serviceName, true) + validB := newPolarisTestInstance("valid-b", "10.0.0.2", 20002, serviceName, true) + invalidA := newPolarisTestInstance("invalid-a", "10.0.0.1", 20001, serviceName, false) + invalidB := newPolarisTestInstance("invalid-b", "10.0.0.2", 20002, serviceName, false) + + tests := []struct { + name string + before model.Instance + after model.Instance + wantRegistry []recordedPolarisEvent + verifyReconnect bool + verifyNoRepeat bool + }{ + { + name: "same key valid to valid updates after", + before: validA, + after: updatedA, + wantRegistry: []recordedPolarisEvent{ + {action: remoting.EventTypeUpdate, host: "10.0.0.1", port: "20001"}, + }, + }, + { + name: "changed key valid to valid deletes before updating", + before: validA, + after: validB, + wantRegistry: []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + {action: remoting.EventTypeUpdate, host: "10.0.0.2", port: "20002"}, + }, + verifyReconnect: true, + }, + { + name: "same key valid to invalid deletes before", + before: validA, + after: invalidA, + wantRegistry: []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + }, + verifyNoRepeat: true, + }, + { + name: "changed key valid to invalid deletes before", + before: validA, + after: invalidB, + wantRegistry: []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + }, + }, + { + name: "invalid to valid only updates after", + before: invalidA, + after: validB, + wantRegistry: []recordedPolarisEvent{ + {action: remoting.EventTypeUpdate, host: "10.0.0.2", port: "20002"}, + }, + }, + { + name: "invalid to invalid has no registry notification", + before: invalidA, + after: invalidB, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + watcher := newStoppedPolarisWatcher(t, nil) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + watcher.handleWatchSnapshot([]model.Instance{tt.before}) + if polarisInstanceURLValidationError(tt.before) == "" { + assertPolarisListenerEvent(t, listener, remoting.EventTypeAdd, tt.before) + } else { + assertNoPolarisListenerEvent(t, listener) + } + application := &recordingPolarisNotifyListener{} + watcher.AddSubscriber(application.recordInstances) + + watcher.handleInstanceEvent(&model.InstanceEvent{ + UpdateEvent: &model.InstanceUpdateEvent{UpdateList: []model.OneInstanceUpdate{{ + Before: tt.before, + After: tt.after, + }}}, + }) + + var registryEvents []recordedPolarisEvent + if len(tt.wantRegistry) == 0 { + assertNoPolarisListenerEvent(t, listener) + } else { + registryEvents = assertPolarisIncrementalListenerEvents(t, listener, tt.wantRegistry) + } + for _, event := range registryEvents { + if event.action == remoting.EventTypeUpdate { + assertPolarisEventInstance(t, event, remoting.EventTypeUpdate, tt.after) + } + } + if len(application.recordedEvents()) != 1 { + t.Fatalf("application event count = %d, want 1", len(application.recordedEvents())) + } + assertPolarisEventInstance(t, application.recordedEvents()[0], remoting.EventTypeUpdate, tt.after) + if len(watcher.currentInstances) != 1 { + t.Fatalf("current instance count = %d, want 1", len(watcher.currentInstances)) + } + assertPolarisInstanceIdentity(t, watcher.currentInstances[0], tt.after) + + if tt.verifyReconnect { + watcher.handleWatchSnapshot([]model.Instance{tt.after}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{{ + action: remoting.EventTypeAdd, + host: tt.after.GetHost(), + port: strconv.Itoa(int(tt.after.GetPort())), + }}) + } + if tt.verifyNoRepeat { + watcher.handleWatchSnapshot([]model.Instance{tt.after}) + assertNoPolarisListenerEvent(t, listener) + } + }) + } +} + +func TestPolarisChangedKeyUpdateBatchPreservesEveryAfterInstance(t *testing.T) { + serviceName := "com.test.ChangedKeyUpdateBatchService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + instanceB := newPolarisTestInstance("instance-b", "10.0.0.2", 20002, serviceName, true) + movedA := newPolarisTestInstance("moved-a", "10.0.0.2", 20002, serviceName, true) + movedB := newPolarisTestInstance("moved-b", "10.0.0.3", 20003, serviceName, true) + watcher := newStoppedPolarisWatcher(t, nil) + listener, err := newPolarisListener(watcher, nil, reconcileWithBaseline) + if err != nil { + t.Fatalf("newPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + + watcher.handleWatchSnapshot([]model.Instance{instanceA, instanceB}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeAdd, host: "10.0.0.1", port: "20001"}, + {action: remoting.EventTypeAdd, host: "10.0.0.2", port: "20002"}, + }) + application := &recordingPolarisNotifyListener{} + watcher.AddSubscriber(application.recordInstances) + + watcher.handleInstanceEvent(&model.InstanceEvent{ + UpdateEvent: &model.InstanceUpdateEvent{UpdateList: []model.OneInstanceUpdate{ + {Before: instanceA, After: movedA}, + {Before: instanceB, After: movedB}, + }}, + }) + + registryEvents := assertPolarisIncrementalListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + {action: remoting.EventTypeDel, host: "10.0.0.2", port: "20002"}, + {action: remoting.EventTypeUpdate, host: "10.0.0.2", port: "20002"}, + {action: remoting.EventTypeUpdate, host: "10.0.0.3", port: "20003"}, + }) + assertPolarisEventInstance(t, registryEvents[2], remoting.EventTypeUpdate, movedA) + assertPolarisEventInstance(t, registryEvents[3], remoting.EventTypeUpdate, movedB) + if len(application.recordedEvents()) != 2 { + t.Fatalf("application event count = %d, want 2", len(application.recordedEvents())) + } + assertPolarisEventInstance(t, application.recordedEvents()[0], remoting.EventTypeUpdate, movedA) + assertPolarisEventInstance(t, application.recordedEvents()[1], remoting.EventTypeUpdate, movedB) + if len(watcher.currentInstances) != 2 { + t.Fatalf("current instance count = %d, want 2", len(watcher.currentInstances)) + } + assertPolarisInstanceIdentity(t, watcher.currentInstances[0], movedA) + assertPolarisInstanceIdentity(t, watcher.currentInstances[1], movedB) + + watcher.handleWatchSnapshot([]model.Instance{movedB}) + assertPolarisListenerEvents(t, listener, []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.2", port: "20002"}, + {action: remoting.EventTypeAdd, host: "10.0.0.3", port: "20003"}, + }) +} + +func TestPolarisReusedWatcherReconcilesSecondSubscriberAfterDelete(t *testing.T) { + serviceName := "com.test.ReusedWatcherService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + consumer := &fakePolarisConsumer{ + instanceResponses: [][]model.Instance{{instanceA}}, + watchInstances: []model.Instance{instanceA}, + watchEvents: []model.SubScribeEvent{&model.InstanceEvent{ + DeleteEvent: &model.InstanceDeleteEvent{Instances: []model.Instance{instanceA}}, + }}, + } + pr := newTestPolarisRegistry(consumer) + watcher := mustStoppedRegistryWatcher(t, pr, serviceName) + + firstNotify := &recordingPolarisNotifyListener{} + firstListener, err := pr.createPolarisListener(serviceName, firstNotify) + if err != nil { + t.Fatalf("createPolarisListener(first) error = %v", err) + } + defer closeTestPolarisListener(firstListener) + watcher.handleWatchSnapshot([]model.Instance{instanceA}) + assertPolarisListenerEvent(t, firstListener, remoting.EventTypeAdd, instanceA) + + secondNotify := &recordingPolarisNotifyListener{} + if loadErr := pr.LoadSubscribeInstances(newPolarisConsumerURL(serviceName), secondNotify); loadErr != nil { + t.Fatalf("LoadSubscribeInstances() error = %v", loadErr) + } + if watchErr := watcher.watchOnce(); watchErr != nil { + t.Fatalf("watchOnce() error = %v", watchErr) + } + assertPolarisListenerEvent(t, firstListener, remoting.EventTypeAdd, instanceA) + assertPolarisListenerEvent(t, firstListener, remoting.EventTypeDel, instanceA) + + secondListener, err := pr.createPolarisListener(serviceName, secondNotify) + if err != nil { + t.Fatalf("createPolarisListener(second) error = %v", err) + } + defer closeTestPolarisListener(secondListener) + assertPolarisListenerEvent(t, secondListener, remoting.EventTypeDel, instanceA) + assertNoPolarisListenerEvent(t, firstListener) + + watcher.handleWatchSnapshot(nil) + assertNoPolarisListenerEvent(t, secondListener) + if len(watcher.subscribers) != 2 || !watcher.subscribers[1].reconciled || watcher.subscribers[1].initialSnapshot != nil { + t.Fatalf("second subscriber state = %#v, want reconciled with cleared baseline", watcher.subscribers[1]) + } +} + +func TestPolarisInitialSubscribeInstancesAreListenerScoped(t *testing.T) { + serviceName := "com.test.ListenerScopedBaselineService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + instanceB := newPolarisTestInstance("instance-b", "10.0.0.2", 20002, serviceName, true) + newerA := newPolarisTestInstance("newer-a", "10.0.0.1", 20001, serviceName, true) + newerA.GetMetadata()[testPolarisRevisionKey] = "newer" + + for _, tt := range []struct { + name string + responses [][]model.Instance + current []model.Instance + wantPending int + want []recordedPolarisEvent + }{ + { + name: "Load A then Load B keeps A for first Watch B", + responses: [][]model.Instance{{instanceA}, {instanceB}}, + current: []model.Instance{instanceB}, + wantPending: 2, + want: []recordedPolarisEvent{ + {action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}, + {action: remoting.EventTypeAdd, host: "10.0.0.2", port: "20002"}, + }, + }, + { + name: "empty Load preserves prior pending baseline", + responses: [][]model.Instance{{instanceA}, nil}, + wantPending: 1, + want: []recordedPolarisEvent{{action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}}, + }, + { + name: "duplicate InstanceKey keeps latest instance once", + responses: [][]model.Instance{{instanceA}, {newerA}}, + wantPending: 1, + want: []recordedPolarisEvent{{action: remoting.EventTypeDel, host: "10.0.0.1", port: "20001"}}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + consumer := &fakePolarisConsumer{instanceResponses: tt.responses} + pr := newTestPolarisRegistry(consumer) + notify := &recordingPolarisNotifyListener{} + url := newPolarisConsumerURL(serviceName) + for range tt.responses { + if err := pr.LoadSubscribeInstances(url, notify); err != nil { + t.Fatalf("LoadSubscribeInstances() error = %v", err) + } + } + + key := mustInitialSubscribeInstancesKey(t, serviceName, notify) + entry, pending := pr.loadInitialSubscribeInstances(key) + if entry == nil || len(pending) != tt.wantPending { + t.Fatalf("pending baseline = (%p, %v), want %d instances", entry, pending, tt.wantPending) + } + if tt.name == "duplicate InstanceKey keeps latest instance once" { + assertPolarisInstanceIdentity(t, pending[0], newerA) + } + + watcher := mustStoppedRegistryWatcher(t, pr, serviceName) + watcher.handleWatchSnapshot(tt.current) + listener, err := pr.createPolarisListener(serviceName, notify) + if err != nil { + t.Fatalf("createPolarisListener() error = %v", err) + } + defer closeTestPolarisListener(listener) + events := assertPolarisListenerEvents(t, listener, tt.want) + if tt.name == "duplicate InstanceKey keeps latest instance once" { + assertPolarisEventInstance(t, events[0], remoting.EventTypeDel, newerA) + } + if pendingEntry, _ := pr.loadInitialSubscribeInstances(key); pendingEntry != nil { + t.Fatalf("pending entry after listener success = %p, want nil", pendingEntry) + } + }) + } + + t.Run("baseline is stored before Notify", func(t *testing.T) { + pr := newTestPolarisRegistry(&fakePolarisConsumer{instanceResponses: [][]model.Instance{{instanceA}}}) + baselineWasReady := false + notify := &recordingPolarisNotifyListener{} + notify.onNotify = func(*registry.ServiceEvent) { + key := mustInitialSubscribeInstancesKey(t, serviceName, notify) + entry, instances := pr.loadInitialSubscribeInstances(key) + baselineWasReady = entry != nil && len(instances) == 1 + } + if err := pr.LoadSubscribeInstances(newPolarisConsumerURL(serviceName), notify); err != nil { + t.Fatalf("LoadSubscribeInstances() error = %v", err) + } + if !baselineWasReady { + t.Fatal("pending baseline was not stored before Notify(ADD)") + } + }) + + t.Run("stored baseline is isolated from input mutation", func(t *testing.T) { + baselineA := newPolarisTestInstance("baseline-a", "10.0.0.1", 20001, serviceName, true) + wantInterface := baselineA.GetMetadata()["interface"] + wantPath := baselineA.GetMetadata()["path"] + pr := newTestPolarisRegistry(nil) + notify := &recordingPolarisNotifyListener{} + key := mustInitialSubscribeInstancesKey(t, serviceName, notify) + + pr.storeInitialSubscribeInstances(key, []model.Instance{baselineA}) + delete(baselineA.GetMetadata(), "interface") + delete(baselineA.GetMetadata(), "path") + _, baseline := pr.loadInitialSubscribeInstances(key) + if len(baseline) != 1 { + t.Fatalf("pending baseline count = %d, want 1", len(baseline)) + } + metadata := baseline[0].GetMetadata() + if metadata["interface"] != wantInterface || metadata["path"] != wantPath { + t.Fatalf("pending baseline metadata = %v, want interface=%q path=%q", metadata, wantInterface, wantPath) + } + }) + + t.Run("Destroy clears pending baseline", func(t *testing.T) { + pr := newTestPolarisRegistry(nil) + notify := &recordingPolarisNotifyListener{} + key := mustInitialSubscribeInstancesKey(t, serviceName, notify) + pr.storeInitialSubscribeInstances(key, []model.Instance{instanceA}) + pr.Destroy() + pr.Destroy() + if len(pr.initialSubscribeInstances) != 0 { + t.Fatalf("pending entry count after Destroy = %d, want 0", len(pr.initialSubscribeInstances)) + } + }) +} + +func TestPolarisInitialSubscribeCompletionPreservesNewerEntry(t *testing.T) { + serviceName := "com.test.InitialSubscribeCompletionService" + instanceA := newPolarisTestInstance("instance-a", "10.0.0.1", 20001, serviceName, true) + instanceB := newPolarisTestInstance("instance-b", "10.0.0.2", 20002, serviceName, true) + pr := newTestPolarisRegistry(nil) + notify := &recordingPolarisNotifyListener{} + key := mustInitialSubscribeInstancesKey(t, serviceName, notify) + + pr.storeInitialSubscribeInstances(key, []model.Instance{instanceA}) + oldEntry, oldBaseline := pr.loadInitialSubscribeInstances(key) + if oldEntry == nil || len(oldBaseline) != 1 { + t.Fatalf("old pending entry = (%p, %v), want instance A", oldEntry, oldBaseline) + } + assertPolarisInstanceIdentity(t, oldBaseline[0], instanceA) + + pr.storeInitialSubscribeInstances(key, []model.Instance{instanceB}) + newEntry, newBaseline := pr.loadInitialSubscribeInstances(key) + if newEntry == nil || newEntry == oldEntry { + t.Fatalf("new pending entry = %p, want non-nil entry distinct from %p", newEntry, oldEntry) + } + if len(newBaseline) != 2 { + t.Fatalf("new pending baseline = %v, want instances A and B", newBaseline) + } + assertPolarisInstanceIdentity(t, newBaseline[0], instanceA) + assertPolarisInstanceIdentity(t, newBaseline[1], instanceB) + + pr.completeInitialSubscribeInstances(key, oldEntry) + remainingEntry, remainingBaseline := pr.loadInitialSubscribeInstances(key) + if remainingEntry != newEntry || len(remainingBaseline) != 2 { + t.Fatalf("pending entry after old completion = (%p, %v), want newer entry %p with A and B", remainingEntry, remainingBaseline, newEntry) + } + assertPolarisInstanceIdentity(t, remainingBaseline[0], instanceA) + assertPolarisInstanceIdentity(t, remainingBaseline[1], instanceB) + + pr.completeInitialSubscribeInstances(key, newEntry) + entry, instances := pr.loadInitialSubscribeInstances(key) + if entry != nil || len(instances) != 0 { + t.Fatalf("pending entry after new completion = (%p, %v), want empty", entry, instances) + } +} + +func TestPolarisSubscribeConsumesTaggedNotifications(t *testing.T) { + const childEnv = "DUBBO_GO_POLARIS_SUBSCRIBE_TEST_CHILD" + if os.Getenv(childEnv) == "" { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) Review Comment: [P2] 不要用固定 10 秒限制测试子进程启动 这个超时把整个子进程启动和包初始化也算在内，并非只约束被测订阅流程。在 Windows 上当前 Head 可稳定复现：父测试约 11.8 秒后以 exit status 1 终止子进程；直接设置 `DUBBO_GO_POLARIS_SUBSCRIBE_TEST_CHILD=1` 运行时，三个子用例均在 0.01 秒内通过，但包总耗时约 24 秒。相同用例在 WSL/Linux 为 0.403 秒，因此这是新增测试的机器/平台速度敏感门限，会让合法的 `go test ./registry/polaris` 失败。建议不要为已生成的测试二进制设置这么短的固定墙钟超时，或使用项目统一且留有慢机余量的测试超时；同时在失败信息中区分 `ctx.Err()==context.DeadlineExceeded`，便于诊断。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
