enhance: streamline broadcaster ACK and tombstone GC (#54015)

Delete broadcaster tombstones in batches: a GC pass marks the selected
tasks DONE and removes their keys with one catalog call,
`RemoveBroadcastTasks`, which commits them in transactions of at most
`metastore.maxEtcdTxnNum` keys instead of one etcd write per tombstone.
`FastLock` fails with the plain sentinel error, and the ACK scheduler
logs one rate-limited summary of blocked tasks instead of a warning per
blocked task on every pass. The native metrics text parsing also moves
into a helper with tests; its behavior is unchanged. ACK wakeup
coalescing, late-ACK suppression and the checkpoint-gated tombstone GC
landed separately in #54319 and #54432.

issue: #40451

---------

Signed-off-by: sunby <sunbingyi1992@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Codex <noreply@openai.com>

docs/agent_guides/streaming-system/coordination/broadcaster.md

@@ -23,6 +23,8 @@ exchange; task inspection, resource locking, callbacks, and GC run outside that
 mutex. ACK enqueue and callback completion send coalesced, nonblocking wakeups.
 The ACK loop drains its inbox before each scheduling pass, appending new tasks
 after existing blocked tasks to preserve FIFO scheduling for conflicting keys.
+Each pass retries every blocked task: a held key fails `FastLock` with a plain
+sentinel error, and the pass logs one rate-limited summary of blocked tasks.
 Tombstone GC remains timer-driven.
 
 ## Idempotent Broadcast
@@ -122,9 +124,14 @@ do not count toward `maxCount`, so the total retained count can exceed that limi
 Expired eligible tombstones leave the quota before count eviction is selected,
 avoiding unnecessary eviction of younger tombstones. Count eviction orders by
 BroadcastID, not callback completion or enqueue time. A blocked tombstone
-does not prevent eligible, unrelated tombstones from being collected. Failed
-catalog deletion retains the entry for retry. Task and idempotency-index removal
-continue through the existing DONE transition.
+does not prevent eligible, unrelated tombstones from being collected.
+A pass marks the selected tasks DONE and deletes their keys with one catalog
+call, which commits them in transactions of at most `metastore.maxEtcdTxnNum`
+keys instead of one etcd write per tombstone. A save racing the deletion writes
+DONE, which the catalog turns into a removal. The catalog retries until its
+context ends, so deletion fails only on shutdown; the entries are then retained
+and recovery collects the keys that remain. Task and idempotency-index removal
+follow a successful deletion.
 
 The existing retention clock starts when the scheduler receives the persisted
 TOMBSTONE. It is not stored in the task proto: coordinator recovery starts a new

internal/metastore/kv/streamingcoord/kv_catalog.go

@@ -203,6 +203,20 @@ func (c *catalog) SaveBroadcastTask(ctx context.Context, broadcastID uint64, tas
 	return c.metaKV.Save(ctx, key, string(v))
 }
 
+func (c *catalog) RemoveBroadcastTasks(ctx context.Context, broadcastIDs []uint64) error {
+	keys := make([]string, 0, len(broadcastIDs))
+	for _, id := range broadcastIDs {
+		keys = append(keys, buildBroadcastTaskPath(id))
+	}
+	maxTxnNum := max(1, paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt())
+	return etcd.RemoveByBatchWithLimit(keys, maxTxnNum, func(batch []string) error {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		return c.metaKV.MultiRemove(ctx, batch)
+	})
+}
+
 // buildPChannelInfoPath builds the path for pchannel info.
 func buildPChannelInfoPath(name string) string {
 	return PChannelMetaPrefix + name

internal/metastore/kv/streamingcoord/kv_catalog_batch_test.go

@@ -0,0 +1,130 @@
+package streamingcoord
+
+import (
+	"context"
+	"slices"
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
+	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
+)
+
+func setBroadcastDeletionBatchSize(t *testing.T, value string) {
+	t.Helper()
+	paramtable.Init()
+	limit := &paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum
+	previous := limit.SwapTempValue(value)
+	t.Cleanup(func() { limit.SwapTempValue(previous) })
+}
+
+func TestRemoveBroadcastTasksUsesBoundedExactKeys(t *testing.T) {
+	setBroadcastDeletionBatchSize(t, "2")
+	catalog, stored, kv := newTestCatalog(t)
+	ctx := context.Background()
+	ids := []uint64{1, 2, 3, 4, 5}
+	for _, id := range append(slices.Clone(ids), 11) {
+		require.NoError(t, catalog.SaveBroadcastTask(ctx, id, &streamingpb.BroadcastTask{
+			State: streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE,
+		}))
+	}
+	stored["querycoord-collection-loadinfo/1"] = "keep"
+	var batches [][]string
+	kv.EXPECT().MultiRemove(mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, keys []string) error {
+			batches = append(batches, slices.Clone(keys))
+			for _, key := range keys {
+				delete(stored, key)
+			}
+			return nil
+		}).Times(3)
+	require.NoError(t, catalog.RemoveBroadcastTasks(ctx, nil))
+	require.NoError(t, catalog.RemoveBroadcastTasks(ctx, ids))
+	require.Equal(t, [][]string{
+		{buildBroadcastTaskPath(1), buildBroadcastTaskPath(2)},
+		{buildBroadcastTaskPath(3), buildBroadcastTaskPath(4)},
+		{buildBroadcastTaskPath(5)},
+	}, batches)
+	require.Len(t, stored, 2)
+	require.Contains(t, stored, buildBroadcastTaskPath(11), "deletion must not match key prefixes")
+	require.Equal(t, "keep", stored["querycoord-collection-loadinfo/1"])
+}
+
+func TestRemoveBroadcastTasksRetriesLostCommitResponse(t *testing.T) {
+	setBroadcastDeletionBatchSize(t, "2")
+	catalog, stored, kv := newTestCatalog(t)
+	ids := []uint64{1, 2, 3}
+	for _, id := range ids {
+		stored[buildBroadcastTaskPath(id)] = "tombstone"
+	}
+	var attempts [][]string
+	kv.EXPECT().MultiRemove(mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, keys []string) error {
+			attempts = append(attempts, slices.Clone(keys))
+			for _, key := range keys {
+				delete(stored, key)
+			}
+			if len(attempts) == 1 {
+				return merr.WrapErrServiceUnavailable("commit response lost")
+			}
+			return nil
+		}).Times(3)
+	require.NoError(t, catalog.RemoveBroadcastTasks(context.Background(), ids))
+	require.Equal(t, attempts[0], attempts[1], "the real reliable-write wrapper must retry the same deletion")
+	require.Equal(t, []string{buildBroadcastTaskPath(3)}, attempts[2])
+	require.Empty(t, stored)
+}
+
+func TestRemoveBroadcastTasksCancellationAfterPartialProgress(t *testing.T) {
+	setBroadcastDeletionBatchSize(t, "2")
+	catalog, stored, kv := newTestCatalog(t)
+	ids := []uint64{1, 2, 3, 4, 5}
+	for _, id := range ids {
+		require.NoError(t, catalog.SaveBroadcastTask(context.Background(), id, &streamingpb.BroadcastTask{
+			State: streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE,
+		}))
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var attempts [][]string
+	kv.EXPECT().MultiRemove(mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, keys []string) error {
+			attempts = append(attempts, slices.Clone(keys))
+			if len(attempts) == 2 {
+				cancel()
+				return ctx.Err()
+			}
+			for _, key := range keys {
+				delete(stored, key)
+			}
+			return nil
+		})
+	require.ErrorIs(t, catalog.RemoveBroadcastTasks(ctx, ids), context.Canceled)
+	require.Len(t, attempts, 2, "do not issue further batches after cancellation")
+	tasks, err := catalog.ListBroadcastTask(context.Background())
+	require.NoError(t, err)
+	require.Len(t, tasks, 3)
+	for _, task := range tasks {
+		require.Equal(t, streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE, task.State)
+	}
+	require.NoError(t, catalog.RemoveBroadcastTasks(context.Background(), ids))
+	require.Equal(t, attempts[0], attempts[2], "retry tolerates already removed IDs")
+	require.Empty(t, stored)
+	require.ErrorIs(t, catalog.RemoveBroadcastTasks(ctx, ids), context.Canceled)
+	require.Len(t, attempts, 5, "an already canceled request must not access the KV store")
+}
+
+func TestRemoveBroadcastTasksUsesPositiveBatchLimit(t *testing.T) {
+	for _, limit := range []string{"0", "-1"} {
+		t.Run(limit, func(t *testing.T) {
+			setBroadcastDeletionBatchSize(t, limit)
+			catalog, _, kv := newTestCatalog(t)
+			kv.EXPECT().MultiRemove(mock.Anything, []string{buildBroadcastTaskPath(1)}).
+				Return(nil).Once()
+			require.NoError(t, catalog.RemoveBroadcastTasks(context.Background(), []uint64{1}))
+		})
+	}
+}

