simplify code

Author: Zettat123

models/actions/run.go

@@ -374,38 +374,3 @@ func UpdateRun(ctx context.Context, run *ActionRun, cols ...string) error {
 }
 
 type ActionRunIndex db.ResourceIndex
-
-func CancelConcurrentJobs(ctx context.Context, actionRunJob *ActionRunJob) error {
-	// cancel previous jobs in the same concurrency group
-	previousJobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
-		RepoID:           actionRunJob.RepoID,
-		ConcurrencyGroup: actionRunJob.ConcurrencyGroup,
-		Statuses: []Status{
-			StatusRunning,
-			StatusWaiting,
-			StatusBlocked,
-		},
-	})
-	if err != nil {
-		return fmt.Errorf("find previous jobs: %w", err)
-	}
-
-	return CancelJobs(ctx, previousJobs)
-}
-
-func ShouldJobBeBlockedByConcurrentJobs(ctx context.Context, actionRunJob *ActionRunJob) (bool, error) {
-	if actionRunJob.ConcurrencyCancel {
-		return false, CancelConcurrentJobs(ctx, actionRunJob)
-	}
-
-	concurrentJobsNum, err := db.Count[ActionRunJob](ctx, FindRunJobOptions{
-		RepoID:           actionRunJob.RepoID,
-		ConcurrencyGroup: actionRunJob.ConcurrencyGroup,
-		Statuses:         []Status{StatusRunning, StatusWaiting},
-	})
-	if err != nil {
-		return false, fmt.Errorf("count waiting jobs: %w", err)
-	}
-
-	return concurrentJobsNum > 0, nil
-}

models/actions/run_job.go

@@ -34,10 +34,11 @@ type ActionRunJob struct {
 	TaskID            int64    // the latest task of the job
 	Status            Status   `xorm:"index"`
 
-	RawConcurrencyGroup  string // raw concurrency.group
-	RawConcurrencyCancel string // raw concurrency.cancel-in-progress
-	ConcurrencyGroup     string // evaluated concurrency.group
-	ConcurrencyCancel    bool   // evaluated concurrency.cancel-in-progress
+	RawConcurrencyGroup    string // raw concurrency.group
+	RawConcurrencyCancel   string // raw concurrency.cancel-in-progress
+	IsConcurrencyEvaluated bool   // whether RawConcurrencyGroup have been evaluated, only valid when RawConcurrencyGroup is not empty
+	ConcurrencyGroup       string // evaluated concurrency.group
+	ConcurrencyCancel      bool   // evaluated concurrency.cancel-in-progress
 
 	Started timeutil.TimeStamp
 	Stopped timeutil.TimeStamp
@@ -190,3 +191,47 @@ func AggregateJobStatus(jobs []*ActionRunJob) Status {
 		return StatusUnknown // it shouldn't happen
 	}
 }
+
+func ShouldJobBeBlockedByConcurrentJobs(ctx context.Context, actionRunJob *ActionRunJob) (bool, error) {
+	if len(actionRunJob.RawConcurrencyGroup) == 0 {
+		return false, nil
+	}
+	if !actionRunJob.IsConcurrencyEvaluated {
+		return false, fmt.Errorf("the raw concurrency group has not been evaluated")
+	}
+	if len(actionRunJob.ConcurrencyGroup) == 0 {
+		return false, nil
+	}
+	if actionRunJob.ConcurrencyCancel {
+		return false, CancelConcurrentJobs(ctx, actionRunJob)
+	}
+
+	concurrentJobsNum, err := db.Count[ActionRunJob](ctx, FindRunJobOptions{
+		RepoID:           actionRunJob.RepoID,
+		ConcurrencyGroup: actionRunJob.ConcurrencyGroup,
+		Statuses:         []Status{StatusRunning, StatusWaiting},
+	})
+	if err != nil {
+		return false, fmt.Errorf("count waiting jobs: %w", err)
+	}
+
+	return concurrentJobsNum > 0, nil
+}
+
+func CancelConcurrentJobs(ctx context.Context, actionRunJob *ActionRunJob) error {
+	// cancel previous jobs in the same concurrency group
+	previousJobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
+		RepoID:           actionRunJob.RepoID,
+		ConcurrencyGroup: actionRunJob.ConcurrencyGroup,
+		Statuses: []Status{
+			StatusRunning,
+			StatusWaiting,
+			StatusBlocked,
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("find previous jobs: %w", err)
+	}
+
+	return CancelJobs(ctx, previousJobs)
+}

models/migrations/v1_23/v312.go

