enhance: batch broadcast tombstone GC

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

@@ -17,6 +17,8 @@ Non-primary clusters reject all broadcasts with `ErrNotPrimary`.
 5. **AckCallback**: CChannel ACK enqueues the task into `ackCallbackScheduler`. The callback executes only after all VChannels are ACKed. For tasks with conflicting ResourceKeys, callbacks execute in CChannel TimeTick order. Callbacks retry with exponential backoff until success.
 6. **Tombstone & GC**: After callbacks complete and TOMBSTONE is persisted, release the callback resource locks before handing the task to `tombstoneScheduler`. Handoff appends to an in-memory queue and coalesces wakeups; catalog deletion runs outside the queue lock. Every queued ID is retained until drained. Recovery rebuilds this queue from durable TOMBSTONE tasks if shutdown interrupts handoff. GC applies the existing count and lifetime limits; a sustained deletion deficit can still grow the queue.
 
+GC removes eligible tombstones in batches bounded by `metastore.maxEtcdTxnNum` (default 64), using exact broadcast-task keys. Each successful batch retires its in-memory tasks and advances the queue; failures retain the batch for idempotent retry, including when the deletion result is ambiguous. Late ACKs on TOMBSTONE or DONE tasks are ignored, so GC deletion holds neither task nor manager locks. Manager shutdown cancels an in-flight deletion; recovery only enqueues records still present in the catalog and does not replay their completed callbacks.
+
 ## Resource Key Locking
 
 Each ResourceKey has: **Domain** (resource type), **Key** (entity identifier), **Shared** (read vs exclusive). Every broadcast automatically acquires SharedCluster.

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 := 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,118 @@
+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")
+}

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/streamingcoord/server/broadcaster/ack_callback_gc_order_test.go

