fix(queue): close TOCTOU race between schedule() and start()

queue.go

@@ -30,6 +30,7 @@ type (
 		metric        *metric       // Metrics collector for tracking queue and worker stats
 		logger        Logger        // Logger for queue events and errors
 		workerCount   int64         // Number of worker goroutines to process jobs
+		activeSlots   int64         // Number of reserved worker slots (scheduling guard)
 		routineGroup  *routineGroup // Group to manage and wait for goroutines
 		quit          chan struct{} // Channel to signal shutdown to all goroutines
 		ready         chan struct{} // Channel to signal worker readiness
@@ -185,6 +186,9 @@ func (q *Queue) work(task core.TaskMessage) {
 	// Defer block to handle panics, update metrics, and run afterFn callback.
 	defer func() {
 		q.metric.DecBusyWorker()
+		q.Lock()
+		q.activeSlots--
+		q.Unlock()
 		e := recover()
 		if e != nil {
 			q.logger.Fatalf("panic error: %v", e)
@@ -313,12 +317,12 @@ func (q *Queue) UpdateWorkerCount(num int64) {
 	q.schedule()
 }
 
-// schedule checks if more workers can be started based on the current busy count.
+// schedule checks if more workers can be started based on reserved slots.
 // If so, it signals readiness to start a new worker.
 func (q *Queue) schedule() {
 	q.Lock()
 	defer q.Unlock()
-	if q.BusyWorkers() >= q.workerCount {
+	if q.activeSlots >= q.workerCount {
 		return
 	}
 
@@ -351,6 +355,15 @@ func (q *Queue) start() {
 			return
 		}
 
+		// Reserve a slot under the mutex to close the TOCTOU gap.
+		q.Lock()
+		if q.activeSlots >= q.workerCount {
+			q.Unlock()
+			continue
+		}
+		q.activeSlots++
+		q.Unlock()
+
 		// Request a task from the worker in a background goroutine.
 		q.routineGroup.Run(func() {
 			for {
@@ -386,6 +399,9 @@ func (q *Queue) start() {
 
 		task, ok := <-tasks
 		if !ok {
+			q.Lock()
+			q.activeSlots--
+			q.Unlock()
 			return
 		}
 

ring_test.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"log"
 	"runtime"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -550,3 +552,65 @@ func BenchmarkRingQueue(b *testing.B) {
 		}
 	})
 }
+
+func TestBusyWorkersNeverExceedsWorkerCount(t *testing.T) {
+	const workerCount = 4
+	const totalTasks = 100
+
+	var maxObserved int64
+	var wg sync.WaitGroup
+	wg.Add(totalTasks)
+	gate := make(chan struct{})
+
+	w := NewRing(
+		WithFn(func(ctx context.Context, m core.TaskMessage) error {
+			<-gate
+			wg.Done()
+			return nil
+		}),
+	)
+	q, err := NewQueue(
+		WithWorker(w),
+		WithWorkerCount(workerCount),
+	)
+	assert.NoError(t, err)
+
+	q.Start()
+	for i := 0; i < totalTasks; i++ {
+		assert.NoError(t, q.Queue(mockMessage{message: "task"}))
+	}
+
+	// Continuously monitor BusyWorkers while tasks execute.
+	stop := make(chan struct{})
+	monitorDone := make(chan struct{})
+	go func() {
+		defer close(monitorDone)
+		ticker := time.NewTicker(100 * time.Microsecond)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-stop:
+				return
+			case <-ticker.C:
+				busy := q.BusyWorkers()
+				for {
+					old := atomic.LoadInt64(&maxObserved)
+					if busy <= old || atomic.CompareAndSwapInt64(&maxObserved, old, busy) {
+						break
+					}
+				}
+			}
+		}
+	}()
+
+	// Release tasks in batches to create scheduling pressure.
+	for i := 0; i < totalTasks; i++ {
+		gate <- struct{}{}
+	}
+	wg.Wait()
+	close(stop)
+	<-monitorDone
+	q.Release()
+
+	assert.LessOrEqual(t, maxObserved, int64(workerCount))
+}