fix cancelling jobs

Author: Zettat123

models/actions/run.go

@@ -402,3 +402,37 @@ func ShouldBlockRunByConcurrency(ctx context.Context, actionRun *ActionRun) (boo
 
 	return len(previousRuns) > 0, nil
 }
+
+func CancelPreviousJobsByRunConcurrency(ctx context.Context, actionRun *ActionRun) ([]*ActionRunJob, error) {
+	var cancelledJobs []*ActionRunJob
+
+	if actionRun.ConcurrencyGroup != "" && actionRun.ConcurrencyCancel {
+		// cancel previous runs in the same concurrency group
+		runs, err := db.Find[ActionRun](ctx, &FindRunOptions{
+			RepoID:           actionRun.RepoID,
+			ConcurrencyGroup: actionRun.ConcurrencyGroup,
+			Status:           []Status{StatusRunning, StatusWaiting, StatusBlocked},
+		})
+		if err != nil {
+			return cancelledJobs, fmt.Errorf("find runs: %w", err)
+		}
+		for _, run := range runs {
+			if run.ID == actionRun.ID {
+				continue
+			}
+			jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
+				RunID: run.ID,
+			})
+			if err != nil {
+				return cancelledJobs, fmt.Errorf("find run %d jobs: %w", run.ID, err)
+			}
+			cjs, err := CancelJobs(ctx, jobs)
+			if err != nil {
+				return cancelledJobs, fmt.Errorf("cancel run %d jobs: %w", run.ID, err)
+			}
+			cancelledJobs = append(cancelledJobs, cjs...)
+		}
+	}
+
+	return cancelledJobs, nil
+}

models/actions/run_job.go

@@ -238,7 +238,7 @@ func ShouldBlockJobByConcurrency(ctx context.Context, job *ActionRunJob) (bool,
 	return ShouldBlockRunByConcurrency(ctx, job.Run)
 }
 
-func CancelPreviousJobsByConcurrency(ctx context.Context, job *ActionRunJob) ([]*ActionRunJob, error) {
+func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) ([]*ActionRunJob, error) {
 	var cancelledJobs []*ActionRunJob
 
 	if job.RawConcurrencyGroup != "" {
@@ -270,33 +270,12 @@ func CancelPreviousJobsByConcurrency(ctx context.Context, job *ActionRunJob) ([]
 	if err := job.LoadRun(ctx); err != nil {
 		return cancelledJobs, fmt.Errorf("load run: %w", err)
 	}
-	if job.Run.ConcurrencyGroup != "" && job.Run.ConcurrencyCancel {
-		// cancel previous runs in the same concurrency group
-		runs, err := db.Find[ActionRun](ctx, &FindRunOptions{
-			RepoID:           job.RepoID,
-			ConcurrencyGroup: job.Run.ConcurrencyGroup,
-			Status:           []Status{StatusRunning, StatusWaiting, StatusBlocked},
-		})
-		if err != nil {
-			return cancelledJobs, fmt.Errorf("find runs: %w", err)
-		}
-		for _, run := range runs {
-			if run.ID == job.Run.ID {
-				continue
-			}
-			jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
-				RunID: run.ID,
-			})
-			if err != nil {
-				return cancelledJobs, fmt.Errorf("find run %d jobs: %w", run.ID, err)
-			}
-			cjs, err := CancelJobs(ctx, jobs)
-			if err != nil {
-				return cancelledJobs, fmt.Errorf("cancel run %d jobs: %w", run.ID, err)
-			}
-			cancelledJobs = append(cancelledJobs, cjs...)
-		}
+
+	cancelledJobsByRun, err := CancelPreviousJobsByRunConcurrency(ctx, job.Run)
+	if err != nil {
+		return cancelledJobs, fmt.Errorf("cancel runs: %w", err)
 	}
+	cancelledJobs = append(cancelledJobs, cancelledJobsByRun...)
 
 	return cancelledJobs, nil
 }

routers/web/repo/actions/view.go

@@ -470,6 +470,8 @@ func rerunJob(ctx *context_module.Context, job *actions_model.ActionRunJob, shou
 		}
 		if blockByConcurrency {
 			job.Status = actions_model.StatusBlocked
+		} else if err := actions_service.CancelJobsByJobConcurrency(ctx, job); err != nil {
+			return fmt.Errorf("cancel jobs: %w", err)
 		}
 	}
 
@@ -585,6 +587,9 @@ func Approve(ctx *context_module.Context) {
 				return err
 			}
 			if len(job.Needs) == 0 && job.Status.IsBlocked() && !blockJobByConcurrency {
+				if err := actions_service.CancelJobsByJobConcurrency(ctx, job); err != nil {
+					return fmt.Errorf("cancel jobs: %w", err)
+				}
 				job.Status = actions_model.StatusWaiting
 				n, err := actions_model.UpdateRunJob(ctx, job, nil, "status")
 				if err != nil {

services/actions/clear_tasks.go

@@ -57,8 +57,14 @@ func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository) er
 	return err
 }
 
-func CancelJobs(ctx context.Context, job *actions_model.ActionRunJob) error {
-	jobs, err := actions_model.CancelPreviousJobsByConcurrency(ctx, job)
+func CancelJobsByJobConcurrency(ctx context.Context, job *actions_model.ActionRunJob) error {
+	jobs, err := actions_model.CancelPreviousJobsByJobConcurrency(ctx, job)
+	notifyWorkflowJobStatusUpdate(ctx, jobs)
+	return err
+}
+
+func CancelJobsByRunConcurrency(ctx context.Context, run *actions_model.ActionRun) error {
+	jobs, err := actions_model.CancelPreviousJobsByRunConcurrency(ctx, run)
 	notifyWorkflowJobStatusUpdate(ctx, jobs)
 	return err
 }

services/actions/job_emitter.go

@@ -275,6 +275,11 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
 				continue
 			}
 
+			if err := CancelJobsByJobConcurrency(ctx, r.jobMap[id]); err != nil {
+				log.Error("Cancel previous jobs for job %d: %v. This job will stay blocked.", id, err)
+				continue
+			}
+
 			if allSucceed {
 				ret[id] = actions_model.StatusWaiting
 			} else {

services/actions/run.go

@@ -38,6 +38,8 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, jobs []*jobpar
 	}
 	if blockRunByConcurrency {
 		run.Status = actions_model.StatusBlocked
+	} else if err := CancelJobsByRunConcurrency(ctx, run); err != nil {
+		return fmt.Errorf("cancel jobs: %w", err)
 	}
 
 	if err := db.Insert(ctx, run); err != nil {
@@ -111,6 +113,8 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, jobs []*jobpar
 				}
 				if blockByConcurrency {
 					runJob.Status = actions_model.StatusBlocked
+				} else if err := CancelJobsByJobConcurrency(ctx, runJob); err != nil {
+					return fmt.Errorf("cancel jobs: %w", err)
 				}
 			}
 		}

services/actions/task.go

@@ -53,7 +53,7 @@ func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv
 			return nil
 		}
 
-		if err := CancelJobs(ctx, t.Job); err != nil {
+		if err := CancelJobsByJobConcurrency(ctx, t.Job); err != nil {
 			return fmt.Errorf("CancelJobs: %w", err)
 		}