From da119e36c7489410ec77eb198df06270c00f5b32 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 19 Jun 2026 11:13:06 +0000 Subject: [PATCH] fix: critical runner async completion, workflow restore, and HA workflow progression - Runner job pool: respect Executor.Async() after Run returns so Kubernetes (and other async) executors are not falsely marked successful while still running on the cluster. - OSS workflow store stub: return ErrWorkflowsProOnly from CreateWorkflowTemplate instead of silently dropping workflow definitions during project restore. - HA finalize: when duplicate finalization observes End already persisted, still call HandleWorkflowTaskCompletion so a crash between saveStatus and workflow progression does not leave runs stuck. Co-authored-by: Denis Gukov --- pro/db/sql/workflow.go | 8 ++- services/runners/job_pool.go | 90 ++++++++++++++------------ services/runners/job_pool_test.go | 43 +++++++++++++ services/tasks/TaskPool.go | 7 ++ services/tasks/TaskPool_test.go | 103 ++++++++++++++++++++++++++++++ 5 files changed, 210 insertions(+), 41 deletions(-) diff --git a/pro/db/sql/workflow.go b/pro/db/sql/workflow.go index 4853675f1f..d5e9a15bc8 100644 --- a/pro/db/sql/workflow.go +++ b/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) { diff --git a/services/runners/job_pool.go b/services/runners/job_pool.go index d1d7bbea74..03a5182d70 100644 --- a/services/runners/job_pool.go +++ b/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") +} diff --git a/services/runners/job_pool_test.go b/services/runners/job_pool_test.go index f6914898a7..2fe3cd32e5 100644 --- a/services/runners/job_pool_test.go +++ b/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()) +} diff --git a/services/tasks/TaskPool.go b/services/tasks/TaskPool.go index 44ded1b763..3b48f56098 100644 --- a/services/tasks/TaskPool.go +++ b/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 } } diff --git a/services/tasks/TaskPool_test.go b/services/tasks/TaskPool_test.go index 27d2870c32..94b4ac4e01 100644 --- a/services/tasks/TaskPool_test.go +++ b/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") +}