github/gitea
1
0
Fork 0
mirror of https://github.com/go-gitea/gitea.git synced 2025-07-29 15:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 42c2ca2c91 into 82c9589faa

Browse Source
This commit is contained in:
Zettat123 2025-07-26 13:22:35 -06:00 committed by GitHub
commit 9c160abfef
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 2367 additions and 230 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

224
models/actions/run.go
View File

@ -22,7 +22,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"
)
@ -30,7 +29,7 @@ import (
type ActionRun struct {
ID int64
Title string
RepoID int64 `xorm:"index unique(repo_index)"`
RepoID int64 `xorm:"index 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 +48,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)"`
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
@ -181,7 +183,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 +240,73 @@ 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
}
run.Index = index
run.Title = util.EllipsisDisplayString(run.Title, 255)
if err := db.Insert(ctx, run); 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
}
if run.Repo == nil {
repo, err := repo_model.GetRepositoryByID(ctx, run.RepoID)
// 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 err
return cancelledJobs, err
}
run.Repo = repo
// 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 err := updateRepoRunsNumbers(ctx, run.Repo); 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 cancelledJobs, err
}
updatedJob, err := GetRunJobByID(ctx, job.ID)
if err != nil {
return cancelledJobs, fmt.Errorf("get job: %w", err)
}
cancelledJobs = append(cancelledJobs, updatedJob)
}
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 err := db.Insert(ctx, runJobs); err != nil {
return err
}
// Return nil to indicate successful cancellation of all running and waiting jobs.
return cancelledJobs, nil
}
// 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
})
func GetRunByID(ctx context.Context, id int64) (*ActionRun, error) {
var run ActionRun
has, err := db.GetEngine(ctx).Where("id=?", id).Get(&run)
if err != nil {
return nil, err
} else if !has {
return nil, fmt.Errorf("run with id %d: %w", id, util.ErrNotExist)
}
return &run, nil
}
func GetRunByRepoAndID(ctx context.Context, repoID, runID int64) (*ActionRun, error) {
@ -432,7 +391,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 +400,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)
}

91
models/actions/run_job.go
View File

@ -22,7 +22,7 @@ type ActionRunJob struct {
ID int64
RunID int64 `xorm:"index"`
Run *ActionRun `xorm:"-"`
RepoID int64 `xorm:"index"`
RepoID int64 `xorm:"index index(repo_concurrency)"`
Repo *repo_model.Repository `xorm:"-"`
OwnerID int64 `xorm:"index"`
CommitSHA string `xorm:"index"`
@ -35,10 +35,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"`
RawConcurrency string // raw concurrency
IsConcurrencyEvaluated bool // whether RawConcurrency has been evaluated, only valid when RawConcurrency is not empty
ConcurrencyGroup string `xorm:"index(repo_concurrency)"` // 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 +203,78 @@ func AggregateJobStatus(jobs []*ActionRunJob) Status {
return StatusUnknown // it shouldn't happen
}
}
func ShouldBlockJobByConcurrency(ctx context.Context, job *ActionRunJob) (bool, error) {
if job.RawConcurrency == "" {
return false, nil
}
if !job.IsConcurrencyEvaluated {
return false, ErrUnevaluatedConcurrency{
RawConcurrency: job.RawConcurrency,
}
}
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("find concurrent runs and jobs: %w", err)
}
return len(runs) > 0 || len(jobs) > 0, nil
}
func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) ([]*ActionRunJob, error) {
if job.RawConcurrency == "" {
return nil, nil
}
var jobsToCancel []*ActionRunJob
if !job.IsConcurrencyEvaluated {
return nil, ErrUnevaluatedConcurrency{
RawConcurrency: job.RawConcurrency,
}
}
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)
}
type ErrUnevaluatedConcurrency struct {
RawConcurrency string
}
func IsErrUnevaluatedConcurrency(err error) bool {
_, ok := err.(ErrUnevaluatedConcurrency)
return ok
}
func (err ErrUnevaluatedConcurrency) Error() string {
return fmt.Sprintf("the raw concurrency [%s] is not evaluated", err.RawConcurrency)
}

16
models/actions/run_job_list.go
View File

@ -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 {
@ -94,6 +95,9 @@ func (opts FindRunJobOptions) ToConds() builder.Cond {
if opts.UpdatedBefore > 0 {
cond = cond.And(builder.Lt{"`action_run_job`.updated": opts.UpdatedBefore})
}
if opts.ConcurrencyGroup != "" {
cond = cond.And(builder.Eq{"concurrency_group": opts.ConcurrencyGroup})
}
return cond
}

22
models/actions/run_list.go
View File

@ -64,15 +64,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
CommitSHA string
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
CommitSHA string
}
func (opts FindRunOptions) ToConds() builder.Cond {
@ -101,6 +102,9 @@ func (opts FindRunOptions) ToConds() builder.Cond {
if opts.CommitSHA != "" {
cond = cond.And(builder.Eq{"`action_run`.commit_sha": opts.CommitSHA})
}
if len(opts.ConcurrencyGroup) > 0 {
cond = cond.And(builder.Eq{"concurrency_group": opts.ConcurrencyGroup})
}
return cond
}

1
models/migrations/migrations.go
View File

@ -386,6 +386,7 @@ func prepareMigrationTasks() []*migration {
// Gitea 1.24.0 ends at database version 321
newMigration(321, "Use LONGTEXT for some columns and fix review_state.updated_files column", v1_25.UseLongTextInSomeColumnsAndFixBugs),
newMigration(322, "Add support for actions concurrency", v1_25.AddActionsConcurrency),
}
return preparedMigrations
}

