diff --git a/internal/querycoordv2/job/job_release_test.go b/internal/querycoordv2/job/job_release_test.go index 7f26ace198..d5509a3182 100644 --- a/internal/querycoordv2/job/job_release_test.go +++ b/internal/querycoordv2/job/job_release_test.go @@ -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() diff --git a/internal/querycoordv2/observers/target_observer.go b/internal/querycoordv2/observers/target_observer.go index 1ee4d4bf65..0fe6aa17b2 100644 --- a/internal/querycoordv2/observers/target_observer.go +++ b/internal/querycoordv2/observers/target_observer.go @@ -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] diff --git a/internal/querycoordv2/observers/target_observer_test.go b/internal/querycoordv2/observers/target_observer_test.go index 549e229806..c1052a2ade 100644 --- a/internal/querycoordv2/observers/target_observer_test.go +++ b/internal/querycoordv2/observers/target_observer_test.go @@ -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