fix: Avoid blocking compactor executor slot queries (#50789)

## Summary

Fixes #50788

The DataNode compactor executor used the same mutex for task state, slot
accounting, channel enqueueing, completion callbacks, and filesystem
metric publication. When the task queue was full, `Enqueue` could block
while holding the mutex, which also blocked `Slots()` and made DataNode
`QuerySlot` time out.

This PR keeps the executor mutex scoped to in-memory state updates only:

- release the mutex before sending to `taskCh`
- update terminal task state and release slot usage before completion
callbacks
- publish filesystem metrics outside the executor lock
- add regression tests for full task queues and completion callbacks
querying slots

## Test Plan

- `go test -v -count=1 -tags dynamic,test -gcflags="all=-N -l"
-ldflags="-r ${TEST_MILVUS_WORK_DIR}/cmake_build/lib -r
${TEST_MILVUS_WORK_DIR}/internal/core/output/lib"
./internal/datanode/compactor -run "TestCompactionExecutor" -timeout
300s`
- `gofumpt -l internal/datanode/compactor/executor.go
internal/datanode/compactor/executor_test.go`
- `git diff --check`

Signed-off-by: sijie-ni-0214 <sijie.ni@zilliz.com>

internal/datanode/compactor/executor.go

@@ -104,12 +104,12 @@ func getTaskSlotUsage(task Compactor) int64 {
 
 func (e *executor) Enqueue(task Compactor) (bool, error) {
 	e.mu.Lock()
-	defer e.mu.Unlock()
 
 	planID := task.GetPlanID()
 
 	// Check for duplicate task
 	if _, exists := e.tasks[planID]; exists {
+		e.mu.Unlock()
 		mlog.Warn(context.TODO(), "duplicated compaction task",
 			mlog.Int64("planID", planID),
 			mlog.String("channel", task.GetChannelName()))
@@ -123,6 +123,7 @@ func (e *executor) Enqueue(task Compactor) (bool, error) {
 		state:     datapb.CompactionTaskState_executing,
 		result:    nil,
 	}
+	e.mu.Unlock()
 
 	e.taskCh <- task
 	return true, nil
@@ -138,11 +139,8 @@ func (e *executor) Slots() int64 {
 // completeTask updates task state to completed and adjusts slot usage
 func (e *executor) completeTask(planID int64, result *datapb.CompactionPlanResult) {
 	e.mu.Lock()
-	defer e.mu.Unlock()
 
 	if task, exists := e.tasks[planID]; exists {
-		task.compactor.Complete()
-
 		// Update state based on result
 		if result != nil {
 			task.state = datapb.CompactionTaskState_completed
@@ -151,18 +149,24 @@ func (e *executor) completeTask(planID int64, result *datapb.CompactionPlanResul
 			task.state = datapb.CompactionTaskState_failed
 		}
 
-		// Publish filesystem metrics after compaction task completion
-		storageConfig := task.compactor.GetStorageConfig()
-		if _, err := storagev2.PublishFilesystemMetricsWithConfig(storageConfig); err != nil {
-			mlog.Warn(context.TODO(), "failed to publish filesystem metrics", mlog.Err(err))
-		}
-
 		// Adjust slot usage
 		e.usingSlots -= getTaskSlotUsage(task.compactor)
 		if e.usingSlots < 0 {
 			e.usingSlots = 0
 		}
+		e.mu.Unlock()
+
+		task.compactor.Complete()
+
+		// Publish filesystem metrics after compaction task completion
+		storageConfig := task.compactor.GetStorageConfig()
+		if _, err := storagev2.PublishFilesystemMetricsWithConfig(storageConfig); err != nil {
+			mlog.Warn(context.TODO(), "failed to publish filesystem metrics", mlog.Err(err))
+		}
+		return
 	}
+
+	e.mu.Unlock()
 }
 
 func (e *executor) RemoveTask(planID int64) {

internal/datanode/compactor/executor_test.go

@@ -72,6 +72,72 @@ func TestCompactionExecutor(t *testing.T) {
 		assert.Equal(t, 1, len(ex.taskCh))
 	})
 
+	t.Run("Test_Slots_NotBlocked_WhenEnqueueWaitsOnFullQueue", func(t *testing.T) {
+		ex := NewExecutor()
+		for i := 0; i < cap(ex.taskCh); i++ {
+			ex.taskCh <- nil
+		}
+
+		enqueueHoldingLock := make(chan struct{})
+		mockC := NewMockCompactor(t)
+		mockC.EXPECT().GetPlanID().Return(int64(100))
+		mockC.EXPECT().GetSlotUsage().Run(func() {
+			close(enqueueHoldingLock)
+		}).Return(int64(8))
+
+		enqueueDone := make(chan struct{})
+		go func() {
+			defer close(enqueueDone)
+			succeed, err := ex.Enqueue(mockC)
+			assert.True(t, succeed)
+			assert.NoError(t, err)
+		}()
+
+		require.Eventually(t, func() bool {
+			select {
+			case <-enqueueHoldingLock:
+				return true
+			default:
+				return false
+			}
+		}, time.Second, 10*time.Millisecond)
+
+		slotsDone := make(chan int64, 1)
+		go func() {
+			slotsDone <- ex.Slots()
+		}()
+
+		var slotsBlocked bool
+		select {
+		case slots := <-slotsDone:
+			assert.Equal(t, int64(8), slots)
+		case <-time.After(100 * time.Millisecond):
+			slotsBlocked = true
+		}
+
+		<-ex.taskCh
+		require.Eventually(t, func() bool {
+			select {
+			case <-enqueueDone:
+				return true
+			default:
+				return false
+			}
+		}, time.Second, 10*time.Millisecond)
+
+		if slotsBlocked {
+			require.Eventually(t, func() bool {
+				select {
+				case <-slotsDone:
+					return true
+				default:
+					return false
+				}
+			}, time.Second, 10*time.Millisecond)
+			require.Fail(t, "Slots blocked while Enqueue waited on a full task queue")
+		}
+	})
+
 	t.Run("Test_Enqueue_DefaultSlotUsage", func(t *testing.T) {
 		testCases := []struct {
 			name              string
@@ -395,6 +461,44 @@ func TestCompactionExecutor(t *testing.T) {
 		assert.Equal(t, int64(0), ex.Slots())
 	})
 
+	t.Run("Test_CompleteTask_DoesNotHoldLockDuringCallbacks", func(t *testing.T) {
+		ex := NewExecutor()
+		mockC := NewMockCompactor(t)
+		planID := int64(10)
+		slotUsage := int64(8)
+
+		ex.tasks[planID] = &taskState{
+			compactor: mockC,
+			state:     datapb.CompactionTaskState_executing,
+		}
+		ex.usingSlots = slotUsage
+
+		callbackSlots := make(chan int64, 2)
+		mockC.EXPECT().GetSlotUsage().Return(slotUsage)
+		mockC.EXPECT().Complete().Run(func() {
+			callbackSlots <- ex.Slots()
+		}).Return()
+		mockC.EXPECT().GetStorageConfig().Run(func() {
+			callbackSlots <- ex.Slots()
+		}).Return(nil)
+
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			ex.completeTask(planID, &datapb.CompactionPlanResult{PlanID: planID})
+		}()
+
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("completeTask blocked while invoking compactor callbacks")
+		}
+
+		require.Equal(t, int64(0), <-callbackSlots)
+		require.Equal(t, int64(0), <-callbackSlots)
+		assert.Equal(t, int64(0), ex.Slots())
+	})
+
 	t.Run("Test_Task_State_Transitions", func(t *testing.T) {
 		ex := NewExecutor()
 		mockC := NewMockCompactor(t)