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,7 @@ import (
 	"fmt"
 	"log"
 	"runtime"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -550,3 +551,38 @@ func BenchmarkRingQueue(b *testing.B) {
 		}
 	})
 }
+
+func TestBusyWorkersNeverExceedsWorkerCount(t *testing.T) {
+	const workerCount = 4
+	const totalTasks = 100
+
+	var maxObserved int64
+
+	w := NewRing(
+		WithFn(func(ctx context.Context, m core.TaskMessage) error {
+			runtime.Gosched()
+			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"}))
+		busy := q.BusyWorkers()
+		for {
+			old := atomic.LoadInt64(&maxObserved)
+			if busy <= old || atomic.CompareAndSwapInt64(&maxObserved, old, busy) {
+				break
+			}
+		}
+	}
+	time.Sleep(2 * time.Second)
+	q.Release()
+
+	assert.LessOrEqual(t, maxObserved, int64(workerCount))
+}