fix(tasks,runners): critical requeue and finalization bugs

services/runners/job_pool.go

@@ -320,39 +320,23 @@ func (p *JobPool) Run() {
 				err := running.job.Run(t.username, t.incomingVersion, t.alias)
 
 				if err != nil {
-
 					log.WithFields(log.Fields{
 						"context":     "job_running",
 						"task_id":     t.taskID,
 						"task_status": t.status,
 					}).WithError(err).Error("launch job failed")
 
 					running.Log("Unable to launch the application. Please contact your system administrator for assistance.")
-
-					if running.getStatus() == task_logger.TaskStoppingStatus {
-						running.SetStatus(task_logger.TaskStoppedStatus)
-					} else {
-						running.SetStatus(task_logger.TaskFailStatus)
-					}
 				} else {
-
 					log.WithFields(log.Fields{
 						"context": "job_running",
 						"task_id": running.taskID,
 						"status":  string(running.getStatus()),
 					}).Debug("Job run returned")
-
-					if running.getStatus().IsFinished() {
-						return
-					}
-
-					if running.getStatus() == task_logger.TaskStoppingStatus {
-						running.SetStatus(task_logger.TaskStoppedStatus)
-					} else {
-						running.SetStatus(task_logger.TaskSuccessStatus)
-					}
 				}
 
+				running.finalizeAfterRun(err)
+
 				log.WithFields(log.Fields{
 					"context": "job_running",
 					"task_id": running.taskID,

services/runners/running_job.go

@@ -129,6 +129,30 @@ func (p *runningJob) getStatus() task_logger.TaskStatus {
 	return p.status
 }
 
+// finalizeAfterRun applies the terminal status after job.Run returns. If the job
+// was already brought to a terminal state (e.g. emergency-stopped via
+// terminated_jobs while Run was still unwinding), the status is left unchanged.
+func (p *runningJob) finalizeAfterRun(err error) {
+	if p.getStatus().IsFinished() {
+		return
+	}
+
+	if err != nil {
+		if p.getStatus() == task_logger.TaskStoppingStatus {
+			p.SetStatus(task_logger.TaskStoppedStatus)
+		} else {
+			p.SetStatus(task_logger.TaskFailStatus)
+		}
+		return
+	}
+
+	if p.getStatus() == task_logger.TaskStoppingStatus {
+		p.SetStatus(task_logger.TaskStoppedStatus)
+	} else {
+		p.SetStatus(task_logger.TaskSuccessStatus)
+	}
+}
+
 // getProgress atomically snapshots the data needed to report progress to the
 // server. The returned slice is a copy, so the caller can read it freely while
 // the job keeps appending records.

services/runners/running_job_test.go

@@ -1,6 +1,7 @@
 package runners
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"testing"
@@ -152,3 +153,39 @@ func TestRunningJob_ConcurrentAccess(t *testing.T) {
 	close(start)
 	wg.Wait()
 }
+
+func TestRunningJob_finalizeAfterRun_KeepsFinishedStatusOnRunError(t *testing.T) {
+	rj := newTestRunningJob(1)
+	rj.SetStatus(task_logger.TaskStoppedStatus)
+
+	rj.finalizeAfterRun(errors.New("process killed"))
+
+	assert.Equal(t, task_logger.TaskStoppedStatus, rj.getStatus())
+}
+
+func TestRunningJob_finalizeAfterRun_SetsFailOnRunError(t *testing.T) {
+	rj := newTestRunningJob(2)
+	rj.SetStatus(task_logger.TaskRunningStatus)
+
+	rj.finalizeAfterRun(errors.New("ansible failed"))
+
+	assert.Equal(t, task_logger.TaskFailStatus, rj.getStatus())
+}
+
+func TestRunningJob_finalizeAfterRun_SetsSuccessOnCleanReturn(t *testing.T) {
+	rj := newTestRunningJob(3)
+	rj.SetStatus(task_logger.TaskRunningStatus)
+
+	rj.finalizeAfterRun(nil)
+
+	assert.Equal(t, task_logger.TaskSuccessStatus, rj.getStatus())
+}
+
+func TestRunningJob_finalizeAfterRun_StoppingBecomesStopped(t *testing.T) {
+	rj := newTestRunningJob(4)
+	rj.SetStatus(task_logger.TaskStoppingStatus)
+
+	rj.finalizeAfterRun(nil)
+
+	assert.Equal(t, task_logger.TaskStoppedStatus, rj.getStatus())
+}

services/tasks/TaskPool.go

@@ -463,13 +463,14 @@ func (p *TaskPool) FinalizeRemoteTask(tsk *TaskRunner, runner *db.Runner) {
 func (p *TaskPool) finalizeRemoteTaskLocked(tsk *TaskRunner, runner *db.Runner) {
 	if util.HAEnabled() {
 		p.refreshTaskStatusFromDB(tsk)
-		if tsk.Task.End != nil {
-			// Another node may have persisted End before onTaskStop ran (e.g.
-			// crash between saveStatus and the queue drain). Release any stale
-			// shared pool state without re-running finish or autorun.
-			p.onTaskStop(tsk)
-			return
-		}
+	}
+
+	if tsk.Task.End != nil {
+		// Another node may have persisted End before onTaskStop ran (e.g.
+		// crash between saveStatus and the queue drain). Release any stale
+		// shared pool state without re-running finish or autorun.
+		p.onTaskStop(tsk)
+		return
 	}
 
 	if runner != nil {

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

@@ -195,6 +195,11 @@ func (p *TaskPool) failTaskRunnerLost(tsk *TaskRunner, runner *db.Runner, reason
 	}
 
 	if tsk.Task.Status.IsFinished() {
+		// Another node (or the runner report on this node) already persisted a
+		// terminal status. Complete finalization instead of bailing: if we won
+		// the finalize lock over FinalizeRemoteTask, that path will not run and
+		// pool/Redis state (running set, claims, End, autorun) would leak.
+		p.finalizeRemoteTaskLocked(tsk, runner)
 		return
 	}
 
@@ -264,6 +269,22 @@ func (p *TaskPool) requeueTaskRunnerOffline(tsk *TaskRunner, runnerID int, reaso
 
 	tsk.Logf("Runner #%d lost the task: %s. Returning task to queue.", runnerID, reason)
 
+	// Re-check the DB immediately before mutating: another node may have
+	// received a concurrent "running" report while we held the finalize lock.
+	if util.HAEnabled() {
+		p.refreshTaskStatusFromDB(tsk)
+		if tsk.Task.Status != task_logger.TaskStartingStatus &&
+			tsk.Task.Status != task_logger.TaskWaitingStatus {
+			return
+		}
+		if tsk.Task.RunnerID == nil || *tsk.Task.RunnerID != runnerID {
+			return
+		}
+	}
+
+	prevRunnerID := tsk.Task.RunnerID
+	prevStatus := tsk.Task.Status
+
 	tsk.Task.RunnerID = nil
 	tsk.SetStatus(task_logger.TaskWaitingStatus)
 
@@ -274,12 +295,17 @@ func (p *TaskPool) requeueTaskRunnerOffline(tsk *TaskRunner, runnerID int, reaso
 			"task_id": tsk.Task.ID,
 			"context": "runner_reconciler",
 		}).Error("failed to persist requeued task")
+		// Roll back in-memory changes so the next reconcile tick retries via
+		// the runner-liveness path instead of mis-routing as undispatched.
+		tsk.Task.RunnerID = prevRunnerID
+		tsk.Task.Status = prevStatus
 		return
 	}
 
 	// 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}
 }
@@ -334,8 +360,9 @@ func (p *TaskPool) requeueUndispatchedTask(tsk *TaskRunner) {
 		return
 	}
 
-	// EventTypeRequeued -> onTaskStop releases the running/active bookkeeping and
-	// the claim, so the task can be re-claimed from the queue by any live node.
+	// Release running/active bookkeeping before enqueueing, then notify the
+	// pool 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}
 }