support concurrency

Author: Zettat123

models/actions/run.go

@@ -21,7 +21,6 @@ import (
 	"code.gitea.io/gitea/modules/util"
 	webhook_module "code.gitea.io/gitea/modules/webhook"
 
-	"github.com/nektos/act/pkg/jobparser"
 	"xorm.io/builder"
 )
 
@@ -48,6 +47,8 @@ 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
+	ConcurrencyGroup  string
+	ConcurrencyCancel bool
 	// 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
@@ -169,7 +170,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).
 		SetExpr("num_action_runs",
 			builder.Select("count(*)").From("action_run").
@@ -266,80 +267,41 @@ func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID strin
 	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 {
-	ctx, committer, err := db.TxContext(ctx)
-	if err != nil {
-		return err
-	}
-	defer committer.Close()
+func CancelJobs(ctx context.Context, jobs []*ActionRunJob) error {
+	// 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
+		}
 
-	index, err := db.GetNextResourceIndex(ctx, "action_run_index", run.RepoID)
-	if err != nil {
-		return err
-	}
-	run.Index = index
-	run.Title = util.EllipsisDisplayString(run.Title, 255)
+		// 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 err := db.Insert(ctx, run); err != nil {
-		return err
-	}
-
-	if run.Repo == nil {
-		repo, err := repo_model.GetRepositoryByID(ctx, run.RepoID)
-		if err != nil {
-			return err
-		}
-		run.Repo = repo
-	}
+			// 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
+			}
 
-	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, so we need to try again.
+			if n == 0 {
+				return fmt.Errorf("job has changed, try again")
+			}
 
-	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
+			// Continue with the next job.
+			continue
 		}
-		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 err := db.Insert(ctx, runJobs); err != nil {
-		return err
-	}
 
-	// if there is a job in the waiting status, increase tasks version.
-	if hasWaiting {
-		if err := IncreaseTaskVersion(ctx, run.OwnerID, run.RepoID); err != nil {
+		// 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 err
 		}
 	}
-
-	return committer.Commit()
+	return nil
 }
 
 func GetRunByID(ctx context.Context, id int64) (*ActionRun, error) {
@@ -431,7 +393,7 @@ func UpdateRun(ctx context.Context, run *ActionRun, cols ...string) error {
 			}
 			run.Repo = repo
 		}
-		if err := updateRepoRunsNumbers(ctx, run.Repo); err != nil {
+		if err := UpdateRepoRunsNumbers(ctx, run.Repo); err != nil {
 			return err
 		}
 	}
@@ -440,3 +402,21 @@ 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
+	}
+
+	concurrentRuns, err := db.Find[ActionRun](ctx, &FindRunOptions{
+		RepoID:           actionRun.RepoID,
+		ConcurrencyGroup: actionRun.ConcurrencyGroup,
+		Status:           []Status{StatusWaiting, StatusRunning},
+	})
+	if err != nil {
+		return false, fmt.Errorf("find running and waiting runs: %w", err)
+	}
+	previousRuns := slices.DeleteFunc(concurrentRuns, func(r *ActionRun) bool { return r.ID == actionRun.ID })
+
+	return len(previousRuns) > 0, nil
+}

models/actions/run_job.go

@@ -35,10 +35,17 @@ type ActionRunJob struct {
 	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"`
+
+	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
+	Created timeutil.TimeStamp `xorm:"created"`
+	Updated timeutil.TimeStamp `xorm:"updated index"`
 }
 
 func init() {
@@ -197,3 +204,107 @@ func AggregateJobStatus(jobs []*ActionRunJob) Status {
 		return StatusUnknown // it shouldn't happen
 	}
 }
