Support Actions concurrency syntax

models/actions/run.go

@@ -16,21 +16,21 @@ import (
 	user_model "code.gitea.io/gitea/models/user"
 	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/json"
+	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/setting"
 	api "code.gitea.io/gitea/modules/structs"
 	"code.gitea.io/gitea/modules/timeutil"
 	"code.gitea.io/gitea/modules/util"
 	webhook_module "code.gitea.io/gitea/modules/webhook"
 
-	"github.com/nektos/act/pkg/jobparser"
 	"xorm.io/builder"
 )
 
 // ActionRun represents a run of a workflow file
 type ActionRun struct {
 	ID                int64
 	Title             string
-	RepoID            int64                  `xorm:"index unique(repo_index)"`
+	RepoID            int64                  `xorm:"unique(repo_index) index(repo_concurrency)"`
 	Repo              *repo_model.Repository `xorm:"-"`
 	OwnerID           int64                  `xorm:"index"`
 	WorkflowID        string                 `xorm:"index"`                    // the name of workflow file
@@ -49,6 +49,9 @@ type ActionRun struct {
 	TriggerEvent      string                       // the trigger event defined in the `on` configuration of the triggered workflow
 	Status            Status                       `xorm:"index"`
 	Version           int                          `xorm:"version default 0"` // Status could be updated concomitantly, so an optimistic lock is needed
+	RawConcurrency    string                       // raw concurrency
+	ConcurrencyGroup  string                       `xorm:"index(repo_concurrency) NOT NULL DEFAULT ''"`
+	ConcurrencyCancel bool                         `xorm:"NOT NULL DEFAULT FALSE"`
 	// Started and Stopped is used for recording last run time, if rerun happened, they will be reset to 0
 	Started timeutil.TimeStamp
 	Stopped timeutil.TimeStamp
@@ -181,7 +184,7 @@ func (run *ActionRun) IsSchedule() bool {
 	return run.ScheduleID > 0
 }
 
-func updateRepoRunsNumbers(ctx context.Context, repo *repo_model.Repository) error {
+func UpdateRepoRunsNumbers(ctx context.Context, repo *repo_model.Repository) error {
 	_, err := db.GetEngine(ctx).ID(repo.ID).
 		NoAutoTime().
 		SetExpr("num_action_runs",
@@ -238,116 +241,62 @@ func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID strin
 			return cancelledJobs, err
 		}
 
-		// Iterate over each job and attempt to cancel it.
-		for _, job := range jobs {
-			// Skip jobs that are already in a terminal state (completed, cancelled, etc.).
-			status := job.Status
-			if status.IsDone() {
-				continue
-			}
-
-			// If the job has no associated task (probably an error), set its status to 'Cancelled' and stop it.
-			if job.TaskID == 0 {
-				job.Status = StatusCancelled
-				job.Stopped = timeutil.TimeStampNow()
-
-				// Update the job's status and stopped time in the database.
-				n, err := UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped")
-				if err != nil {
-					return cancelledJobs, err
-				}
-
-				// If the update affected 0 rows, it means the job has changed in the meantime, so we need to try again.
-				if n == 0 {
-					return cancelledJobs, errors.New("job has changed, try again")
-				}
-
-				cancelledJobs = append(cancelledJobs, job)
-				// Continue with the next job.
-				continue
-			}
-
-			// If the job has an associated task, try to stop the task, effectively cancelling the job.
-			if err := StopTask(ctx, job.TaskID, StatusCancelled); err != nil {
-				return cancelledJobs, err
-			}
-			cancelledJobs = append(cancelledJobs, job)
+		cjs, err := CancelJobs(ctx, jobs)
+		if err != nil {
+			return cancelledJobs, err
 		}
+		cancelledJobs = append(cancelledJobs, cjs...)
 	}
 
 	// Return nil to indicate successful cancellation of all running and waiting jobs.
 	return cancelledJobs, nil
 }
 
-// InsertRun inserts a run
-// The title will be cut off at 255 characters if it's longer than 255 characters.
-func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWorkflow) error {
-	return db.WithTx(ctx, func(ctx context.Context) error {
-		index, err := db.GetNextResourceIndex(ctx, "action_run_index", run.RepoID)
-		if err != nil {
-			return err
+func CancelJobs(ctx context.Context, jobs []*ActionRunJob) ([]*ActionRunJob, error) {
+	cancelledJobs := make([]*ActionRunJob, 0, len(jobs))
+	// Iterate over each job and attempt to cancel it.
+	for _, job := range jobs {
+		// Skip jobs that are already in a terminal state (completed, cancelled, etc.).
+		status := job.Status
+		if status.IsDone() {
+			continue
 		}
-		run.Index = index
-		run.Title = util.EllipsisDisplayString(run.Title, 255)
 
-		if err := db.Insert(ctx, run); err != nil {
-			return err
-		}
+		// If the job has no associated task (probably an error), set its status to 'Cancelled' and stop it.
+		if job.TaskID == 0 {
+			job.Status = StatusCancelled
+			job.Stopped = timeutil.TimeStampNow()
 
-		if run.Repo == nil {
-			repo, err := repo_model.GetRepositoryByID(ctx, run.RepoID)
+			// Update the job's status and stopped time in the database.
+			n, err := UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped")
 			if err != nil {
-				return err
+				return cancelledJobs, err
 			}
-			run.Repo = repo
-		}
 
-		if err := updateRepoRunsNumbers(ctx, run.Repo); err != nil {
-			return err
+			// If the update affected 0 rows, it means the job has changed in the meantime
+			if n == 0 {
+				log.Error("Failed to cancel job %d because it has changed", job.ID)
+				continue
+			}
+
+			cancelledJobs = append(cancelledJobs, job)
+			// Continue with the next job.
+			continue
 		}
 
-		runJobs := make([]*ActionRunJob, 0, len(jobs))
-		var hasWaiting bool
-		for _, v := range jobs {
-			id, job := v.Job()
-			needs := job.Needs()
-			if err := v.SetJob(id, job.EraseNeeds()); err != nil {
-				return err
-			}
-			payload, _ := v.Marshal()
-			status := StatusWaiting
-			if len(needs) > 0 || run.NeedApproval {
-				status = StatusBlocked
-			} else {
-				hasWaiting = true
-			}
-			job.Name = util.EllipsisDisplayString(job.Name, 255)
-			runJobs = append(runJobs, &ActionRunJob{
-				RunID:             run.ID,
-				RepoID:            run.RepoID,
-				OwnerID:           run.OwnerID,
-				CommitSHA:         run.CommitSHA,
-				IsForkPullRequest: run.IsForkPullRequest,
-				Name:              job.Name,
-				WorkflowPayload:   payload,
-				JobID:             id,
-				Needs:             needs,
-				RunsOn:            job.RunsOn(),
-				Status:            status,
-			})
+		// If the job has an associated task, try to stop the task, effectively cancelling the job.
+		if err := StopTask(ctx, job.TaskID, StatusCancelled); err != nil {
+			return cancelledJobs, err
 		}
-		if err := db.Insert(ctx, runJobs); err != nil {
-			return err
+		updatedJob, err := GetRunJobByID(ctx, job.ID)
+		if err != nil {
+			return cancelledJobs, fmt.Errorf("get job: %w", err)
 		}
+		cancelledJobs = append(cancelledJobs, updatedJob)
+	}
 
-		// if there is a job in the waiting status, increase tasks version.
-		if hasWaiting {
-			if err := IncreaseTaskVersion(ctx, run.OwnerID, run.RepoID); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
+	// Return nil to indicate successful cancellation of all running and waiting jobs.
+	return cancelledJobs, nil
 }
 
 func GetRunByRepoAndID(ctx context.Context, repoID, runID int64) (*ActionRun, error) {
@@ -432,7 +381,7 @@ func UpdateRun(ctx context.Context, run *ActionRun, cols ...string) error {
 		if err = run.LoadRepo(ctx); err != nil {
 			return err
 		}
-		if err := updateRepoRunsNumbers(ctx, run.Repo); err != nil {
+		if err := UpdateRepoRunsNumbers(ctx, run.Repo); err != nil {
 			return err
 		}
 	}
@@ -441,3 +390,72 @@ func UpdateRun(ctx context.Context, run *ActionRun, cols ...string) error {
 }
 
 type ActionRunIndex db.ResourceIndex
+
+func ShouldBlockRunByConcurrency(ctx context.Context, actionRun *ActionRun) (bool, error) {
+	if actionRun.ConcurrencyGroup == "" || actionRun.ConcurrencyCancel {
+		return false, nil
+	}
+
+	runs, jobs, err := GetConcurrentRunsAndJobs(ctx, actionRun.RepoID, actionRun.ConcurrencyGroup, []Status{StatusRunning})
+	if err != nil {
+		return false, fmt.Errorf("find concurrent runs and jobs: %w", err)
+	}
+
+	return len(runs) > 0 || len(jobs) > 0, nil
+}
+
+func GetConcurrentRunsAndJobs(ctx context.Context, repoID int64, concurrencyGroup string, status []Status) ([]*ActionRun, []*ActionRunJob, error) {
+	runs, err := db.Find[ActionRun](ctx, &FindRunOptions{
+		RepoID:           repoID,
+		ConcurrencyGroup: concurrencyGroup,
+		Status:           status,
+	})
+	if err != nil {
+		return nil, nil, fmt.Errorf("find runs: %w", err)
+	}
+
+	jobs, err := db.Find[ActionRunJob](ctx, &FindRunJobOptions{
+		RepoID:           repoID,
+		ConcurrencyGroup: concurrencyGroup,
+		Statuses:         status,
+	})
+	if err != nil {
+		return nil, nil, fmt.Errorf("find jobs: %w", err)
+	}
+
+	return runs, jobs, nil
+}
+
+func CancelPreviousJobsByRunConcurrency(ctx context.Context, actionRun *ActionRun) ([]*ActionRunJob, error) {
+	if actionRun.ConcurrencyGroup == "" {
+		return nil, nil
+	}
+
+	var jobsToCancel []*ActionRunJob
+
+	statusFindOption := []Status{StatusWaiting, StatusBlocked}
+	if actionRun.ConcurrencyCancel {
+		statusFindOption = append(statusFindOption, StatusRunning)
+	}
+	runs, jobs, err := GetConcurrentRunsAndJobs(ctx, actionRun.RepoID, actionRun.ConcurrencyGroup, statusFindOption)
+	if err != nil {
+		return nil, fmt.Errorf("find concurrent runs and jobs: %w", err)
+	}
+	jobsToCancel = append(jobsToCancel, jobs...)
+
+	// cancel runs in the same concurrency group
+	for _, run := range runs {
+		if run.ID == actionRun.ID {
+			continue
+		}
+		jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
+			RunID: run.ID,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("find run %d jobs: %w", run.ID, err)
+		}
+		jobsToCancel = append(jobsToCancel, jobs...)
+	}
+
+	return CancelJobs(ctx, jobsToCancel)
+}

models/actions/run_job.go

@@ -22,23 +22,38 @@ type ActionRunJob struct {
 	ID                int64
 	RunID             int64                  `xorm:"index"`
 	Run               *ActionRun             `xorm:"-"`
-	RepoID            int64                  `xorm:"index"`
+	RepoID            int64                  `xorm:"index(repo_concurrency)"`
 	Repo              *repo_model.Repository `xorm:"-"`
 	OwnerID           int64                  `xorm:"index"`
 	CommitSHA         string                 `xorm:"index"`
 	IsForkPullRequest bool
 	Name              string `xorm:"VARCHAR(255)"`
 	Attempt           int64
-	WorkflowPayload   []byte
-	JobID             string   `xorm:"VARCHAR(255)"` // job id in workflow, not job's id
-	Needs             []string `xorm:"JSON TEXT"`
-	RunsOn            []string `xorm:"JSON TEXT"`
-	TaskID            int64    // the latest task of the job
-	Status            Status   `xorm:"index"`
-	Started           timeutil.TimeStamp
-	Stopped           timeutil.TimeStamp
-	Created           timeutil.TimeStamp `xorm:"created"`
-	Updated           timeutil.TimeStamp `xorm:"updated index"`
+
+	// WorkflowPayload is act/jobparser.SingleWorkflow for act/jobparser.Parse
+	// it should contain exactly one job with global workflow fields for this model
+	WorkflowPayload []byte
+
+	JobID  string   `xorm:"VARCHAR(255)"` // job id in workflow, not job's id
+	Needs  []string `xorm:"JSON TEXT"`
+	RunsOn []string `xorm:"JSON TEXT"`
+	TaskID int64    // the latest task of the job
+	Status Status   `xorm:"index"`
+
+	RawConcurrency string // raw concurrency from job YAML's "concurrency" section
+
+	// IsConcurrencyEvaluated is only valid/needed when this job's RawConcurrency is not empty.
+	// If RawConcurrency can't be evaluated (e.g. depend on other job's outputs or have errors), this field will be false.
+	// If RawConcurrency has been successfully evaluated, this field will be true, ConcurrencyGroup and ConcurrencyCancel are also set.
+	IsConcurrencyEvaluated bool
+
+	ConcurrencyGroup  string `xorm:"index(repo_concurrency) NOT NULL DEFAULT ''"` // evaluated concurrency.group
+	ConcurrencyCancel bool   `xorm:"NOT NULL DEFAULT FALSE"`                      // evaluated concurrency.cancel-in-progress
+
+	Started timeutil.TimeStamp
+	Stopped timeutil.TimeStamp
+	Created timeutil.TimeStamp `xorm:"created"`
+	Updated timeutil.TimeStamp `xorm:"updated index"`
 }
 
 func init() {
@@ -125,7 +140,7 @@ func UpdateRunJob(ctx context.Context, job *ActionRunJob, cond builder.Cond, col
 		return affected, nil
 	}
 
-	if affected != 0 && slices.Contains(cols, "status") && job.Status.IsWaiting() {
+	if slices.Contains(cols, "status") && job.Status.IsWaiting() {
 		// if the status of job changes to waiting again, increase tasks version.
 		if err := IncreaseTaskVersion(ctx, job.OwnerID, job.RepoID); err != nil {
 			return 0, err
@@ -197,3 +212,60 @@ func AggregateJobStatus(jobs []*ActionRunJob) Status {
 		return StatusUnknown // it shouldn't happen
 	}
 }
+
+func ShouldWaitJobForConcurrencyEvaluation(job *ActionRunJob) bool {
+	return job.RawConcurrency != "" && !job.IsConcurrencyEvaluated
+}
+
+func ShouldBlockJobByConcurrency(ctx context.Context, job *ActionRunJob) (bool, error) {
+	if ShouldWaitJobForConcurrencyEvaluation(job) {
+		panic("job concurrency not evaluated, this function shouldn't be called")
+	}
+
+	if job.ConcurrencyGroup == "" || job.ConcurrencyCancel {
+		return false, nil
+	}
+
+	runs, jobs, err := GetConcurrentRunsAndJobs(ctx, job.RepoID, job.ConcurrencyGroup, []Status{StatusRunning})
+	if err != nil {
+		return false, fmt.Errorf("GetConcurrentRunsAndJobs: %w", err)
+	}
+
+	return len(runs) > 0 || len(jobs) > 0, nil
+}
+
+func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) (jobsToCancel []*ActionRunJob, _ error) {
+	if job.RawConcurrency == "" {
+		return nil, nil
+	}
+	if !job.IsConcurrencyEvaluated {
+		return nil, nil
+	}
+	if job.ConcurrencyGroup == "" {
+		return nil, nil
+	}
+
+	statusFindOption := []Status{StatusWaiting, StatusBlocked}
+	if job.ConcurrencyCancel {
+		statusFindOption = append(statusFindOption, StatusRunning)
+	}
+	runs, jobs, err := GetConcurrentRunsAndJobs(ctx, job.RepoID, job.ConcurrencyGroup, statusFindOption)
+	if err != nil {
+		return nil, fmt.Errorf("find concurrent runs and jobs: %w", err)
+	}
+	jobs = slices.DeleteFunc(jobs, func(j *ActionRunJob) bool { return j.ID == job.ID })
+	jobsToCancel = append(jobsToCancel, jobs...)
+
+	// cancel runs in the same concurrency group
+	for _, run := range runs {
+		jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
+			RunID: run.ID,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("find run %d jobs: %w", run.ID, err)
+		}
+		jobsToCancel = append(jobsToCancel, jobs...)
+	}
+
+	return CancelJobs(ctx, jobsToCancel)
+}