Fix/8787 cancel pipeline keeps running (#8832)

* fix(pipeline): cancel pipeline returns 200 OK but keeps running (#8787)

Three independent bugs caused pipeline cancellation to silently fail:

1. CancelPipeline discarded errors from CancelTask and never cancelled
   TASK_CREATED tasks in future stages. Now running tasks are cancelled
   via context, non-running tasks are batch-updated to TASK_CANCELLED in
   the DB, and errors are logged and returned.

2. gitextractor's storeRepoSnapshot (go-git path) had no ctx.Done()
   checks in its commit/blame loops, making it unresponsive to
   cancellation for 30+ minutes on large repos. Added cancellation
   checkpoints following the pattern already used elsewhere in the file.

3. Cancelled tasks were marked TASK_FAILED instead of TASK_CANCELLED,
   and ComputePipelineStatus never returned TASK_CANCELLED. Now RunTask
   checks for context cancellation and writes TASK_CANCELLED, and
   ComputePipelineStatus returns TASK_CANCELLED when the pipeline was
   cancelled by the user.

Test gaps: RunTask deferred status logic, CancelPipeline flow, and
storeRepoSnapshot have no existing unit tests. These are pre-existing
gaps not introduced by this change. The only existing test,
TestComputePipelineStatus, has been extended to cover isCancelled=true.

Closes #8787
Related: #5585, #4188

* refactor(pipeline): improve CancelPipeline readability and correctness

Extract cancelRunningTasks and cancelPendingTasksInDB helpers from
CancelPipeline for better separation of concerns. Also fixes:
- .As(NotFound) replaced with .GetType() to prevent swallowing wrapped errors
- Added TASK_RESUME to cancelPendingTasksInDB status filter
- Error message now includes pipeline ID for traceability
- Added TestCancelPipeline e2e tests with 3 subtests

Author: danielemoraschi

backend/core/runner/run_task.go

@@ -77,7 +77,12 @@ func RunTask(
 		}
 		finishedAt := time.Now()
 		spentSeconds := finishedAt.Unix() - beganAt.Unix()
+		isCancelled := errors.Is(err, gocontext.Canceled) || ctx.Err() == gocontext.Canceled
 		if err != nil {
+			taskStatus := models.TASK_FAILED
+			if isCancelled {
+				taskStatus = models.TASK_CANCELLED
+			}
 			lakeErr := errors.AsLakeErrorType(err)
 			subTaskName := "unknown"
 			if lakeErr = lakeErr.As(errors.SubtaskErr); lakeErr != nil {
@@ -87,16 +92,19 @@ func RunTask(
 			} else {
 				lakeErr = errors.Convert(err)
 			}
-			dbe := db.UpdateColumns(task, []dal.DalSet{
-				{ColumnName: "status", Value: models.TASK_FAILED},
+			columns := []dal.DalSet{
+				{ColumnName: "status", Value: taskStatus},
 				{ColumnName: "message", Value: lakeErr.Error()},
 				{ColumnName: "error_name", Value: lakeErr.Messages().Format()},
 				{ColumnName: "finished_at", Value: finishedAt},
 				{ColumnName: "spent_seconds", Value: spentSeconds},
-				{ColumnName: "failed_sub_task", Value: subTaskName},
-			})
+			}
+			if taskStatus == models.TASK_FAILED {
+				columns = append(columns, dal.DalSet{ColumnName: "failed_sub_task", Value: subTaskName})
+			}
+			dbe := db.UpdateColumns(task, columns)
 			if dbe != nil {
-				logger.Error(dbe, "failed to finalize task status into db (task failed)")
+				logger.Error(dbe, "failed to finalize task status into db (task %s)", taskStatus)
 			}
 		} else {
 			dbe := db.UpdateColumns(task, []dal.DalSet{
@@ -116,7 +124,7 @@ func RunTask(
 			dal.Where("id=?", task.PipelineId),
 		))
 		// not return err if the `SkipOnFail` is true and the error is not canceled
-		if dbPipeline.SkipOnFail && !errors.Is(err, gocontext.Canceled) {
+		if dbPipeline.SkipOnFail && !isCancelled {
 			err = nil
 		}
 	}()

backend/plugins/gitextractor/parser/repo_gogit.go

@@ -524,6 +524,11 @@ func (r *GogitRepoCollector) storeRepoSnapshot(subtaskCtx plugin.SubTaskContext,
 	ctx := subtaskCtx.GetContext()
 	snapshot := make(map[string][]string) // {"filePathAndName": ["line1 commit sha", "line2 commit sha"]}
 	for _, commit := range commitList {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
 		commitTree, firstParentTree, err := r.getCurrentAndParentTree(ctx, commit)
 		if err != nil {
 			return err
@@ -533,6 +538,11 @@ func (r *GogitRepoCollector) storeRepoSnapshot(subtaskCtx plugin.SubTaskContext,
 			return err
 		}
 		for _, p := range patch.Stats() {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			default:
+			}
 			fileName := p.Name
 			if _, ok := snapshot[fileName]; !ok {
 				snapshot[fileName] = []string{}

backend/server/services/pipeline.go

@@ -461,10 +461,54 @@ func CancelPipeline(pipelineId uint64) errors.Error {
 	if count == 0 {
 		return nil
 	}
-	for _, pendingTask := range pendingTasks {
-		_ = CancelTask(pendingTask.ID)
+	var runningTaskIds []uint64
+	var pendingTaskIds []uint64
+	for _, task := range pendingTasks {
+		if task.Status == models.TASK_RUNNING {
+			runningTaskIds = append(runningTaskIds, task.ID)
+		} else {
+			pendingTaskIds = append(pendingTaskIds, task.ID)
+		}
+	}
+	failedCancels := cancelRunningTasks(runningTaskIds) + cancelPendingTasksInDB(pendingTaskIds)
+	if failedCancels > 0 {
+		return errors.Default.New(fmt.Sprintf("failed to cancel %d task(s) for pipeline #%d", failedCancels, pipelineId))
+	}
+	return nil
+}
+
+// cancelRunningTasks cancels tasks that are actively running by triggering
+// their context cancellation. Returns the number of tasks that failed to cancel.
+func cancelRunningTasks(taskIds []uint64) int {
+	failCount := 0
+	for _, taskId := range taskIds {
+		if err := CancelTask(taskId); err != nil {
+			if err.GetType() == errors.NotFound {
+				continue // task no longer tracked in-memory (finished or context lost after restart)
+			}
+			globalPipelineLog.Error(err, "failed to cancel running task #%d", taskId)
+			failCount++
+		}
+	}
+	return failCount
+}
+
+// cancelPendingTasksInDB marks non-running pending tasks as cancelled directly
+// in the database. Returns the number of tasks that failed to update.
+func cancelPendingTasksInDB(taskIds []uint64) int {
+	if len(taskIds) == 0 {
+		return 0
+	}
+	err := db.UpdateColumn(
+		&models.Task{},
+		"status", models.TASK_CANCELLED,
+		dal.Where("id IN ? AND status IN ?", taskIds, []string{models.TASK_CREATED, models.TASK_RERUN, models.TASK_RESUME}),
+	)
+	if err != nil {
+		globalPipelineLog.Error(err, "failed to cancel %d pending tasks in DB", len(taskIds))
+		return len(taskIds)
 	}
-	return errors.Convert(err)
+	return 0
 }
 
 // getPipelineLogsPath gets the logs directory of this pipeline

backend/server/services/pipeline_runner.go

@@ -107,9 +107,10 @@ func runPipeline(pipelineId uint64) errors.Error {
 }
 
 // ComputePipelineStatus determines pipeline status by its latest(rerun included) tasks statuses
-// 1. TASK_COMPLETED: all tasks were executed successfully
-// 2. TASK_FAILED: SkipOnFail=false with failed task(s)
-// 3. TASK_PARTIAL: SkipOnFail=true with failed task(s)
+// 1. TASK_CANCELLED: pipeline was cancelled by the user (takes priority)
+// 2. TASK_COMPLETED: all tasks were executed successfully
+// 3. TASK_FAILED: SkipOnFail=false with failed task(s)
+// 4. TASK_PARTIAL: SkipOnFail=true with failed task(s)
 func ComputePipelineStatus(pipeline *models.Pipeline, isCancelled bool) (string, errors.Error) {
 	tasks, err := GetLatestTasksOfPipeline(pipeline)
 	if err != nil {
@@ -134,6 +135,9 @@ func ComputePipelineStatus(pipeline *models.Pipeline, isCancelled bool) (string,
 		return "", errors.Default.New("unexpected status, did you call computePipelineStatus at a wrong timing?")
 	}
 
+	if isCancelled {
+		return models.TASK_CANCELLED, nil
+	}
 	if failed == 0 {
 		return models.TASK_COMPLETED, nil
 	}

backend/test/e2e/services/pipeline_runner_test.go

@@ -18,11 +18,14 @@ limitations under the License.
 package services
 
 import (
+	"testing"
+	"time"
+
+	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/models"
 	"github.com/apache/incubator-devlake/server/services"
 	"github.com/apache/incubator-devlake/test/helper"
 	"github.com/stretchr/testify/assert"
-	"testing"
 )
 
 func TestComputePipelineStatus(t *testing.T) {
@@ -163,4 +166,121 @@ func TestComputePipelineStatus(t *testing.T) {
 	status, err = services.ComputePipelineStatus(pipeline, false)
 	assert.Nil(t, err)
 	assert.Equal(t, models.TASK_PARTIAL, status)
+
+	// pipeline.status == "cancelled" if the pipeline was cancelled by the user
+	// regardless of individual task statuses
+	task_row1_col1_rerun.Status = models.TASK_COMPLETED
+	err = db.Update(task_row1_col1_rerun)
+	assert.Nil(t, err)
+	status, err = services.ComputePipelineStatus(pipeline, true)
+	assert.Nil(t, err)
+	assert.Equal(t, models.TASK_CANCELLED, status)
+
+	// pipeline.status == "cancelled" even when some tasks failed
+	task_row1_col1_rerun.Status = models.TASK_FAILED
+	err = db.Update(task_row1_col1_rerun)
+	assert.Nil(t, err)
+	status, err = services.ComputePipelineStatus(pipeline, true)
+	assert.Nil(t, err)
+	assert.Equal(t, models.TASK_CANCELLED, status)
+}
+
+func TestCancelPipeline(t *testing.T) {
+	client := helper.StartDevLakeServer(t, nil)
+	db := client.GetDal()
+
+	t.Run("cancels pending pipeline and all its tasks", func(t *testing.T) {
+		pipeline := &models.Pipeline{
+			TotalTasks: 2,
+			Status:     models.TASK_CREATED,
+		}
+		err := db.Create(pipeline)
+		assert.Nil(t, err)
+		assert.NotZero(t, pipeline.ID)
+
+		task1 := &models.Task{
+			PipelineId:  pipeline.ID,
+			PipelineRow: 1,
+			PipelineCol: 1,
+			Plugin:      "github",
+			Status:      models.TASK_CREATED,
+		}
+		task2 := &models.Task{
+			PipelineId:  pipeline.ID,
+			PipelineRow: 1,
+			PipelineCol: 2,
+			Plugin:      "gitextractor",
+			Status:      models.TASK_CREATED,
+		}
+		err = db.Create(task1)
+		assert.Nil(t, err)
+		assert.NotZero(t, task1.ID)
+		err = db.Create(task2)
+		assert.Nil(t, err)
+		assert.NotZero(t, task2.ID)
+
+		err = services.CancelPipeline(pipeline.ID)
+		assert.Nil(t, err)
+
+		cancelledPipeline := &models.Pipeline{}
+		err = db.First(cancelledPipeline, dal.Where("id = ?", pipeline.ID))
+		assert.Nil(t, err)
+		assert.Equal(t, models.TASK_CANCELLED, cancelledPipeline.Status)
+
+		cancelledTask1, err := services.GetTask(task1.ID)
+		assert.Nil(t, err)
+		assert.Equal(t, models.TASK_CANCELLED, cancelledTask1.Status)
+		cancelledTask2, err := services.GetTask(task2.ID)
+		assert.Nil(t, err)
+		assert.Equal(t, models.TASK_CANCELLED, cancelledTask2.Status)
+	})
+
+	t.Run("cancels pending tasks but leaves completed tasks unchanged", func(t *testing.T) {
+		pipeline := &models.Pipeline{
+			TotalTasks: 2,
+			Status:     models.TASK_RUNNING,
+		}
+		err := db.Create(pipeline)
+		assert.Nil(t, err)
+		assert.NotZero(t, pipeline.ID)
+
+		finishedAt := time.Now()
+		completedTask := &models.Task{
+			PipelineId:  pipeline.ID,
+			PipelineRow: 1,
+			PipelineCol: 1,
+			Plugin:      "github",
+			Status:      models.TASK_COMPLETED,
+			FinishedAt:  &finishedAt,
+		}
+		pendingTask := &models.Task{
+			PipelineId:  pipeline.ID,
+			PipelineRow: 2,
+			PipelineCol: 1,
+			Plugin:      "refdiff",
+			Status:      models.TASK_CREATED,
+		}
+		err = db.Create(completedTask)
+		assert.Nil(t, err)
+		assert.NotZero(t, completedTask.ID)
+		err = db.Create(pendingTask)
+		assert.Nil(t, err)
+		assert.NotZero(t, pendingTask.ID)
+
+		err = services.CancelPipeline(pipeline.ID)
+		assert.Nil(t, err)
+
+		reloadedCompleted, err := services.GetTask(completedTask.ID)
+		assert.Nil(t, err)
+		assert.Equal(t, models.TASK_COMPLETED, reloadedCompleted.Status)
+
+		reloadedPending, err := services.GetTask(pendingTask.ID)
+		assert.Nil(t, err)
+		assert.Equal(t, models.TASK_CANCELLED, reloadedPending.Status)
+	})
+
+	t.Run("returns error for non-existent pipeline", func(t *testing.T) {
+		err := services.CancelPipeline(999999)
+		assert.NotNil(t, err)
+	})
 }