ah2

api/tasks/tasks.go

@@ -1,9 +1,10 @@
 package tasks
 
 import (
-	"github.com/semaphoreui/semaphore/pkg/task_logger"
 	"net/http"
 
+	"github.com/semaphoreui/semaphore/pkg/task_logger"
+
 	"github.com/semaphoreui/semaphore/api/helpers"
 	"github.com/semaphoreui/semaphore/db"
 	task2 "github.com/semaphoreui/semaphore/services/tasks"
@@ -44,7 +45,7 @@ func GetTasks(w http.ResponseWriter, r *http.Request) {
 
 	res := []taskRes{}
 
-	for _, task := range pool.Queue {
+	for _, task := range pool.GetQueuedTasks() {
 		res = append(res, taskRes{
 			TaskID:    task.Task.ID,
 			ProjectID: task.Task.ProjectID,
@@ -55,7 +56,7 @@ func GetTasks(w http.ResponseWriter, r *http.Request) {
 		})
 	}
 
-	for _, task := range pool.RunningTasks {
+	for _, task := range pool.GetRunningTasks() {
 		res = append(res, taskRes{
 			TaskID:    task.Task.ID,
 			ProjectID: task.Task.ProjectID,
@@ -77,15 +78,15 @@ func DeleteTask(w http.ResponseWriter, r *http.Request) {
 
 	var task *db.Task
 
-	for _, t := range pool.Queue {
+	for _, t := range pool.GetQueuedTasks() {
 		if t.Task.ID == taskID {
 			task = &t.Task
 			break
 		}
 	}
 
 	if task == nil {
-		for _, t := range pool.RunningTasks {
+		for _, t := range pool.GetRunningTasks() {
 			if t.Task.ID == taskID {
 				task = &t.Task
 				break

cli/cmd/root.go

@@ -17,6 +17,7 @@ import (
 	"github.com/semaphoreui/semaphore/db/factory"
 	proFactory "github.com/semaphoreui/semaphore/pro/db/factory"
 	proServer "github.com/semaphoreui/semaphore/pro/services/server"
+	proTasks "github.com/semaphoreui/semaphore/pro/services/tasks"
 	"github.com/semaphoreui/semaphore/services/schedules"
 	"github.com/semaphoreui/semaphore/services/tasks"
 	"github.com/semaphoreui/semaphore/util"
@@ -72,6 +73,7 @@ func Execute() {
 
 func runService() {
 	store := createStore("root")
+	state := proTasks.NewTaskStateStore()
 	terraformStore := proFactory.NewTerraformStore(store)
 	ansibleTaskRepo := proFactory.NewAnsibleTaskRepository(store)
 
@@ -93,6 +95,7 @@ func runService() {
 
 	taskPool := tasks.CreateTaskPool(
 		store,
+		state,
 		ansibleTaskRepo,
 		inventoryService,
 		encryptionService,

go.mod

@@ -18,6 +18,7 @@
 	github.com/lib/pq v1.10.9
 	github.com/mdp/qrterminal/v3 v3.2.1
 	github.com/pquerna/otp v1.4.0
+	github.com/redis/go-redis/v9 v9.7.0
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/semaphoreui/semaphore/pro v0.0.0
 	github.com/sirupsen/logrus v1.9.3
@@ -41,9 +42,11 @@
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/ProtonMail/go-crypto v1.1.6 // indirect
 	github.com/boombuler/barcode v1.0.1-0.20190219062509-6c824513bacc // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/cloudflare/circl v1.6.1 // indirect
 	github.com/cyphar/filepath-securejoin v0.4.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/emirpasic/gods v1.18.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect

go.sum

@@ -19,6 +19,12 @@ github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPd
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
 github.com/boombuler/barcode v1.0.1-0.20190219062509-6c824513bacc h1:biVzkmvwrH8WK8raXaxBx6fRVTlJILwEwQGL1I/ByEI=
 github.com/boombuler/barcode v1.0.1-0.20190219062509-6c824513bacc/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cloudflare/circl v1.6.1 h1:zqIqSPIndyBh1bjLVVDHMPpVKqp8Su/V+6MeDzzQBQ0=
 github.com/cloudflare/circl v1.6.1/go.mod h1:uddAzsPgqdMAYatqJ0lsjX1oECcQLIlRpzZh3pJrofs=
 github.com/coreos/go-oidc/v3 v3.14.1 h1:9ePWwfdwC4QKRlCXsJGou56adA/owXczOzwKdOumLqk=
@@ -31,6 +37,8 @@ github.com/cyphar/filepath-securejoin v0.4.1/go.mod h1:Sdj7gXlvMcPZsbhwhQ33GguGL
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/elazarl/goproxy v1.7.2 h1:Y2o6urb7Eule09PjlhQRGNsqRfPmYI3KKQLFpCAV3+o=
@@ -135,6 +143,8 @@ github.com/poy/onpar v1.1.2 h1:QaNrNiZx0+Nar5dLgTVp5mXkyoVFIbepjyEoGSnhbAY=
 github.com/poy/onpar v1.1.2/go.mod h1:6X8FLNoxyr9kkmnlqpK6LSoiOtrO6MICtWwEuWkLjzg=
 github.com/pquerna/otp v1.4.0 h1:wZvl1TIVxKRThZIBiwOOHOGP/1+nZyWBil9Y2XNEDzg=
 github.com/pquerna/otp v1.4.0/go.mod h1:dkJfzwRKNiegxyNb54X/3fLwhCynbMspSyWKnvi1AEg=
+github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
+github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=

pro/services/tasks/task_state_store_factory.go

@@ -0,0 +1,9 @@
+package tasks
+
+import (
+	"github.com/semaphoreui/semaphore/services/tasks"
+)
+
+func NewTaskStateStore() tasks.TaskStateStore {
+	return tasks.NewMemoryTaskStateStore()
+}

services/tasks/RemoteJob.go

@@ -4,11 +4,12 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"github.com/semaphoreui/semaphore/pkg/tz"
-	log "github.com/sirupsen/logrus"
 	"net/http"
 	"time"
 
+	"github.com/semaphoreui/semaphore/pkg/tz"
+	log "github.com/sirupsen/logrus"
+
 	"github.com/semaphoreui/semaphore/db"
 	"github.com/semaphoreui/semaphore/pkg/task_logger"
 	"github.com/semaphoreui/semaphore/util"
@@ -93,6 +94,9 @@ func (t *RemoteJob) Run(username string, incomingVersion *string, alias string)
 	tsk.IncomingVersion = incomingVersion
 	tsk.Username = username
 	tsk.Alias = alias
+	if t.taskPool != nil && t.taskPool.state != nil {
+		t.taskPool.state.UpdateRuntimeFields(tsk)
+	}
 
 	var runners []db.Runner
 	db.StoreSession(t.taskPool.store, "run remote job", func() {
@@ -141,6 +145,9 @@ func (t *RemoteJob) Run(username string, incomingVersion *string, alias string)
 	}
 
 	tsk.RunnerID = runner.ID
+	if t.taskPool != nil && t.taskPool.state != nil {
+		t.taskPool.state.UpdateRuntimeFields(tsk)
+	}
 
 	startTime := tz.Now()
 
@@ -154,6 +161,12 @@ func (t *RemoteJob) Run(username string, incomingVersion *string, alias string)
 
 		time.Sleep(1_000_000_000)
 		tsk = t.taskPool.GetTask(t.Task.ID)
+
+		if tsk == nil {
+			err = fmt.Errorf("task %d not found", t.Task.ID)
+			return
+		}
+
 		if tsk.Task.Status == task_logger.TaskSuccessStatus ||
 			tsk.Task.Status == task_logger.TaskStoppedStatus ||
 			tsk.Task.Status == task_logger.TaskFailStatus {