fix: bind file resource refCnt to collection lifecycle to prevent panic (#48893)

relate: #48612
## Summary
- Increment `fileResourceRefCnt` during `validateSchema` instead of in
the async ack callback's `AddCollection`, closing the TOCTOU race where
`RemoveFileResource` could delete a resource between validation and
`AddCollection`
- On failure before `Broadcast`, refCnt is decremented immediately; on
restart, refCnt for pending broadcast tasks is recovered from etcd
before rootcoord becomes Healthy
- Remove refCnt++ from `addCollectionMeta` since it's now done at
validation time (reload path unchanged)

---------

Signed-off-by: aoiasd <zhicheng.yue@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: aoiasd

internal/mocks/streamingcoord/server/mock_broadcaster/mock_Broadcaster.go

@@ -51,6 +51,19 @@ type createCollectionTask struct {
 	header          *message.CreateCollectionMessageHeader
 	body            *message.CreateCollectionRequest
 	preserveFieldID bool
+
+	// heldFileResourceIds tracks file resources whose refCnt was incremented
+	// during validation, to be released if the task fails before Broadcast.
+	heldFileResourceIds []int64
+}
+
+// releaseFileResources decrements refCnt for file resources that were
+// incremented during validation. Called when the task fails before Broadcast.
+func (t *createCollectionTask) releaseFileResources() {
+	if len(t.heldFileResourceIds) > 0 {
+		t.meta.DecFileResourceRefCnt(t.heldFileResourceIds)
+		t.heldFileResourceIds = nil
+	}
 }
 
 func (t *createCollectionTask) validate(ctx context.Context) error {
@@ -246,6 +259,15 @@ func (t *createCollectionTask) validateSchema(ctx context.Context, schema *schem
 			return err
 		}
 		schema.FileResourceIds = resp.GetResourceIds()
+
+		// Bind file resources to collection lifecycle: refCnt++ now, refCnt-- on
+		// drop. Under ddLock, atomic with RemoveFileResource. See #48612.
+		if len(schema.FileResourceIds) > 0 {
+			if err := t.meta.IncFileResourceRefCnt(schema.FileResourceIds); err != nil {
+				return err
+			}
+			t.heldFileResourceIds = schema.FileResourceIds
+		}
 	}
 
 	return validateFieldDataType(schema.GetFields())

internal/rootcoord/create_collection_task.go

@@ -1468,6 +1468,9 @@ func Test_createCollectionTask_prepareSchema(t *testing.T) {
 			ResourceIds: []int64{1, 2, 3},
 		}, nil)
 
+		meta := mockrootcoord.NewIMetaTable(t)
+		meta.EXPECT().IncFileResourceRefCnt(mock.Anything).Return(nil)
+
 		collectionName := funcutil.GenRandomStr()
 		field1 := funcutil.GenRandomStr()
 		field2 := funcutil.GenRandomStr()
