furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 02:05:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: revert target sync readiness change (#51547)

Browse Source 
issue: #48903

- QueryCoord target observer: revert the stricter all-delegator
readiness gate introduced by #50686
- target observer tests: restore the previous partial-ready delegator
sync coverage
- release job tests: remove the target-version mock added for the
reverted target observer path

Signed-off-by: chyezh <chyezh@outlook.com>
This commit is contained in:
Zhen Ye
2026-07-17 23:52:41 +08:00
committed by GitHub
parent c8dd73314c
commit e228dd3bc7
3 changed files with 63 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/querycoordv2/job/job_release_test.go
-1
View File
@@ -147,7 +147,6 @@ func newReleaseJobDeps(t *testing.T, m *meta.Meta, dist *meta.DistributionManage
targetMgr.EXPECT().IsNextTargetExist(mock.Anything, mock.Anything).Return(true).Maybe()
targetMgr.EXPECT().IsCurrentTargetExist(mock.Anything, mock.Anything, mock.Anything).Return(true).Maybe()
targetMgr.EXPECT().GetDmChannelsByCollection(mock.Anything, mock.Anything, mock.Anything).Return(map[string]*meta.DmChannel{}).Maybe()
targetMgr.EXPECT().GetCollectionTargetVersion(mock.Anything, mock.Anything, mock.Anything).Return(int64(0)).Maybe()
targetMgr.EXPECT().UpdateCollectionNextTarget(mock.Anything, mock.Anything).Return(nil).Maybe()
targetMgr.EXPECT().RemoveCollection(mock.Anything, mock.Anything).Return().Maybe()
nodeMgr := session.NewNodeManager()
internal/querycoordv2/observers/target_observer.go
+41 -91
View File
@@ -86,12 +86,9 @@ type TargetObserver struct {
initChan chan initRequest
// nextTargetLastUpdate map[int64]time.Time
nextTargetLastUpdate *typeutil.ConcurrentMap[int64, time.Time]
// nextTargetSyncStarted records the NextTarget version after the observer
// enters SyncTargetVersion commit phase for a collection.
nextTargetSyncStarted *typeutil.ConcurrentMap[int64, int64]
updateChan chan targetUpdateRequest
mut sync.Mutex // Guard readyNotifiers
readyNotifiers map[int64][]chan struct{} // CollectionID -> Notifiers
updateChan chan targetUpdateRequest
mut sync.Mutex // Guard readyNotifiers
readyNotifiers map[int64][]chan struct{} // CollectionID -> Notifiers
// loadingDispatcher updates targets for collections that are loading (also collections without a current target).
loadingDispatcher *taskDispatcher[int64]
@@ -113,18 +110,17 @@ func NewTargetObserver(
nodeMgr *session.NodeManager,
) *TargetObserver {
result := &TargetObserver{
meta: meta,
targetMgr: targetMgr,
distMgr: distMgr,
broker: broker,
cluster: cluster,
nodeMgr: nodeMgr,
nextTargetLastUpdate: typeutil.NewConcurrentMap[int64, time.Time](),
nextTargetSyncStarted: typeutil.NewConcurrentMap[int64, int64](),
updateChan: make(chan targetUpdateRequest, 10),
readyNotifiers: make(map[int64][]chan struct{}),
initChan: make(chan initRequest),
keylocks: lock.NewKeyLock[int64](),
meta: meta,
targetMgr: targetMgr,
distMgr: distMgr,
broker: broker,
cluster: cluster,
nodeMgr: nodeMgr,
nextTargetLastUpdate: typeutil.NewConcurrentMap[int64, time.Time](),
updateChan: make(chan targetUpdateRequest, 10),
readyNotifiers: make(map[int64][]chan struct{}),
initChan: make(chan initRequest),
keylocks: lock.NewKeyLock[int64](),
}
result.loadingDispatcher = newTaskDispatcher(result.check)
@@ -404,7 +400,6 @@ func (ob *TargetObserver) clean() {
ob.nextTargetLastUpdate.Range(func(collectionID int64, _ time.Time) bool {
if !collectionSet.Contain(collectionID) {
ob.nextTargetLastUpdate.Remove(collectionID)
ob.nextTargetSyncStarted.Remove(collectionID)
}
return true
})
@@ -422,14 +417,7 @@ func (ob *TargetObserver) clean() {
}
func (ob *TargetObserver) shouldUpdateNextTarget(ctx context.Context, collectionID int64) bool {
if !ob.targetMgr.IsNextTargetExist(ctx, collectionID) {
return true
}
if !ob.isNextTargetExpired(collectionID) {
return false
}
return !ob.isNextTargetSyncStarted(ctx, collectionID)
return !ob.targetMgr.IsNextTargetExist(ctx, collectionID) || ob.isNextTargetExpired(collectionID)
}
func (ob *TargetObserver) isNextTargetExpired(collectionID int64) bool {
@@ -451,7 +439,6 @@ func (ob *TargetObserver) updateNextTarget(ctx context.Context, collectionID int
return err
}
ob.updateNextTargetTimestamp(collectionID)
ob.nextTargetSyncStarted.Remove(collectionID)
return nil
}
@@ -459,28 +446,6 @@ func (ob *TargetObserver) updateNextTargetTimestamp(collectionID int64) {
ob.nextTargetLastUpdate.Insert(collectionID, time.Now())
}
func (ob *TargetObserver) isNextTargetSyncStarted(ctx context.Context, collectionID int64) bool {
startedVersion, hasStartedVersion := ob.nextTargetSyncStarted.Get(collectionID)
delegators := ob.distMgr.ChannelDistManager.GetByFilter(meta.WithCollectionID2Channel(collectionID))
if !hasStartedVersion && len(delegators) == 0 {
return false
}
nextVersion := ob.targetMgr.GetCollectionTargetVersion(ctx, collectionID, meta.NextTarget)
if nextVersion <= 0 {
return false
}
if hasStartedVersion && startedVersion == nextVersion {
return true
}
return lo.ContainsBy(delegators, func(delegator *meta.DmChannel) bool {
return delegator != nil &&
delegator.View != nil &&
delegator.View.TargetVersion == nextVersion
})
}
func (ob *TargetObserver) shouldUpdateCurrentTarget(ctx context.Context, collectionID int64) bool {
replicaNum := ob.meta.GetReplicaNumber(ctx, collectionID)
log := mlog.With(
@@ -521,90 +486,76 @@ func (ob *TargetObserver) shouldUpdateCurrentTarget(ctx context.Context, collect
return (newVersion == channel.View.TargetVersion && channel.IsServiceable()) || dataReadyForNextTarget
}
// Iterate through each replica to check if all its delegators are ready.
// this approach ensures each replica has at least one ready delegator for every channel.
// This prevents the issue where some replicas may lack nodes during dynamic replica scaling,
// while the total count still meets the threshold.
readyDelegatorsInCollection := make([]*meta.DmChannel, 0)
replicas := ob.meta.GetByCollection(ctx, collectionID)
if len(replicas) == 0 {
return false
}
// Prepare phase: verify all required replica/channel delegators are ready
// before SyncTargetVersion changes any readable view. This prevents a
// partially-synced NextTarget from being discarded later as an orphan.
readyDelegatorsToSync := make([]*meta.DmChannel, 0)
for _, replica := range replicas {
readyDelegatorsInReplica := make([]*meta.DmChannel, 0)
for channel := range channelNames {
// Filter delegators by replica to ensure we only check delegators belonging to this replica
delegatorList := ob.distMgr.ChannelDistManager.GetByFilter(meta.WithReplica2Channel(replica), meta.WithChannelName2Channel(channel))
if len(delegatorList) == 0 {
return false
}
for _, delegator := range delegatorList {
if delegator == nil || delegator.View == nil {
return false
}
if delegator.View.TargetVersion == newVersion && delegator.IsServiceable() {
continue
}
if checkDelegatorDataReady(replica, delegator) {
readyDelegatorsToSync = append(readyDelegatorsToSync, delegator)
continue
}
return false
readyDelegatorsInChannel := lo.Filter(delegatorList, func(ch *meta.DmChannel, _ int) bool {
return checkDelegatorDataReady(replica, ch)
})
if len(readyDelegatorsInChannel) > 0 {
readyDelegatorsInReplica = append(readyDelegatorsInReplica, readyDelegatorsInChannel...)
}
}
readyDelegatorsInCollection = append(readyDelegatorsInCollection, readyDelegatorsInReplica...)
}
// Collection-level segment readiness is the final prepare barrier. Do not
// sync readable views until the target can be promoted if sync succeeds.
if paramtable.Get().QueryCoordCfg.UpdateTargetNeedSegmentDataReady.GetAsBool() &&
utils.CheckSegmentDataReady(ctx, collectionID, ob.distMgr, ob.targetMgr, meta.NextTarget) != nil {
syncSuccess := ob.syncNextTargetToDelegator(ctx, collectionID, readyDelegatorsInCollection, newVersion)
syncedChannelNames := lo.Uniq(lo.Map(readyDelegatorsInCollection, func(ch *meta.DmChannel, _ int) string { return ch.ChannelName }))
// only after all channel are synced, we can consider the current target is ready
if !syncSuccess || !lo.Every(syncedChannelNames, lo.Keys(channelNames)) {
return false
}
syncSuccess, syncStarted := ob.syncNextTargetToDelegator(ctx, collectionID, readyDelegatorsToSync, newVersion)
if syncStarted {
ob.nextTargetSyncStarted.Insert(collectionID, newVersion)
}
return syncSuccess
// segment data satisfies next target spec
return !paramtable.Get().QueryCoordCfg.UpdateTargetNeedSegmentDataReady.GetAsBool() ||
utils.CheckSegmentDataReady(ctx, collectionID, ob.distMgr, ob.targetMgr, meta.NextTarget) == nil
}
// sync next target info to delegator as readable snapshot
// 1. if next target is changed before delegator becomes serviceable, we need to sync the new next target to delegator to support partial search
// 2. if next target is ready to read, we need to sync the next target to delegator to support full search
func (ob *TargetObserver) syncNextTargetToDelegator(ctx context.Context, collectionID int64, collReadyDelegatorList []*meta.DmChannel, newVersion int64) (bool, bool) {
func (ob *TargetObserver) syncNextTargetToDelegator(ctx context.Context, collectionID int64, collReadyDelegatorList []*meta.DmChannel, newVersion int64) bool {
var partitions []int64
var indexInfo []*indexpb.IndexInfo
var err error
syncStarted := false
for _, d := range collReadyDelegatorList {
updateVersionAction := ob.genSyncAction(ctx, d.View, newVersion)
replica := ob.meta.GetByCollectionAndNode(ctx, collectionID, d.Node)
if replica == nil {
mlog.Warn(ctx, "replica not found", mlog.FieldNodeID(d.Node), mlog.FieldCollectionID(collectionID))
// should not happen, don't update current target if replica not found
return false, syncStarted
return false
}
// init all the meta information
if partitions == nil {
partitions, err = utils.GetPartitions(ctx, ob.targetMgr, collectionID)
if err != nil {
mlog.Warn(ctx, "failed to get partitions", mlog.Err(err))
return false, syncStarted
return false
}
// Get collection index info
indexInfo, err = ob.broker.ListIndexes(ctx, collectionID)
if err != nil {
mlog.Warn(ctx, "fail to get index info of collection", mlog.Err(err))
return false, syncStarted
return false
}
}
if !ob.syncToDelegator(ctx, replica, d.View, updateVersionAction, partitions, indexInfo) {
return false, syncStarted
return false
}
syncStarted = true
}
return true, syncStarted
return true
}
func (ob *TargetObserver) syncToDelegator(ctx context.Context, replica *meta.Replica, LeaderView *meta.LeaderView, action *querypb.SyncAction,
@@ -734,7 +685,6 @@ func (ob *TargetObserver) updateAllReplicasCheckpointMetric(ctx context.Context,
func (ob *TargetObserver) updateCurrentTarget(ctx context.Context, collectionID int64) {
mlog.RatedInfo(ctx, rate.Limit(10), "observer trigger update current target", mlog.FieldCollectionID(collectionID))
if ob.targetMgr.UpdateCollectionCurrentTarget(ctx, collectionID) {
ob.nextTargetSyncStarted.Remove(collectionID)
ob.mut.Lock()
defer ob.mut.Unlock()
notifiers := ob.readyNotifiers[collectionID]
internal/querycoordv2/observers/target_observer_test.go
+22 -123
View File
@@ -603,10 +603,10 @@ func TestShouldUpdateCurrentTarget_ReplicaReadiness(t *testing.T) {
assert.False(t, result)
}
// TestShouldUpdateCurrentTarget_PartialReplicaReadinessDoesNotSync verifies that
// NextTarget readiness is a prepare phase only. A ready delegator must not receive
// SyncTargetVersion until every required delegator is ready.
func TestShouldUpdateCurrentTarget_PartialReplicaReadinessDoesNotSync(t *testing.T) {
// TestShouldUpdateCurrentTarget_OnlyReadyDelegatorsSynced verifies that only ready delegators
// are included in the sync operation. This test specifically validates the fix for the bug where
// all delegators (including non-ready ones) were being added to readyDelegatorsInReplica.
func TestShouldUpdateCurrentTarget_OnlyReadyDelegatorsSynced(t *testing.T) {
paramtable.Init()
ctx := context.Background()
collectionID := int64(1000)
@@ -669,6 +669,16 @@ func TestShouldUpdateCurrentTarget_PartialReplicaReadinessDoesNotSync(t *testing
// Return segments for CheckSegmentDataReady
targetMgr.EXPECT().GetSealedSegmentsByCollection(mock.Anything, collectionID, meta.NextTarget).Return(targetSegments).Maybe()
broker.EXPECT().ListIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
// Track which nodes receive SyncDistribution calls
syncedNodes := make([]int64, 0)
cluster.EXPECT().SyncDistribution(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
func(ctx context.Context, nodeID int64, req *querypb.SyncDistributionRequest) (*commonpb.Status, error) {
syncedNodes = append(syncedNodes, nodeID)
return merr.Success(), nil
}).Maybe()
// Node 1: READY delegator
// - Has the target segment loaded (segment 100)
// - CheckDelegatorDataReady will return nil (ready)
@@ -719,126 +729,15 @@ func TestShouldUpdateCurrentTarget_PartialReplicaReadinessDoesNotSync(t *testing
// Execute the function under test
result := observer.shouldUpdateCurrentTarget(ctx, collectionID)
assert.False(t, result)
cluster.AssertNotCalled(t, "SyncDistribution", mock.Anything, mock.Anything, mock.Anything)
broker.AssertNotCalled(t, "ListIndexes", mock.Anything, mock.Anything)
}
// Verify the result is true (at least one ready delegator exists)
assert.True(t, result)
func TestShouldUpdateNextTarget_DoesNotExpireAfterDelegatorSyncStarted(t *testing.T) {
paramtable.Init()
paramtable.Get().Save(Params.QueryCoordCfg.NextTargetSurviveTime.Key, "1")
ctx := context.Background()
collectionID := int64(1000)
nextVersion := int64(100)
nodeMgr := session.NewNodeManager()
nodeMgr.Add(session.NewNodeInfo(session.ImmutableNodeInfo{NodeID: 1}))
targetMgr := meta.NewMockTargetManager(t)
distMgr := meta.NewDistributionManager(nodeMgr)
observer := NewTargetObserver(
&meta.Meta{},
targetMgr,
distMgr,
meta.NewMockBroker(t),
session.NewMockCluster(t),
nodeMgr,
)
observer.nextTargetLastUpdate.Insert(collectionID, time.Now().Add(-time.Hour))
observer.nextTargetSyncStarted.Insert(collectionID, nextVersion)
targetMgr.EXPECT().IsNextTargetExist(mock.Anything, collectionID).Return(true).Once()
targetMgr.EXPECT().GetCollectionTargetVersion(mock.Anything, collectionID, meta.NextTarget).Return(nextVersion).Once()
assert.False(t, observer.shouldUpdateNextTarget(ctx, collectionID))
}
func TestShouldUpdateCurrentTarget_SyncMetadataFailureDoesNotPinNextTarget(t *testing.T) {
paramtable.Init()
paramtable.Get().Save(Params.QueryCoordCfg.NextTargetSurviveTime.Key, "1")
ctx := context.Background()
collectionID := int64(1000)
newVersion := int64(100)
segmentID := int64(100)
nodeMgr := session.NewNodeManager()
nodeMgr.Add(session.NewNodeInfo(session.ImmutableNodeInfo{NodeID: 1}))
targetMgr := meta.NewMockTargetManager(t)
distMgr := meta.NewDistributionManager(nodeMgr)
broker := meta.NewMockBroker(t)
cluster := session.NewMockCluster(t)
replica := meta.NewReplica(&querypb.Replica{
ID: 1,
CollectionID: collectionID,
ResourceGroup: meta.DefaultResourceGroupName,
Nodes: []int64{1},
})
mockCatalog := mocks.NewQueryCoordCatalog(t)
mockCatalog.EXPECT().SaveReplica(mock.Anything, mock.Anything).Return(nil).Maybe()
replicaMgr := meta.NewReplicaManager(nil, mockCatalog)
assert.NoError(t, replicaMgr.Put(ctx, replica))
metaInstance := &meta.Meta{
CollectionManager: meta.NewCollectionManager(nil),
ReplicaManager: replicaMgr,
}
observer := NewTargetObserver(metaInstance, targetMgr, distMgr, broker, cluster, nodeMgr)
channelNames := map[string]*meta.DmChannel{
"channel-1": {
VchannelInfo: &datapb.VchannelInfo{CollectionID: collectionID, ChannelName: "channel-1"},
},
}
targetSegments := map[int64]*datapb.SegmentInfo{
segmentID: {ID: segmentID, CollectionID: collectionID, InsertChannel: "channel-1"},
}
targetMgr.EXPECT().GetDmChannelsByCollection(mock.Anything, collectionID, meta.NextTarget).Return(channelNames).Maybe()
targetMgr.EXPECT().GetCollectionTargetVersion(mock.Anything, collectionID, meta.NextTarget).Return(newVersion).Maybe()
targetMgr.EXPECT().GetSealedSegmentsByChannel(mock.Anything, collectionID, "channel-1", mock.Anything).Return(targetSegments).Maybe()
targetMgr.EXPECT().GetGrowingSegmentsByChannel(mock.Anything, collectionID, mock.Anything, mock.Anything).Return(nil).Maybe()
targetMgr.EXPECT().GetDroppedSegmentsByChannel(mock.Anything, collectionID, mock.Anything, mock.Anything).Return(nil).Maybe()
targetMgr.EXPECT().GetDmChannel(mock.Anything, collectionID, mock.Anything, mock.Anything).Return(nil).Maybe()
targetMgr.EXPECT().GetPartitions(mock.Anything, collectionID, mock.Anything).Return([]int64{}, nil).Maybe()
targetMgr.EXPECT().GetSealedSegmentsByCollection(mock.Anything, collectionID, meta.NextTarget).Return(targetSegments).Maybe()
targetMgr.EXPECT().IsNextTargetExist(mock.Anything, collectionID).Return(true).Once()
broker.EXPECT().ListIndexes(mock.Anything, collectionID).Return(nil, assert.AnError).Once()
distMgr.ChannelDistManager.Update(1, &meta.DmChannel{
VchannelInfo: &datapb.VchannelInfo{
CollectionID: collectionID,
ChannelName: "channel-1",
},
Node: 1,
View: &meta.LeaderView{
ID: 1,
CollectionID: collectionID,
Channel: "channel-1",
Segments: map[int64]*querypb.SegmentDist{
segmentID: {NodeID: 1},
},
Status: &querypb.LeaderViewStatus{Serviceable: true},
},
})
distMgr.SegmentDistManager.Update(1, &meta.Segment{
SegmentInfo: &datapb.SegmentInfo{
ID: segmentID,
CollectionID: collectionID,
InsertChannel: "channel-1",
},
Node: 1,
})
assert.False(t, observer.shouldUpdateCurrentTarget(ctx, collectionID))
cluster.AssertNotCalled(t, "SyncDistribution", mock.Anything, mock.Anything, mock.Anything)
observer.nextTargetLastUpdate.Insert(collectionID, time.Now().Add(-time.Hour))
assert.True(t, observer.shouldUpdateNextTarget(ctx, collectionID))
// Verify that ONLY node 1 received SyncDistribution call
// This is the key assertion: if the bug existed (using delegatorList instead of readyDelegatorsInChannel),
// node 2 would also receive a SyncDistribution call
assert.Equal(t, 1, len(syncedNodes), "Expected only 1 SyncDistribution call for the ready delegator")
assert.Contains(t, syncedNodes, int64(1), "Expected node 1 (ready delegator) to receive SyncDistribution")
assert.NotContains(t, syncedNodes, int64(2), "Node 2 (not ready delegator) should NOT receive SyncDistribution")
}
// TestShouldUpdateCurrentTarget_AllChannelsSynced tests that shouldUpdateCurrentTarget returns true