fix: runner async completion, workflow restore data loss, HA workflow progression

pro/db/sql/workflow.go

@@ -1,9 +1,15 @@
 package sql
 
 import (
+	"errors"
+
 	"github.com/semaphoreui/semaphore/db"
 )
 
+// ErrWorkflowsProOnly is returned by the OSS workflow store stub when a caller
+// attempts to persist workflow data without the Pro implementation wired.
+var ErrWorkflowsProOnly = errors.New("workflows require Semaphore Pro")
+
 // WorkflowStoreImpl is the open-source no-op stub for the Pro workflow store.
 // Workflows are a Pro feature; the real implementation lives in
 // pro_impl/db/sql/workflow.go. The stub keeps the open build compiling while
@@ -24,7 +30,7 @@ func (d *WorkflowStoreImpl) GetWorkflowTemplate(projectID int, workflowID int) (
 }
 
 func (d *WorkflowStoreImpl) CreateWorkflowTemplate(workflow db.WorkflowTemplate) (res db.WorkflowTemplate, err error) {
-	return
+	return res, ErrWorkflowsProOnly
 }
 
 func (d *WorkflowStoreImpl) UpdateWorkflowTemplate(workflow db.WorkflowTemplate) (err error) {

services/runners/job_pool.go

@@ -321,46 +321,7 @@ func (p *JobPool) Run() {
 				}).Debug("Running job")
 
 				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)
-					}
-				}
-
-				log.WithFields(log.Fields{
-					"context": "job_running",
-					"task_id": running.taskID,
-					"status":  string(running.getStatus()),
-				}).Info("Task finished")
+				applyJobRunResult(running, err, t.taskID, t.status)
 			}(rj)
 
 			log.WithFields(log.Fields{
@@ -1017,3 +978,52 @@ func (p *JobPool) checkNewJobs() {
 		}).Info("Task enqueued")
 	}
 }
+
+// applyJobRunResult updates the running job after executor.Run returns.
+func applyJobRunResult(running *runningJob, err error, queuedTaskID int, queuedStatus task_logger.TaskStatus) {
+	if err != nil {
+		log.WithFields(log.Fields{
+			"context":     "job_running",
+			"task_id":     queuedTaskID,
+			"task_status": queuedStatus,
+		}).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")
+
+		// Async executors (e.g. Kubernetes) return once the workload is
+		// dispatched; completion is reported later via progress polls.
+		// Marking success here would tell the server the task finished
+		// while it is still running.
+		if running.job.Async() {
+			return
+		}
+
+		if running.getStatus().IsFinished() {
+			return
+		}
+
+		if running.getStatus() == task_logger.TaskStoppingStatus {
+			running.SetStatus(task_logger.TaskStoppedStatus)
+		} else {
+			running.SetStatus(task_logger.TaskSuccessStatus)
+		}
+	}
+
+	log.WithFields(log.Fields{
+		"context": "job_running",
+		"task_id": running.taskID,
+		"status":  string(running.getStatus()),
+	}).Info("Task finished")
+}

services/runners/job_pool_test.go

@@ -260,3 +260,46 @@ func TestJobPool_checkNewJobs_ExecutorErrorWithoutCacheCleanProjectID(t *testing
 	})
 	assert.Equal(t, 0, p.queueLen())
 }
+
+type stubExecutor struct {
+	tasks.LocalExecutor
+	async bool
+}
+
+func (s *stubExecutor) Async() bool {
+	return s.async
+}
+
+func TestApplyJobRunResult_AsyncExecutorKeepsRunningStatus(t *testing.T) {
+	exec := &stubExecutor{
+		LocalExecutor: tasks.LocalExecutor{Task: db.Task{ID: 1}},
+		async:         true,
+	}
+	rj := &runningJob{
+		job:    exec,
+		taskID: 1,
+		status: task_logger.TaskRunningStatus,
+	}
+	exec.Logger = rj
+
+	applyJobRunResult(rj, nil, 1, task_logger.TaskStartingStatus)
+
+	assert.Equal(t, task_logger.TaskRunningStatus, rj.getStatus())
+}
+
+func TestApplyJobRunResult_SyncExecutorMarksSuccess(t *testing.T) {
+	exec := &stubExecutor{
+		LocalExecutor: tasks.LocalExecutor{Task: db.Task{ID: 2}},
+		async:         false,
+	}
+	rj := &runningJob{
+		job:    exec,
+		taskID: 2,
+		status: task_logger.TaskRunningStatus,
+	}
+	exec.Logger = rj
+
+	applyJobRunResult(rj, nil, 2, task_logger.TaskStartingStatus)
+
+	assert.Equal(t, task_logger.TaskSuccessStatus, rj.getStatus())
+}

services/tasks/TaskPool.go