+
+func ShouldBlockJobByConcurrency(ctx context.Context, job *ActionRunJob) (bool, error) {
+	if job.RawConcurrencyGroup == "" {
+		return false, nil
+	}
+	if !job.IsConcurrencyEvaluated {
+		return false, ErrUnevaluatedConcurrency{
+			Group:            job.RawConcurrencyGroup,
+			CancelInProgress: job.RawConcurrencyCancel,
+		}
+	}
+	if job.ConcurrencyGroup == "" || job.ConcurrencyCancel {
+		return false, nil
+	}
+
+	concurrentJobsNum, err := db.Count[ActionRunJob](ctx, FindRunJobOptions{
+		RepoID:           job.RepoID,
+		ConcurrencyGroup: job.ConcurrencyGroup,
+		Statuses:         []Status{StatusRunning, StatusWaiting},
+	})
+	if err != nil {
+		return false, fmt.Errorf("count running and waiting jobs: %w", err)
+	}
+	if concurrentJobsNum > 0 {
+		return true, nil
+	}
+
+	if err := job.LoadRun(ctx); err != nil {
+		return false, fmt.Errorf("load run: %w", err)
+	}
+
+	return ShouldBlockRunByConcurrency(ctx, job.Run)
+}
+
+func CancelPreviousJobsByConcurrency(ctx context.Context, job *ActionRunJob) error {
+	if job.RawConcurrencyGroup != "" {
+		if !job.IsConcurrencyEvaluated {
+			return ErrUnevaluatedConcurrency{
+				Group:            job.RawConcurrencyGroup,
+				CancelInProgress: job.RawConcurrencyCancel,
+			}
+		}
+		if job.ConcurrencyGroup != "" && job.ConcurrencyCancel {
+			// cancel previous jobs in the same concurrency group
+			previousJobs, err := db.Find[ActionRunJob](ctx, &FindRunJobOptions{
+				RepoID:           job.RepoID,
+				ConcurrencyGroup: job.ConcurrencyGroup,
+				Statuses:         []Status{StatusRunning, StatusWaiting, StatusBlocked},
+			})
+			if err != nil {
+				return fmt.Errorf("find previous jobs: %w", err)
+			}
+			previousJobs = slices.DeleteFunc(previousJobs, func(j *ActionRunJob) bool { return j.ID == job.ID })
+			if err := CancelJobs(ctx, previousJobs); err != nil {
+				return fmt.Errorf("cancel previous jobs: %w", err)
+			}
+		}
+	}
+
+	if err := job.LoadRun(ctx); err != nil {
+		return 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 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 fmt.Errorf("find run %d jobs: %w", run.ID, err)
+			}
+			if err := CancelJobs(ctx, jobs); err != nil {
+				return fmt.Errorf("cancel run %d jobs: %w", run.ID, err)
+			}
+		}
+	}
+
+	return nil
+}
+
+type ErrUnevaluatedConcurrency struct {
+	Group            string
+	CancelInProgress string
+}
+
+func IsErrUnevaluatedConcurrency(err error) bool {
+	_, ok := err.(ErrUnevaluatedConcurrency)
+	return ok
+}
+
+func (err ErrUnevaluatedConcurrency) Error() string {
+	return fmt.Sprintf("the raw concurrency [group=%s, cancel-in-progress=%s] is not evaluated", err.Group, err.CancelInProgress)
+}

models/actions/run_job_list.go

@@ -69,12 +69,13 @@ func (jobs ActionJobList) LoadAttributes(ctx context.Context, withRepo bool) err
 
 type FindRunJobOptions struct {
 	db.ListOptions
-	RunID         int64
-	RepoID        int64
-	OwnerID       int64
-	CommitSHA     string
-	Statuses      []Status
-	UpdatedBefore timeutil.TimeStamp
+	RunID            int64
+	RepoID           int64
+	OwnerID          int64
+	CommitSHA        string
+	Statuses         []Status
+	UpdatedBefore    timeutil.TimeStamp
+	ConcurrencyGroup string
 }
 
 func (opts FindRunJobOptions) ToConds() builder.Cond {
@@ -97,5 +98,8 @@ func (opts FindRunJobOptions) ToConds() builder.Cond {
 	if opts.UpdatedBefore > 0 {
 		cond = cond.And(builder.Lt{"updated": opts.UpdatedBefore})
 	}
+	if opts.ConcurrencyGroup != "" {
+		cond = cond.And(builder.Eq{"concurrency_group": opts.ConcurrencyGroup})
+	}
 	return cond
 }

models/actions/run_list.go

@@ -64,14 +64,15 @@ func (runs RunList) LoadRepos(ctx context.Context) error {
 
 type FindRunOptions struct {
 	db.ListOptions
-	RepoID        int64
-	OwnerID       int64
-	WorkflowID    string
-	Ref           string // the commit/tag/… that caused this workflow
-	TriggerUserID int64
-	TriggerEvent  webhook_module.HookEventType
-	Approved      bool // not util.OptionalBool, it works only when it's true
-	Status        []Status
+	RepoID           int64
+	OwnerID          int64
+	WorkflowID       string
+	Ref              string // the commit/tag/… that caused this workflow
+	TriggerUserID    int64
+	TriggerEvent     webhook_module.HookEventType
+	Approved         bool // not util.OptionalBool, it works only when it's true
+	Status           []Status
+	ConcurrencyGroup string
 }
 
 func (opts FindRunOptions) ToConds() builder.Cond {
@@ -100,6 +101,9 @@ func (opts FindRunOptions) ToConds() builder.Cond {
 	if opts.TriggerEvent != "" {
 		cond = cond.And(builder.Eq{"trigger_event": opts.TriggerEvent})
 	}
+	if len(opts.ConcurrencyGroup) > 0 {
+		cond = cond.And(builder.Eq{"concurrency_group": opts.ConcurrencyGroup})
+	}
 	return cond
 }
 

models/actions/task.go

@@ -321,6 +321,10 @@ func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask
 		return nil, false, nil
 	}
 
+	if err := CancelPreviousJobsByConcurrency(ctx, job); err != nil {
+		return nil, false, err
+	}
+
 	task.Job = job
 
 	if err := committer.Commit(); err != nil {

models/migrations/migrations.go

@@ -381,6 +381,7 @@ func prepareMigrationTasks() []*migration {
 		newMigration(317, "Add new index for action for heatmap", v1_24.AddNewIndexForUserDashboard),
 		newMigration(318, "Add anonymous_access_mode for repo_unit", v1_24.AddRepoUnitAnonymousAccessMode),
 		newMigration(319, "Add ExclusiveOrder to Label table", v1_24.AddExclusiveOrderColumnToLabelTable),
+		newMigration(320, "Add support for actions concurrency", v1_24.AddActionsConcurrency),
 	}
 	return preparedMigrations
 }