WIP: Support concurrency for Actions

models/actions/run.go

@@ -20,7 +20,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"
 )
 
@@ -47,6 +46,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
@@ -196,13 +197,20 @@ func updateRepoRunsNumbers(ctx context.Context, repo *repo_model.Repository) err
 // It's useful when a new run is triggered, and all previous runs needn't be continued anymore.
 func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error {
 	// Find all runs in the specified repository, reference, and workflow with non-final status
-	runs, total, err := db.FindAndCount[ActionRun](ctx, FindRunOptions{
+	opts := &FindRunOptions{
 		RepoID:       repoID,
 		Ref:          ref,
 		WorkflowID:   workflowID,
 		TriggerEvent: event,
 		Status:       []Status{StatusRunning, StatusWaiting, StatusBlocked},
-	})
+	}
+	return CancelPreviousJobsWithOpts(ctx, opts)
+}
+
+// CancelPreviousJobs cancels all previous jobs with opts
+func CancelPreviousJobsWithOpts(ctx context.Context, opts *FindRunOptions) error {
+	// Find all runs by opts
+	runs, total, err := db.FindAndCount[ActionRun](ctx, opts)
 	if err != nil {
 		return err
 	}
@@ -222,48 +230,55 @@ func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID strin
 			return 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 err := CancelJobs(ctx, jobs); 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()
+	// Return nil to indicate successful cancellation of all running and waiting jobs.
+	return nil
+}
 
-				// 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
-				}
+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
+		}
 
-				// 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")
-				}
+		// 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()
 
-				// Continue with the next job.
-				continue
+			// 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 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
+			// 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")
 			}
+
+			// Continue with the next job.
+			continue
 		}
-	}
 
-	// Return nil to indicate successful cancellation of all running and waiting jobs.
+		// 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 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 {
+func InsertRun(ctx context.Context, run *ActionRun, runJobs []*ActionRunJob) error {
 	ctx, committer, err := db.TxContext(ctx)
 	if err != nil {
 		return err
@@ -277,6 +292,32 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
 	run.Index = index
 	run.Title, _ = util.SplitStringAtByteN(run.Title, 255)
 
+	blockedByWorkflowConcurrency := false
+	if len(run.ConcurrencyGroup) > 0 {
+		if run.ConcurrencyCancel {
+			if err := CancelPreviousJobsWithOpts(ctx, &FindRunOptions{
+				RepoID:           run.RepoID,
+				ConcurrencyGroup: run.ConcurrencyGroup,
+				Status:           []Status{StatusRunning, StatusWaiting, StatusBlocked},
+			}); err != nil {
+				return err
+			}
+		} else {
+			runningConcurrentRunsNum, err := db.Count[ActionRun](ctx, &FindRunOptions{
+				RepoID:           run.RepoID,
+				ConcurrencyGroup: run.ConcurrencyGroup,
+				Status:           []Status{StatusRunning},
+			})
+			if err != nil {
+				return err
+			}
+			blockedByWorkflowConcurrency = runningConcurrentRunsNum > 0
+		}
+	}
+	if blockedByWorkflowConcurrency {
+		run.Status = StatusBlocked
+	}
+
 	if err := db.Insert(ctx, run); err != nil {
 		return err
 	}
@@ -293,35 +334,28 @@ func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWork
 		return err
 	}
 
-	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
+	for _, job := range runJobs {
+		job.RunID = run.ID
+		if job.Status != StatusBlocked {
+			if blockedByWorkflowConcurrency {
+				// the job should also be blocked when the run is blocked
+				job.Status = StatusBlocked
+			} else if len(job.ConcurrencyGroup) > 0 {
+				// check if the job should be blocked by job concurrency
+				shouldBlock, err := ShouldJobBeBlockedByConcurrentJobs(ctx, job)
+				if err != nil {
+					return err
+				}
+				if shouldBlock {
+					job.Status = StatusBlocked
+				}
+			}
 		}
-		payload, _ := v.Marshal()
-		status := StatusWaiting
-		if len(needs) > 0 || run.NeedApproval {
-			status = StatusBlocked
-		} else {
+
+		if job.Status == StatusWaiting {
 			hasWaiting = true
 		}
-		job.Name, _ = util.SplitStringAtByteN(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
@@ -435,3 +469,34 @@ 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) {
+	waitingConcurrentJobsNum, err := db.Count[ActionRunJob](ctx, FindRunJobOptions{
+		RepoID:           actionRunJob.RepoID,
+		ConcurrencyGroup: actionRunJob.ConcurrencyGroup,
+		Statuses:         []Status{StatusWaiting},
+	})
+	if err != nil {
+		return false, fmt.Errorf("count waiting jobs: %w", err)
+	}
+
+	return waitingConcurrentJobsNum > 0, nil
+}

models/actions/run_job.go

@@ -33,10 +33,16 @@ 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
+	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() {

models/actions/run_job_list.go

@@ -48,12 +48,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 {
@@ -76,5 +77,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

@@ -63,14 +63,16 @@ 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
+	SortType         string
+	ConcurrencyGroup string
 }
 
 func (opts FindRunOptions) ToConds() builder.Cond {
@@ -99,11 +101,21 @@ 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
 }
 
 func (opts FindRunOptions) ToOrders() string {
-	return "`id` DESC"
+	switch opts.SortType {
+	case "oldest":
+		return "created ASC"
+	case "newest":
+		return "created DESC"
+	default:
+		return "`id` DESC"
+	}
 }
 
 type StatusInfo struct {

models/migrations/migrations.go

@@ -369,6 +369,7 @@ func prepareMigrationTasks() []*migration {
 		newMigration(309, "Improve Notification table indices", v1_23.ImproveNotificationTableIndices),
 		newMigration(310, "Add Priority to ProtectedBranch", v1_23.AddPriorityToProtectedBranch),
 		newMigration(311, "Add TimeEstimate to Issue table", v1_23.AddTimeEstimateColumnToIssueTable),
+		// TODO: AddActionsConcurrency
 	}
 	return preparedMigrations
 }

models/migrations/v1_23/v312.go

@@ -0,0 +1,28 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package v1_23 //nolint
+
+import (
+	"xorm.io/xorm"
+)
+
+func AddActionsConcurrency(x *xorm.Engine) error {
+	type ActionRun struct {
+		ConcurrencyGroup  string
+		ConcurrencyCancel bool
+	}
+
+	if err := x.Sync(new(ActionRun)); err != nil {
+		return err
+	}
+
+	type ActionRunJob struct {
+		RawConcurrencyGroup  string
+		RawConcurrencyCancel string
+		ConcurrencyGroup     string
+		ConcurrencyCancel    bool
+	}
+
+	return x.Sync(new(ActionRunJob))
+}