feat(runners): flag all runners busy (#3519)

* feat(runners): flag all runners busy

* Initial plan

* Update services/tasks/TaskRunner.go

Co-authored-by: Copilot <[email protected]>

* refactor: use sentinel error for all runners busy check

Co-authored-by: fiftin <[email protected]>

* Use sentinel error for all runners busy condition

Co-authored-by: fiftin <[email protected]>

* feat(runners): requene flag

* chore: remove extra check

---------

Co-authored-by: copilot-swe-agent[bot] <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: fiftin <[email protected]>

Author: 3 people

pro/go.mod

@@ -38,7 +38,11 @@ require (
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/skeema/knownhosts v1.3.1 // indirect
+	github.com/xanzy/ssh-agent v0.3.3 // indirect
 	golang.org/x/crypto v0.45.0 // indirect
+	golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b // indirect
+	golang.org/x/net v0.47.0 // indirect
 	golang.org/x/sys v0.38.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/warnings.v0 v0.1.2 // indirect

pro/go.sum

@@ -111,12 +111,29 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
-github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8=
-golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw=
-golang.org/x/crypto v0.44.0/go.mod h1:013i+Nw79BMiQiMsOPcVCB5ZIJbYkerPrGnOa00tvmc=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM=
+github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw=
+go.etcd.io/bbolt v1.4.1 h1:5mOV+HWjIPLEAlUGMsveaUvK2+byZMFOzojoi7bh7uI=
+go.etcd.io/bbolt v1.4.1/go.mod h1:c8zu2BnXWTu2XM4XcICtbGSl9cFwsXtcf9zLt2OncM8=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
 golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
+golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o=
+golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8=
+golang.org/x/mod v0.27.0 h1:kb+q2PyFnEADO2IEF935ehFUXlWiNjJWtRNgBLSfbxQ=
+golang.org/x/mod v0.27.0/go.mod h1:rWI627Fq0DEoudcK+MBkNkCe0EetEaDSwJJkCcjpazc=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
+golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
+golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
+golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=

services/tasks/RemoteJob.go

@@ -3,6 +3,7 @@ package tasks
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"time"
@@ -15,6 +16,9 @@ import (
 	"github.com/semaphoreui/semaphore/util"
 )
 
+// ErrAllRunnersBusy is returned when all available runners are busy
+var ErrAllRunnersBusy = errors.New("all runners busy")
+
 type RemoteJob struct {
 	RunnerTag *string
 	Task      db.Task
@@ -72,7 +76,7 @@ func callRunnerWebhook(runner *db.Runner, tsk *TaskRunner, action string) (err e
 	if resp != nil {
 		defer resp.Body.Close() //nolint:errcheck
 	}
-	
+
 	if resp.StatusCode != 200 && resp.StatusCode != 204 {
 		err = fmt.Errorf("webhook returned incorrect status")
 		return
@@ -136,7 +140,7 @@ func (t *RemoteJob) Run(username string, incomingVersion *string, alias string)
 	}
 
 	if runner == nil {
-		err = fmt.Errorf("no runners available")
+		err = ErrAllRunnersBusy
 		return
 	}
 

services/tasks/TaskPool.go

@@ -31,10 +31,11 @@ type logRecord struct {
 type EventType uint
 
 const (
-	EventTypeNew      EventType = 0
-	EventTypeFinished EventType = 1
-	EventTypeFailed   EventType = 2
-	EventTypeEmpty    EventType = 3
+	EventTypeNew      EventType = 0 // EventTypeNew represents an event when a new task is created, typically sent during a periodic check or timer.
+	EventTypeFinished EventType = 1 // EventTypeFinished represents an event when a task finishes, typically sent during a periodic check or timer.
+	EventTypeFailed   EventType = 2 // EventTypeFailed represents an event when a task fails, typically sent during a periodic check or timer.
+	EventTypeEmpty    EventType = 3 // EventTypeEmpty represents an event when the queue is empty, typically sent during a periodic check or timer.
+	EventTypeRequeued EventType = 4 // EventTypeRequeued represents an event when a task is moved back to the waiting state for reprocessing.
 )
 
 const (
@@ -190,7 +191,21 @@ func getTaskName(t *TaskRunner) string {
 
 func (p *TaskPool) handleQueue() {
 	for t := range p.queueEvents {
+		// When a task is re-queued (e.g., no remote runner available), we should
+		// clean up its "running" bookkeeping but avoid immediately retrying it in
+		// the same queue pass to prevent hot retry loops.
+		skipTaskID := 0
+
 		switch t.eventType {
+		case EventTypeRequeued:
+			// Task was started but moved back to waiting. It must not remain in
+			// running/active sets and must release its claim so it can be picked
+			// up again later.
+			p.onTaskStop(t.task)
+			// Avoid immediate retry in this same event handling iteration; it
+			// will be retried on the next periodic tick or when another event
+			// triggers queue processing.
+			skipTaskID = t.task.Task.ID
 		case EventTypeNew:
 			p.state.Enqueue(t.task)
 		case EventTypeFinished:
@@ -209,6 +224,12 @@ func (p *TaskPool) handleQueue() {
 				continue
 			}
 
+			// When handling a requeue event, don't immediately start the same task again.
+			if skipTaskID != 0 && curr.Task.ID == skipTaskID {
+				i = i + 1
+				continue
+			}
+
 			if curr.Task.Status == task_logger.TaskFailStatus {
 				//delete failed TaskRunner from queue
 				_ = p.state.DequeueAt(i)

services/tasks/TaskPool_test.go

@@ -0,0 +1,89 @@
+package tasks
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/semaphoreui/semaphore/db"
+	"github.com/semaphoreui/semaphore/db/bolt"
+	"github.com/semaphoreui/semaphore/pkg/task_logger"
+	"github.com/semaphoreui/semaphore/util"
+	"github.com/stretchr/testify/assert"
+)
+
+type spyTaskStateStore struct {
+	*MemoryTaskStateStore
+	tryClaimCalls int
+}
+
+func newSpyTaskStateStore() *spyTaskStateStore {
+	return &spyTaskStateStore{
+		MemoryTaskStateStore: NewMemoryTaskStateStore(),
+	}
+}
+
+// TryClaim returns false to ensure tests don't actually start tasks; we only want to
+// observe whether the queue loop attempted to claim a task.
+func (s *spyTaskStateStore) TryClaim(_ int) bool {
+	s.tryClaimCalls++
+	return false
+}
+
+func TestTaskPool_RequeuedEventCleansRunningStateAndSkipsImmediateRetry(t *testing.T) {
+	// Ensure util.Config is non-nil for p.blocks() checks.
+	prevCfg := util.Config
+	t.Cleanup(func() { util.Config = prevCfg })
+	util.Config = &util.ConfigType{MaxParallelTasks: 0}
+
+	store := bolt.CreateTestStore()
+	proj, err := store.CreateProject(db.Project{})
+	assert.NoError(t, err)
+
+	state := newSpyTaskStateStore()
+
+	pool := TaskPool{
+		queueEvents: make(chan PoolEvent),
+		state:       state,
+		store:       store,
+	}
+
+	tr := &TaskRunner{
+		Task: db.Task{
+			ID:         42,
+			ProjectID:  proj.ID,
+			TemplateID: 7,
+			Status:     task_logger.TaskWaitingStatus,
+		},
+		Template: db.Template{
+			ID:   7,
+			Name: "Test Template",
+		},
+		Alias: "alias-42",
+	}
+
+	// Simulate a task that was marked as running and then re-queued (the state that
+	// exists right before EventTypeRequeued is handled).
+	state.SetRunning(tr)
+	state.AddActive(tr.Task.ProjectID, tr)
+	state.SetAlias(tr.Alias, tr)
+	state.Enqueue(tr)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		pool.handleQueue()
+	}()
+
+	pool.queueEvents <- PoolEvent{EventTypeRequeued, tr}
+	close(pool.queueEvents)
+	wg.Wait()
+
+	assert.Equal(t, 0, state.RunningCount(), "requeued task must be removed from running set")
+	assert.Equal(t, 0, state.ActiveCount(tr.Task.ProjectID), "requeued task must be removed from active-by-project set")
+	assert.Nil(t, state.GetByAlias(tr.Alias), "requeued task alias mapping must be cleared")
+	assert.Equal(t, 1, state.QueueLen(), "requeued task must remain queued")
+	assert.Equal(t, 0, state.tryClaimCalls, "requeued task should not be immediately retried in the same queue pass")
+}
+
+

services/tasks/TaskRunner.go

@@ -171,7 +171,17 @@ func (t *TaskRunner) run() {
 		defer t.pool.store.Close("run task " + strconv.Itoa(t.Task.ID))
 	}
 
+	// requeued indicates task should go back to waiting state (e.g., all runners busy)
+	requeued := false
+
 	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}
+			return
+		}
+
 		log.Info("Stopped running TaskRunner " + strconv.Itoa(t.Task.ID))
 		log.Info("Release resource locker with TaskRunner " + strconv.Itoa(t.Task.ID))
 
@@ -214,6 +224,14 @@ func (t *TaskRunner) run() {
 	err = t.job.Run(username, incomingVersion, t.Alias)
 
 	if err != nil {
+		if errors.Is(err, ErrAllRunnersBusy) {
+			// No runners available right now, put task back in waiting state
+			t.SetStatus(task_logger.TaskWaitingStatus)
+			t.pool.state.Enqueue(t)
+			requeued = true
+			return
+		}
+
 		if t.job.IsKilled() {
 			t.SetStatus(task_logger.TaskStoppedStatus)
 		} else {