31
models/migrations/v1_25/v322.go Normal file
View File

@ -0,0 +1,31 @@
// Copyright 2025 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package v1_25
import (
"xorm.io/xorm"
)
func AddActionsConcurrency(x *xorm.Engine) error {
type ActionRun struct {
RepoID int64 `xorm:"index unique(repo_index) index(repo_concurrency)"`
RawConcurrency string
ConcurrencyGroup string `xorm:"index(repo_concurrency)"`
ConcurrencyCancel bool
}
if err := x.Sync(new(ActionRun)); err != nil {
return err
}
type ActionRunJob struct {
RepoID int64 `xorm:"index index(repo_concurrency)"`
RawConcurrency string
IsConcurrencyEvaluated bool
ConcurrencyGroup string `xorm:"index(repo_concurrency)"`
ConcurrencyCancel bool
}
return x.Sync(new(ActionRunJob))
}

2
routers/api/actions/runner/runner.go
View File

@ -227,7 +227,7 @@ func (s *Service) UpdateTask(
}
if req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED {
if err := actions_service.EmitJobsIfReady(task.Job.RunID); err != nil {
if err := actions_service.EmitJobsIfReadyByRun(task.Job.RunID); err != nil {
log.Error("Emit ready jobs of run %d: %v", task.Job.RunID, err)
}
}

140
routers/web/repo/actions/view.go
View File

@ -27,7 +27,6 @@ import (
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/storage"
"code.gitea.io/gitea/modules/templates"
"code.gitea.io/gitea/modules/timeutil"
"code.gitea.io/gitea/modules/util"
"code.gitea.io/gitea/modules/web"
"code.gitea.io/gitea/routers/common"
@ -36,6 +35,7 @@ import (
notify_service "code.gitea.io/gitea/services/notify"
"github.com/nektos/act/pkg/model"
"gopkg.in/yaml.v3"
"xorm.io/builder"
)
@ -420,26 +420,68 @@ func Rerun(ctx *context_module.Context) {
return
}
// reset run's start and stop time when it is done
if run.Status.IsDone() {
run.PreviousDuration = run.Duration()
run.Started = 0
run.Stopped = 0
if err := actions_model.UpdateRun(ctx, run, "started", "stopped", "previous_duration"); err != nil {
ctx.ServerError("UpdateRun", err)
return
}
}
// check run (workflow-level) concurrency
job, jobs := getRunJobs(ctx, runIndex, jobIndex)
if ctx.Written() {
return
}
var blockRunByConcurrency bool
// reset run's start and stop time when it is done
if run.Status.IsDone() {
run.PreviousDuration = run.Duration()
run.Started = 0
run.Stopped = 0
vars, err := actions_model.GetVariablesOfRun(ctx, run)
if err != nil {
ctx.ServerError("GetVariablesOfRun", fmt.Errorf("get run %d variables: %w", run.ID, err))
return
}
if run.RawConcurrency != "" {
var rawConcurrency model.RawConcurrency
if err := yaml.Unmarshal([]byte(run.RawConcurrency), &rawConcurrency); err != nil {
ctx.ServerError("UnmarshalRawConcurrency", fmt.Errorf("unmarshal raw concurrency: %w", err))
return
}
wfConcurrencyGroup, wfConcurrencyCancel, err := actions_service.EvaluateWorkflowConcurrency(ctx, run, &rawConcurrency, vars)
if err != nil {
ctx.ServerError("EvaluateWorkflowConcurrency", fmt.Errorf("evaluate workflow concurrency: %w", err))
return
}
if wfConcurrencyGroup != "" {
run.ConcurrencyGroup = wfConcurrencyGroup
run.ConcurrencyCancel = wfConcurrencyCancel
}
blockRunByConcurrency, err = actions_model.ShouldBlockRunByConcurrency(ctx, run)
if err != nil {
ctx.ServerError("ShouldBlockRunByConcurrency", err)
return
}
if blockRunByConcurrency {
run.Status = actions_model.StatusBlocked
} else {
run.Status = actions_model.StatusRunning
}
if err := actions_service.CancelJobsByRunConcurrency(ctx, run); err != nil {
ctx.ServerError("cancel jobs", err)
return
}
}
if err := actions_model.UpdateRun(ctx, run, "started", "stopped", "previous_duration", "status", "concurrency_group", "concurrency_cancel"); err != nil {
ctx.ServerError("UpdateRun", err)
return
}
}
if jobIndexStr == "" { // rerun all jobs
for _, j := range jobs {
// if the job has needs, it should be set to "blocked" status to wait for other jobs
shouldBlock := len(j.Needs) > 0
shouldBlock := len(j.Needs) > 0 || blockRunByConcurrency
if err := rerunJob(ctx, j, shouldBlock); err != nil {
ctx.ServerError("RerunJob", err)
return
@ -453,7 +495,7 @@ func Rerun(ctx *context_module.Context) {
for _, j := range rerunJobs {
// jobs other than the specified one should be set to "blocked" status
shouldBlock := j.JobID != job.JobID
shouldBlock := j.JobID != job.JobID || blockRunByConcurrency
if err := rerunJob(ctx, j, shouldBlock); err != nil {
ctx.ServerError("RerunJob", err)
return
@ -477,8 +519,38 @@ func rerunJob(ctx *context_module.Context, job *actions_model.ActionRunJob, shou
job.Started = 0
job.Stopped = 0
job.ConcurrencyGroup = ""
job.ConcurrencyCancel = false
job.IsConcurrencyEvaluated = false
if err := job.LoadRun(ctx); err != nil {
return err
}
vars, err := actions_model.GetVariablesOfRun(ctx, job.Run)
if err != nil {
return fmt.Errorf("get run %d variables: %w", job.Run.ID, err)
}
if job.RawConcurrency != "" && job.Status != actions_model.StatusBlocked {
var err error
job.ConcurrencyGroup, job.ConcurrencyCancel, err = actions_service.EvaluateJobConcurrency(ctx, job.Run, job, vars, nil)
if err != nil {
return fmt.Errorf("evaluate job concurrency: %w", err)
}
job.IsConcurrencyEvaluated = true
blockByConcurrency, err := actions_model.ShouldBlockJobByConcurrency(ctx, job)
if err != nil {
return err
}
if blockByConcurrency {
job.Status = actions_model.StatusBlocked
}
if err := actions_service.CancelJobsByJobConcurrency(ctx, job); err != nil {
return fmt.Errorf("cancel jobs: %w", err)
}
}
if err := db.WithTx(ctx, func(ctx context.Context) error {
_, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": status}, "task_id", "status", "started", "stopped")
updateCols := []string{"task_id", "status", "started", "stopped", "concurrency_group", "concurrency_cancel", "is_concurrency_evaluated"}
_, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": status}, updateCols...)
return err
}); err != nil {
return err
@ -521,30 +593,11 @@ func Cancel(ctx *context_module.Context) {
var updatedjobs []*actions_model.ActionRunJob
if err := db.WithTx(ctx, func(ctx context.Context) error {
for _, job := range jobs {
status := job.Status
if status.IsDone() {
continue
}
if job.TaskID == 0 {
job.Status = actions_model.StatusCancelled
job.Stopped = timeutil.TimeStampNow()
n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped")
if err != nil {
return err
}
if n == 0 {
return errors.New("job has changed, try again")
}
if n > 0 {
updatedjobs = append(updatedjobs, job)
}
continue
}
if err := actions_model.StopTask(ctx, job.TaskID, actions_model.StatusCancelled); err != nil {
return err
}
cancelledJobs, err := actions_model.CancelJobs(ctx, jobs)
if err != nil {
return fmt.Errorf("cancel jobs: %w", err)
}
updatedjobs = append(updatedjobs, cancelledJobs...)
return nil
}); err != nil {
ctx.ServerError("StopTask", err)
@ -552,6 +605,7 @@ func Cancel(ctx *context_module.Context) {
}
actions_service.CreateCommitStatus(ctx, jobs...)
actions_service.EmitJobsIfReadyByJobs(updatedjobs)
for _, job := range updatedjobs {
_ = job.LoadAttributes(ctx)
@ -584,7 +638,17 @@ func Approve(ctx *context_module.Context) {
return err
}
for _, job := range jobs {
if len(job.Needs) == 0 && job.Status.IsBlocked() {
blockJobByConcurrency, err := actions_model.ShouldBlockJobByConcurrency(ctx, job)
if err != nil {
if actions_model.IsErrUnevaluatedConcurrency(err) {
continue
}
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 {

21
services/actions/clear_tasks.go
View File

@ -52,12 +52,26 @@ func notifyWorkflowJobStatusUpdate(ctx context.Context, jobs []*actions_model.Ac
func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error {
jobs, err := actions_model.CancelPreviousJobs(ctx, repoID, ref, workflowID, event)
notifyWorkflowJobStatusUpdate(ctx, jobs)
EmitJobsIfReadyByJobs(jobs)
return err
}
func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository) error {
jobs, err := actions_model.CleanRepoScheduleTasks(ctx, repo)
notifyWorkflowJobStatusUpdate(ctx, jobs)
EmitJobsIfReadyByJobs(jobs)
return err
}
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
}
@ -97,6 +111,7 @@ func stopTasks(ctx context.Context, opts actions_model.FindTaskOptions) error {
}
notifyWorkflowJobStatusUpdate(ctx, jobs)
EmitJobsIfReadyByJobs(jobs)
return nil
}
@ -105,7 +120,7 @@ func stopTasks(ctx context.Context, opts actions_model.FindTaskOptions) error {
func CancelAbandonedJobs(ctx context.Context) error {
jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{
Statuses: []actions_model.Status{actions_model.StatusWaiting, actions_model.StatusBlocked},
UpdatedBefore: timeutil.TimeStamp(time.Now().Add(-setting.Actions.AbandonedJobTimeout).Unix()),
UpdatedBefore: timeutil.TimeStampNow().AddDuration(-setting.Actions.AbandonedJobTimeout),
})
if err != nil {
log.Warn("find abandoned tasks: %v", err)
@ -113,6 +128,7 @@ func CancelAbandonedJobs(ctx context.Context) error {
}
now := timeutil.TimeStampNow()
var updatedJobs []*actions_model.ActionRunJob
for _, job := range jobs {
job.Status = actions_model.StatusCancelled
job.Stopped = now
@ -127,10 +143,13 @@ func CancelAbandonedJobs(ctx context.Context) error {
}
CreateCommitStatus(ctx, job)
if updated {
updatedJobs = append(updatedJobs, job)
NotifyWorkflowRunStatusUpdateWithReload(ctx, job)
notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
}
}
EmitJobsIfReadyByJobs(updatedJobs)
return nil
}

87
services/actions/concurrency.go Normal file
View File

@ -0,0 +1,87 @@
// Copyright 2025 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package actions
import (
"context"
"errors"
"fmt"
actions_model "code.gitea.io/gitea/models/actions"
"code.gitea.io/gitea/modules/json"
api "code.gitea.io/gitea/modules/structs"
"github.com/nektos/act/pkg/jobparser"
act_model "github.com/nektos/act/pkg/model"
"gopkg.in/yaml.v3"
)
func EvaluateWorkflowConcurrency(ctx context.Context, run *actions_model.ActionRun, rc *act_model.RawConcurrency, vars map[string]string) (string, bool, error) {
if err := run.LoadAttributes(ctx); err != nil {
return "", false, fmt.Errorf("run LoadAttributes: %w", err)
}
gitCtx := GenerateGiteaContext(run, nil)
jobResults := map[string]*jobparser.JobResult{"": {}}
inputs, err := getInputsFromRun(run)
if err != nil {
return "", false, fmt.Errorf("get inputs: %w", err)
}
concurrencyGroup, concurrencyCancel, err := jobparser.EvaluateConcurrency(rc, "", nil, gitCtx, jobResults, vars, inputs)
if err != nil {
return "", false, fmt.Errorf("evaluate concurrency: %w", err)
}
return concurrencyGroup, concurrencyCancel, nil
}
func EvaluateJobConcurrency(ctx context.Context, run *actions_model.ActionRun, actionRunJob *actions_model.ActionRunJob, vars map[string]string, jobResults map[string]*jobparser.JobResult) (string, bool, error) {
if err := actionRunJob.LoadAttributes(ctx); err != nil {
return "", false, fmt.Errorf("job LoadAttributes: %w", err)
}
var rawConcurrency act_model.RawConcurrency
if err := yaml.Unmarshal([]byte(actionRunJob.RawConcurrency), &rawConcurrency); err != nil {
return "", false, fmt.Errorf("unmarshal raw concurrency: %w", err)
}
gitCtx := GenerateGiteaContext(run, actionRunJob)
if jobResults == nil {
jobResults = map[string]*jobparser.JobResult{}
}
jobResults[actionRunJob.JobID] = &jobparser.JobResult{
Needs: actionRunJob.Needs,
}
inputs, err := getInputsFromRun(run)
if err != nil {
return "", false, fmt.Errorf("get inputs: %w", err)
}
singleWorkflows, err := jobparser.Parse(actionRunJob.WorkflowPayload)
if err != nil {
return "", false, fmt.Errorf("parse single workflow: %w", err)
} else if len(singleWorkflows) != 1 {
return "", false, errors.New("not single workflow")
}
_, singleWorkflowJob := singleWorkflows[0].Job()
concurrencyGroup, concurrencyCancel, err := jobparser.EvaluateConcurrency(&rawConcurrency, actionRunJob.JobID, singleWorkflowJob, gitCtx, jobResults, vars, inputs)
if err != nil {
return "", false, fmt.Errorf("evaluate concurrency: %w", err)
}
return concurrencyGroup, concurrencyCancel, nil
}
func getInputsFromRun(run *actions_model.ActionRun) (map[string]any, error) {
if run.Event != "workflow_dispatch" {
return map[string]any{}, nil
}
var payload api.WorkflowDispatchPayload
if err := json.Unmarshal([]byte(run.EventPayload), &payload); err != nil {
return nil, err
}
return payload.Inputs, nil
}

225
services/actions/job_emitter.go
View File

@ -10,6 +10,7 @@ import (
actions_model "code.gitea.io/gitea/models/actions"
"code.gitea.io/gitea/models/db"
"code.gitea.io/gitea/modules/container"
"code.gitea.io/gitea/modules/graceful"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/queue"
@ -25,7 +26,7 @@ type jobUpdate struct {
RunID int64
}
func EmitJobsIfReady(runID int64) error {
func EmitJobsIfReadyByRun(runID int64) error {
err := jobEmitterQueue.Push(&jobUpdate{
RunID: runID,
})
@ -35,47 +36,57 @@ func EmitJobsIfReady(runID int64) error {
return err
}
func EmitJobsIfReadyByJobs(jobs []*actions_model.ActionRunJob) {
checkedRuns := make(container.Set[int64])
for _, job := range jobs {
if !job.Status.IsDone() || checkedRuns.Contains(job.RunID) {
continue
}
if err := EmitJobsIfReadyByRun(job.RunID); err != nil {
log.Error("Check jobs of run %d: %v", job.RunID, err)
}
checkedRuns.Add(job.RunID)
}
}
func jobEmitterQueueHandler(items ...*jobUpdate) []*jobUpdate {
ctx := graceful.GetManager().ShutdownContext()
var ret []*jobUpdate
for _, update := range items {
if err := checkJobsOfRun(ctx, update.RunID); err != nil {
if err := checkJobsByRunID(ctx, update.RunID); err != nil {
log.Error("check run %d: %v", update.RunID, err)
ret = append(ret, update)
}
}
return ret
}
func checkJobsOfRun(ctx context.Context, runID int64) error {
jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: runID})
func checkJobsByRunID(ctx context.Context, runID int64) error {
run, err := actions_model.GetRunByID(ctx, runID)
if err != nil {
return err
return fmt.Errorf("get action run: %w", err)
}
var updatedjobs []*actions_model.ActionRunJob
var jobs, updatedJobs []*actions_model.ActionRunJob
if err := db.WithTx(ctx, func(ctx context.Context) error {
idToJobs := make(map[string][]*actions_model.ActionRunJob, len(jobs))
for _, job := range jobs {
idToJobs[job.JobID] = append(idToJobs[job.JobID], job)
// check jobs of the current run
if js, ujs, err := checkJobsOfRun(ctx, run); err != nil {
return err
} else {
jobs = append(jobs, js...)
updatedJobs = append(updatedJobs, ujs...)
}
updates := newJobStatusResolver(jobs).Resolve()
for _, job := range jobs {
if status, ok := updates[job.ID]; ok {
job.Status = status
if n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked}, "status"); err != nil {
return err
} else if n != 1 {
return fmt.Errorf("no affected for updating blocked job %v", job.ID)
}
updatedjobs = append(updatedjobs, job)
}
if js, ujs, err := checkRunConcurrency(ctx, run); err != nil {
return err
} else {
jobs = append(jobs, js...)
updatedJobs = append(updatedJobs, ujs...)
}
return nil
}); err != nil {
return err
}
CreateCommitStatus(ctx, jobs...)
for _, job := range updatedjobs {
for _, job := range updatedJobs {
_ = job.LoadAttributes(ctx)
notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
}
@ -94,6 +105,114 @@ func checkJobsOfRun(ctx context.Context, runID int64) error {
return nil
}
func findBlockedRunByConcurrency(ctx context.Context, repoID int64, concurrencyGroup string) (*actions_model.ActionRun, bool, error) {
if concurrencyGroup == "" {
return nil, false, nil
}
cRuns, cJobs, err := actions_model.GetConcurrentRunsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusBlocked})
if err != nil {
return nil, false, fmt.Errorf("find concurrent runs and jobs: %w", err)
}
// There can be at most one blocked run or job
var concurrentRun *actions_model.ActionRun
if len(cRuns) > 0 {
concurrentRun = cRuns[0]
} else if len(cJobs) > 0 {
jobRun, err := actions_model.GetRunByID(ctx, cJobs[0].RunID)
if err != nil {
return nil, false, fmt.Errorf("get run by job %d: %w", cJobs[0].ID, err)
}
concurrentRun = jobRun
}
return concurrentRun, concurrentRun != nil, nil
}
func checkRunConcurrency(ctx context.Context, run *actions_model.ActionRun) (jobs, updatedJobs []*actions_model.ActionRunJob, err error) {
checkedConcurrencyGroup := make(container.Set[string])
// check run (workflow-level) concurrency
if run.ConcurrencyGroup != "" {
concurrentRun, found, err := findBlockedRunByConcurrency(ctx, run.RepoID, run.ConcurrencyGroup)
if err != nil {
return nil, nil, fmt.Errorf("find blocked run by concurrency: %w", err)
}
if found && !concurrentRun.NeedApproval {
js, ujs, err := checkJobsOfRun(ctx, concurrentRun)
if err != nil {
return nil, nil, err
}
jobs = append(jobs, js...)
updatedJobs = append(updatedJobs, ujs...)
}
checkedConcurrencyGroup.Add(run.ConcurrencyGroup)
}
// check job concurrency
runJobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: run.ID})
if err != nil {
return nil, nil, fmt.Errorf("find run %d jobs: %w", run.ID, err)
}
for _, job := range runJobs {
if !job.Status.IsDone() {
continue
}
if job.ConcurrencyGroup == "" && checkedConcurrencyGroup.Contains(job.ConcurrencyGroup) {
continue
}
concurrentRun, found, err := findBlockedRunByConcurrency(ctx, job.RepoID, job.ConcurrencyGroup)
if err != nil {
return nil, nil, fmt.Errorf("find blocked run by concurrency: %w", err)
}
if found && !concurrentRun.NeedApproval {
js, ujs, err := checkJobsOfRun(ctx, concurrentRun)
if err != nil {
return nil, nil, err
}
jobs = append(jobs, js...)
updatedJobs = append(updatedJobs, ujs...)
}
checkedConcurrencyGroup.Add(job.ConcurrencyGroup)
}
return jobs, updatedJobs, nil
}
func checkJobsOfRun(ctx context.Context, run *actions_model.ActionRun) (jobs, updatedJobs []*actions_model.ActionRunJob, err error) {
jobs, err = db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: run.ID})
if err != nil {
return nil, nil, err
}
vars, err := actions_model.GetVariablesOfRun(ctx, run)
if err != nil {
return nil, nil, err
}
if err = db.WithTx(ctx, func(ctx context.Context) error {
for _, job := range jobs {
job.Run = run
}
updates := newJobStatusResolver(jobs, vars).Resolve(ctx)
for _, job := range jobs {
if status, ok := updates[job.ID]; ok {
job.Status = status
if n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked}, "status"); err != nil {
return err
} else if n != 1 {
return fmt.Errorf("no affected for updating blocked job %v", job.ID)
}
updatedJobs = append(updatedJobs, job)
}
}
return nil
}); err != nil {
return nil, nil, err
}
return jobs, updatedJobs, nil
}
func NotifyWorkflowRunStatusUpdateWithReload(ctx context.Context, job *actions_model.ActionRunJob) {
job.Run = nil
if err := job.LoadAttributes(ctx); err != nil {
@ -107,9 +226,10 @@ type jobStatusResolver struct {
statuses map[int64]actions_model.Status
needs map[int64][]int64
jobMap map[int64]*actions_model.ActionRunJob
vars map[string]string
}
func newJobStatusResolver(jobs actions_model.ActionJobList) *jobStatusResolver {
func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]string) *jobStatusResolver {
idToJobs := make(map[string][]*actions_model.ActionRunJob, len(jobs))
jobMap := make(map[int64]*actions_model.ActionRunJob)
for _, job := range jobs {
@ -131,13 +251,14 @@ func newJobStatusResolver(jobs actions_model.ActionJobList) *jobStatusResolver {
statuses: statuses,
needs: needs,
jobMap: jobMap,
vars: vars,
}
}
func (r *jobStatusResolver) Resolve() map[int64]actions_model.Status {
func (r *jobStatusResolver) Resolve(ctx context.Context) map[int64]actions_model.Status {
ret := map[int64]actions_model.Status{}
for i := 0; i < len(r.statuses); i++ {
updated := r.resolve()
updated := r.resolve(ctx)
if len(updated) == 0 {
return ret
}
@ -149,7 +270,7 @@ func (r *jobStatusResolver) Resolve() map[int64]actions_model.Status {
return ret
}
func (r *jobStatusResolver) resolve() map[int64]actions_model.Status {
func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model.Status {
ret := map[int64]actions_model.Status{}
for id, status := range r.statuses {
if status != actions_model.StatusBlocked {
@ -166,6 +287,21 @@ func (r *jobStatusResolver) resolve() map[int64]actions_model.Status {
}
}
if allDone {
// check concurrency
blockedByJobConcurrency, err := checkConcurrencyForJobWithNeeds(ctx, r.jobMap[id], r.vars)
if err != nil {
log.Error("Check job %d concurrency: %v. This job will stay blocked.", id, err)
continue
}
if blockedByJobConcurrency {
continue
}
if err := CancelJobsByJobConcurrency(ctx, r.jobMap[id]); err != nil {
log.Error("Cancel previous jobs for job %d: %v", id, err)
}
if allSucceed {
ret[id] = actions_model.StatusWaiting
} else {
@ -175,7 +311,6 @@ func (r *jobStatusResolver) resolve() map[int64]actions_model.Status {
_, wfJob := wfJobs[0].Job()
hasIf = len(wfJob.If.Value) > 0
}
if hasIf {
// act_runner will check the "if" condition
ret[id] = actions_model.StatusWaiting
@ -189,3 +324,39 @@ func (r *jobStatusResolver) resolve() map[int64]actions_model.Status {
}
return ret
}
func checkConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_model.ActionRunJob, vars map[string]string) (bool, error) {
if actionRunJob.RawConcurrency == "" {
return false, nil
}
if err := actionRunJob.LoadAttributes(ctx); err != nil {
return false, err
}
if !actionRunJob.IsConcurrencyEvaluated {
taskNeeds, err := FindTaskNeeds(ctx, actionRunJob)
if err != nil {
return false, fmt.Errorf("find task needs: %w", err)
}
jobResults := make(map[string]*jobparser.JobResult, len(taskNeeds))
for jobID, taskNeed := range taskNeeds {
jobResult := &jobparser.JobResult{
Result: taskNeed.Result.String(),
Outputs: taskNeed.Outputs,
}
jobResults[jobID] = jobResult
}
actionRunJob.ConcurrencyGroup, actionRunJob.ConcurrencyCancel, err = EvaluateJobConcurrency(ctx, actionRunJob.Run, actionRunJob, vars, jobResults)
if err != nil {
return false, fmt.Errorf("evaluate job concurrency: %w", err)
}
actionRunJob.IsConcurrencyEvaluated = true
if _, err := actions_model.UpdateRunJob(ctx, actionRunJob, nil, "concurrency_group", "concurrency_cancel", "is_concurrency_evaluated"); err != nil {
return false, fmt.Errorf("update run job: %w", err)
}
}
return actions_model.ShouldBlockJobByConcurrency(ctx, actionRunJob)
}

4
services/actions/job_emitter_test.go
View File

@ -129,8 +129,8 @@ jobs:
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
r := newJobStatusResolver(tt.jobs)
assert.Equal(t, tt.want, r.Resolve())
r := newJobStatusResolver(tt.jobs, nil)
assert.Equal(t, tt.want, r.Resolve(t.Context()))
})
}
}

40
services/actions/notifier_helper.go
View File

@ -31,6 +31,7 @@ import (
"github.com/nektos/act/pkg/jobparser"
"github.com/nektos/act/pkg/model"
"gopkg.in/yaml.v3"
)
type methodCtxKeyType struct{}
@ -357,6 +358,29 @@ func handleWorkflows(
continue
}
wfRawConcurrency, err := jobparser.ReadWorkflowRawConcurrency(dwf.Content)
if err != nil {
log.Error("ReadWorkflowRawConcurrency: %v", err)
continue
}
if wfRawConcurrency != nil {
rawConcurrency, err := yaml.Marshal(wfRawConcurrency)
if err != nil {
log.Error("Marshal raw concurrency: %v", err)
continue
}
run.RawConcurrency = string(rawConcurrency)
wfConcurrencyGroup, wfConcurrencyCancel, err := EvaluateWorkflowConcurrency(ctx, run, wfRawConcurrency, vars)
if err != nil {
log.Error("EvaluateWorkflowConcurrency: %v", err)
continue
}
if wfConcurrencyGroup != "" {
run.ConcurrencyGroup = wfConcurrencyGroup
run.ConcurrencyCancel = wfConcurrencyCancel
}
}
giteaCtx := GenerateGiteaContext(run, nil)
jobs, err := jobparser.Parse(dwf.Content, jobparser.WithVars(vars), jobparser.WithGitContext(giteaCtx.ToGitHubContext()))
@ -369,21 +393,7 @@ func handleWorkflows(
run.Title = jobs[0].RunName
}
// cancel running jobs if the event is push or pull_request_sync
if run.Event == webhook_module.HookEventPush ||
run.Event == webhook_module.HookEventPullRequestSync {
if err := CancelPreviousJobs(
ctx,
run.RepoID,
run.Ref,
run.WorkflowID,
run.Event,
); err != nil {
log.Error("CancelPreviousJobs: %v", err)
}
}
if err := actions_model.InsertRun(ctx, run, jobs); err != nil {
if err := InsertRun(ctx, run, jobs); err != nil {
log.Error("InsertRun: %v", err)
continue
}

146
services/actions/run.go Normal file
View File

@ -0,0 +1,146 @@
// Copyright 2025 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package actions
import (
"context"
"fmt"
actions_model "code.gitea.io/gitea/models/actions"
"code.gitea.io/gitea/models/db"
repo_model "code.gitea.io/gitea/models/repo"
"code.gitea.io/gitea/modules/util"
"github.com/nektos/act/pkg/jobparser"
"gopkg.in/yaml.v3"
)
// 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 *actions_model.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
}
run.Index = index
run.Title = util.EllipsisDisplayString(run.Title, 255)
// check run (workflow-level) concurrency
blockRunByConcurrency, err := actions_model.ShouldBlockRunByConcurrency(ctx, run)
if err != nil {
return err
}
if blockRunByConcurrency {
run.Status = actions_model.StatusBlocked
}
if err := CancelJobsByRunConcurrency(ctx, run); err != nil {
return fmt.Errorf("cancel jobs: %w", err)
}
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
}
if err := actions_model.UpdateRepoRunsNumbers(ctx, run.Repo); err != nil {
return err
}
// query vars for evaluating job concurrency groups
vars, err := actions_model.GetVariablesOfRun(ctx, run)
if err != nil {
return fmt.Errorf("get run %d variables: %w", run.ID, err)
}
runJobs := make([]*actions_model.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 := actions_model.StatusWaiting
if len(needs) > 0 || run.NeedApproval || run.Status == actions_model.StatusBlocked {
status = actions_model.StatusBlocked
} else {
hasWaiting = true
}
job.Name = util.EllipsisDisplayString(job.Name, 255)
runJob := &actions_model.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,
}
// check job concurrency
if job.RawConcurrency != nil {
rawConcurrency, err := yaml.Marshal(job.RawConcurrency)
if err != nil {
return fmt.Errorf("marshal raw concurrency: %w", err)
}
runJob.RawConcurrency = string(rawConcurrency)
// do not evaluate job concurrency when it requires `needs`
if len(needs) == 0 {
var err error
runJob.ConcurrencyGroup, runJob.ConcurrencyCancel, err = EvaluateJobConcurrency(ctx, run, runJob, vars, nil)
if err != nil {
return fmt.Errorf("evaluate job concurrency: %w", err)
}
runJob.IsConcurrencyEvaluated = true
}
// do not need to check job concurrency if the job is blocked because it will be checked by job emitter
if runJob.Status != actions_model.StatusBlocked {
// check if the job should be blocked by job concurrency
blockByConcurrency, err := actions_model.ShouldBlockJobByConcurrency(ctx, runJob)
if err != nil {
return err
}
if blockByConcurrency {
runJob.Status = actions_model.StatusBlocked
}
if err := CancelJobsByJobConcurrency(ctx, runJob); err != nil {
return fmt.Errorf("cancel jobs: %w", err)
}
}
}
if err := db.Insert(ctx, runJob); err != nil {
return err
}
runJobs = append(runJobs, runJob)
}
run.Status = actions_model.AggregateJobStatus(runJobs)
if err := actions_model.UpdateRun(ctx, run, "status"); err != nil {
return err
}
// if there is a job in the waiting status, increase tasks version.
if hasWaiting {
if err := actions_model.IncreaseTaskVersion(ctx, run.OwnerID, run.RepoID); err != nil {
return err
}
}
return nil
})
}

36
services/actions/schedule_tasks.go
View File

@ -18,6 +18,7 @@ import (
notify_service "code.gitea.io/gitea/services/notify"
"github.com/nektos/act/pkg/jobparser"
"gopkg.in/yaml.v3"
)
// StartScheduleTasks start the task
@ -53,20 +54,6 @@ func startTasks(ctx context.Context) error {
// Loop through each spec and create a schedule task for it
for _, row := range specs {
// cancel running jobs if the event is push
if row.Schedule.Event == webhook_module.HookEventPush {
// cancel running jobs of the same workflow
if err := CancelPreviousJobs(
ctx,
row.RepoID,
row.Schedule.Ref,
row.Schedule.WorkflowID,
webhook_module.HookEventSchedule,
); err != nil {
log.Error("CancelPreviousJobs: %v", err)
}
}
if row.Repo.IsArchived {
// Skip if the repo is archived
continue
@ -144,9 +131,28 @@ func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule)
if err != nil {
return err
}
wfRawConcurrency, err := jobparser.ReadWorkflowRawConcurrency(cron.Content)
if err != nil {
return err
}
if wfRawConcurrency != nil {
rawConcurrency, err := yaml.Marshal(wfRawConcurrency)
if err != nil {
return fmt.Errorf("marshal raw concurrency: %w", err)
}
run.RawConcurrency = string(rawConcurrency)
wfConcurrencyGroup, wfConcurrencyCancel, err := EvaluateWorkflowConcurrency(ctx, run, wfRawConcurrency, vars)
if err != nil {
return err
}
if wfConcurrencyGroup != "" {
run.ConcurrencyGroup = wfConcurrencyGroup
run.ConcurrencyCancel = wfConcurrencyCancel
}
}
// Insert the action run and its associated jobs into the database
if err := actions_model.InsertRun(ctx, run, workflows); err != nil {
if err := InsertRun(ctx, run, workflows); err != nil {
return err
}
allJobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: run.ID})

37
services/actions/workflow.go
View File

@ -26,6 +26,7 @@ import (
"github.com/nektos/act/pkg/jobparser"
"github.com/nektos/act/pkg/model"
"gopkg.in/yaml.v3"
)
func EnableOrDisableWorkflow(ctx *context.APIContext, workflowID string, isEnable bool) error {
@ -99,6 +100,7 @@ func DispatchActionWorkflow(ctx reqctx.RequestContext, doer *user_model.User, re
// find workflow from commit
var workflows []*jobparser.SingleWorkflow
var entry *git.TreeEntry
var wfRawConcurrency *model.RawConcurrency
run := &actions_model.ActionRun{
Title: strings.SplitN(runTargetCommit.CommitMessage, "\n", 2)[0],
@ -154,6 +156,11 @@ func DispatchActionWorkflow(ctx reqctx.RequestContext, doer *user_model.User, re
)
}
wfRawConcurrency, err = jobparser.ReadWorkflowRawConcurrency(content)
if err != nil {
return err
}
// get inputs from post
workflow := &model.Workflow{
RawOn: workflows[0].RawOn,
@ -182,19 +189,29 @@ func DispatchActionWorkflow(ctx reqctx.RequestContext, doer *user_model.User, re
}
run.EventPayload = string(eventPayload)
// cancel running jobs of the same workflow
if err := CancelPreviousJobs(
ctx,
run.RepoID,
run.Ref,
run.WorkflowID,
run.Event,
); err != nil {
log.Error("CancelRunningJobs: %v", err)
// cancel running jobs of the same concurrency group
if wfRawConcurrency != nil {
vars, err := actions_model.GetVariablesOfRun(ctx, run)
if err != nil {
return err
}
rawConcurrency, err := yaml.Marshal(wfRawConcurrency)
if err != nil {
return fmt.Errorf("marshal raw concurrency: %w", err)
}
run.RawConcurrency = string(rawConcurrency)
wfConcurrencyGroup, wfConcurrencyCancel, err := EvaluateWorkflowConcurrency(ctx, run, wfRawConcurrency, vars)
if err != nil {
return err
}
if wfConcurrencyGroup != "" {
run.ConcurrencyGroup = wfConcurrencyGroup
run.ConcurrencyCancel = wfConcurrencyCancel
}
}
// Insert the action run and its associated jobs into the database
if err := actions_model.InsertRun(ctx, run, workflows); err != nil {
if err := InsertRun(ctx, run, workflows); err != nil {
return fmt.Errorf("InsertRun: %w", err)
}

1455
tests/integration/actions_concurrency_test.go Normal file
View File

File diff suppressed because it is too large Load Diff

19
tests/integration/actions_runner_test.go
View File

@ -93,7 +93,20 @@ func (r *mockRunner) registerAsRepoRunner(t *testing.T, ownerName, repoName, run
}
func (r *mockRunner) fetchTask(t *testing.T, timeout ...time.Duration) *runnerv1.Task {
fetchTimeout := 10 * time.Second
task := r.tryFetchTask(t, timeout...)
assert.NotNil(t, task, "failed to fetch a task")
return task
}
func (r *mockRunner) fetchNoTask(t *testing.T, timeout ...time.Duration) {
task := r.tryFetchTask(t, timeout...)
assert.Nil(t, task, "a task is fetched")
}
const defaultFetchTaskTimeout = 1 * time.Second
func (r *mockRunner) tryFetchTask(t *testing.T, timeout ...time.Duration) *runnerv1.Task {
fetchTimeout := defaultFetchTaskTimeout
if len(timeout) > 0 {
fetchTimeout = timeout[0]
}
@ -108,9 +121,9 @@ func (r *mockRunner) fetchTask(t *testing.T, timeout ...time.Duration) *runnerv1
task = resp.Msg.Task
break
}
time.Sleep(time.Second)
time.Sleep(200 * time.Millisecond)
}
assert.NotNil(t, task, "failed to fetch a task")
return task
}