internal/metastore/streamingcoord_catalog.go

@@ -43,6 +43,10 @@ type StreamingCoordCataLog interface {
 	// Only return error if the ctx is canceled, otherwise it will retry until success.
 	SaveBroadcastTask(ctx context.Context, broadcastID uint64, task *streamingpb.BroadcastTask) error
 
+	// RemoveBroadcastTasks removes completed tombstones with distinct IDs in bounded batches.
+	// An error may follow partially completed deletion; retrying the same IDs is safe.
+	RemoveBroadcastTasks(ctx context.Context, broadcastIDs []uint64) error
+
 	// SaveReplicateConfiguration saves the replicate configuration to metastore.
 	// Only return error if the ctx is canceled, otherwise it will retry until success.
 	SaveReplicateConfiguration(ctx context.Context, config *streamingpb.ReplicateConfigurationMeta, replicatingTasks []*streamingpb.ReplicatePChannelMeta) error

internal/mocks/mock_metastore/mock_StreamingCoordCataLog.go

@@ -5,9 +5,8 @@ package mock_metastore
 import (
 	context "context"
 
-	mock "github.com/stretchr/testify/mock"
-
 	streamingpb "github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
+	mock "github.com/stretchr/testify/mock"
 )
 
 // MockStreamingCoordCataLog is an autogenerated mock type for the StreamingCoordCataLog type
@@ -313,6 +312,53 @@ func (_c *MockStreamingCoordCataLog_ListPChannel_Call) RunAndReturn(run func(con
 	return _c
 }
 
+// RemoveBroadcastTasks provides a mock function with given fields: ctx, broadcastIDs
+func (_m *MockStreamingCoordCataLog) RemoveBroadcastTasks(ctx context.Context, broadcastIDs []uint64) error {
+	ret := _m.Called(ctx, broadcastIDs)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RemoveBroadcastTasks")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, []uint64) error); ok {
+		r0 = rf(ctx, broadcastIDs)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// MockStreamingCoordCataLog_RemoveBroadcastTasks_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoveBroadcastTasks'
+type MockStreamingCoordCataLog_RemoveBroadcastTasks_Call struct {
+	*mock.Call
+}
+
+// RemoveBroadcastTasks is a helper method to define mock.On call
+//   - ctx context.Context
+//   - broadcastIDs []uint64
+func (_e *MockStreamingCoordCataLog_Expecter) RemoveBroadcastTasks(ctx interface{}, broadcastIDs interface{}) *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call {
+	return &MockStreamingCoordCataLog_RemoveBroadcastTasks_Call{Call: _e.mock.On("RemoveBroadcastTasks", ctx, broadcastIDs)}
+}
+
+func (_c *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call) Run(run func(ctx context.Context, broadcastIDs []uint64)) *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].([]uint64))
+	})
+	return _c
+}
+
+func (_c *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call) Return(_a0 error) *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call) RunAndReturn(run func(context.Context, []uint64) error) *MockStreamingCoordCataLog_RemoveBroadcastTasks_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // SaveBroadcastTask provides a mock function with given fields: ctx, broadcastID, task
 func (_m *MockStreamingCoordCataLog) SaveBroadcastTask(ctx context.Context, broadcastID uint64, task *streamingpb.BroadcastTask) error {
 	ret := _m.Called(ctx, broadcastID, task)

internal/streamingcoord/server/broadcaster/ack_callback_scheduler.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"go.opentelemetry.io/otel/codes"
+	"golang.org/x/time/rate"
 
 	"github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster/registry"
 	"github.com/milvus-io/milvus/pkg/v3/mlog"
@@ -170,7 +171,6 @@ func (s *ackCallbackScheduler) triggerAckCallback() {
 
 		g, err := s.rkLocker.FastLock(task.Header().ResourceKeys.Collect()...)
 		if err != nil {
-			s.Logger().Warn(context.TODO(), "lock is occupied, delay the ack callback", mlog.FieldBroadcastID(task.Header().BroadcastID), mlog.Err(err))
 			pendingTasks = append(pendingTasks, task)
 			continue
 		}
@@ -179,6 +179,13 @@ func (s *ackCallbackScheduler) triggerAckCallback() {
 		go s.doAckCallback(task, g)
 	}
 	s.pendingAckedTasks = pendingTasks
+	if len(pendingTasks) > 0 {
+		// Blocked tasks are retried on every wakeup: log a rate-limited summary,
+		// not one line per task.
+		s.Logger().RatedWarn(s.notifier.Context(), rate.Every(time.Second), "lock is occupied, delay the ack callbacks",
+			mlog.Int("blockedTaskCount", len(pendingTasks)),
+			mlog.Uint64("firstBlockedBroadcastID", pendingTasks[0].Header().BroadcastID))
+	}
 }
 
 // doForcePromoteFixIncompleteBroadcasts handles the full force promote lifecycle:

internal/streamingcoord/server/broadcaster/broadcast_manager.go

@@ -346,26 +346,36 @@ func (bm *broadcastTaskManager) Ack(ctx context.Context, msg message.ImmutableMe
 	return t.Ack(ctx, msg)
 }
 
-// DropTombstone drops the tombstone task from the manager.
-func (bm *broadcastTaskManager) DropTombstone(ctx context.Context, broadcastID uint64, checkpoints map[string]*streamingpb.WALCheckpoint) (bool, error) {
+// DropTombstones drops the tombstones of the broadcast tasks with one catalog
+// call, which splits the keys into bounded transactions, and then removes the
+// tasks from the manager. After the tombstones are dropped, the idempotency and
+// deduplication can not be guaranteed.
+func (bm *broadcastTaskManager) DropTombstones(ctx context.Context, broadcastIDs []uint64) error {
 	if !bm.lifetime.Add(typeutil.LifetimeStateWorking) {
-		return false, status.NewOnShutdownError("broadcaster is closing")
+		return status.NewOnShutdownError("broadcaster is closing")
 	}
 	defer bm.lifetime.Done()
+	ctx, cancel := contextutil.MergeContext(ctx, bm.ctx)
+	defer cancel()
 
-	t, ok := bm.getBroadcastTaskByID(broadcastID)
-	if !ok {
-		bm.Logger().Debug(ctx, "task is not found, ignored the drop tombstone request", mlog.FieldBroadcastID(broadcastID))
-		return true, nil
+	bm.mu.Lock()
+	tasks := make([]*broadcastTask, 0, len(broadcastIDs))
+	for _, id := range broadcastIDs {
+		if t, ok := bm.tasks[id]; ok {
+			tasks = append(tasks, t)
+		}
 	}
-	if !t.canDropTombstone(checkpoints) {
-		return false, nil
+	bm.mu.Unlock()
+	// Mark DONE before deleting: a save that races the deletion then writes DONE,
+	// which the catalog turns into a removal.
+	for _, t := range tasks {
+		t.markTombstoneDropped()
 	}
-	if err := t.DropTombstone(ctx); err != nil {
-		return false, err
+	if err := resource.Resource().StreamingCatalog().RemoveBroadcastTasks(ctx, broadcastIDs); err != nil {
+		return err
 	}
-	bm.removeBroadcastTask(broadcastID)
-	return true, nil
+	bm.removeBroadcastTask(broadcastIDs...)
+	return nil
 }
 
 // Close cancels in-flight requests before waiting for them to exit. An ACK
@@ -473,15 +483,17 @@ func (bm *broadcastTaskManager) getBroadcastTaskByID(broadcastID uint64) (*broad
 	return t, ok
 }
 
-// removeBroadcastTask removes the broadcast task by the broadcastID.
-func (bm *broadcastTaskManager) removeBroadcastTask(broadcastID uint64) {
+// removeBroadcastTask removes the broadcast tasks by the broadcastIDs.
+func (bm *broadcastTaskManager) removeBroadcastTask(broadcastIDs ...uint64) {
 	bm.mu.Lock()
 	defer bm.mu.Unlock()
 
-	if t, ok := bm.tasks[broadcastID]; ok {
-		bm.idempotencyIndex.Remove(t.IdempotencyScope(), broadcastID)
+	for _, broadcastID := range broadcastIDs {
+		if t, ok := bm.tasks[broadcastID]; ok {
+			bm.idempotencyIndex.Remove(t.IdempotencyScope(), broadcastID)
+		}
+		delete(bm.tasks, broadcastID)
 	}
-	delete(bm.tasks, broadcastID)
 }
 
 // getIncompleteBroadcastTasks returns all incomplete broadcast tasks that have pending messages.

internal/streamingcoord/server/broadcaster/broadcast_task.go

@@ -273,12 +273,14 @@ func (b *broadcastTask) MessageTypeWithVersion() message.MessageTypeWithVersion
 }
 
 // PendingSchemaFileResourceSnapshot returns the collection and file resource
-// IDs referenced by a non-tombstone CreateCollection or AlterCollection task.
+// IDs referenced by a CreateCollection or AlterCollection task that is neither
+// a tombstone nor a tombstone being dropped (DONE).
 func (b *broadcastTask) PendingSchemaFileResourceSnapshot() (int64, []int64, bool) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	if b.task.State == streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE {
+	if b.task.State == streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE ||
+		b.task.State == streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_DONE {
 		return 0, nil, false
 	}
 	switch b.messageTypeWithVersion {
@@ -597,16 +599,17 @@ func (b *broadcastTask) canDropTombstone(checkpoints map[string]*streamingpb.WAL
 	return true
 }
 
-// DropTombstone drops the tombstone of the broadcast task.
-// It will remove the tombstone of the broadcast task in recovery storage.
-// After the tombstone is dropped, the idempotency and deduplication can not be guaranteed.
-func (b *broadcastTask) DropTombstone(ctx context.Context) error {
+// markTombstoneDropped marks the task DONE before its tombstone is deleted from
+// recovery storage.
+func (b *broadcastTask) markTombstoneDropped() {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if b.task.State == streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_DONE {
+		return // A retry after a failed deletion.
+	}
 	b.task.State = streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_DONE
-	b.dirty = true
-	return b.saveTaskIfDirty(ctx, b.Logger())
+	b.ObserveStateChanged(b.task.State)
 }
 
 // isAllDone check if all the vchannels are acked.

internal/streamingcoord/server/broadcaster/broadcaster_test.go

@@ -500,6 +500,8 @@ func TestGetPendingSchemaFileResources(t *testing.T) {
 			3: newTask(3, alterCollectionMsg(200, nil), streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_PENDING),
 			4: newTask(4, alterCollectionMsg(300, []int64{40}), streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE),
 			5: newTask(5, createNewBroadcastMsg([]string{"v1"}), streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_PENDING),
+			// A tombstone being dropped is marked DONE until it leaves the manager.
+			6: newTask(6, createCollectionMsg(400, []int64{50}), streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_DONE),
 		},
 	}
 

internal/streamingcoord/server/broadcaster/resource_key_locker.go

@@ -8,7 +8,6 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/proto/messagespb"
 	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
 	"github.com/milvus-io/milvus/pkg/v3/util/lock"
-	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
@@ -100,7 +99,9 @@ func (r *resourceKeyLocker) FastLock(keys ...message.ResourceKey) (*lockGuards,
 			continue
 		}
 		g.Unlock()
-		return nil, merr.Wrapf(errFastLockFailed, "fast lock failed at resource key %s", key.String())
+		// The ACK scheduler retries a blocked task on every pass, so fail without
+		// building an error.
+		return nil, errFastLockFailed
 	}
 	return g, nil
 }

internal/streamingcoord/server/broadcaster/schema_test.go

@@ -149,7 +149,7 @@ func TestSchemaTargetPersistenceBoundary(t *testing.T) {
 		require.Contains(t, bm.tasks, uint64(1), "retirement preserves the tombstone and deduplication")
 		require.NoError(t, task.MarkAckCallbackDone(t.Context()))
 		require.Nil(t, task.schemaForTarget, "completion releases the task's cached schema")
-		require.NoError(t, task.DropTombstone(t.Context()))
+		task.markTombstoneDropped()
 		bm.removeBroadcastTask(1)
 		require.Zero(t, bm.schemaTaskCount(100))
 	})

internal/streamingcoord/server/broadcaster/tombstone_scheduler.go

@@ -2,6 +2,7 @@ package broadcaster
 
 import (
 	"context"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -159,24 +160,25 @@ func (s *tombstoneScheduler) triggerGCTombstone() {
 		toDrop[eligible[i]] = struct{}{}
 	}
 
-	retained := s.tombstones[:0]
-	for idx, tombstone := range s.tombstones {
-		if _, ok := toDrop[tombstone.broadcastID]; !ok {
-			retained = append(retained, tombstone)
-			continue
-		}
-		dropped, err := s.bm.DropTombstone(ctx, tombstone.broadcastID, checkpoints)
-		if err != nil {
-			s.Logger().Warn(ctx, "failed to drop tombstone", mlog.Err(err))
-			retained = append(retained, s.tombstones[idx:]...)
-			break
-		}
-		if !dropped {
-			// A lagging PChannel must not block unrelated eligible tombstones.
-			retained = append(retained, tombstone)
+	if len(toDrop) == 0 {
+		return
+	}
+	ids := make([]uint64, 0, len(toDrop))
+	for _, tombstone := range s.tombstones {
+		if _, ok := toDrop[tombstone.broadcastID]; ok {
+			ids = append(ids, tombstone.broadcastID)
 		}
 	}
-	s.tombstones = retained
+	// The catalog retries until the context ends, so a failure means shutdown:
+	// keep the entries, and recovery collects the keys that remain.
+	if err := s.bm.DropTombstones(ctx, ids); err != nil {
+		s.Logger().Warn(ctx, "failed to drop tombstones", mlog.Int("count", len(ids)), mlog.Err(err))
+		return
+	}
+	s.tombstones = slices.DeleteFunc(s.tombstones, func(tombstone tombstoneItem) bool {
+		_, ok := toDrop[tombstone.broadcastID]
+		return ok
+	})
 }
 
 // loadCheckpoints reads the durable recovery boundary once per PChannel per GC

internal/streamingcoord/server/broadcaster/tombstone_scheduler_test.go

@@ -279,10 +279,10 @@ func TestTombstoneGCFailuresAndLegacyPositions(t *testing.T) {
 	for _, target := range []struct {
 		receiver any
 		method   string
-	}{{coord, "ListPChannel"}, {node, "GetConsumeCheckpoint"}, {coord, "SaveBroadcastTask"}} {
+	}{{coord, "ListPChannel"}, {node, "GetConsumeCheckpoint"}, {coord, "RemoveBroadcastTasks"}} {
 		builder := mockey.Mock(mockey.GetMethod(target.receiver, target.method))
-		if target.method == "SaveBroadcastTask" {
-			builder.To(func(ctx context.Context, _ uint64, _ *streamingpb.BroadcastTask) error {
+		if target.method == "RemoveBroadcastTasks" {
+			builder.To(func(ctx context.Context, _ []uint64) error {
 				s.notifier.Cancel()
 				return ctx.Err()
 			})
@@ -304,9 +304,40 @@ func TestTombstoneGCFailuresAndLegacyPositions(t *testing.T) {
 	s.triggerGCTombstone()
 	require.Len(t, s.tombstones, 1)
 	require.Equal(t, uint64(2), s.tombstones[0].broadcastID, "legacy position is retained")
-	dropped, err := bm.DropTombstone(t.Context(), 99, nil)
+	require.NoError(t, bm.DropTombstones(t.Context(), []uint64{99}))
+}
+
+func TestTombstoneGCDeletesMoreThanOneTransaction(t *testing.T) {
+	coord, node := setupTombstoneGC(t)
+	limit := &paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum
+	old := limit.SwapTempValue("2")
+	t.Cleanup(func() { limit.SwapTempValue(old) })
+	ids := []uint64{1, 2, 3, 4, 5}
+	for _, id := range ids {
+		require.NoError(t, coord.SaveBroadcastTask(t.Context(), id, gcTestTombstone(id, []string{"p0_vcchan"}, false)))
+	}
+	saveGCCheckpoint(t, node, "p0", 1000, 1)
+	bc, err := RecoverBroadcaster(t.Context())
 	require.NoError(t, err)
-	require.True(t, dropped)
+	defer bc.Close()
+	bm := bc.(*broadcastTaskManager)
+	var tasks []*broadcastTask
+	for _, id := range ids {
+		task, ok := bm.getBroadcastTaskByID(id)
+		require.True(t, ok)
+		tasks = append(tasks, task)
+	}
+	s := gcTestScheduler(bm, ids...)
+	s.triggerGCTombstone()
+	require.Empty(t, s.tombstones)
+	persisted, err := coord.ListBroadcastTask(t.Context())
+	require.NoError(t, err)
+	require.Empty(t, persisted)
+	for i, task := range tasks {
+		_, ok := bm.getBroadcastTaskByID(ids[i])
+		require.False(t, ok)
+		require.Equal(t, streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_DONE, task.State())
+	}
 }
 
 func TestTombstoneGCReplayAndRecovery(t *testing.T) {

internal/util/metrics/c_registry.go

@@ -138,8 +138,6 @@ type CRegistry struct {
 
 // Gather implements Gatherer.
 func (r *CRegistry) Gather() (res []*dto.MetricFamily, err error) {
-	var parser expfmt.TextParser
-
 	r.mtx.RLock()
 	defer r.mtx.RUnlock()
 
@@ -147,7 +145,7 @@ func (r *CRegistry) Gather() (res []*dto.MetricFamily, err error) {
 	metricsStr := C.GoString(cMetricsStr)
 	C.free(unsafe.Pointer(cMetricsStr))
 
-	out, err := parser.TextToMetricFamilies(strings.NewReader(metricsStr))
+	out, err := parseTextMetrics(metricsStr)
 	if err != nil {
 		mlog.Error(context.TODO(), "fail to parse knowhere prometheus metrics", mlog.Err(err))
 		return res, err
@@ -157,7 +155,7 @@ func (r *CRegistry) Gather() (res []*dto.MetricFamily, err error) {
 	metricsStr = C.GoString(cMetricsStr)
 	C.free(unsafe.Pointer(cMetricsStr))
 
-	out1, err := parser.TextToMetricFamilies(strings.NewReader(metricsStr))
+	out1, err := parseTextMetrics(metricsStr)
 	if err != nil {
 		mlog.Error(context.TODO(), "fail to parse storage prometheus metrics", mlog.Err(err))
 		return res, err
@@ -175,6 +173,12 @@ func (r *CRegistry) Gather() (res []*dto.MetricFamily, err error) {
 	return res, err
 }
 
+func parseTextMetrics(metricsStr string) (map[string]*dto.MetricFamily, error) {
+	// prometheus/common v0.55.0 uses the zero-value parser with legacy name validation.
+	var parser expfmt.TextParser
+	return parser.TextToMetricFamilies(strings.NewReader(metricsStr))
+}
+
 // gatherJemallocMetrics collects comprehensive jemalloc stats and returns them as metric families.
 // Uses a 10-second cache to avoid expensive mallctl("epoch") calls on every Prometheus scrape.
 func gatherJemallocMetrics() map[string]*dto.MetricFamily {

internal/util/metrics/c_registry_test.go

@@ -0,0 +1,39 @@
+// 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 metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseTextMetrics(t *testing.T) {
+	const metricsText = `# HELP milvus_c_registry_test_total Test metric.
+# TYPE milvus_c_registry_test_total counter
+milvus_c_registry_test_total{source="core"} 1
+`
+
+	metricFamilies, err := parseTextMetrics(metricsText)
+	require.NoError(t, err)
+	require.Contains(t, metricFamilies, "milvus_c_registry_test_total")
+}
+
+func TestParseTextMetricsUsesLegacyValidation(t *testing.T) {
+	_, err := parseTextMetrics(`{"metric.name",source="core"} 1`)
+	require.Error(t, err)
+}