fix: [poc_2.6.17] backport load-path, object-store & CDC fixes (#49769 #50172 #50070 #47507 #50026 #50162 #50342) (#50548)

Backport a small set of load-path, object-store and CDC fixes onto
`poc_2.6.17`, targeted at a self-hosted / S3-compatible deployment with
weak object storage that relies heavily on backup-restore (bulk import)
and runs CDC replication.

### Load-path & object-store
- **#49769** `limit delegator post-load concurrency` — caps
delegator-side post-load work (delete-buffer handling) with a CPU-factor
semaphore so CPU does not spike when many loads complete at once. New
config `queryNode.delegatorPostLoadConcurrencyFactor` (default 1 =
current behavior).
- **#50172** `Optimize dropped segment index GC` — when recycling a
dropped segment, also delete its segment-index object files and meta
even when the field index is already marked deleted (previously filtered
out), preventing index file/meta leakage in object storage.
- **#50070** `clear import segments NumOfRows when import task retries`
— on import retry, stale `NumOfRows` without persisted binlogs caused
later sort-compaction to fail with `unexpected row count` / EOF; clears
it on retry.
- **#47507** `treat 200 and 404 as successful DeleteObject response` —
hand-adapted to the 2.6 storage layer (the original is a 2.5 commit).
Some S3-compatible stores return HTTP 200 instead of 204 for a
successful delete (and 404 means already-gone); treat both as success
before mapping the error.

### CDC / replication (force-promote & salvage)
- **#50026** `refuse to build QueryCoord target when channel checkpoint
is dropped sentinel` — after a CDC force-promote, a channel checkpoint
may be the dropped-sentinel marker; QueryCoord must not build a load
target from it (would load against a stale/invalid checkpoint).
- **#50162** `ack stale alter-load-config on dropped-sentinel channel` —
companion to #50026: acks (drops) a stale alter-load-config DDL callback
that targets a dropped-sentinel channel instead of blocking on it.
Depends on #50026 and is ordered after it.
- **#50342** (cp #50519) `return error instead of panic on malformed
DumpMessages start_message_id` — `DumpMessages` (CDC salvage RPC)
unmarshalled a client-controlled `start_message_id` via
`MustUnmarshalMessageID`, so a malformed id panicked the proxy process;
now returns `InvalidParameter`.

### Notes
- `#47507` is the only non-cherry-pick: the 2.6 storage layer uses
`mapObjectStorageError` (refactored from 2.5's
`checkObjectStorageError`), so the 200/404 success check was re-applied
by hand in `MinioObjectStorage.RemoveObject`.
- `#50026` → `#50162` → `#50342` were cherry-picked in that order (the
first two share `querycoordv2/meta/target_manager.go`; #50162 builds on
#50026).
- `#50167` (sliced index sidecar loading) was intentionally
**excluded**: it depends on nullable-vector support (`VALID_DATA_KEY` /
`VALID_DATA_COUNT_KEY` in `VectorIndex.h`) which is not on this branch,
and its fix targets nullable-vector sidecars not used by this
deployment.

---------

Signed-off-by: Wei Liu <wei.liu@zilliz.com>
Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: wei liu <wei.liu@zilliz.com>
Co-authored-by: cai.zhang <cai.zhang@zilliz.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: yihao.dai <yihao.dai@zilliz.com>

configs/milvus.yaml

@@ -579,6 +579,7 @@ queryNode:
   levelZeroForwardPolicy: FilterByBF # delegator level zero deletion forward policy, possible option["FilterByBF", "RemoteLoad"]
   streamingDeltaForwardPolicy: FilterByBF # delegator streaming deletion forward policy, possible option["FilterByBF", "Direct"]
   forwardBatchSize: 4194304 # the batch size delegator uses for forwarding stream delete in loading procedure
+  delegatorPostLoadConcurrencyFactor: 1 # delegator post-load concurrency factor after worker LoadSegments returns. Concurrency is hardware.GetCPUNum * factor
   exprCache:
     enabled: false # enable expression result cache
     capacityBytes: 268435456 # max capacity in bytes for expression result cache

internal/datacoord/garbage_collector.go

@@ -743,40 +743,148 @@ func (gc *garbageCollector) recycleDroppedSegments(ctx context.Context, signal <
 			continue
 		}
 
-		cloned := segment.Clone()
-		binlog.DecompressBinLogs(cloned.SegmentInfo)
-
-		logs := getLogs(cloned)
-		for key := range getTextLogs(cloned) {
-			logs[key] = struct{}{}
-		}
-
-		for key := range getJSONKeyLogs(cloned, gc) {
-			logs[key] = struct{}{}
-		}
-
-		log.Info("GC segment start...", zap.Int("insert_logs", len(cloned.GetBinlogs())),
-			zap.Int("delta_logs", len(cloned.GetDeltalogs())),
-			zap.Int("stats_logs", len(cloned.GetStatslogs())),
-			zap.Int("bm25_logs", len(cloned.GetBm25Statslogs())),
-			zap.Int("text_logs", len(cloned.GetTextStatsLogs())),
-			zap.Int("json_key_logs", len(cloned.GetJsonKeyStats())))
-		if err := gc.removeObjectFiles(ctx, logs); err != nil {
-			log.Warn("GC segment remove logs failed", zap.Error(err))
-			cloned = nil
-			continue
-		}
-
-		if err := gc.meta.DropSegment(ctx, cloned.GetID()); err != nil {
-			log.Warn("GC segment meta failed to drop segment", zap.Error(err))
-			cloned = nil
-			continue
-		}
-		log.Info("GC segment meta drop segment done")
-		cloned = nil // release memory
+		gc.recycleDroppedSegment(ctx, segmentID, segment)
 	}
 }
 
