AlexStocks commented on code in PR #3482:
URL: https://github.com/apache/dubbo-go/pull/3482#discussion_r3607972447
##########
registry/polaris/core.go:
##########
@@ -72,60 +88,174 @@ func (watcher *PolarisServiceWatcher) lazyRun() {
})
}
+func (watcher *PolarisServiceWatcher) addSubscriberWithInitialSnapshot(
+ initialSnapshot []model.Instance,
+ subscriber item,
+) {
+ state := &subscriberState{
+ notify: subscriber,
+ initialSnapshot: copyInstances(initialSnapshot),
+ }
+
+ func() {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.subscribers = append(watcher.subscribers, state)
+ if watcher.snapshotReady {
+ watcher.reconcileSubscriberLocked(state)
+ }
+ }()
+
+ // Start only after the subscriber is registered and any available
current
+ // snapshot has been replayed.
+ watcher.lazyRun()
+}
+
+// missingInitialInstances returns initial - current by model.InstanceKey while
+// preserving the order of the initial snapshot.
+func missingInitialInstances(initial []model.Instance, current
[]model.Instance) []model.Instance {
+ currentInstances := make(map[model.InstanceKey]struct{}, len(current))
+ for _, instance := range current {
+ currentInstances[instance.GetInstanceKey()] = struct{}{}
+ }
+
+ missing := make([]model.Instance, 0, len(initial))
+ for _, instance := range initial {
+ if _, ok := currentInstances[instance.GetInstanceKey()]; !ok {
+ missing = append(missing, instance)
+ }
+ }
+ return missing
+}
+
+// handleWatchSnapshot replaces the watcher's current state and reconciles each
+// subscriber's own synchronous-load baseline exactly once.
+func (watcher *PolarisServiceWatcher) handleWatchSnapshot(current
[]model.Instance) {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.currentInstances = copyInstances(current)
+ watcher.snapshotReady = true
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.reconciled {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeAdd, watcher.currentInstances)
Review Comment:
[P1] 当前 Head `d36c21a` 仍可稳定复现这个问题。WSL overlay 先执行全量快照 `[A, B]`，再模拟重连全量快照
`[B]`，回调实际只有 `ADD(B)`，断言结果为 `event count = 1, want 2`，缺少 `DEL(A)`。
请在替换 `currentInstances` 前保留 previous snapshot，对已 reconciled subscriber 计算
`previous-current`，先下发删除事件，再发布当前完整快照；并补一条连续两次调用 `handleWatchSnapshot` 的回归测试，覆盖
`[A,B] -> [B]` 和 `[A] -> []`。
##########
registry/polaris/registry.go:
##########
@@ -188,14 +225,91 @@ func (pr *polarisRegistry) LoadSubscribeInstances(url
*common.URL, notify regist
return perrors.New(fmt.Sprintf("could not query the instances
for serviceName=%s,namespace=%s,error=%v",
serviceName, pr.namespace, err))
}
+ initialSubscribeInstances := make([]model.Instance, 0,
len(resp.Instances))
for i := range resp.Instances {
if newUrl := generateUrl(resp.Instances[i]); newUrl != nil {
notify.Notify(®istry.ServiceEvent{Action:
remoting.EventTypeAdd, Service: newUrl})
+ initialSubscribeInstances =
append(initialSubscribeInstances, resp.Instances[i])
}
}
+ pr.storeInitialSubscribeInstances(key, initialSubscribeInstances)
return nil
}
+func newInitialSubscribeInstancesKey(
+ serviceName string,
+ notify registry.NotifyListener,
+) (initialSubscribeInstancesKey, error) {
+ if isNilNotifyListener(notify) {
+ return initialSubscribeInstancesKey{}, fmt.Errorf("notify
listener type %T is nil", notify)
+ }
+ if !reflect.TypeOf(notify).Comparable() {
Review Comment:
[P1] 当前 Head `d36c21a` 尚未实现这里回复中描述的方案：`registry.go:246` 仍会对不可比较的
`NotifyListener` 直接返回错误。WSL overlay 使用合法的具名 slice listener 调用
`LoadSubscribeInstances`，稳定失败为 `notify listener type ... is not comparable`。
按你提出的设计，至少需要满足以下验收条件：不可比较 listener 的同步加载仍调用 `Notify` 且不返回兼容性错误；首个成功 watch
全量快照使用 `NotifyAll` 以当前完整列表替换同步结果；覆盖 `A -> B`、`A -> empty` 两种回归场景；可比较 listener
继续使用现有 per-listener baseline。实现和测试提交到新 Head 后，这个 P1 才能关闭。
--
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]
