fix(runners): preserve stopped status when Run returns after termination

services/runners/job_pool.go

@@ -321,40 +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)
-					}
-				}
+				running.completeAfterRun(err)
 
 				log.WithFields(log.Fields{
 					"context": "job_running",

services/runners/running_job.go

@@ -155,9 +155,53 @@ func (p *runningJob) ackLogRecords(sent int) (pending int) {
 	return len(p.logRecords)
 }
 
-func (p *runningJob) logPipe(reader io.Reader) {
-	p.logWG.Add(1)
-	defer p.logWG.Done()
+// completeAfterRun applies the terminal status transition once Executor.Run
+// returns. It mirrors the server-side TaskRunner.run error handling: a job that
+// was already stopped (e.g. via applyTerminatedJobs or a server-side stop) or
+// killed must not be downgraded to failed just because Run returned an error
+// from the interrupted process.
+func (running *runningJob) completeAfterRun(err error) {
+	if err != nil {
+		if running.getStatus().IsFinished() {
+			return
+		}
+
+		if running.getStatus() == task_logger.TaskStoppingStatus || running.job.IsKilled() {
+			running.SetStatus(task_logger.TaskStoppedStatus)
+			return
+		}
+
+		log.WithFields(log.Fields{
+			"context":     "job_running",
+			"task_id":     running.taskID,
+			"task_status": running.getStatus(),
+		}).WithError(err).Error("launch job failed")
+
+		running.Log("Unable to launch the application. Please contact your system administrator for assistance.")
+		running.SetStatus(task_logger.TaskFailStatus)
+		return
+	}
+
+	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)
+	}
+}
+
+func (running *runningJob) logPipe(reader io.Reader) {
+	running.logWG.Add(1)
+	defer running.logWG.Done()
 
 	scanner := bufio.NewScanner(reader)
 	const maxCapacity = 10 * 1024 * 1024 // 10 MB
@@ -166,7 +210,7 @@ func (p *runningJob) logPipe(reader io.Reader) {
 
 	for scanner.Scan() {
 		line := scanner.Text()
-		p.Log(line)
+		running.Log(line)
 	}
 
 	err := scanner.Err()
@@ -183,13 +227,13 @@ func (p *runningJob) logPipe(reader io.Reader) {
 			msg = "TaskRunner output exceeds the maximum allowed size of 10MB"
 		}
 
-		p.job.Kill() // kill the job because stdout cannot be read.
+		running.job.Kill() // kill the job because stdout cannot be read.
 
 		log.WithError(err).WithFields(log.Fields{
-			"task_id": p.taskID,
+			"task_id": running.taskID,
 			"context": "task_logger",
 		}).Error(msg)
 
-		p.Log("Fatal error: " + msg)
+		running.Log("Fatal error: " + msg)
 	}
 }

services/runners/running_job_test.go

@@ -152,3 +152,72 @@ func TestRunningJob_ConcurrentAccess(t *testing.T) {
 	close(start)
 	wg.Wait()
 }
+
+type stubExecutor struct {
+	tasks.LocalExecutor
+	runErr error
+}
+
+func (s *stubExecutor) Run(string, *string, string) error {
+	return s.runErr
+}
+
+func (s *stubExecutor) Prepare(string, *string, string) error { return nil }
+func (s *stubExecutor) Cleanup()                               {}
+
+func TestRunningJob_completeAfterRun_PreservesStoppedOnError(t *testing.T) {
+	exec := &stubExecutor{runErr: fmt.Errorf("process interrupted")}
+	rj := &runningJob{
+		job:    exec,
+		taskID: 1,
+		status: task_logger.TaskStoppedStatus,
+	}
+	exec.Logger = rj
+
+	rj.completeAfterRun(exec.runErr)
+
+	assert.Equal(t, task_logger.TaskStoppedStatus, rj.getStatus())
+}
+
+func TestRunningJob_completeAfterRun_KilledJobMarkedStopped(t *testing.T) {
+	exec := &stubExecutor{runErr: fmt.Errorf("signal: killed")}
+	rj := &runningJob{
+		job:    exec,
+		taskID: 2,
+		status: task_logger.TaskRunningStatus,
+	}
+	exec.Logger = rj
+	exec.Kill()
+
+	rj.completeAfterRun(exec.runErr)
+
+	assert.Equal(t, task_logger.TaskStoppedStatus, rj.getStatus())
+}
+
+func TestRunningJob_completeAfterRun_LaunchErrorMarksFailed(t *testing.T) {
+	exec := &stubExecutor{runErr: fmt.Errorf("ansible not found")}
+	rj := &runningJob{
+		job:    exec,
+		taskID: 3,
+		status: task_logger.TaskRunningStatus,
+	}
+	exec.Logger = rj
+
+	rj.completeAfterRun(exec.runErr)
+
+	assert.Equal(t, task_logger.TaskFailStatus, rj.getStatus())
+}
+
+func TestRunningJob_completeAfterRun_Success(t *testing.T) {
+	exec := &stubExecutor{}
+	rj := &runningJob{
+		job:    exec,
+		taskID: 4,
+		status: task_logger.TaskRunningStatus,
+	}
+	exec.Logger = rj
+
+	rj.completeAfterRun(nil)
+
+	assert.Equal(t, task_logger.TaskSuccessStatus, rj.getStatus())
+}