@@ -18,10 +18,11 @@ func AddActionsConcurrency(x *xorm.Engine) error {
 	}
 
 	type ActionRunJob struct {
-		RawConcurrencyGroup  string
-		RawConcurrencyCancel string
-		ConcurrencyGroup     string
-		ConcurrencyCancel    bool
+		RawConcurrencyGroup    string
+		RawConcurrencyCancel   string
+		IsConcurrencyEvaluated bool
+		ConcurrencyGroup       string
+		ConcurrencyCancel      bool
 	}
 
 	return x.Sync(new(ActionRunJob))

routers/web/repo/actions/view.go

@@ -560,8 +560,11 @@ func Approve(ctx *context_module.Context) {
 			return err
 		}
 		for _, job := range jobs {
-			if len(job.Needs) == 0 && job.Status.IsBlocked() {
-				// TODO: check concurrency
+			blockedByConcurrency, err := actions_model.ShouldJobBeBlockedByConcurrentJobs(ctx, job)
+			if err != nil {
+				return err
+			}
+			if len(job.Needs) == 0 && job.Status.IsBlocked() && !blockedByConcurrency {
 				job.Status = actions_model.StatusWaiting
 				_, err := actions_model.UpdateRunJob(ctx, job, nil, "status")
 				if err != nil {

services/actions/job_emitter.go

@@ -275,7 +275,7 @@ func checkConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_
 		return false, err
 	}
 
-	if len(actionRunJob.ConcurrencyGroup) == 0 {
+	if !actionRunJob.IsConcurrencyEvaluated {
 		// empty concurrency group means the raw concurrency has not been evaluated
 		task, err := actions_model.GetTaskByID(ctx, actionRunJob.TaskID)
 		if err != nil {
@@ -300,25 +300,14 @@ func checkConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_
 		}
 
 		if _, err := actions_model.UpdateRunJob(ctx, &actions_model.ActionRunJob{
-			ID:                actionRunJob.ID,
-			ConcurrencyGroup:  actionRunJob.ConcurrencyGroup,
-			ConcurrencyCancel: actionRunJob.ConcurrencyCancel,
+			ID:                     actionRunJob.ID,
+			IsConcurrencyEvaluated: true,
+			ConcurrencyGroup:       actionRunJob.ConcurrencyGroup,
+			ConcurrencyCancel:      actionRunJob.ConcurrencyCancel,
 		}, nil); err != nil {
 			return false, fmt.Errorf("update run job: %w", err)
 		}
 	}
 
-	if len(actionRunJob.ConcurrencyGroup) == 0 {
-		// the job should not be blocked by concurrency if its concurrency group is empty
-		return false, nil
-	}
-
-	if actionRunJob.ConcurrencyCancel {
-		if err := actions_model.CancelConcurrentJobs(ctx, actionRunJob); err != nil {
-			return false, fmt.Errorf("cancel concurrent jobs: %w", err)
-		}
-		return false, nil
-	}
-
 	return actions_model.ShouldJobBeBlockedByConcurrentJobs(ctx, actionRunJob)
 }

services/actions/run.go

@@ -116,23 +116,21 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, jobs []*jobpar
 		if job.RawConcurrency != nil && len(job.RawConcurrency.Group) > 0 {
 			runJob.RawConcurrencyGroup = job.RawConcurrency.Group
 			runJob.RawConcurrencyCancel = job.RawConcurrency.CancelInProgress
-			// we do not need to evaluate job concurrency if the job is blocked
-			// because it will be checked by job emitter
+			// we do not need to evaluate job concurrency if the job is blocked because it will be checked by job emitter
 			if runJob.Status != actions_model.StatusBlocked {
 				var err error
 				runJob.ConcurrencyGroup, runJob.ConcurrencyCancel, err = evaluateJobConcurrency(run, runJob, vars, map[string]*jobparser.JobResult{})
 				if err != nil {
 					return fmt.Errorf("evaluate job concurrency: %w", err)
 				}
-				if len(runJob.ConcurrencyGroup) > 0 {
-					// check if the job should be blocked by job concurrency
-					shouldBlock, err := actions_model.ShouldJobBeBlockedByConcurrentJobs(ctx, runJob)
-					if err != nil {
-						return err
-					}
-					if shouldBlock {
-						runJob.Status = actions_model.StatusBlocked
-					}
+				runJob.IsConcurrencyEvaluated = true
+				// check if the job should be blocked by job concurrency
+				shouldBlock, err := actions_model.ShouldJobBeBlockedByConcurrentJobs(ctx, runJob)
+				if err != nil {
+					return err
+				}
+				if shouldBlock {
+					runJob.Status = actions_model.StatusBlocked
 				}
 			}
 		}