+// recycleDroppedSegment deletes a single dropped segment's object files,
+// segment-index files, segment-index meta, and segment meta in that order.
+//
+// The ordering matters: files first so a meta-only retry after partial
+// file deletion can still observe the leftover keys; segment-index meta
+// next so segment meta deletion (the final marker) is the only step that
+// commits the GC; if any step fails the later state is preserved for the
+// next GC cycle to retry.
+//
+// This path can race with recycleUnusedSegIndexes on the same BuildID
+// whenever a dropped segment's parent field index has also been marked
+// IsDeleted — both paths call removeObjectFiles for the same index files
+// and call indexMeta.RemoveSegmentIndex(buildID). The races are safe today
+// only because two invariants hold elsewhere:
+//
+//  1. removeObjectFiles swallows merr.ErrIoKeyNotFound (see the loop in
+//     removeObjectFiles below), so a double file delete is a no-op for the
+//     loser.
+//  2. indexMeta.RemoveSegmentIndex acquires a per-buildID keyLock and
+//     returns nil when segmentBuildInfo.Get(buildID) is !ok (see
+//     index_meta.go), so a double catalog delete is a no-op for the loser.
+//
+// Any future refactor on either side — tightening removeObjectFiles to
+// surface NotFound, or batching RemoveSegmentIndex past the per-buildID
+// keyLock — must preserve these invariants, otherwise dropped-segment GC
+// will silently break under load.
+func (gc *garbageCollector) recycleDroppedSegment(ctx context.Context, segmentID int64, segment *SegmentInfo) {
+	log := log.With(zap.Int64("segmentID", segmentID), zap.Int64("collectionID", segment.GetCollectionID()))
+
+	if ctx.Err() != nil {
+		return
+	}
+
+	segIndexes, indexFiles := gc.getDroppedSegmentIndexFiles(segmentID)
+
+	cloned := segment.Clone()
+	if err := gc.removeDroppedSegmentFiles(ctx, cloned, indexFiles); err != nil {
+		log.Warn("GC segment remove files failed", zap.Error(err))
+		return
+	}
+
+	if ctx.Err() != nil {
+		return
+	}
+
+	if err := gc.removeDroppedSegmentIndexMeta(ctx, segIndexes); err != nil {
+		log.Warn("GC segment index meta failed, wait to retry", zap.Error(err))
+		return
+	}
+
+	if ctx.Err() != nil {
+		return
+	}
+
+	if err := gc.meta.DropSegment(ctx, cloned.GetID()); err != nil {
+		log.Warn("GC segment meta failed to drop segment", zap.Error(err))
+		return
+	}
+	log.Info("GC segment meta drop segment done", zap.Int("indexMetaCleaned", len(segIndexes)))
+}
+
+func (gc *garbageCollector) getDroppedSegmentIndexFiles(segmentID int64) ([]*model.SegmentIndex, map[string]struct{}) {
+	segIndexes := gc.getAllSegmentIndexesForDroppedSegment(segmentID)
+	if len(segIndexes) == 0 {
+		return nil, nil
+	}
+	indexFiles := make(map[string]struct{}, len(segIndexes))
+	for _, segIdx := range segIndexes {
+		for key := range gc.getAllIndexFilesOfIndex(segIdx) {
+			indexFiles[key] = struct{}{}
+		}
+	}
+	return segIndexes, indexFiles
+}
+
+// getAllSegmentIndexesForDroppedSegment wraps indexMeta.GetAllSegmentIndexes
+// with a defensive nil guard. Production newMeta always wires indexMeta, but
+// the guard is cheap and turns any unexpected nil into "no index records"
+// instead of a panic during a GC sweep — keeping a single misbuilt gc
+// instance from taking the whole datacoord down.
+func (gc *garbageCollector) getAllSegmentIndexesForDroppedSegment(segmentID int64) []*model.SegmentIndex {
+	if gc.meta == nil || gc.meta.indexMeta == nil {
+		return nil
+	}
+	return gc.meta.indexMeta.GetAllSegmentIndexes(segmentID)
+}
+
+func (gc *garbageCollector) removeDroppedSegmentFiles(ctx context.Context, cloned *SegmentInfo, indexFiles map[string]struct{}) error {
+	log := log.With(zap.Int64("segmentID", cloned.GetID()))
+
+	binlog.DecompressBinLogs(cloned.SegmentInfo)
+	logs := getLogs(cloned)
+	for key := range getTextLogs(cloned) {
+		logs[key] = struct{}{}
+	}
+	for key := range getJSONKeyLogs(cloned, gc) {
+		logs[key] = struct{}{}
+	}
+	for key := range indexFiles {
+		logs[key] = struct{}{}
+	}
+
+	log.Info("GC segment start...", zap.Int("insert_logs", len(cloned.GetBinlogs())),
+		zap.Int("delta_logs", len(cloned.GetDeltalogs())),
+		zap.Int("stats_logs", len(cloned.GetStatslogs())),
+		zap.Int("bm25_logs", len(cloned.GetBm25Statslogs())),
+		zap.Int("text_logs", len(cloned.GetTextStatsLogs())),
+		zap.Int("json_key_logs", len(cloned.GetJsonKeyStats())),
+		zap.Int("index_files", len(indexFiles)))
+	if err := gc.removeObjectFiles(ctx, logs); err != nil {
+		log.Warn("GC segment remove logs failed", zap.Error(err))
+		return err
+	}
+	return nil
+}
+
+// removeDroppedSegmentIndexMeta removes the segment-index meta for every
+// build of a dropped segment, one RemoveSegmentIndex at a time.
+//
+// Partial failure converges: RemoveSegmentIndex updates in-memory state
+// (segmentIndexes + segmentBuildInfo) only after its catalog write succeeds,
+// so if the k-th call succeeds but the (k+1)-th fails, the first k entries are
+// already gone from meta. recycleDroppedSegment then bails out (segment meta is
+// kept), and on the next GC cycle GetAllSegmentIndexes returns only the
+// remaining entries — the retry covers exactly what is left and eventually
+// drains to empty.
+func (gc *garbageCollector) removeDroppedSegmentIndexMeta(ctx context.Context, segIndexes []*model.SegmentIndex) error {
+	if len(segIndexes) == 0 {
+		return nil
+	}
+	for _, segIdx := range segIndexes {
+		if err := gc.meta.indexMeta.RemoveSegmentIndex(ctx, segIdx.BuildID); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (gc *garbageCollector) recycleChannelCPMeta(ctx context.Context, signal <-chan gcCmd) {
 	log := log.Ctx(ctx)
 	channelCPs, err := gc.meta.catalog.ListChannelCheckpoint(ctx)

internal/datacoord/garbage_collector_test.go

@@ -25,9 +25,11 @@ import (
 	"path"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/bytedance/mockey"
 	"github.com/cockroachdb/errors"
 	"github.com/google/uuid"
 	"github.com/minio/minio-go/v7"
@@ -36,6 +38,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
+	"go.uber.org/atomic"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
@@ -43,17 +46,20 @@ import (
 	broker2 "github.com/milvus-io/milvus/internal/datacoord/broker"
 	kvmocks "github.com/milvus-io/milvus/internal/kv/mocks"
 	"github.com/milvus-io/milvus/internal/metastore"
+	"github.com/milvus-io/milvus/internal/metastore/kv/binlog"
 	"github.com/milvus-io/milvus/internal/metastore/kv/datacoord"
 	catalogmocks "github.com/milvus-io/milvus/internal/metastore/mocks"
 	"github.com/milvus-io/milvus/internal/metastore/model"
 	"github.com/milvus-io/milvus/internal/mocks"
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/pkg/v2/common"
+	"github.com/milvus-io/milvus/pkg/v2/objectstorage"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/workerpb"
 	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/lock"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
+	"github.com/milvus-io/milvus/pkg/v2/util/metautil"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
@@ -858,6 +864,13 @@ func TestGarbageCollector_clearETCD(t *testing.T) {
 		mock.Anything,
 		mock.Anything,
 	).Return(nil)
+	catalog.On("DropSegmentIndex",
+		mock.Anything,
+		mock.Anything,
+		mock.Anything,
+		mock.Anything,
+		mock.Anything,
+	).Return(nil).Maybe()
 
 	channelCPs := newChannelCps()
 	channelCPs.checkpoints["dmlChannel"] = &msgpb.MsgPosition{
@@ -1389,6 +1402,7 @@ func TestGarbageCollector_clearETCD(t *testing.T) {
 
 	cm := &mocks.ChunkManager{}
 	cm.EXPECT().Remove(mock.Anything, mock.Anything).Return(nil)
+	cm.EXPECT().RootPath().Return("").Maybe()
 	signal := make(chan gcCmd)
 	gc := newGarbageCollector(
 		m,
@@ -1810,3 +1824,352 @@ func (s *GarbageCollectorSuite) TestAvoidGCLoadedSegments() {
 func TestGarbageCollector(t *testing.T) {
 	suite.Run(t, new(GarbageCollectorSuite))
 }
+
+func setupDroppedSegmentWithIndexForGC(t *testing.T) (*meta, *SegmentInfo, *model.SegmentIndex, string) {
+	t.Helper()
+	ctx := context.Background()
+	const (
+		collID  = int64(100)
+		partID  = int64(10)
+		segID   = int64(1001)
+		fieldID = int64(20)
+		indexID = int64(30)
+		buildID = int64(40)
+		logID   = int64(1)
+	)
+	// Build the binlog path the same way DecompressBinLog will rebuild it at GC
+	// time so assertions stay paramtable-agnostic.
+	binlogPath, err := binlog.BuildLogPath(storage.InsertBinlog, collID, partID, segID, fieldID, logID)
+	require.NoError(t, err)
+
+	m, err := newMemoryMeta(t)
+	require.NoError(t, err)
+	segment := &SegmentInfo{
+		SegmentInfo: &datapb.SegmentInfo{
+			ID:            segID,
+			CollectionID:  collID,
+			PartitionID:   partID,
+			InsertChannel: "ch1",
+			State:         commonpb.SegmentState_Dropped,
+			DroppedAt:     uint64(time.Now().Add(-time.Hour).UnixNano()),
+			Binlogs:       []*datapb.FieldBinlog{getFieldBinlogPaths(fieldID, binlogPath)},
+		},
+	}
+	require.NoError(t, m.AddSegment(ctx, segment))
+
+	// Write through fieldIndexLock so this helper does not race with the GC
+	// goroutine that reads indexMeta.indexes under the same lock (go test -race).
+	m.indexMeta.fieldIndexLock.Lock()
+	if m.indexMeta.indexes == nil {
+		m.indexMeta.indexes = make(map[UniqueID]map[UniqueID]*model.Index)
+	}
+	m.indexMeta.indexes[collID] = map[UniqueID]*model.Index{
+		indexID: {
+			CollectionID: collID,
+			FieldID:      fieldID,
+			IndexID:      indexID,
+			IndexName:    "deleted-index",
+			IsDeleted:    true,
+		},
+	}
+	m.indexMeta.fieldIndexLock.Unlock()
+	segIdx := &model.SegmentIndex{
+		SegmentID:     segID,
+		CollectionID:  collID,
+		PartitionID:   partID,
+		IndexID:       indexID,
+		BuildID:       buildID,
+		IndexVersion:  1,
+		IndexState:    commonpb.IndexState_Finished,
+		IndexFileKeys: []string{"idx-file"},
+	}
+	require.NoError(t, m.indexMeta.AddSegmentIndex(ctx, segIdx))
+	return m, segment, segIdx, binlogPath
+}
+
+func TestGarbageCollector_recycleDroppedSegments_RecyclesSegmentIndexMeta(t *testing.T) {
+	ctx := context.Background()
+	m, segment, segIdx, binlogPath := setupDroppedSegmentWithIndexForGC(t)
+
+	expectedIndexFile := metautil.BuildSegmentIndexFilePath("root", segIdx.BuildID, segIdx.IndexVersion,
+		segIdx.PartitionID, segIdx.SegmentID, "idx-file")
+
+	cm := mocks.NewChunkManager(t)
+	cm.EXPECT().RootPath().Return("root").Maybe()
+	var mu sync.Mutex
+	removed := make([]string, 0, 2)
+	cm.EXPECT().Remove(mock.Anything, mock.Anything).RunAndReturn(
+		func(ctx context.Context, filePath string) error {
+			mu.Lock()
+			defer mu.Unlock()
+			removed = append(removed, filePath)
+			return nil
+		}).Maybe()
+
+	gc := newGarbageCollector(m, newMockHandler(), GcOption{
+		cli:           cm,
+		dropTolerance: 0,
+	})
+
+	assert.Empty(t, m.indexMeta.GetSegmentIndexes(segment.CollectionID, segment.ID),
+		"deleted field indexes are filtered by the existing query API")
+	assert.Len(t, m.indexMeta.GetAllSegmentIndexes(segment.ID), 1)
+
+	gc.recycleDroppedSegments(ctx, nil)
+
+	assert.Nil(t, m.GetSegment(ctx, segment.ID))
+	assert.Empty(t, m.indexMeta.GetAllSegmentIndexes(segment.ID))
+	mu.Lock()
+	assert.ElementsMatch(t, []string{binlogPath, expectedIndexFile}, removed)
+	mu.Unlock()
+}
+
+func TestGarbageCollector_recycleDroppedSegments_FileDeleteFailureKeepsMeta(t *testing.T) {
+	ctx := context.Background()
+	m, segment, _, _ := setupDroppedSegmentWithIndexForGC(t)
+
+	cli := storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test"))
+	gc := newGarbageCollector(m, newMockHandler(), GcOption{
+		cli:           cli,
+		dropTolerance: 0,
+	})
+
+	mockRemoveObjectFiles := mockey.Mock((*garbageCollector).removeObjectFiles).Return(errors.New("remove failed")).Build()
+	defer mockRemoveObjectFiles.UnPatch()
+
+	gc.recycleDroppedSegments(ctx, nil)
+
+	assert.NotNil(t, m.GetSegment(ctx, segment.ID))
+	assert.Len(t, m.indexMeta.GetAllSegmentIndexes(segment.ID), 1)
+}
+
+func TestGarbageCollector_recycleDroppedSegments_IndexMetaFailureKeepsSegmentMeta(t *testing.T) {
+	ctx := context.Background()
+	m, segment, _, _ := setupDroppedSegmentWithIndexForGC(t)
+
+	cli := storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test"))
+	gc := newGarbageCollector(m, newMockHandler(), GcOption{
+		cli:           cli,
+		dropTolerance: 0,
+	})
+
+	mockRemoveObjectFiles := mockey.Mock((*garbageCollector).removeObjectFiles).Return(nil).Build()
+	defer mockRemoveObjectFiles.UnPatch()
+	mockRemoveSegmentIndex := mockey.Mock((*indexMeta).RemoveSegmentIndex).Return(errors.New("meta failed")).Build()
+	defer mockRemoveSegmentIndex.UnPatch()
+
+	gc.recycleDroppedSegments(ctx, nil)
+
+	assert.NotNil(t, m.GetSegment(ctx, segment.ID))
+	assert.Len(t, m.indexMeta.GetAllSegmentIndexes(segment.ID), 1)
+}
+
+func TestGarbageCollector_recycleDroppedSegment_DropSegmentFailure(t *testing.T) {
+	ctx := context.Background()
+	m, segment, _, _ := setupDroppedSegmentWithIndexForGC(t)
+	cli := storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test"))
+	gc := newGarbageCollector(m, newMockHandler(), GcOption{
+		cli:           cli,
+		dropTolerance: 0,
+	})
+
+	mockRemoveObjectFiles := mockey.Mock((*garbageCollector).removeObjectFiles).Return(nil).Build()
+	defer mockRemoveObjectFiles.UnPatch()
+	mockDropSegment := mockey.Mock((*meta).DropSegment).Return(errors.New("drop segment failed")).Build()
+	defer mockDropSegment.UnPatch()
+
+	gc.recycleDroppedSegment(ctx, segment.ID, segment)
+
+	assert.NotNil(t, m.GetSegment(ctx, segment.ID))
+}
+
+func TestGarbageCollector_DroppedSegmentIndexHelpers(t *testing.T) {
+	ctx := context.Background()
+	m, segment, segIdx, _ := setupDroppedSegmentWithIndexForGC(t)
+
+	gc := newGarbageCollector(m, newMockHandler(), GcOption{
+		cli: storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test")),
+	})
+
+	segIndexes, indexFiles := gc.getDroppedSegmentIndexFiles(segment.ID)
+	require.Len(t, segIndexes, 1)
+	assert.Equal(t, segIdx.BuildID, segIndexes[0].BuildID)
+	expectedIndexFile := metautil.BuildSegmentIndexFilePath(gc.option.cli.RootPath(), segIdx.BuildID,
+		segIdx.IndexVersion, segIdx.PartitionID, segIdx.SegmentID, "idx-file")
+	assert.Contains(t, indexFiles, expectedIndexFile)
+
+	assert.Nil(t, (&garbageCollector{}).getAllSegmentIndexesForDroppedSegment(segment.ID))
+	assert.NoError(t, gc.removeDroppedSegmentIndexMeta(ctx, nil))
+	require.NoError(t, gc.removeDroppedSegmentIndexMeta(ctx, segIndexes))
+	assert.Empty(t, m.indexMeta.GetAllSegmentIndexes(segment.ID))
+}
+
+func TestGarbageCollector_recycleDroppedSegment_CancellationShortCircuit(t *testing.T) {
+	t.Run("cancel before files step", func(t *testing.T) {
+		m, segment, _, _ := setupDroppedSegmentWithIndexForGC(t)
+		gc := newGarbageCollector(m, newMockHandler(), GcOption{
+			cli:           storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test-gc-cancel-1")),
+			dropTolerance: 0,
+		})
+
+		var removeCalled atomic.Bool
+		mockRemove := mockey.Mock((*garbageCollector).removeObjectFiles).To(
+			func(gc *garbageCollector, ctx context.Context, files map[string]struct{}) error {
+				removeCalled.Store(true)
+				return nil
+			}).Build()
+		defer mockRemove.UnPatch()
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		gc.recycleDroppedSegment(ctx, segment.ID, segment)
+
+		assert.False(t, removeCalled.Load(), "file removal must be skipped when ctx is canceled before the files step")
+		assert.NotNil(t, m.GetSegment(context.Background(), segment.ID))
+	})
+
+	t.Run("cancel between files and index meta", func(t *testing.T) {
+		m, segment, _, _ := setupDroppedSegmentWithIndexForGC(t)
+		gc := newGarbageCollector(m, newMockHandler(), GcOption{
+			cli:           storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test-gc-cancel-2")),
+			dropTolerance: 0,
+		})
+
+		ctx, cancel := context.WithCancel(context.Background())
+		mockRemove := mockey.Mock((*garbageCollector).removeObjectFiles).To(
+			func(gc *garbageCollector, c context.Context, files map[string]struct{}) error {
+				cancel()
+				return nil
+			}).Build()
+		defer mockRemove.UnPatch()
+
+		var indexMetaCalled atomic.Bool
+		mockIdx := mockey.Mock((*garbageCollector).removeDroppedSegmentIndexMeta).To(
+			func(gc *garbageCollector, c context.Context, idx []*model.SegmentIndex) error {
+				indexMetaCalled.Store(true)
+				return nil
+			}).Build()
+		defer mockIdx.UnPatch()
+
+		gc.recycleDroppedSegment(ctx, segment.ID, segment)
+
+		assert.False(t, indexMetaCalled.Load(), "index meta step must be skipped when ctx is canceled mid-segment")
+		// segment meta should still exist since DropSegment never ran.
+		assert.NotNil(t, m.GetSegment(context.Background(), segment.ID))
+	})
+}
+
+func TestGarbageCollector_getDroppedSegmentIndexFiles_EdgeCases(t *testing.T) {
+	t.Run("empty segIndexes returns nil maps", func(t *testing.T) {
+		m, err := newMemoryMeta(t)
+		require.NoError(t, err)
+		gc := newGarbageCollector(m, newMockHandler(), GcOption{
+			cli: storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test")),
+		})
+		segIndexes, indexFiles := gc.getDroppedSegmentIndexFiles(9999)
+		assert.Nil(t, segIndexes)
+		assert.Nil(t, indexFiles)
+	})
+
+	t.Run("segment with index returns full file set", func(t *testing.T) {
+		m, segment, segIdx, _ := setupDroppedSegmentWithIndexForGC(t)
+		gc := newGarbageCollector(m, newMockHandler(), GcOption{
+			cli: storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test")),
+		})
+		segIndexes, indexFiles := gc.getDroppedSegmentIndexFiles(segment.ID)
+		require.Len(t, segIndexes, 1)
+		assert.Equal(t, segIdx.BuildID, segIndexes[0].BuildID)
+		assert.NotEmpty(t, indexFiles)
+	})
+}
+
+// TestGarbageCollector_recycleDroppedSegment_CtxCanceledBeforeDrop covers
+// the ctx.Err() guard before DropSegment by canceling ctx inside
+// removeDroppedSegmentIndexMeta so DropSegment must NOT run.
+func TestGarbageCollector_recycleDroppedSegment_CtxCanceledBeforeDrop(t *testing.T) {
+	m, segment, _, _ := setupDroppedSegmentWithIndexForGC(t)
+	cli := storage.NewLocalChunkManager(objectstorage.RootPath("/tmp/test-gc-cancel-drop"))
+	gc := newGarbageCollector(m, newMockHandler(), GcOption{
+		cli:           cli,
+		dropTolerance: 0,
+	})
+
+	mockRemove := mockey.Mock((*garbageCollector).removeObjectFiles).Return(nil).Build()
+	defer mockRemove.UnPatch()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	var indexMetaCalls atomic.Int32
+	mockIdx := mockey.Mock((*garbageCollector).removeDroppedSegmentIndexMeta).To(
+		func(gc *garbageCollector, c context.Context, idx []*model.SegmentIndex) error {
+			indexMetaCalls.Inc()
+			cancel()
+			return nil
+		}).Build()
+	defer mockIdx.UnPatch()
+
+	var dropCalls atomic.Int32
+	mockDrop := mockey.Mock((*meta).DropSegment).To(
+		func(m *meta, ctx context.Context, sid int64) error {
+			dropCalls.Inc()
+			return nil
+		}).Build()
+	defer mockDrop.UnPatch()
+
+	gc.recycleDroppedSegment(ctx, segment.ID, segment)
+
+	assert.Equal(t, int32(1), indexMetaCalls.Load())
+	assert.Equal(t, int32(0), dropCalls.Load(),
+		"DropSegment must NOT run when ctx is canceled between index-meta and segment-meta steps")
+	assert.NotNil(t, m.GetSegment(context.Background(), segment.ID))
+}
+
+// TestGarbageCollector_removeDroppedSegmentFiles_TextAndJSONLogs covers the
+// for-range loops that merge getTextLogs and getJSONKeyLogs into the file
+// deletion set in the V1/V2 path.
+func TestGarbageCollector_removeDroppedSegmentFiles_TextAndJSONLogs(t *testing.T) {
+	ctx := context.Background()
+	cm := mocks.NewChunkManager(t)
+	cm.EXPECT().RootPath().Return("root").Maybe()
+	var mu sync.Mutex
+	removed := make(map[string]struct{})
+	cm.EXPECT().Remove(mock.Anything, mock.Anything).RunAndReturn(
+		func(ctx context.Context, filePath string) error {
+			mu.Lock()
+			defer mu.Unlock()
+			removed[filePath] = struct{}{}
+			return nil
+		}).Maybe()
+
+	gc := newGarbageCollector(nil, nil, GcOption{cli: cm})
+	const textFile = "text/log/file-1"
+	const jsonFile = "json-file"
+	const indexFile = "idx/extra/file-99"
+	segment := &SegmentInfo{
+		SegmentInfo: &datapb.SegmentInfo{
+			ID:           7001,
+			CollectionID: 100,
+			PartitionID:  10,
+			TextStatsLogs: map[int64]*datapb.TextIndexStats{
+				101: {Files: []string{textFile}},
+			},
+			JsonKeyStats: map[int64]*datapb.JsonKeyStats{
+				102: {
+					FieldID: 102,
+					BuildID: 11,
+					Version: 1,
+					Files:   []string{jsonFile},
+				},
+			},
+		},
+	}
+
+	indexFiles := map[string]struct{}{indexFile: {}}
+	require.NoError(t, gc.removeDroppedSegmentFiles(ctx, segment, indexFiles))
+
+	expectedJSON := path.Join("root", common.JSONIndexPath, "11", "1", "100", "10", "7001", "102", jsonFile)
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Contains(t, removed, textFile)
+	assert.Contains(t, removed, expectedJSON)
+	assert.Contains(t, removed, indexFile)
+}

internal/datacoord/import_task_import.go

@@ -163,6 +163,18 @@ func (t *importTask) QueryTaskOnWorker(cluster session.Cluster) {
 	}
 	resp, err := cluster.QueryImport(t.GetNodeID(), req)
 	if err != nil || resp.GetState() == datapb.ImportTaskStateV2_Retry {
+		// Clear partial progress recorded from the failed attempt. Otherwise,
+		// if the retried attempt's PickSegment lands on a different subset of
+		// the preallocated segments, the segments it skips will keep stale
+		// NumOfRows without insert binlogs — causing sort compaction to fail
+		// with "unexpected row count" or EOF.
+		if segmentIDs := t.GetSegmentIDs(); len(segmentIDs) > 0 {
+			if resetErr := t.meta.UpdateSegmentsInfo(context.TODO(), ResetImportingSegmentRows(segmentIDs...)); resetErr != nil {
+				log.Warn("failed to reset import segment row counts on retry",
+					WrapTaskLog(t, zap.Error(resetErr))...)
+				return
+			}
+		}
 		updateErr := t.importMeta.UpdateTask(context.TODO(), t.GetTaskID(), UpdateState(datapb.ImportTaskStateV2_Pending))
 		if updateErr != nil {
 			log.Warn("failed to update import task state to pending", WrapTaskLog(t, zap.Error(updateErr))...)

internal/datacoord/import_task_import_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
 	"github.com/milvus-io/milvus/internal/datacoord/session"
 	"github.com/milvus-io/milvus/internal/metastore/mocks"
@@ -271,8 +272,11 @@ func TestImportTask_QueryTaskOnWorker(t *testing.T) {
 			State:        datapb.ImportTaskStateV2_InProgress,
 		}
 		task := &importTask{
-			alloc:      nil,
-			meta:       &meta{collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo]()},
+			alloc: nil,
+			meta: &meta{
+				collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo](),
+				segments:    NewSegmentsInfo(),
+			},
 			importMeta: im,
 			tr:         timerecord.NewTimeRecorder(""),
 		}
@@ -289,6 +293,66 @@ func TestImportTask_QueryTaskOnWorker(t *testing.T) {
 		assert.Equal(t, datapb.ImportTaskStateV2_InProgress, task.GetState())
 	})
 
+	t.Run("QueryImport rpc failed resets NumOfRows", func(t *testing.T) {
+		catalog := mocks.NewDataCoordCatalog(t)
+		catalog.EXPECT().ListImportJobs(mock.Anything).Return(nil, nil)
+		catalog.EXPECT().ListPreImportTasks(mock.Anything).Return(nil, nil)
+		catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
+		catalog.EXPECT().SaveImportTask(mock.Anything, mock.Anything).Return(nil)
+
+		im, err := NewImportMeta(context.TODO(), catalog, nil, nil)
+		assert.NoError(t, err)
+
+		taskProto := &datapb.ImportTaskV2{
+			JobID:        1,
+			TaskID:       2,
+			CollectionID: 3,
+			SegmentIDs:   []int64{5, 6},
+			NodeID:       7,
+			State:        datapb.ImportTaskStateV2_InProgress,
+		}
+		segCatalog := mocks.NewDataCoordCatalog(t)
+		segCatalog.EXPECT().AlterSegments(mock.Anything, mock.Anything).Return(nil)
+		task := &importTask{
+			alloc: nil,
+			meta: &meta{
+				catalog:     segCatalog,
+				collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo](),
+				segments:    NewSegmentsInfo(),
+			},
+			importMeta: im,
+			tr:         timerecord.NewTimeRecorder(""),
+		}
+		task.task.Store(taskProto)
+		err = im.AddTask(context.TODO(), task)
+		assert.NoError(t, err)
+
+		task.meta.segments.SetSegment(5, &SegmentInfo{
+			SegmentInfo: &datapb.SegmentInfo{
+				ID:        5,
+				State:     commonpb.SegmentState_Importing,
+				NumOfRows: 100,
+				MaxRowNum: 100,
+			},
+		})
+		task.meta.segments.SetSegment(6, &SegmentInfo{
+			SegmentInfo: &datapb.SegmentInfo{
+				ID:        6,
+				State:     commonpb.SegmentState_Importing,
+				NumOfRows: 50,
+				MaxRowNum: 50,
+			},
+		})
+
+		cluster := session.NewMockCluster(t)
+		cluster.EXPECT().QueryImport(mock.Anything, mock.Anything).Return(nil, errors.New("mock err"))
+		task.QueryTaskOnWorker(cluster)
+
+		assert.Equal(t, datapb.ImportTaskStateV2_Pending, task.GetState())
+		assert.EqualValues(t, 0, task.meta.segments.GetSegment(5).GetNumOfRows())
+		assert.EqualValues(t, 0, task.meta.segments.GetSegment(6).GetNumOfRows())
+	})
+
 	t.Run("import failed", func(t *testing.T) {
 		catalog := mocks.NewDataCoordCatalog(t)
 		catalog.EXPECT().ListImportJobs(mock.Anything).Return(nil, nil)

internal/datacoord/index_meta.go

@@ -783,6 +783,33 @@ func (m *indexMeta) GetSegmentIndexes(collectionID UniqueID, segID UniqueID) map
 	return m.getSegmentIndexes(collectionID, segID)
 }
 
+// GetAllSegmentIndexes returns a deep-clone of every SegmentIndex recorded
+// for segID, including ones whose parent field index has been marked
+// IsDeleted — that is the case dropped-segment GC needs to clean up.
+//
+// Intentionally does NOT acquire fieldIndexLock: this method only reads
+// m.segmentIndexes (a ConcurrentMap, self-synchronized) and never touches
+// m.indexes (which fieldIndexLock protects). Do not add an IsDeleted
+// filter here without first taking fieldIndexLock — that change would
+// silently start requiring the lock, unlike GetSegmentIndexes above which
+// already holds it.
+func (m *indexMeta) GetAllSegmentIndexes(segID UniqueID) []*model.SegmentIndex {
+	if m.segmentIndexes == nil {
+		return nil
+	}
+	segIndexInfos, ok := m.segmentIndexes.Get(segID)
+	if !ok || segIndexInfos.Len() == 0 {
+		return nil
+	}
+
+	segIndexes := segIndexInfos.Values()
+	ret := make([]*model.SegmentIndex, 0, len(segIndexes))
+	for _, segIdx := range segIndexes {
+		ret = append(ret, model.CloneSegmentIndex(segIdx))
+	}
+	return ret
+}
+
 // Note: thread-unsafe, don't call it outside indexMeta
 func (m *indexMeta) getSegmentIndexes(collectionID UniqueID, segID UniqueID) map[UniqueID]*model.SegmentIndex {
 	ret := make(map[UniqueID]*model.SegmentIndex, 0)

internal/datacoord/index_meta_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus/internal/json"
@@ -985,6 +986,42 @@ func TestMeta_GetSegmentIndexes(t *testing.T) {
 	})
 }
 
+func TestMeta_GetAllSegmentIndexes(t *testing.T) {
+	assert.Nil(t, (&indexMeta{}).GetAllSegmentIndexes(segID))
+
+	m := &indexMeta{
+		segmentIndexes: typeutil.NewConcurrentMap[UniqueID, *typeutil.ConcurrentMap[UniqueID, *model.SegmentIndex]](),
+		indexes: map[UniqueID]map[UniqueID]*model.Index{
+			collID: {
+				indexID: {
+					CollectionID: collID,
+					FieldID:      fieldID,
+					IndexID:      indexID,
+					IndexName:    indexName,
+					IsDeleted:    true,
+				},
+			},
+		},
+	}
+	segIdxes := typeutil.NewConcurrentMap[UniqueID, *model.SegmentIndex]()
+	segIdxes.Insert(indexID, &model.SegmentIndex{
+		CollectionID: collID,
+		SegmentID:    segID,
+		IndexID:      indexID,
+		BuildID:      buildID,
+		IndexState:   commonpb.IndexState_Finished,
+	})
+	m.segmentIndexes.Insert(segID, segIdxes)
+
+	assert.Empty(t, m.GetSegmentIndexes(collID, segID))
+	segIndexes := m.GetAllSegmentIndexes(segID)
+	require.Len(t, segIndexes, 1)
+	assert.Equal(t, buildID, segIndexes[0].BuildID)
+	origin, ok := segIdxes.Get(indexID)
+	require.True(t, ok)
+	assert.False(t, origin == segIndexes[0])
+}
+
 func TestMeta_GetFieldIDByIndexID(t *testing.T) {
 	m := newSegmentIndexMeta(nil)
 	m.indexes = map[UniqueID]map[UniqueID]*model.Index{

internal/datacoord/meta.go

@@ -336,7 +336,7 @@ func (m *meta) reloadFromKV(ctx context.Context, collectionIDs []int64) error {
 		// for 2.2.2 issue https://github.com/milvus-io/milvus/issues/22181
 		pos.ChannelName = vChannel
 		m.channelCPs.checkpoints[vChannel] = pos
-		if pos.Timestamp != math.MaxUint64 {
+		if !funcutil.IsDroppedChannelCheckpoint(pos) {
 			// Should not be set as metric since it's a tombstone value.
 			ts, _ := tsoutil.ParseTS(pos.Timestamp)
 			metrics.DataCoordCheckpointUnixSeconds.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()), vChannel).
@@ -1252,6 +1252,35 @@ func UpdateImportedRows(segmentID int64, rows int64) UpdateOperator {
 	}
 }
 
+// ResetImportingSegmentRows clears NumOfRows and MaxRowNum on each given
+// segment that is still in the Importing state. Used to discard partial
+// progress reported by a failed import attempt before the task is rescheduled,
+// so segments the retried attempt skips do not keep stale row counts that
+// would break sort compaction.
+func ResetImportingSegmentRows(segmentIDs ...int64) UpdateOperator {
+	return func(modPack *updateSegmentPack) bool {
+		anyReset := false
+		for _, segmentID := range segmentIDs {
+			segment := modPack.Get(segmentID)
+			if segment == nil {
+				log.Ctx(context.TODO()).Warn("meta update: reset importing segment rows failed - segment not found",
+					zap.Int64("segmentID", segmentID))
+				continue
+			}
+			if segment.GetState() != commonpb.SegmentState_Importing {
+				log.Ctx(context.TODO()).Warn("meta update: reset importing segment rows skipped - segment not in Importing state",
+					zap.Int64("segmentID", segmentID),
+					zap.String("state", segment.GetState().String()))
+				continue
+			}
+			segment.NumOfRows = 0
+			segment.MaxRowNum = 0
+			anyReset = true
+		}
+		return anyReset
+	}
+}
+
 func UpdateIsImporting(segmentID int64, isImporting bool) UpdateOperator {
 	return func(modPack *updateSegmentPack) bool {
 		segment := modPack.Get(segmentID)
@@ -2106,15 +2135,17 @@ func (m *meta) UpdateChannelCheckpoint(ctx context.Context, vChannel string, pos
 	return nil
 }
 
-// MarkChannelCheckpointDropped set channel checkpoint to MaxUint64 preventing future update
-// and remove the metrics for channel checkpoint lag.
+// MarkChannelCheckpointDropped writes the dropped-channel sentinel
+// (funcutil.DroppedChannelCheckpointTimestamp) so no later
+// UpdateChannelCheckpoint can overwrite it, and removes the channel-checkpoint
+// lag metric.
 func (m *meta) MarkChannelCheckpointDropped(ctx context.Context, channel string) error {
 	m.channelCPs.Lock()
 	defer m.channelCPs.Unlock()
 
 	cp := &msgpb.MsgPosition{
 		ChannelName: channel,
-		Timestamp:   math.MaxUint64,
+		Timestamp:   funcutil.DroppedChannelCheckpointTimestamp,
 	}
 
 	err := m.catalog.SaveChannelCheckpoints(ctx, []*msgpb.MsgPosition{cp})

internal/proxy/impl.go

@@ -7131,7 +7131,13 @@ func (node *Proxy) DumpMessages(req *milvuspb.DumpMessagesRequest, stream milvus
 		return merr.WrapErrParameterMissing("start_message_id")
 	}
 
-	startMsgID := message.MustUnmarshalMessageID(req.GetStartMessageId())
+	// Unmarshal the message id without panicking: the id bytes are
+	// client-controlled, so a malformed value must be rejected as an invalid
+	// parameter rather than crashing the process.
+	startMsgID, err := message.UnmarshalMessageID(req.GetStartMessageId())
+	if err != nil {
+		return merr.WrapErrParameterInvalidMsg("invalid start_message_id: %s", err.Error())
+	}
 
 	// Use exclusive start position (dump messages AFTER start_message_id)
 	// This is appropriate for salvage scenarios where start_message_id is the last synced message

internal/querycoordv2/ddl_callbacks_alter_load_info.go

@@ -19,9 +19,14 @@ package querycoordv2
 import (
 	"context"
 
+	"github.com/cockroachdb/errors"
+	"go.uber.org/zap"
+
 	"github.com/milvus-io/milvus/internal/querycoordv2/job"
 	"github.com/milvus-io/milvus/internal/querycoordv2/meta"
+	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 )
 
 // alterLoadConfigV2AckCallback is called when the put load config message is acknowledged
@@ -30,6 +35,18 @@ func (s *Server) alterLoadConfigV2AckCallback(ctx context.Context, result messag
 	// TODO: after we support query view in 3.0, we should broadcast the put load config message to all vchannels.
 	job := job.NewLoadCollectionJob(ctx, result, s.dist, s.meta, s.broker, s.targetMgr, s.targetObserver, s.collectionObserver, s.checkerController, s.nodeMgr, s.proxyClientManager)
 	if err := job.Execute(); err != nil {
+		// The collection's channel is already in dropped-sentinel state, meaning
+		// a concurrent DropCollection has landed at the channel level on this
+		// cluster (typically on a CDC replica). The load can never succeed; ack
+		// the broadcast so the broadcaster stops retrying forever and the
+		// pending rootcoord meta cleanup can complete. Mirrors the
+		// ErrCollectionNotFound early-return inside LoadCollectionJob.Execute().
+		if errors.Is(err, merr.ErrChannelDroppedSentinel) {
+			log.Ctx(ctx).Warn("ack alter load config: collection channel is dropped sentinel",
+				zap.Int64("collectionID", result.Message.Header().GetCollectionId()),
+				zap.Error(err))
+			return nil
+		}
 		return err
 	}
 	meta.GlobalFailedLoadCache.Remove(result.Message.Header().GetCollectionId())

internal/querycoordv2/ddl_callbacks_alter_load_info_test.go

@@ -0,0 +1,80 @@
+// Licensed to the LF AI & Data foundation 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 querycoordv2
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bytedance/mockey"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/milvus-io/milvus/internal/querycoordv2/job"
+	"github.com/milvus-io/milvus/pkg/v2/proto/messagespb"
+	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
+	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+)
+
+func buildAlterLoadConfigBroadcastResult(collectionID int64) message.BroadcastResultAlterLoadConfigMessageV2 {
+	controlChannel := "_ctrl_channel"
+	broadcastMsg := message.NewAlterLoadConfigMessageBuilderV2().
+		WithHeader(&messagespb.AlterLoadConfigMessageHeader{
+			CollectionId: collectionID,
+			Replicas: []*messagespb.LoadReplicaConfig{
+				{ReplicaId: 1, ResourceGroupName: "__default_resource_group"},
+			},
+		}).
+		WithBody(&messagespb.AlterLoadConfigMessageBody{}).
+		WithBroadcast([]string{controlChannel}).
+		MustBuildBroadcast()
+
+	return message.BroadcastResultAlterLoadConfigMessageV2{
+		Message: message.MustAsBroadcastAlterLoadConfigMessageV2(broadcastMsg),
+		Results: map[string]*message.AppendResult{
+			controlChannel: {},
+		},
+	}
+}
+
+// TestAlterLoadConfigV2AckCallback verifies that the ack callback swallows the
+// dropped-sentinel error (so the broadcaster stops retrying forever) while still
+// propagating any other error from the load job.
+func TestAlterLoadConfigV2AckCallback(t *testing.T) {
+	paramtable.Init()
+	ctx := context.Background()
+	s := &Server{}
+	result := buildAlterLoadConfigBroadcastResult(1000)
+
+	mockey.PatchConvey("dropped sentinel is acked with no-op", t, func() {
+		mockey.Mock((*job.LoadCollectionJob).Execute).
+			Return(merr.WrapErrChannelDroppedSentinel("_ctrl_channel")).Build()
+
+		err := s.alterLoadConfigV2AckCallback(ctx, result)
+		assert.NoError(t, err)
+	})
+
+	mockey.PatchConvey("generic error is propagated", t, func() {
+		expectedErr := errors.New("broker unavailable")
+		mockey.Mock((*job.LoadCollectionJob).Execute).Return(expectedErr).Build()
+
+		err := s.alterLoadConfigV2AckCallback(ctx, result)
+		assert.Error(t, err)
+		assert.True(t, errors.Is(err, expectedErr))
+	})
+}

internal/querycoordv2/meta/target_manager.go

@@ -33,6 +33,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/querypb"
 	"github.com/milvus-io/milvus/pkg/v2/util/conc"
+	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/retry"
@@ -153,6 +154,22 @@ func (mgr *TargetManager) UpdateCollectionNextTarget(ctx context.Context, collec
 		return err
 	}
 
+	// A dropped checkpoint sentinel is not a valid seek position; do not
+	// build a next target that could dispatch WatchDmChannels with it.
+	for _, channelInfo := range vChannelInfos {
+		if funcutil.IsDroppedChannelCheckpoint(channelInfo.GetSeekPosition()) {
+			log.Warn("refuse to build next target: channel checkpoint is a dropped sentinel; sticky until collection meta is fully dropped",
+				zap.Int64("collectionID", collectionID),
+				zap.String("channel", channelInfo.GetChannelName()),
+				zap.Uint64("seekTs", channelInfo.GetSeekPosition().GetTimestamp()),
+			)
+			return merr.WrapErrChannelDroppedSentinel(
+				channelInfo.GetChannelName(),
+				"refuse to build next target",
+			)
+		}
+	}
+
 	partitions := mgr.meta.GetPartitionsByCollection(ctx, collectionID)
 	partitionIDs := lo.Map(partitions, func(partition *Partition, i int) int64 {
 		return partition.PartitionID

internal/querycoordv2/meta/target_manager_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
 	"github.com/milvus-io/milvus/internal/json"
 	etcdkv "github.com/milvus-io/milvus/internal/kv/etcd"
 	"github.com/milvus-io/milvus/internal/metastore"
@@ -38,6 +39,8 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/querypb"
 	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
+	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metricsinfo"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -776,6 +779,131 @@ func (suite *TargetManagerSuite) TestGetTargetJSON() {
 	assert.Len(suite.T(), currentTarget2, 0)
 }
 
+func (suite *TargetManagerSuite) TestUpdateNextTarget_DroppedSentinelRejected() {
+	ctx := suite.ctx
+	collectionID := int64(2001)
+
+	suite.meta.PutCollection(ctx, &Collection{
+		CollectionLoadInfo: &querypb.CollectionLoadInfo{
+			CollectionID:  collectionID,
+			ReplicaNumber: 1,
+		},
+	})
+	suite.meta.PutPartition(ctx, &Partition{
+		PartitionLoadInfo: &querypb.PartitionLoadInfo{
+			CollectionID: collectionID,
+			PartitionID:  1,
+		},
+	})
+
+	sentinelChannels := []*datapb.VchannelInfo{
+		{
+			CollectionID: collectionID,
+			ChannelName:  "sentinel-channel",
+			SeekPosition: &msgpb.MsgPosition{
+				ChannelName: "sentinel-channel",
+				Timestamp:   funcutil.DroppedChannelCheckpointTimestamp,
+			},
+		},
+	}
+	suite.broker.EXPECT().GetRecoveryInfoV2(mock.Anything, collectionID).
+		Return(sentinelChannels, nil, nil).Once()
+
+	err := suite.mgr.UpdateCollectionNextTarget(ctx, collectionID)
+	suite.Require().Error(err)
+	suite.True(errors.Is(err, merr.ErrChannelDroppedSentinel),
+		"error should wrap ErrChannelDroppedSentinel, got: %v", err)
+	suite.Contains(err.Error(), "sentinel-channel")
+
+	// Next target must NOT be advanced.
+	suite.assertChannels([]string{}, suite.mgr.GetDmChannelsByCollection(ctx, collectionID, NextTarget))
+}
+
+func (suite *TargetManagerSuite) TestUpdateNextTarget_SentinelAmongMultiple() {
+	ctx := suite.ctx
+	collectionID := int64(2002)
+
+	suite.meta.PutCollection(ctx, &Collection{
+		CollectionLoadInfo: &querypb.CollectionLoadInfo{
+			CollectionID:  collectionID,
+			ReplicaNumber: 1,
+		},
+	})
+	suite.meta.PutPartition(ctx, &Partition{
+		PartitionLoadInfo: &querypb.PartitionLoadInfo{
+			CollectionID: collectionID,
+			PartitionID:  1,
+		},
+	})
+
+	mixedChannels := []*datapb.VchannelInfo{
+		{
+			CollectionID: collectionID,
+			ChannelName:  "normal-channel",
+			SeekPosition: &msgpb.MsgPosition{
+				ChannelName: "normal-channel",
+				Timestamp:   450000000000000000,
+			},
+		},
+		{
+			CollectionID: collectionID,
+			ChannelName:  "poisoned-channel",
+			SeekPosition: &msgpb.MsgPosition{
+				ChannelName: "poisoned-channel",
+				Timestamp:   funcutil.DroppedChannelCheckpointTimestamp,
+			},
+		},
+	}
+	suite.broker.EXPECT().GetRecoveryInfoV2(mock.Anything, collectionID).
+		Return(mixedChannels, nil, nil).Once()
+
+	err := suite.mgr.UpdateCollectionNextTarget(ctx, collectionID)
+	suite.Require().Error(err)
+	suite.True(errors.Is(err, merr.ErrChannelDroppedSentinel))
+	suite.Contains(err.Error(), "poisoned-channel")
+
+	// Next target must remain empty — no partial build.
+	suite.assertChannels([]string{}, suite.mgr.GetDmChannelsByCollection(ctx, collectionID, NextTarget))
+}
+
+func (suite *TargetManagerSuite) TestUpdateNextTarget_SentinelDoesNotBurnRetries() {
+	ctx := suite.ctx
+	collectionID := int64(2003)
+
+	suite.meta.PutCollection(ctx, &Collection{
+		CollectionLoadInfo: &querypb.CollectionLoadInfo{
+			CollectionID:  collectionID,
+			ReplicaNumber: 1,
+		},
+	})
+	suite.meta.PutPartition(ctx, &Partition{
+		PartitionLoadInfo: &querypb.PartitionLoadInfo{
+			CollectionID: collectionID,
+			PartitionID:  1,
+		},
+	})
+
+	sentinelChannels := []*datapb.VchannelInfo{
+		{
+			CollectionID: collectionID,
+			ChannelName:  "sentinel-once",
+			SeekPosition: &msgpb.MsgPosition{
+				ChannelName: "sentinel-once",
+				Timestamp:   funcutil.DroppedChannelCheckpointTimestamp,
+			},
+		},
+	}
+
+	// .Once() asserts that GetRecoveryInfoV2 is called exactly once: the
+	// sentinel must not re-enter the retry.Handle loop.
+	suite.broker.EXPECT().GetRecoveryInfoV2(mock.Anything, collectionID).
+		Return(sentinelChannels, nil, nil).Once()
+
+	err := suite.mgr.UpdateCollectionNextTarget(ctx, collectionID)
+	suite.Require().Error(err)
+	suite.broker.AssertExpectations(suite.T())
+}
+
 func BenchmarkTargetManager(b *testing.B) {
 	paramtable.Init()
 	config := GenerateEtcdConfig()

internal/querynodev2/delegator/delegator.go

@@ -47,6 +47,7 @@ import (
 	"github.com/milvus-io/milvus/internal/util/searchutil/optimizers"
 	"github.com/milvus-io/milvus/internal/util/streamrpc"
 	"github.com/milvus-io/milvus/pkg/v2/common"
+	"github.com/milvus-io/milvus/pkg/v2/config"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/metrics"
 	"github.com/milvus-io/milvus/pkg/v2/proto/internalpb"
@@ -167,6 +168,10 @@ type shardDelegator struct {
 	schemaChangeMutex sync.RWMutex
 	schemaVersion     uint64
 
+	// limits delegator-side post-load work after worker LoadSegments returns.
+	postLoadSem           *syncutil.Semaphore
+	postLoadConfigHandler config.EventHandler
+
 	// streaming data catch-up state
 	catchingUpStreamingData *atomic.Bool
 
@@ -1240,6 +1245,10 @@ func (sd *shardDelegator) Close() {
 		sd.idfOracle.Close()
 	}
 
+	if sd.postLoadConfigHandler != nil {
+		paramtable.Get().Unwatch(paramtable.Get().QueryNodeCfg.DelegatorPostLoadConcurrencyFactor.Key, sd.postLoadConfigHandler)
+	}
+
 	if sd.functionRunners != nil {
 		for _, function := range sd.functionRunners {
 			function.Close()
@@ -1326,6 +1335,14 @@ func NewShardDelegator(ctx context.Context, collectionID UniqueID, replicaID Uni
 
 	policy := paramtable.Get().QueryNodeCfg.LevelZeroForwardPolicy.GetValue()
 	log.Info("shard delegator setup l0 forward policy", zap.String("policy", policy))
+	postLoadSem := syncutil.NewSemaphore(paramtable.Get().QueryNodeCfg.DelegatorPostLoadConcurrencyFactor.GetAsInt())
+	postLoadConfigHandler := config.NewHandler(fmt.Sprintf("qn.delegator.postload.%s.%p", channel, postLoadSem), func(event *config.Event) {
+		if event.HasUpdated {
+			concurrency := paramtable.Get().QueryNodeCfg.DelegatorPostLoadConcurrencyFactor.GetAsInt()
+			postLoadSem.SetCapacity(concurrency)
+			log.Info("resize delegator post-load concurrency", zap.Int("concurrency", concurrency))
+		}
+	})
 
 	sd := &shardDelegator{
 		collectionID:   collectionID,
@@ -1349,6 +1366,8 @@ func NewShardDelegator(ctx context.Context, collectionID UniqueID, replicaID Uni
 		analyzerRunners:            make(map[UniqueID]function.Analyzer),
 		isBM25Field:                make(map[int64]bool),
 		l0ForwardPolicy:            policy,
+		postLoadSem:                postLoadSem,
+		postLoadConfigHandler:      postLoadConfigHandler,
 		catchingUpStreamingData:    atomic.NewBool(true),
 		latestRequiredMVCCTimeTick: atomic.NewUint64(0),
 	}
@@ -1386,6 +1405,7 @@ func NewShardDelegator(ctx context.Context, collectionID UniqueID, replicaID Uni
 	}
 
 	sd.tsCond = syncutil.NewContextCond(&sync.Mutex{})
+	paramtable.Get().Watch(paramtable.Get().QueryNodeCfg.DelegatorPostLoadConcurrencyFactor.Key, postLoadConfigHandler)
 	log.Info("finish build new shardDelegator")
 	return sd, nil
 }

internal/querynodev2/delegator/delegator_data.go

@@ -523,56 +523,76 @@ func (sd *shardDelegator) LoadSegments(ctx context.Context, req *querypb.LoadSeg
 		return nil
 	}
 
-	infos := lo.Filter(req.GetInfos(), func(info *querypb.SegmentLoadInfo, _ int) bool {
-		return !sd.distribution.SealedSegmentExistsOnNode(info.GetSegmentID(), targetNodeID)
-	})
+	return sd.withPostLoadLimit(ctx, func() error {
+		infos := lo.Filter(req.GetInfos(), func(info *querypb.SegmentLoadInfo, _ int) bool {
+			return !sd.distribution.SealedSegmentExistsOnNode(info.GetSegmentID(), targetNodeID)
+		})
 
-	candidates, err := sd.loader.LoadBloomFilterSet(ctx, req.GetCollectionID(), infos...)
-	if err != nil {
-		log.Warn("failed to load bloom filter set for segment", zap.Error(err))
-		return err
-	}
-
-	// Load BM25 stats BEFORE loadStreamDelete so stats are ready before segment becomes visible
-	err = sd.loadBM25Stats(ctx, infos, req)
-	if err != nil {
-		log.Warn("failed to load BM25 stats", zap.Error(err))
-		return err
-	}
-
-	// Build a map from segmentID to BloomFilterSet
-	bfMap := make(map[int64]pkoracle.Candidate)
-	for _, candidate := range candidates {
-		log.Info("loaded bloom filter set for sealed segment",
-			zap.Int64("segmentID", candidate.ID()),
-		)
-		bfMap[candidate.ID()] = candidate
-	}
-
-	// Build entries with Candidate before loadStreamDelete, which will atomically add them to distribution
-	entries := make([]SegmentEntry, 0, len(infos))
-	for _, info := range infos {
-		entry := SegmentEntry{
-			SegmentID:   info.GetSegmentID(),
-			PartitionID: info.GetPartitionID(),
-			NodeID:      req.GetDstNodeID(),
-			Version:     req.GetVersion(),
-			Level:       info.GetLevel(),
-			Candidate:   bfMap[info.GetSegmentID()],
+		candidates, err := sd.loader.LoadBloomFilterSet(ctx, req.GetCollectionID(), infos...)
+		if err != nil {
+			log.Warn("failed to load bloom filter set for segment", zap.Error(err))
+			return err
 		}
-		entries = append(entries, entry)
+
+		// Load BM25 stats BEFORE loadStreamDelete so stats are ready before segment becomes visible
+		err = sd.loadBM25Stats(ctx, infos, req)
+		if err != nil {
+			log.Warn("failed to load BM25 stats", zap.Error(err))
+			return err
+		}
+
+		// Build a map from segmentID to BloomFilterSet
+		bfMap := make(map[int64]pkoracle.Candidate)
+		for _, candidate := range candidates {
+			log.Info("loaded bloom filter set for sealed segment",
+				zap.Int64("segmentID", candidate.ID()),
+			)
+			bfMap[candidate.ID()] = candidate
+		}
+
+		// Build entries with Candidate before loadStreamDelete, which will atomically add them to distribution
+		entries := make([]SegmentEntry, 0, len(infos))
+		for _, info := range infos {
+			entries = append(entries, SegmentEntry{
+				SegmentID:   info.GetSegmentID(),
+				PartitionID: info.GetPartitionID(),
+				NodeID:      req.GetDstNodeID(),
+				Version:     req.GetVersion(),
+				Level:       info.GetLevel(),
+				Candidate:   bfMap[info.GetSegmentID()],
+			})
+		}
+
+		log.Debug("load delete...")
+		// loadStreamDelete now handles distribution add atomically in Phase 3
+		err = sd.loadStreamDelete(ctx, candidates, infos, req, targetNodeID, worker,
+			entries, req.GetLoadMeta().GetSchemaVersion())
+		if err != nil {
+			log.Warn("load stream delete failed", zap.Error(err))
+			return err
+		}
+
+		return nil
+	})
+}
+
+func (sd *shardDelegator) withPostLoadLimit(ctx context.Context, fn func() error) error {
+	if sd.postLoadSem == nil {
+		return fn()
 	}
 
-	log.Debug("load delete...")
-	// loadStreamDelete now handles distribution add atomically in Phase 3
-	err = sd.loadStreamDelete(ctx, candidates, infos, req, targetNodeID, worker,
-		entries, req.GetLoadMeta().GetSchemaVersion())
-	if err != nil {
-		log.Warn("load stream delete failed", zap.Error(err))
+	start := time.Now()
+	if err := sd.postLoadSem.Acquire(ctx); err != nil {
 		return err
 	}
+	defer sd.postLoadSem.Release()
 
-	return nil
+	log.Ctx(ctx).Debug("delegator acquired post-load slot",
+		zap.Duration("wait", time.Since(start)),
+		zap.Int("capacity", sd.postLoadSem.Cap()),
+		zap.Int("current", sd.postLoadSem.Current()))
+
+	return fn()
 }
 
 func (sd *shardDelegator) addDistributionIfVersionOK(version uint64, entries ...SegmentEntry) error {

internal/querynodev2/delegator/delegator_data_test.go

@@ -60,6 +60,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/util/metautil"
 	"github.com/milvus-io/milvus/pkg/v2/util/metric"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/syncutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
@@ -974,6 +975,65 @@ func (s *DelegatorDataSuite) TestLoadSegments() {
 	})
 }
 
+func (s *DelegatorDataSuite) TestPostLoadLimiter() {
+	s.Run("serializes_post_load_work", func() {
+		sd := &shardDelegator{
+			postLoadSem: syncutil.NewSemaphore(1),
+		}
+		var current int32
+		var maxConcurrent int32
+		start := make(chan struct{})
+		errCh := make(chan error, 8)
+		wg := sync.WaitGroup{}
+		for i := 0; i < 8; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				<-start
+				err := sd.withPostLoadLimit(context.Background(), func() error {
+					running := atomic.AddInt32(&current, 1)
+					for {
+						maxValue := atomic.LoadInt32(&maxConcurrent)
+						if running <= maxValue || atomic.CompareAndSwapInt32(&maxConcurrent, maxValue, running) {
+							break
+						}
+					}
+					time.Sleep(5 * time.Millisecond)
+					atomic.AddInt32(&current, -1)
+					return nil
+				})
+				errCh <- err
+			}()
+		}
+		close(start)
+		wg.Wait()
+		close(errCh)
+		for err := range errCh {
+			s.NoError(err)
+		}
+		s.Equal(int32(1), maxConcurrent)
+		s.Equal(0, sd.postLoadSem.Current())
+	})
+
+	s.Run("returns_context_error_while_waiting", func() {
+		sd := &shardDelegator{
+			postLoadSem: syncutil.NewSemaphore(1),
+		}
+		s.NoError(sd.postLoadSem.Acquire(context.Background()))
+		defer sd.postLoadSem.Release()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Millisecond)
+		defer cancel()
+		called := false
+		err := sd.withPostLoadLimit(ctx, func() error {
+			called = true
+			return nil
+		})
+		s.ErrorIs(err, context.DeadlineExceeded)
+		s.False(called)
+	})
+}
+
 func (s *DelegatorDataSuite) waitTargetVersion(targetVersion int64) {
 	for {
 		if s.delegator.idfOracle.TargetVersion() >= targetVersion {

internal/storage/minio_object_storage.go

@@ -19,6 +19,7 @@ package storage
 import (
 	"context"
 	"io"
+	"net/http"
 
 	"github.com/minio/minio-go/v7"
 	"go.uber.org/zap"
@@ -105,7 +106,27 @@ func (minioObjectStorage *MinioObjectStorage) WalkWithObjects(ctx context.Contex
 }
 
 func (minioObjectStorage *MinioObjectStorage) RemoveObject(ctx context.Context, bucketName, objectName string) error {
+	// isDeleteSuccessful treats 200 and 404 as successful deletion.
+	// Some S3-compatible storage doesn't fully comply with the S3 spec (minio SDK expects 204);
+	// 404 also means the object is already gone, so the delete is effectively successful.
+	isDeleteSuccessful := func(err error) bool {
+		if err == nil {
+			return true
+		}
+		statusCode := minio.ToErrorResponse(err).StatusCode
+		return statusCode == http.StatusOK || statusCode == http.StatusNotFound
+	}
+
 	err := minioObjectStorage.Client.RemoveObject(ctx, bucketName, objectName, minio.RemoveObjectOptions{})
+	if isDeleteSuccessful(err) {
+		if err != nil {
+			log.Debug("object already removed or storage returned non-standard success code",
+				zap.String("bucket", bucketName),
+				zap.String("object", objectName),
+				zap.Int("statusCode", minio.ToErrorResponse(err).StatusCode))
+		}
+		return nil
+	}
 	return mapObjectStorageError(objectName, err)
 }
 

pkg/streaming/util/message/message_id.go

@@ -49,13 +49,25 @@ func MustUnmarshalMessageID(msgID *commonpb.MessageID) MessageID {
 
 // UnmsarshalMessageID unmarshal the message id.
 func UnmarshalMessageID(msgID *commonpb.MessageID) (MessageID, error) {
+	if msgID == nil {
+		return nil, errors.Wrap(ErrInvalidMessageID, "nil message id")
+	}
+	// wal_name is a client-controlled proto enum; proto3 allows any int32 on the
+	// wire. Return errors instead of panicking so a malformed value cannot crash
+	// the process. MustUnmarshalMessageID keeps the panic contract for trusted
+	// internal callers.
 	name := WALName(msgID.WALName)
 	if name == WALNameUnknown {
-		name = MustGetDefaultWALName()
+		// Use the non-panicking lookup: an unregistered default WAL must not
+		// nil-deref on client input.
+		name = GetDefaultWALName()
+		if name == WALNameUnknown {
+			return nil, errors.Wrapf(ErrInvalidMessageID, "default wal name is not registered, wal: %s", msgID.WALName.String())
+		}
 	}
 	unmarshaler, ok := messageIDUnmarshaler.Get(name)
 	if !ok {
-		panic("MessageID Unmarshaler not registered: " + name.String())
+		return nil, errors.Wrapf(ErrInvalidMessageID, "message id unmarshaler not registered for wal: %s", msgID.WALName.String())
 	}
 	return unmarshaler(msgID.Id)
 }

pkg/streaming/util/message/message_id_test.go

@@ -22,8 +22,23 @@ func TestRegisterMessageIDUnmarshaler(t *testing.T) {
 	assert.Nil(t, id)
 	assert.Error(t, err)
 
+	// An unregistered wal_name (a client can put any int32 on the wire for a
+	// proto enum) must return an error instead of panicking, so it cannot crash
+	// the process. 101 is not a registered WAL. MustUnmarshalMessageID keeps the
+	// panic contract.
+	id, err = message.UnmarshalMessageID(&commonpb.MessageID{WALName: commonpb.WALName(101), Id: "123"})
+	assert.Nil(t, id)
+	assert.Error(t, err)
+	assert.ErrorIs(t, err, message.ErrInvalidMessageID)
+	assert.Contains(t, err.Error(), "101")
+
+	id, err = message.UnmarshalMessageID(nil)
+	assert.Nil(t, id)
+	assert.Error(t, err)
+	assert.ErrorIs(t, err, message.ErrInvalidMessageID)
+
 	assert.Panics(t, func() {
-		message.UnmarshalMessageID(&commonpb.MessageID{WALName: commonpb.WALName(101), Id: "123"})
+		message.MustUnmarshalMessageID(&commonpb.MessageID{WALName: commonpb.WALName(101), Id: "123"})
 	})
 
 	assert.Panics(t, func() {

pkg/util/funcutil/checkpoint.go

@@ -0,0 +1,47 @@
+// Licensed to the LF AI & Data foundation 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 funcutil
+
+import (
+	"math"
+
+	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
+)
+
+// DroppedChannelCheckpointTimestamp is the reserved timestamp value used to
+// mark a channel checkpoint as dropped. No legitimate channel checkpoint may
+// carry this value.
+//
+// The sole writer is datacoord.(*meta).MarkChannelCheckpointDropped, invoked
+// when a vchannel is dropped. The sentinel is sticky:
+// datacoord.(*meta).UpdateChannelCheckpoint refuses to overwrite it because
+// oldTs < newTs fails when oldTs == DroppedChannelCheckpointTimestamp. A
+// sentinel therefore remains visible to readers (e.g. GetRecoveryInfo) until
+// the collection meta is fully dropped, at which point the channel ceases to
+// be returned at all.
+const DroppedChannelCheckpointTimestamp uint64 = math.MaxUint64
+
+// IsDroppedChannelCheckpoint reports whether the given checkpoint position is
+// the dropped-channel sentinel.
+//
+// Callers that consume channel seek positions (notably QueryCoord's target
+// builder) MUST treat a sentinel as an abnormal metadata state — not a normal
+// "start from this position" instruction — and refuse to use it as a seek
+// point.
+func IsDroppedChannelCheckpoint(pos *msgpb.MsgPosition) bool {
+	return pos != nil && pos.GetTimestamp() == DroppedChannelCheckpointTimestamp
+}

pkg/util/funcutil/checkpoint_test.go

@@ -0,0 +1,79 @@
+// Licensed to the LF AI & Data foundation 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 funcutil
+
+import (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
+)
+
+func TestIsDroppedChannelCheckpoint(t *testing.T) {
+	cases := []struct {
+		name     string
+		pos      *msgpb.MsgPosition
+		expected bool
+	}{
+		{
+			name:     "nil position is not sentinel",
+			pos:      nil,
+			expected: false,
+		},
+		{
+			name:     "zero timestamp is not sentinel",
+			pos:      &msgpb.MsgPosition{Timestamp: 0},
+			expected: false,
+		},
+		{
+			// A typical Milvus TSO-encoded timestamp (physical ms shifted left 18 bits).
+			name:     "normal recent timestamp is not sentinel",
+			pos:      &msgpb.MsgPosition{Timestamp: 450000000000000000},
+			expected: false,
+		},
+		{
+			name:     "exact sentinel timestamp is sentinel",
+			pos:      &msgpb.MsgPosition{Timestamp: math.MaxUint64},
+			expected: true,
+		},
+		{
+			name:     "sentinel timestamp with MsgID set is still sentinel",
+			pos:      &msgpb.MsgPosition{Timestamp: math.MaxUint64, MsgID: []byte{1, 2, 3}},
+			expected: true,
+		},
+		{
+			name:     "timestamp one below sentinel is not sentinel",
+			pos:      &msgpb.MsgPosition{Timestamp: math.MaxUint64 - 1},
+			expected: false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, IsDroppedChannelCheckpoint(tc.pos))
+		})
+	}
+}
+
+func TestDroppedChannelCheckpointTimestamp(t *testing.T) {
+	// Contract: the sentinel value is math.MaxUint64. Anything else would
+	// break MarkChannelCheckpointDropped's intent that the sentinel can
+	// never be overwritten by a normal checkpoint update.
+	assert.Equal(t, uint64(math.MaxUint64), DroppedChannelCheckpointTimestamp)
+}

pkg/util/merr/errors.go

@@ -100,6 +100,7 @@ var (
 	ErrChannelNotAvailable     = newMilvusError("channel not available", 503, false)
 	ErrChannelCPExceededMaxLag = newMilvusError("channel checkpoint exceed max lag", 504, false)
 	ErrChannelTSafeStalled     = newMilvusError("channel tsafe stalled", 505, true)
+	ErrChannelDroppedSentinel  = newMilvusError("channel checkpoint is dropped sentinel", 506, false)
 
 	// Segment related
 	ErrSegmentNotFound    = newMilvusError("segment not found", 600, false)

pkg/util/merr/errors_test.go

@@ -111,6 +111,10 @@ func (s *ErrSuite) TestWrap() {
 	s.ErrorIs(WrapErrChannelNotFound("test_Channel", "failed to get Channel"), ErrChannelNotFound)
 	s.ErrorIs(WrapErrChannelLack("test_Channel", "failed to get Channel"), ErrChannelLack)
 	s.ErrorIs(WrapErrChannelReduplicate("test_Channel", "failed to get Channel"), ErrChannelReduplicate)
+	s.ErrorIs(WrapErrChannelDroppedSentinel("test_Channel", "refuse to build next target"), ErrChannelDroppedSentinel)
+	// Dropped-sentinel and not-available are distinct: matching one must not match the other.
+	s.False(errors.Is(WrapErrChannelDroppedSentinel("test_Channel"), ErrChannelNotAvailable))
+	s.False(errors.Is(WrapErrChannelNotAvailable("test_Channel"), ErrChannelDroppedSentinel))
 
 	// Segment related
 	s.ErrorIs(WrapErrSegmentNotFound(1, "failed to get Segment"), ErrSegmentNotFound)

pkg/util/merr/utils.go

@@ -191,7 +191,12 @@ func oldCode(code int32) commonpb.ErrorCode {
 	case ErrPartitionNotFound.code(), ErrReplicaNotFound.code():
 		return commonpb.ErrorCode_MetaFailed
 
-	case ErrReplicaNotAvailable.code(), ErrChannelNotAvailable.code(), ErrNodeNotAvailable.code():
+	case ErrReplicaNotAvailable.code(), ErrChannelNotAvailable.code(), ErrChannelDroppedSentinel.code(), ErrNodeNotAvailable.code():
+		// ErrChannelDroppedSentinel is an internal-only signal that is currently
+		// swallowed by the alter-load-config ack callback and never serialized to a
+		// client Status. It is mapped alongside its ErrChannelNotAvailable sibling so
+		// that, if it is ever surfaced to a client, an old SDK keeps seeing
+		// NoReplicaAvailable instead of falling through to UnexpectedError.
 		return commonpb.ErrorCode_NoReplicaAvailable
 
 	case ErrServiceMemoryLimitExceeded.code():
@@ -771,6 +776,10 @@ func WrapErrChannelNotAvailable(name string, msg ...string) error {
 	return warpChannelErr(ErrChannelNotAvailable, name, msg...)
 }
 
+func WrapErrChannelDroppedSentinel(name string, msg ...string) error {
+	return warpChannelErr(ErrChannelDroppedSentinel, name, msg...)
+}
+
 // Segment related
 func WrapErrSegmentNotFound(id int64, msg ...string) error {
 	err := wrapFields(ErrSegmentNotFound, value("segment", id))

pkg/util/paramtable/component_param.go

@@ -3483,9 +3483,10 @@ type queryNodeConfig struct {
 	DeleteBufferBlockSize  ParamItem `refreshable:"false"`
 
 	// delta forward
-	LevelZeroForwardPolicy      ParamItem `refreshable:"true"`
-	StreamingDeltaForwardPolicy ParamItem `refreshable:"true"`
-	ForwardBatchSize            ParamItem `refreshable:"true"`
+	LevelZeroForwardPolicy             ParamItem `refreshable:"true"`
+	StreamingDeltaForwardPolicy        ParamItem `refreshable:"true"`
+	ForwardBatchSize                   ParamItem `refreshable:"true"`
+	DelegatorPostLoadConcurrencyFactor ParamItem `refreshable:"true"`
 
 	// loader
 	IoPoolSize                  ParamItem `refreshable:"false"`
@@ -4572,6 +4573,23 @@ Max read concurrency must greater than or equal to 1, and less than or equal to
 	}
 	p.ForwardBatchSize.Init(base.mgr)
 
+	p.DelegatorPostLoadConcurrencyFactor = ParamItem{
+		Key:          "queryNode.delegatorPostLoadConcurrencyFactor",
+		Version:      "2.6.16",
+		Doc:          "delegator post-load concurrency factor after worker LoadSegments returns. Concurrency is hardware.GetCPUNum * factor",
+		DefaultValue: "1",
+		Formatter: func(v string) string {
+			factor := getAsInt(v)
+			if factor < 1 {
+				factor = 1
+			}
+			concurrency := hardware.GetCPUNum() * factor
+			return strconv.FormatInt(int64(concurrency), 10)
+		},
+		Export: true,
+	}
+	p.DelegatorPostLoadConcurrencyFactor.Init(base.mgr)
+
 	p.IoPoolSize = ParamItem{
 		Key:          "queryNode.ioPoolSize",
 		Version:      "2.3.0",

pkg/util/paramtable/component_param_test.go

@@ -539,6 +539,13 @@ func TestComponentParam(t *testing.T) {
 		assert.Equal(t, 3*time.Second, Params.LazyLoadRequestResourceRetryInterval.GetAsDuration(time.Millisecond))
 
 		assert.Equal(t, 2, Params.BloomFilterApplyParallelFactor.GetAsInt())
+		assert.Equal(t, hardware.GetCPUNum(), Params.DelegatorPostLoadConcurrencyFactor.GetAsInt())
+		params.Save(Params.DelegatorPostLoadConcurrencyFactor.Key, "2")
+		assert.Equal(t, hardware.GetCPUNum()*2, Params.DelegatorPostLoadConcurrencyFactor.GetAsInt())
+		params.Save(Params.DelegatorPostLoadConcurrencyFactor.Key, "0")
+		assert.Equal(t, hardware.GetCPUNum(), Params.DelegatorPostLoadConcurrencyFactor.GetAsInt())
+		params.Reset(Params.DelegatorPostLoadConcurrencyFactor.Key)
+
 		assert.Equal(t, true, Params.SkipGrowingSegmentBF.GetAsBool())
 		assert.Equal(t, true, Params.EnableSegmentFilter.GetAsBool())
 

tests/integration/replication/data_salvage_test.go

@@ -265,3 +265,31 @@ func (s *DataSalvageSuite) TestDumpMessagesWithMissingStartMessageId() {
 
 	s.Error(err)
 }
+
+// TestDumpMessagesWithMalformedStartMessageId verifies that DumpMessages
+// returns an error (instead of panicking and crashing the process) when
+// start_message_id is non-empty but cannot be unmarshaled. See issue #50341.
+func (s *DataSalvageSuite) TestDumpMessagesWithMalformedStartMessageId() {
+	ctx := context.Background()
+
+	stream, err := s.Cluster.MilvusClient.DumpMessages(ctx, &milvuspb.DumpMessagesRequest{
+		Pchannel: "test-pchannel",
+		StartMessageId: &commonpb.MessageID{
+			Id:      "not-a-valid-base64-msgid!!!", // non-empty but undecodable
+			WALName: commonpb.WALName_Pulsar,
+		},
+	})
+
+	// The error might come during stream creation or first Recv()
+	if err == nil {
+		_, err = stream.Recv()
+	}
+
+	s.Error(err)
+
+	// The server must stay up: a follow-up RPC should still succeed.
+	_, err = s.Cluster.MilvusClient.GetReplicateInfo(ctx, &milvuspb.GetReplicateInfoRequest{
+		TargetPchannel: "test-pchannel",
+	})
+	s.NoError(err)
+}