@@ -0,0 +1,169 @@
+package broadcaster
+
+import (
+	"context"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/milvus-io/milvus/internal/mocks/mock_metastore"
+	"github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster/registry"
+	"github.com/milvus-io/milvus/internal/streamingcoord/server/resource"
+	"github.com/milvus-io/milvus/pkg/v3/mlog"
+	"github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
+	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
+)
+
+// Exercises the real ACK scheduler and load/release message types with controlled
+// business callbacks. It deliberately overlaps a retry, blocked persistence,
+// queued same-key successors, and blocked GC catalog deletion.
+func TestAckCallbacksSameCollectionRemainOrderedWithBatchGC(t *testing.T) {
+	configureTombstoneGCTest(t)
+	registry.ResetRegistration()
+	defer registry.ResetRegistration()
+	s := newAckCallbackScheduler(mlog.With())
+	bm := newTombstoneGCTestManager()
+	gcStarted := make(chan struct{})
+	gcGate := make(chan struct{})
+	persistStarted := make(chan struct{})
+	persistGate := make(chan struct{})
+	releaseGC := sync.OnceFunc(func() { close(gcGate) })
+	releasePersist := sync.OnceFunc(func() { close(persistGate) })
+	defer releaseGC()
+	defer releasePersist()
+
+	var eventsMu sync.Mutex
+	var appliedIDs []uint64
+	var appliedStates []bool
+	var active atomic.Int32
+	var firstAttempts atomic.Int32
+	apply := func(id uint64, loaded bool) error {
+		if active.Add(1) != 1 {
+			t.Error("same-collection business callbacks overlapped")
+		}
+		defer active.Add(-1)
+		if id == 1 && firstAttempts.Add(1) == 1 {
+			return context.DeadlineExceeded
+		}
+		eventsMu.Lock()
+		appliedIDs = append(appliedIDs, id)
+		appliedStates = append(appliedStates, loaded)
+		eventsMu.Unlock()
+		return nil
+	}
+	registry.RegisterAlterLoadConfigV2AckCallback(func(_ context.Context, result message.BroadcastResultAlterLoadConfigMessageV2) error {
+		return apply(result.Message.BroadcastHeader().BroadcastID, true)
+	})
+	registry.RegisterDropLoadConfigV2AckCallback(func(_ context.Context, result message.BroadcastResultDropLoadConfigMessageV2) error {
+		return apply(result.Message.BroadcastHeader().BroadcastID, false)
+	})
+
+	meta := mock_metastore.NewMockStreamingCoordCataLog(t)
+	meta.EXPECT().SaveBroadcastTask(mock.Anything, mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, id uint64, task *streamingpb.BroadcastTask) error {
+			if id == 1 && task.State == streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE {
+				close(persistStarted)
+				select {
+				case <-persistGate:
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			}
+			return nil
+		})
+	meta.EXPECT().RemoveBroadcastTasks(mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, ids []uint64) error {
+			if len(ids) == 1 && ids[0] == 0 {
+				close(gcStarted)
+				select {
+				case <-gcGate:
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			}
+			return nil
+		})
+	resource.InitForTest(resource.OptStreamingCatalog(meta))
+
+	rk := message.NewExclusiveCollectionNameResourceKey("db", "same_collection")
+	tasks := make([]*broadcastTask, 0, 3)
+	for id := uint64(0); id <= 3; id++ {
+		var msg message.BroadcastMutableMessage
+		if id == 2 {
+			msg = message.NewDropLoadConfigMessageBuilderV2().
+				WithHeader(&message.DropLoadConfigMessageHeader{CollectionId: 42}).
+				WithBody(&message.DropLoadConfigMessageBody{}).
+				WithBroadcast([]string{"by-dev-0_vcchan"}).MustBuildBroadcast()
+		} else {
+			msg = message.NewAlterLoadConfigMessageBuilderV2().
+				WithHeader(&message.AlterLoadConfigMessageHeader{CollectionId: 42}).
+				WithBody(&message.AlterLoadConfigMessageBody{}).
+				WithBroadcast([]string{"by-dev-0_vcchan"}).MustBuildBroadcast()
+		}
+		msg = msg.OverwriteBroadcastHeader(id, rk)
+		state := streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_PENDING
+		if id == 0 {
+			state = streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE
+		}
+		p := createNewWaitAckBroadcastTaskFromMessage(msg, state, []byte{1})
+		p.AckedCheckpoints[0].TimeTick = id + 100
+		task := newBroadcastTaskFromProto(p, newBroadcasterMetrics(), s)
+		task.SetLogger(mlog.With())
+		bm.tasks[id] = task
+		if id != 0 {
+			tasks = append(tasks, task)
+		}
+	}
+	s.Initialize([]*broadcastTask{tasks[2], tasks[0], tasks[1]}, []uint64{0}, bm)
+	defer func() {
+		releaseGC()
+		releasePersist()
+		s.Close()
+	}()
+	for _, gate := range []<-chan struct{}{gcStarted, persistStarted} {
+		select {
+		case <-gate:
+		case <-time.After(5 * time.Second):
+			t.Fatal("expected blocked phase was not reached")
+		}
+	}
+
+	guards, ok := s.rkLocker.TryLock(rk)
+	if ok {
+		guards.Unlock()
+		t.Fatal("ACK lock released before durable completion")
+	}
+	eventsMu.Lock()
+	ids := slices.Clone(appliedIDs)
+	eventsMu.Unlock()
+	require.Equal(t, []uint64{1}, ids)
+	releasePersist()
+
+	// GC is still blocked, but all three same-collection operations must finish
+	// business execution in WAL order and enqueue all of their GC work.
+	require.Eventually(t, func() bool {
+		s.tombstoneScheduler.pendingMu.Lock()
+		defer s.tombstoneScheduler.pendingMu.Unlock()
+		return len(s.tombstoneScheduler.pending) == 3
+	}, 5*time.Second, time.Millisecond)
+	eventsMu.Lock()
+	ids = slices.Clone(appliedIDs)
+	states := slices.Clone(appliedStates)
+	eventsMu.Unlock()
+	require.Equal(t, []uint64{1, 2, 3}, ids)
+	require.Equal(t, []bool{true, false, true}, states)
+	require.EqualValues(t, 2, firstAttempts.Load())
+	require.Zero(t, active.Load())
+
+	releaseGC()
+	require.Eventually(t, func() bool {
+		bm.mu.Lock()
+		defer bm.mu.Unlock()
+		return len(bm.tasks) == 0
+	}, 5*time.Second, time.Millisecond)
+}