@@ -1498,7 +1501,7 @@ func Test_createCollectionTask_prepareSchema(t *testing.T) {
 		assert.NoError(t, err)
 
 		task := createCollectionTask{
-			Core: newTestCore(withMixCoord(mixcoord)),
+			Core: newTestCore(withMixCoord(mixcoord), withMeta(meta)),
 			Req: &milvuspb.CreateCollectionRequest{
 				Base:           &commonpb.MsgBase{MsgType: commonpb.MsgType_CreateCollection},
 				CollectionName: collectionName,
@@ -1522,6 +1525,9 @@ func Test_createCollectionTask_prepareSchema(t *testing.T) {
 			ResourceIds: []int64{1, 2, 3},
 		}, nil)
 
+		meta := mockrootcoord.NewIMetaTable(t)
+		meta.EXPECT().IncFileResourceRefCnt(mock.Anything).Return(nil)
+
 		collectionName := funcutil.GenRandomStr()
 		field1 := funcutil.GenRandomStr()
 		field2 := funcutil.GenRandomStr()
@@ -1552,7 +1558,7 @@ func Test_createCollectionTask_prepareSchema(t *testing.T) {
 		assert.NoError(t, err)
 
 		task := createCollectionTask{
-			Core: newTestCore(withMixCoord(mixcoord)),
+			Core: newTestCore(withMixCoord(mixcoord), withMeta(meta)),
 			Req: &milvuspb.CreateCollectionRequest{
 				Base:           &commonpb.MsgBase{MsgType: commonpb.MsgType_CreateCollection},
 				CollectionName: collectionName,

internal/rootcoord/create_collection_task_test.go

@@ -80,6 +80,7 @@ func (c *Core) broadcastCreateCollectionV1(ctx context.Context, req *milvuspb.Cr
 		preserveFieldID: preserveFieldID == "true",
 	}
 	if err := createCollectionTask.Prepare(ctx); err != nil {
+		createCollectionTask.releaseFileResources()
 		return err
 	}
 
@@ -95,6 +96,9 @@ func (c *Core) broadcastCreateCollectionV1(ctx context.Context, req *milvuspb.Cr
 		WithBroadcast(broadcastChannel).
 		MustBuildBroadcast()
 	if _, err := broadcaster.Broadcast(ctx, msg); err != nil {
+		// Do NOT release file resources here: the broadcast task is already in the
+		// scheduler and will retry until success. refCnt will be decremented when
+		// the collection is eventually dropped.
 		return err
 	}
 	return nil

internal/rootcoord/ddl_callbacks_create_collection.go

@@ -152,6 +152,9 @@ type IMetaTable interface {
 	AddFileResource(ctx context.Context, resource *internalpb.FileResourceInfo) error
 	RemoveFileResource(ctx context.Context, name string) (error, bool)
 	ListFileResource(ctx context.Context) ([]*internalpb.FileResourceInfo, uint64)
+	IncFileResourceRefCnt(ids []int64) error
+	DecFileResourceRefCnt(ids []int64)
+	RecoverFileResourceRefCnt(pendingCollections map[int64][]int64)
 }
 
 // MetaTable is a persistent meta set of all databases, collections and partitions.
@@ -557,9 +560,6 @@ func (mt *MetaTable) AddCollection(ctx context.Context, coll *model.Collection)
 
 	mt.collID2Meta[coll.CollectionID] = coll.Clone()
 	mt.names.insert(coll.DBName, coll.Name, coll.CollectionID)
-	for _, fileResourceID := range coll.FileResourceIds {
-		mt.fileResourceRefCnt[fileResourceID]++
-	}
 
 	pn := coll.GetPartitionNum(true)
 	mt.generalCnt += pn * int(coll.ShardsNum)
@@ -598,7 +598,12 @@ func (mt *MetaTable) DropCollection(ctx context.Context, collectionID UniqueID,
 	}
 	mt.collID2Meta[collectionID] = clone
 	for _, fileResourceID := range coll.FileResourceIds {
-		mt.fileResourceRefCnt[fileResourceID]--
+		if mt.fileResourceRefCnt[fileResourceID] > 0 {
+			mt.fileResourceRefCnt[fileResourceID]--
+		} else {
+			log.Warn("DropCollection: file resource refCnt underflow",
+				zap.Int64("collectionID", collectionID), zap.Int64("fileResourceID", fileResourceID))
+		}
 	}
 
 	log.Ctx(ctx).Info("update coll state to dropping",
@@ -2326,3 +2331,55 @@ func (mt *MetaTable) ListFileResource(ctx context.Context) ([]*internalpb.FileRe
 
 	return lo.Values(mt.fileResourceID2Meta), mt.fileResourceVersion
 }
+
+// IncFileResourceRefCnt increments refCnt for file resources, binding them to a
+// collection being created. Under ddLock, atomic with RemoveFileResource.
+// Returns error if any resource ID does not exist.
+func (mt *MetaTable) IncFileResourceRefCnt(ids []int64) error {
+	mt.ddLock.Lock()
+	defer mt.ddLock.Unlock()
+	for _, id := range ids {
+		if _, ok := mt.fileResourceID2Meta[id]; !ok {
+			return merr.WrapErrParameterInvalidMsg("file resource %d not found", id)
+		}
+	}
+	for _, id := range ids {
+		mt.fileResourceRefCnt[id]++
+	}
+	return nil
+}
+
+// DecFileResourceRefCnt decrements refCnt. Used for early release when
+// CreateCollection fails after validation.
+func (mt *MetaTable) DecFileResourceRefCnt(ids []int64) {
+	mt.ddLock.Lock()
+	defer mt.ddLock.Unlock()
+	for _, id := range ids {
+		if mt.fileResourceRefCnt[id] > 0 {
+			mt.fileResourceRefCnt[id]--
+		} else {
+			log.Warn("DecFileResourceRefCnt underflow", zap.Int64("id", id))
+		}
+	}
+}
+
+// RecoverFileResourceRefCnt re-increments refCnt for file resources referenced by
+// pending CreateCollection broadcast tasks whose collections have not yet been
+// persisted. Called during startup before rootcoord becomes Healthy.
+func (mt *MetaTable) RecoverFileResourceRefCnt(pendingCollections map[int64][]int64) {
+	mt.ddLock.Lock()
+	defer mt.ddLock.Unlock()
+	for collID, resourceIds := range pendingCollections {
+		if _, exists := mt.collID2Meta[collID]; exists {
+			continue // collection already persisted, reload already counted it
+		}
+		for _, id := range resourceIds {
+			if _, ok := mt.fileResourceID2Meta[id]; ok {
+				mt.fileResourceRefCnt[id]++
+			} else {
+				log.Warn("RecoverFileResourceRefCnt: pending task references missing file resource",
+					zap.Int64("collectionID", collID), zap.Int64("resourceID", id))
+			}
+		}
+	}
+}

internal/rootcoord/meta_table.go

@@ -47,6 +47,7 @@ import (
 	"github.com/milvus-io/milvus/internal/rootcoord/tombstone"
 	"github.com/milvus-io/milvus/internal/storage"
 	streamingcoord "github.com/milvus-io/milvus/internal/streamingcoord/server"
+	"github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster/broadcast"
 	tso2 "github.com/milvus-io/milvus/internal/tso"
 	"github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/internal/util/dependency"
@@ -560,6 +561,14 @@ func (c *Core) Init() error {
 
 	c.initOnce.Do(func() {
 		initError = c.initInternal()
+		// Recover file resource refCnt for pending CreateCollection broadcast tasks
+		// before registering DDL callbacks, so ack callbacks won't race with recovery.
+		// See #48612.
+		pending := broadcast.GetPendingCreateCollectionResources()
+		if len(pending) > 0 {
+			c.meta.RecoverFileResourceRefCnt(pending)
+			log.Info("recovered file resource refCnt from pending broadcast tasks", zap.Int("count", len(pending)))
+		}
 		RegisterDDLCallbacks(c)
 	})
 	log.Info("RootCoord init successfully")

internal/rootcoord/mocks/meta_table.go

@@ -62,6 +62,15 @@ func StartBroadcastWithSecondaryClusterResourceKey(ctx context.Context) (broadca
 	return broadcaster.WithSecondaryClusterResourceKey(ctx)
 }
 
+// GetPendingCreateCollectionResources returns pending CreateCollection file resource
+// IDs from the broadcaster. Must be called after Register.
+func GetPendingCreateCollectionResources() map[int64][]int64 {
+	if !singleton.Ready() {
+		return nil
+	}
+	return singleton.Get().GetPendingCreateCollectionResources()
+}
+
 // Release releases the broadcaster.
 func Release() {
 	if !singleton.Ready() {