@@ -464,6 +464,13 @@ func (p *TaskPool) finalizeRemoteTaskLocked(tsk *TaskRunner, runner *db.Runner)
 			// crash between saveStatus and the queue drain). Release any stale
 			// shared pool state without re-running finish or autorun.
 			p.onTaskStop(tsk)
+			// If the first finalizer crashed after persisting End but before
+			// HandleWorkflowTaskCompletion (see TaskRunner.finishRun), still
+			// progress the workflow run. The workflow service must treat this as
+			// idempotent.
+			if err := p.HandleWorkflowTaskCompletion(tsk.Task); err != nil {
+				log.WithError(err).WithField("task_id", tsk.Task.ID).Warn("workflow progression failed after duplicate finalize")
+			}
 			return
 		}
 	}

services/tasks/TaskPool_test.go

@@ -210,3 +210,106 @@ func TestTaskPool_StopTasksByTemplate_DequeuesWaitingTasksByID(t *testing.T) {
 	assert.Equal(t, 1, state.QueueLen(), "only the targeted template's waiting task should be dequeued")
 	assert.Equal(t, keepMe.Task.ID, state.QueueGet(0).Task.ID)
 }
+
+type spyWorkflowService struct {
+	completionCalls int
+}
+
+func (s *spyWorkflowService) StartWorkflow(workflow db.WorkflowTemplate, user *db.User) (db.WorkflowRun, error) {
+	return db.WorkflowRun{}, nil
+}
+
+func (s *spyWorkflowService) ProgressWorkflowRun(projectID int, runID int, user *db.User) error {
+	return nil
+}
+
+func (s *spyWorkflowService) StopWorkflowRun(projectID int, runID int, user *db.User) (db.WorkflowRun, error) {
+	return db.WorkflowRun{}, nil
+}
+
+func (s *spyWorkflowService) ResolveWorkflowApproval(projectID int, workflowID int, runID int, nodeID int, status db.WorkflowApprovalStatus, user *db.User) (db.WorkflowApproval, error) {
+	return db.WorkflowApproval{}, nil
+}
+
+func (s *spyWorkflowService) HandleWorkflowTaskCompletion(task db.Task) error {
+	s.completionCalls++
+	return nil
+}
+
+func (s *spyWorkflowService) GetWorkflowRunArtifacts(projectID int, runID int, currentTaskID *int) (map[string]any, error) {
+	return map[string]any{}, nil
+}
+
+func TestTaskPool_FinalizeRemoteTask_HA_ProgressesWorkflowOnDuplicateFinalize(t *testing.T) {
+	prevCfg := util.Config
+	t.Cleanup(func() { util.Config = prevCfg })
+
+	store := sql.CreateTestStore()
+	util.Config.MaxParallelTasks = 0
+	util.Config.HA = &util.HAConfig{Enabled: true}
+	proj, err := store.CreateProject(db.Project{})
+	require.NoError(t, err)
+
+	key, err := store.CreateAccessKey(db.AccessKey{
+		ProjectID: &proj.ID,
+		Name:      "test-key",
+		Type:      db.AccessKeyNone,
+	})
+	require.NoError(t, err)
+
+	repo, err := store.CreateRepository(db.Repository{
+		ProjectID: proj.ID,
+		SSHKeyID:  key.ID,
+		Name:      "test-repo",
+		GitURL:    "git@example.com:test/test",
+		GitBranch: "master",
+	})
+	require.NoError(t, err)
+
+	tpl, err := store.CreateTemplate(db.Template{
+		ProjectID:    proj.ID,
+		Name:         "remote tpl",
+		Playbook:     "pb.yml",
+		RepositoryID: repo.ID,
+	})
+	require.NoError(t, err)
+
+	runID := 99
+	task, err := store.CreateTask(db.Task{
+		ProjectID:      proj.ID,
+		TemplateID:     tpl.ID,
+		Status:         task_logger.TaskSuccessStatus,
+		WorkflowRunID:  &runID,
+	}, 0)
+	require.NoError(t, err)
+
+	end := tz.Now()
+	task.End = &end
+	require.NoError(t, store.UpdateTask(task))
+
+	state := NewMemoryTaskStateStore()
+	spy := &spyWorkflowService{}
+	pool := TaskPool{
+		queueEvents:     make(chan PoolEvent, 1),
+		state:           state,
+		store:           store,
+		workflowService: spy,
+	}
+
+	tr := &TaskRunner{
+		Task: db.Task{
+			ID:            task.ID,
+			ProjectID:     task.ProjectID,
+			TemplateID:    task.TemplateID,
+			Status:        task_logger.TaskSuccessStatus,
+			WorkflowRunID: &runID,
+		},
+		Template: tpl,
+		job:      &RemoteJob{Task: task},
+	}
+	state.SetRunning(tr)
+
+	pool.FinalizeRemoteTask(tr, nil)
+
+	assert.Equal(t, 1, spy.completionCalls, "duplicate finalize must still progress the workflow when End is already persisted")
+}