diff --git a/services/tasks/TaskRunner.go b/services/tasks/TaskRunner.go index 4d67d8b1fb..07fe4bf012 100644 --- a/services/tasks/TaskRunner.go +++ b/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 } diff --git a/services/tasks/TaskRunner_test.go b/services/tasks/TaskRunner_test.go index 120765767a..7b6fe03650 100644 --- a/services/tasks/TaskRunner_test.go +++ b/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() diff --git a/services/tasks/runner_reconciler.go b/services/tasks/runner_reconciler.go index c08ced66ac..6d44549447 100644 --- a/services/tasks/runner_reconciler.go +++ b/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} }