fix(tasks): prevent duplicate dispatch on all-runners-busy requeue

services/tasks/TaskRunner.go

@@ -191,9 +191,7 @@ func (t *TaskRunner) run() {
 
 	defer func() {
 		if requeued {
-			// Task is being re-queued, don't mark as finished
-			log.Info("Task " + strconv.Itoa(t.Task.ID) + " re-queued (waiting for available runner)")
-			t.pool.queueEvents <- PoolEvent{EventTypeRequeued, t}
+			// EventTypeRequeued was sent synchronously before return.
 			return
 		}
 
@@ -282,9 +280,16 @@ func (t *TaskRunner) run() {
 
 	if err != nil {
 		if errors.Is(err, ErrAllRunnersBusy) {
-			// No runners available right now, put task back in waiting state
+			// No runners available right now, put task back in waiting state.
+			// Release running/active bookkeeping before enqueueing so a concurrent
+			// queue tick cannot ClaimAndDequeue the task while it is still in the
+			// running set (duplicate dispatch). Notify the pool synchronously so
+			// the current queue pass skips an immediate retry.
 			t.SetStatus(task_logger.TaskWaitingStatus)
+			t.pool.onTaskStop(t)
 			t.pool.state.Enqueue(t)
+			log.Info("Task " + strconv.Itoa(t.Task.ID) + " re-queued (waiting for available runner)")
+			t.pool.queueEvents <- PoolEvent{EventTypeRequeued, t}
 			requeued = true
 			return
 		}

services/tasks/TaskRunner_test.go

@@ -5,6 +5,7 @@ import (
 	"os"
 	"path"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/semaphoreui/semaphore/db/sql"
@@ -71,6 +72,47 @@ func (l *mockLogWriteService) WriteResult(task any) error {
 	return nil
 }
 
+func TestTaskRunner_ErrAllRunnersBusy_ReleasesRunningBeforeEnqueue(t *testing.T) {
+	state := NewMemoryTaskStateStore()
+	pool := TaskPool{
+		queueEvents: make(chan PoolEvent),
+		state:       state,
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		pool.handleQueue()
+	}()
+
+	tr := &TaskRunner{
+		Task: db.Task{
+			ID:         42,
+			ProjectID:  1,
+			TemplateID: 7,
+			Status:     task_logger.TaskStartingStatus,
+		},
+		Template: db.Template{ID: 7, Name: "tpl"},
+		pool:     &pool,
+	}
+
+	state.SetRunning(tr)
+	state.AddActive(tr.Task.ProjectID, tr)
+
+	// Mirrors the ErrAllRunnersBusy path in TaskRunner.run.
+	tr.Task.Status = task_logger.TaskWaitingStatus
+	pool.onTaskStop(tr)
+	state.Enqueue(tr)
+	pool.queueEvents <- PoolEvent{EventTypeRequeued, tr}
+
+	assert.Equal(t, 0, state.RunningCount(), "requeued task must not remain in running set")
+	assert.Equal(t, 1, state.QueueLen(), "requeued task must remain in the queue")
+
+	close(pool.queueEvents)
+	wg.Wait()
+}
+
 func TestTaskRunnerRun(t *testing.T) {
 
 	store := sql.CreateTestStore()

services/tasks/runner_reconciler.go

@@ -278,8 +278,9 @@ func (p *TaskPool) requeueTaskRunnerOffline(tsk *TaskRunner, runnerID int, reaso
 	}
 
 	// Same flow as the ErrAllRunnersBusy requeue in TaskRunner.run:
-	// put the task back into the queue, then let the pool release its
-	// running/active bookkeeping (EventTypeRequeued -> onTaskStop).
+	// release running/active bookkeeping before enqueueing, then notify the
+	// pool so the current queue pass skips an immediate retry.
+	p.onTaskStop(tsk)
 	p.state.Enqueue(tsk)
 	p.queueEvents <- PoolEvent{EventTypeRequeued, tsk}
 }
@@ -336,6 +337,7 @@ func (p *TaskPool) requeueUndispatchedTask(tsk *TaskRunner) {
 
 	// EventTypeRequeued -> onTaskStop releases the running/active bookkeeping and
 	// the claim, so the task can be re-claimed from the queue by any live node.
+	p.onTaskStop(tsk)
 	p.state.Enqueue(tsk)
 	p.queueEvents <- PoolEvent{EventTypeRequeued, tsk}
 }