mirror of
https://github.com/go-gitea/gitea.git
synced 2025-07-22 07:52:04 +02:00
Merge 91fccb7cc87a8b5f7c012b4b3017884178b698f8 into 6599efb3b1400ac06d06e1c8b68ae6037fbb7952
This commit is contained in:
commit
411d4fdc32
@ -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"
|
||||
)
|
||||
|
||||
@ -49,6 +48,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 `xorm:"index"`
|
||||
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 +182,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,121 +239,69 @@ 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 {
|
||||
ctx, committer, err := db.TxContext(ctx)
|
||||
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 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)
|
||||
}
|
||||
|
||||
// Return nil to indicate successful cancellation of all running and waiting jobs.
|
||||
return cancelledJobs, 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 err
|
||||
}
|
||||
defer committer.Close()
|
||||
|
||||
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
|
||||
return nil, err
|
||||
} else if !has {
|
||||
return nil, fmt.Errorf("run with id %d: %w", id, util.ErrNotExist)
|
||||
}
|
||||
|
||||
if run.Repo == nil {
|
||||
repo, err := repo_model.GetRepositoryByID(ctx, run.RepoID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
run.Repo = repo
|
||||
}
|
||||
|
||||
if err := updateRepoRunsNumbers(ctx, run.Repo); err != nil {
|
||||
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
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// 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 committer.Commit()
|
||||
return &run, nil
|
||||
}
|
||||
|
||||
func GetRunByRepoAndID(ctx context.Context, repoID, runID int64) (*ActionRun, error) {
|
||||
@ -437,7 +386,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
|
||||
}
|
||||
}
|
||||
@ -446,3 +395,55 @@ 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
|
||||
}
|
||||
|
||||
func CancelPreviousJobsByRunConcurrency(ctx context.Context, actionRun *ActionRun) ([]*ActionRunJob, error) {
|
||||
var cancelledJobs []*ActionRunJob
|
||||
|
||||
if actionRun.ConcurrencyGroup != "" && actionRun.ConcurrencyCancel {
|
||||
// cancel previous runs in the same concurrency group
|
||||
runs, err := db.Find[ActionRun](ctx, &FindRunOptions{
|
||||
RepoID: actionRun.RepoID,
|
||||
ConcurrencyGroup: actionRun.ConcurrencyGroup,
|
||||
Status: []Status{StatusRunning, StatusWaiting, StatusBlocked},
|
||||
})
|
||||
if err != nil {
|
||||
return cancelledJobs, fmt.Errorf("find runs: %w", err)
|
||||
}
|
||||
for _, run := range runs {
|
||||
if run.ID == actionRun.ID {
|
||||
continue
|
||||
}
|
||||
jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{
|
||||
RunID: run.ID,
|
||||
})
|
||||
if err != nil {
|
||||
return cancelledJobs, fmt.Errorf("find run %d jobs: %w", run.ID, err)
|
||||
}
|
||||
cjs, err := CancelJobs(ctx, jobs)
|
||||
if err != nil {
|
||||
return cancelledJobs, fmt.Errorf("cancel run %d jobs: %w", run.ID, err)
|
||||
}
|
||||
cancelledJobs = append(cancelledJobs, cjs...)
|
||||
}
|
||||
}
|
||||
|
||||
return cancelledJobs, nil
|
||||
}
|
||||
|
@ -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 `xorm:"index"` // 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,92 @@ 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 CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) ([]*ActionRunJob, error) {
|
||||
var cancelledJobs []*ActionRunJob
|
||||
|
||||
if job.RawConcurrencyGroup != "" {
|
||||
if !job.IsConcurrencyEvaluated {
|
||||
return cancelledJobs, 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 cancelledJobs, fmt.Errorf("find previous jobs: %w", err)
|
||||
}
|
||||
previousJobs = slices.DeleteFunc(previousJobs, func(j *ActionRunJob) bool { return j.ID == job.ID })
|
||||
cjs, err := CancelJobs(ctx, previousJobs)
|
||||
if err != nil {
|
||||
return cancelledJobs, fmt.Errorf("cancel previous jobs: %w", err)
|
||||
}
|
||||
cancelledJobs = append(cancelledJobs, cjs...)
|
||||
}
|
||||
}
|
||||
|
||||
if err := job.LoadRun(ctx); err != nil {
|
||||
return cancelledJobs, fmt.Errorf("load run: %w", err)
|
||||
}
|
||||
|
||||
cancelledJobsByRun, err := CancelPreviousJobsByRunConcurrency(ctx, job.Run)
|
||||
if err != nil {
|
||||
return cancelledJobs, fmt.Errorf("cancel runs: %w", err)
|
||||
}
|
||||
cancelledJobs = append(cancelledJobs, cancelledJobsByRun...)
|
||||
|
||||
return cancelledJobs, 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)
|
||||
}
|
||||
|
@ -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
|
||||
}
|
||||
|
||||
|
@ -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
|
||||
}
|
||||
|
||||
|
@ -24,6 +24,7 @@ import (
|
||||
"code.gitea.io/gitea/models/migrations/v1_22"
|
||||
"code.gitea.io/gitea/models/migrations/v1_23"
|
||||
"code.gitea.io/gitea/models/migrations/v1_24"
|
||||
"code.gitea.io/gitea/models/migrations/v1_25"
|
||||
"code.gitea.io/gitea/models/migrations/v1_6"
|
||||
"code.gitea.io/gitea/models/migrations/v1_7"
|
||||
"code.gitea.io/gitea/models/migrations/v1_8"
|
||||
@ -382,6 +383,10 @@ func prepareMigrationTasks() []*migration {
|
||||
newMigration(318, "Add anonymous_access_mode for repo_unit", v1_24.AddRepoUnitAnonymousAccessMode),
|
||||
newMigration(319, "Add ExclusiveOrder to Label table", v1_24.AddExclusiveOrderColumnToLabelTable),
|
||||
newMigration(320, "Migrate two_factor_policy to login_source table", v1_24.MigrateSkipTwoFactor),
|
||||
|
||||
// Gitea 1.24.0-rc0 ends at migration ID number 320 (database version 321)
|
||||
|
||||
newMigration(321, "Add support for actions concurrency", v1_25.AddActionsConcurrency),
|
||||
}
|
||||
return preparedMigrations
|
||||
}
|
||||
|
29
models/migrations/v1_25/v321.go
Normal file
29
models/migrations/v1_25/v321.go
Normal file
@ -0,0 +1,29 @@
|
||||
// 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 {
|
||||
ConcurrencyGroup string `xorm:"index"`
|
||||
ConcurrencyCancel bool
|
||||
}
|
||||
|
||||
if err := x.Sync(new(ActionRun)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
type ActionRunJob struct {
|
||||
RawConcurrencyGroup string
|
||||
RawConcurrencyCancel string
|
||||
IsConcurrencyEvaluated bool
|
||||
ConcurrencyGroup string `xorm:"index"`
|
||||
ConcurrencyCancel bool
|
||||
}
|
||||
|
||||
return x.Sync(new(ActionRunJob))
|
||||
}
|
@ -420,26 +420,45 @@ 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
|
||||
}
|
||||
}
|
||||
// TODO evaluate concurrency expression again, vars may change after the run is done
|
||||
// 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
|
||||
|
||||
blockRunByConcurrency, err = actions_model.ShouldBlockRunByConcurrency(ctx, run)
|
||||
if err != nil {
|
||||
ctx.ServerError("ShouldBlockRunByConcurrency", err)
|
||||
return
|
||||
}
|
||||
if blockRunByConcurrency {
|
||||
run.Status = actions_model.StatusBlocked
|
||||
} else if err := actions_service.CancelJobsByRunConcurrency(ctx, run); err != nil {
|
||||
ctx.ServerError("cancel jobs", err)
|
||||
return
|
||||
} else {
|
||||
run.Status = actions_model.StatusRunning
|
||||
}
|
||||
if err := actions_model.UpdateRun(ctx, run, "started", "stopped", "previous_duration", "status"); 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 +472,8 @@ 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
|
||||
// TODO respect blockRunByConcurrency here?
|
||||
shouldBlock := j.JobID != job.JobID || blockRunByConcurrency
|
||||
if err := rerunJob(ctx, j, shouldBlock); err != nil {
|
||||
ctx.ServerError("RerunJob", err)
|
||||
return
|
||||
@ -477,8 +497,37 @@ 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.RawConcurrencyGroup != "" && 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
|
||||
} else 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
|
||||
@ -584,7 +633,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 {
|
||||
|
@ -61,6 +61,18 @@ func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository) er
|
||||
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
|
||||
}
|
||||
|
||||
func stopTasks(ctx context.Context, opts actions_model.FindTaskOptions) error {
|
||||
tasks, err := db.Find[actions_model.ActionTask](ctx, opts)
|
||||
if err != nil {
|
||||
|
86
services/actions/concurrency.go
Normal file
86
services/actions/concurrency.go
Normal file
@ -0,0 +1,86 @@
|
||||
// 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"
|
||||
)
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
rawConcurrency := &act_model.RawConcurrency{
|
||||
Group: actionRunJob.RawConcurrencyGroup,
|
||||
CancelInProgress: actionRunJob.RawConcurrencyCancel,
|
||||
}
|
||||
|
||||
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
|
||||
}
|
@ -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"
|
||||
@ -39,35 +40,116 @@ 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...)
|
||||
}
|
||||
// check run (workflow-level) concurrency
|
||||
concurrentRunIDs := make(container.Set[int64])
|
||||
concurrentRunIDs.Add(run.ID)
|
||||
if run.ConcurrencyGroup != "" {
|
||||
concurrentRuns, err := db.Find[actions_model.ActionRun](ctx, actions_model.FindRunOptions{
|
||||
RepoID: run.RepoID,
|
||||
ConcurrencyGroup: run.ConcurrencyGroup,
|
||||
Status: []actions_model.Status{actions_model.StatusBlocked},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, concurrentRun := range concurrentRuns {
|
||||
if concurrentRunIDs.Contains(concurrentRun.ID) {
|
||||
continue
|
||||
}
|
||||
concurrentRunIDs.Add(concurrentRun.ID)
|
||||
if concurrentRun.NeedApproval {
|
||||
continue
|
||||
}
|
||||
if js, ujs, err := checkJobsOfRun(ctx, concurrentRun); err != nil {
|
||||
return err
|
||||
} else {
|
||||
jobs = append(jobs, js...)
|
||||
updatedJobs = append(updatedJobs, ujs...)
|
||||
}
|
||||
updatedRun, err := actions_model.GetRunByID(ctx, concurrentRun.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if updatedRun.Status == actions_model.StatusWaiting {
|
||||
// only run one blocked action run in the same concurrency group
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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 {
|
||||
// check job concurrency
|
||||
runJobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: run.ID})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, job := range runJobs {
|
||||
if job.Status.IsDone() && job.ConcurrencyGroup != "" {
|
||||
waitingConcurrentJobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{
|
||||
RepoID: job.RepoID,
|
||||
ConcurrencyGroup: job.ConcurrencyGroup,
|
||||
Statuses: []actions_model.Status{actions_model.StatusWaiting},
|
||||
})
|
||||
if 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 len(waitingConcurrentJobs) == 0 {
|
||||
blockedConcurrentJobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{
|
||||
RepoID: job.RepoID,
|
||||
ConcurrencyGroup: job.ConcurrencyGroup,
|
||||
Statuses: []actions_model.Status{actions_model.StatusBlocked},
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, concurrentJob := range blockedConcurrentJobs {
|
||||
if concurrentRunIDs.Contains(concurrentJob.RunID) {
|
||||
continue
|
||||
}
|
||||
concurrentRunIDs.Add(concurrentJob.RunID)
|
||||
concurrentRun, err := actions_model.GetRunByID(ctx, concurrentJob.RunID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if concurrentRun.NeedApproval {
|
||||
continue
|
||||
}
|
||||
if js, ujs, err := checkJobsOfRun(ctx, concurrentRun); err != nil {
|
||||
return err
|
||||
} else {
|
||||
jobs = append(jobs, js...)
|
||||
updatedJobs = append(updatedJobs, ujs...)
|
||||
}
|
||||
updatedJob, err := actions_model.GetRunJobByID(ctx, concurrentJob.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if updatedJob.Status == actions_model.StatusWaiting {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
@ -75,7 +157,7 @@ func checkJobsOfRun(ctx context.Context, runID int64) error {
|
||||
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 +176,41 @@ func checkJobsOfRun(ctx context.Context, runID int64) error {
|
||||
return 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 +224,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 +249,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 +268,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 +285,19 @@ 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
|
||||
} else 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 {
|
||||
@ -189,3 +321,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.RawConcurrencyGroup == "" {
|
||||
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)
|
||||
}
|
||||
|
@ -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()))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
@ -357,6 +357,23 @@ func handleWorkflows(
|
||||
continue
|
||||
}
|
||||
|
||||
wfRawConcurrency, err := jobparser.ReadWorkflowRawConcurrency(dwf.Content)
|
||||
if err != nil {
|
||||
log.Error("ReadWorkflowRawConcurrency: %v", err)
|
||||
continue
|
||||
}
|
||||
if wfRawConcurrency != nil {
|
||||
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 +386,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
|
||||
}
|
||||
|
142
services/actions/run.go
Normal file
142
services/actions/run.go
Normal file
@ -0,0 +1,142 @@
|
||||
// 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"
|
||||
)
|
||||
|
||||
// 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 {
|
||||
ctx, committer, err := db.TxContext(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer committer.Close()
|
||||
|
||||
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
|
||||
} else 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 && job.RawConcurrency.Group != "" {
|
||||
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
|
||||
if runJob.Status != actions_model.StatusBlocked {
|
||||
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
|
||||
// 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
|
||||
} else 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 committer.Commit()
|
||||
}
|
@ -53,20 +53,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 +130,23 @@ 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 {
|
||||
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})
|
||||
|
@ -53,6 +53,10 @@ func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := CancelJobsByJobConcurrency(ctx, t.Job); err != nil {
|
||||
return fmt.Errorf("CancelJobs: %w", err)
|
||||
}
|
||||
|
||||
if err := t.LoadAttributes(ctx); err != nil {
|
||||
return fmt.Errorf("task LoadAttributes: %w", err)
|
||||
}
|
||||
|
@ -99,6 +99,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 +155,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 +188,24 @@ 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
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
|
1156
tests/integration/actions_concurrency_test.go
Normal file
1156
tests/integration/actions_concurrency_test.go
Normal file
File diff suppressed because it is too large
Load Diff
@ -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
|
||||
}
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user