From 10c678a0a09a417f3ab75df54d117382fa776c18 Mon Sep 17 00:00:00 2001 From: Pascal Zimmermann Date: Mon, 27 Jul 2026 18:18:43 +0200 Subject: [PATCH] feat: Add max-parallel Support for Gitea Actions (#36357) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add support for `strategy.max-parallel` on Gitea Actions matrix jobs. **How it works** Jobs over the limit are inserted as `Blocked` instead of `Waiting`, so runners never see them. When a job finishes, the job-status resolver promotes one `Blocked` job per freed slot, in job order. Slots are counted per `JobID` and scoped by reusable-workflow caller. A `Cancelling` job still owns its runner, so it keeps its slot. The cap is applied wherever a job can become `Waiting`: initial insert, rerun, approval, and resolver promotion. Best effort, not a hard invariant: two concurrent emitter passes can each promote into the last slot, overshooting by one. It does not compound, since every later pass recounts. **Parsing** Any YAML number, cast to an int as GitHub does (`1.5` → 1). `0` or negative means unlimited. Expressions (`${{ ... }}`) are not evaluated yet and fall back to unlimited. **Migration** Adds the `max_parallel` column on `action_run_job`. No index or constraint changes. **Compatibility** Existing rows default to `0`, so behaviour is unchanged. No runner changes needed: the runner protocol is untouched, and since the server splits the matrix each runner still receives a single job. Closes https://github.com/go-gitea/gitea/issues/35561 Signed-off-by: Pascal Zimmermann Signed-off-by: ZPascal Signed-off-by: wxiaoguang Co-authored-by: silverwind Co-authored-by: wxiaoguang Co-authored-by: Zettat123 --- modelmigration/migrations.go | 3 + modelmigration/v1_28/main_test.go | 14 ++ modelmigration/v1_28/v343.go | 22 +++ modelmigration/v1_28/v343_test.go | 37 ++++ models/actions/run_job.go | 2 + services/actions/approve.go | 16 ++ services/actions/job_emitter.go | 38 +++- services/actions/job_emitter_test.go | 166 ++++++++++++++++- services/actions/max_parallel.go | 78 ++++++++ services/actions/max_parallel_test.go | 251 ++++++++++++++++++++++++++ services/actions/rerun.go | 6 +- services/actions/rerun_test.go | 7 + services/actions/reusable_workflow.go | 1 + services/actions/run.go | 13 +- 14 files changed, 637 insertions(+), 17 deletions(-) create mode 100644 modelmigration/v1_28/main_test.go create mode 100644 modelmigration/v1_28/v343.go create mode 100644 modelmigration/v1_28/v343_test.go create mode 100644 services/actions/max_parallel.go create mode 100644 services/actions/max_parallel_test.go diff --git a/modelmigration/migrations.go b/modelmigration/migrations.go index d82b474d96..987ceec270 100644 --- a/modelmigration/migrations.go +++ b/modelmigration/migrations.go @@ -28,6 +28,7 @@ import ( "gitea.dev/modelmigration/v1_25" "gitea.dev/modelmigration/v1_26" "gitea.dev/modelmigration/v1_27" + "gitea.dev/modelmigration/v1_28" "gitea.dev/modelmigration/v1_6" "gitea.dev/modelmigration/v1_7" "gitea.dev/modelmigration/v1_8" @@ -421,6 +422,8 @@ func prepareMigrationTasks() []*migration { newMigration(341, "Convert legacy MSSQL DATETIME columns to DATETIME2", v1_27.FixLegacyMSSQLDateTimeColumns), newMigration(342, "Add scoped workflows schema", v1_27.AddScopedWorkflowsSchema), // Gitea 1.27.0 ends at migration ID number 342 (database version 343) + + newMigration(343, "Add max_parallel column to action_run_job", v1_28.AddMaxParallelToActionRunJob), } return preparedMigrations } diff --git a/modelmigration/v1_28/main_test.go b/modelmigration/v1_28/main_test.go new file mode 100644 index 0000000000..fe62b934ce --- /dev/null +++ b/modelmigration/v1_28/main_test.go @@ -0,0 +1,14 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package v1_28 + +import ( + "testing" + + "gitea.dev/modelmigration/migrationtest" +) + +func TestMain(m *testing.M) { + migrationtest.MainTest(m) +} diff --git a/modelmigration/v1_28/v343.go b/modelmigration/v1_28/v343.go new file mode 100644 index 0000000000..9b72617c71 --- /dev/null +++ b/modelmigration/v1_28/v343.go @@ -0,0 +1,22 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package v1_28 + +import ( + "gitea.dev/modelmigration/base" + + "xorm.io/xorm" +) + +func AddMaxParallelToActionRunJob(x base.EngineMigration) error { + type ActionRunJob struct { + MaxParallel int `xorm:"NOT NULL DEFAULT 0"` + } + + _, err := x.SyncWithOptions(xorm.SyncOptions{ + IgnoreConstrains: true, + IgnoreDropIndices: true, + }, new(ActionRunJob)) + return err +} diff --git a/modelmigration/v1_28/v343_test.go b/modelmigration/v1_28/v343_test.go new file mode 100644 index 0000000000..4eb3e7ffde --- /dev/null +++ b/modelmigration/v1_28/v343_test.go @@ -0,0 +1,37 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package v1_28 + +import ( + "testing" + + "gitea.dev/modelmigration/migrationtest" + + "github.com/stretchr/testify/require" +) + +func TestAddMaxParallelToActionRunJob(t *testing.T) { + type ActionRunJob struct { + ID int64 `xorm:"pk autoincr"` + Name string `xorm:"VARCHAR(255)"` + } + + x, deferable := migrationtest.PrepareTestEnv(t, 0, new(ActionRunJob)) + defer deferable() + if x == nil || t.Failed() { + return + } + + _, err := x.Insert(&ActionRunJob{Name: "job-a"}) + require.NoError(t, err) + + require.NoError(t, AddMaxParallelToActionRunJob(x)) + + // pre-existing rows must default to unlimited + var maxParallel int + has, err := x.SQL("SELECT max_parallel FROM action_run_job WHERE id = ?", 1).Get(&maxParallel) + require.NoError(t, err) + require.True(t, has) + require.Equal(t, 0, maxParallel) +} diff --git a/models/actions/run_job.go b/models/actions/run_job.go index f00c5da51b..83dabff535 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -69,6 +69,8 @@ type ActionRunJob struct { // Org/repo clamps are enforced when the token is used at runtime. // It is JSON-encoded repo_model.ActionsTokenPermissions and may be empty if not specified. TokenPermissions *repo_model.ActionsTokenPermissions `xorm:"JSON TEXT"` + // MaxParallel is strategy.max-parallel, shared by all matrix jobs with the same JobID (0 = unlimited). + MaxParallel int `xorm:"NOT NULL DEFAULT 0"` // RunAttemptID identifies the ActionRunAttempt this job belongs to. // A value of 0 indicates a legacy job created before ActionRunAttempt existed. diff --git a/services/actions/approve.go b/services/actions/approve.go index cb657f90db..c8755434bf 100644 --- a/services/actions/approve.go +++ b/services/actions/approve.go @@ -41,18 +41,34 @@ func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_mo return err } + // approval unblocks every job at once, so max-parallel has to cap them here too + slots := maxParallelSlots{} + for _, job := range jobs { + slots.hold(job, job.Status) + } + for _, job := range jobs { // Skip jobs with `needs`: they stay blocked until their dependencies finish, // at which point job_emitter will evaluate and start them. if len(job.Needs) > 0 { continue } + // Only a job this approval unblocks competes for a slot, one that is already + // active was counted by the seeding loop above and must not take a second. + isUnblocking := job.Status == actions_model.StatusBlocked + // A slot-starved job cannot start, skip the following checks. + if isUnblocking && !slots.available(job) { + continue + } var jobsToCancel []*actions_model.ActionRunJob job.Status, jobsToCancel, err = PrepareToStartJobWithConcurrency(ctx, job) if err != nil { return err } cancelledConcurrencyJobs = append(cancelledConcurrencyJobs, jobsToCancel...) + if isUnblocking { + applyMaxParallel(job, slots) + } if job.Status != actions_model.StatusWaiting { continue } diff --git a/services/actions/job_emitter.go b/services/actions/job_emitter.go index e1b56c3687..434ba9eac1 100644 --- a/services/actions/job_emitter.go +++ b/services/actions/job_emitter.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "slices" actions_model "gitea.dev/models/actions" "gitea.dev/models/db" @@ -343,7 +344,10 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action } type jobStatusResolver struct { - statuses map[int64]actions_model.Status + statuses map[int64]actions_model.Status + // sortedIDs are the keys of statuses, so blocked jobs are resolved in insertion order. + // Resolve only ever rewrites statuses values, never its key set. + sortedIDs []int64 needs map[int64][]int64 jobMap map[int64]*actions_model.ActionRunJob vars map[string]string @@ -367,8 +371,10 @@ func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]stri statuses := make(map[int64]actions_model.Status, len(jobs)) needs := make(map[int64][]int64, len(jobs)) + sortedIDs := make([]int64, 0, len(jobs)) for _, job := range jobs { statuses[job.ID] = job.Status + sortedIDs = append(sortedIDs, job.ID) scope := scopedIDToJobs[job.ParentJobID] for _, need := range job.Needs { for _, v := range scope[need] { @@ -376,11 +382,13 @@ func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]stri } } } + slices.Sort(sortedIDs) return &jobStatusResolver{ - statuses: statuses, - needs: needs, - jobMap: jobMap, - vars: vars, + statuses: statuses, + sortedIDs: sortedIDs, + needs: needs, + jobMap: jobMap, + vars: vars, } } @@ -420,11 +428,22 @@ func (r *jobStatusResolver) resolveCheckNeeds(id int64) (allDone, allSucceed boo func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model.Status { ret := map[int64]actions_model.Status{} + + slots := maxParallelSlots{} for id, status := range r.statuses { + slots.hold(r.jobMap[id], status) + } + + for _, id := range r.sortedIDs { + status := r.statuses[id] actionRunJob := r.jobMap[id] if status != actions_model.StatusBlocked { continue } + // An expanded caller has been resolved in an earlier pass, skip. + if actionRunJob.IsReusableCaller && actionRunJob.IsExpanded { + continue + } // A child of a caller cannot start until the caller has become "ready" (children inserted, CallPayload populated). if actionRunJob.ParentJobID > 0 { if parent, ok := r.jobMap[actionRunJob.ParentJobID]; ok && !parent.IsExpanded { @@ -436,6 +455,11 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model continue } + // A slot-starved job cannot start, skip the following checks. + if !slots.available(actionRunJob) { + continue + } + // update concurrency and check whether the job can run now err := updateConcurrencyEvaluationForJobWithNeeds(ctx, actionRunJob, r.vars) if err != nil { @@ -464,6 +488,10 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model } } + if newStatus == actions_model.StatusWaiting && !slots.take(actionRunJob) { + continue // no free slot, leave blocked + } + if newStatus != actions_model.StatusBlocked { ret[id] = newStatus } diff --git a/services/actions/job_emitter_test.go b/services/actions/job_emitter_test.go index dba21bfb44..4f3c65855e 100644 --- a/services/actions/job_emitter_test.go +++ b/services/actions/job_emitter_test.go @@ -16,6 +16,17 @@ import ( "github.com/stretchr/testify/assert" ) +func minimalWorkflowPayload(jobID string) []byte { + return fmt.Appendf(nil, `name: test +on: push +jobs: + %s: + runs-on: ubuntu-latest + steps: + - run: echo +`, jobID) +} + func Test_jobStatusResolver_Resolve(t *testing.T) { tests := []struct { name string @@ -131,6 +142,68 @@ jobs: }, want: map[int64]actions_model.Status{2: actions_model.StatusSkipped}, }, + { + name: "max-parallel: a freed slot promotes the lowest blocked job id", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "build", Status: actions_model.StatusSuccess, Needs: []string{}, MaxParallel: 1}, + {ID: 2, JobID: "build", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1}, + {ID: 3, JobID: "build", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1}, + }, + want: map[int64]actions_model.Status{2: actions_model.StatusWaiting}, + }, + { + name: "max-parallel: a cancelling job still holds its slot", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "test", Status: actions_model.StatusRunning, Needs: []string{}, MaxParallel: 2}, + {ID: 2, JobID: "test", Status: actions_model.StatusCancelling, Needs: []string{}, MaxParallel: 2}, + {ID: 3, JobID: "test", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 2}, + }, + want: map[int64]actions_model.Status{}, + }, + { + name: "max-parallel: two freed slots promote two jobs", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "test", Status: actions_model.StatusCancelled, Needs: []string{}, MaxParallel: 2}, + {ID: 2, JobID: "test", Status: actions_model.StatusSuccess, Needs: []string{}, MaxParallel: 2}, + {ID: 3, JobID: "test", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 2}, + {ID: 4, JobID: "test", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 2}, + }, + want: map[int64]actions_model.Status{3: actions_model.StatusWaiting, 4: actions_model.StatusWaiting}, + }, + { + name: "max-parallel: slots are counted per job id", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "build", Status: actions_model.StatusRunning, Needs: []string{}, MaxParallel: 1}, + {ID: 2, JobID: "build", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1}, + {ID: 3, JobID: "test", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1}, + }, + want: map[int64]actions_model.Status{3: actions_model.StatusWaiting}, + }, + { + name: "max-parallel: slots are scoped per reusable workflow caller", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "build", ParentJobID: 10, Status: actions_model.StatusRunning, Needs: []string{}, MaxParallel: 1}, + {ID: 2, JobID: "build", ParentJobID: 10, Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1}, + {ID: 3, JobID: "build", ParentJobID: 20, Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1}, + }, + want: map[int64]actions_model.Status{3: actions_model.StatusWaiting}, + }, + { + name: "max-parallel: a caller promoted in an earlier round keeps its slot", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "call", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1, IsReusableCaller: true}, + {ID: 2, JobID: "call", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1, IsReusableCaller: true}, + }, + want: map[int64]actions_model.Status{1: actions_model.StatusWaiting}, + }, + { + name: "max-parallel: an expanded caller aggregated back to Blocked keeps its slot", + jobs: actions_model.ActionJobList{ + {ID: 1, JobID: "call", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1, IsReusableCaller: true, IsExpanded: true}, + {ID: 2, JobID: "call", Status: actions_model.StatusBlocked, Needs: []string{}, MaxParallel: 1, IsReusableCaller: true}, + }, + want: map[int64]actions_model.Status{}, + }, { name: "`if` is empty and a failed need has continue-on-error", jobs: actions_model.ActionJobList{ @@ -171,14 +244,7 @@ jobs: // The resolver evaluates Blocked jobs via evaluateJobIf, which needs a valid YAML payload; // supply a minimal one when the case didn't. if j.Status == actions_model.StatusBlocked && len(j.WorkflowPayload) == 0 { - j.WorkflowPayload = fmt.Appendf(nil, `name: test -on: push -jobs: - %s: - runs-on: ubuntu-latest - steps: - - run: echo -`, j.JobID) + j.WorkflowPayload = minimalWorkflowPayload(j.JobID) } assert.NoError(t, db.Insert(ctx, j)) @@ -196,6 +262,45 @@ jobs: } } +// Test_maxParallelConverges covers the liveness property the status table cannot express: +// repeated resolve cycles never stall and never overshoot the cap. +func Test_maxParallelConverges(t *testing.T) { + ctx := t.Context() + + const totalJobs, maxParallel = 5, 2 + jobs := make(actions_model.ActionJobList, totalJobs) + for i := range jobs { + jobs[i] = &actions_model.ActionRunJob{ + ID: int64(i + 1), JobID: "matrix", Status: actions_model.StatusBlocked, MaxParallel: maxParallel, + WorkflowPayload: minimalWorkflowPayload("matrix"), + } + } + + for cycle := range totalJobs + 1 { + for id, status := range newJobStatusResolver(jobs, nil).Resolve(ctx) { + jobs[id-1].Status = status + } + counts := statusCounts(jobs) + remaining := totalJobs - counts[actions_model.StatusSuccess] + assert.Equal(t, min(remaining, maxParallel), counts[actions_model.StatusWaiting]+counts[actions_model.StatusRunning], + "cycle %d: active jobs must fill every free slot without exceeding max-parallel", cycle) + + for _, job := range jobs { // a runner picks up every waiting job + if job.Status == actions_model.StatusWaiting { + job.Status = actions_model.StatusRunning + } + } + for _, job := range jobs { // the first running job finishes + if job.Status == actions_model.StatusRunning { + job.Status = actions_model.StatusSuccess + break + } + } + } + + assert.Equal(t, totalJobs, statusCounts(jobs)[actions_model.StatusSuccess]) +} + // Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck verifies that when a run's // ConcurrencyGroup has already been checked at the run level, the same group is not // re-checked for individual jobs. @@ -433,3 +538,48 @@ func Test_findConcurrencyWaiterToWake(t *testing.T) { assert.NoError(t, err) assert.Equal(t, int64(0), id) } + +func Test_maxParallelReusableCallerLifecycle(t *testing.T) { + ctx := t.Context() + + const callerJobNum = 3 + callers := make(actions_model.ActionJobList, callerJobNum) + idToCaller := make(map[int64]*actions_model.ActionRunJob, callerJobNum) + for i := range callers { + callers[i] = &actions_model.ActionRunJob{ + ID: int64(i + 1), JobID: "call", Status: actions_model.StatusBlocked, MaxParallel: 1, + IsReusableCaller: true, WorkflowPayload: minimalWorkflowPayload("call"), + } + idToCaller[callers[i].ID] = callers[i] + } + underway := func() (n int) { + for _, caller := range callers { + if caller.IsExpanded && !caller.Status.IsDone() { + n++ + } + } + return n + } + + for cycle := range 2 * len(callers) { + promoted := newJobStatusResolver(callers, nil).Resolve(ctx) + for id, status := range promoted { + caller := idToCaller[id] + assert.False(t, caller.IsExpanded, "cycle %d: resolver re-promoted already-expanded caller %d", cycle, id) + assert.Equal(t, actions_model.StatusWaiting, status, "cycle %d: caller %d", cycle, id) + caller.IsExpanded = true // the emitter expands the caller; children insert as Blocked, so the aggregate keeps it Blocked + } + assert.LessOrEqual(t, underway(), 1, "cycle %d: at most one caller may be underway", cycle) + + if len(promoted) == 0 { // steady state: the underway caller's children finish and cascade Success to it + for _, caller := range callers { + if caller.IsExpanded && !caller.Status.IsDone() { + caller.Status = actions_model.StatusSuccess + break + } + } + } + } + + assert.Equal(t, len(callers), statusCounts(callers)[actions_model.StatusSuccess]) +} diff --git a/services/actions/max_parallel.go b/services/actions/max_parallel.go new file mode 100644 index 0000000000..16be7b0356 --- /dev/null +++ b/services/actions/max_parallel.go @@ -0,0 +1,78 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package actions + +import ( + "math" + "strconv" + + actions_model "gitea.dev/models/actions" + "gitea.dev/modules/log" +) + +// parseMaxParallel returns strategy.max-parallel for a job, 0 meaning unlimited. +// GitHub accepts any YAML number here and casts it to an int, so 1.5 truncates to 1. +// Expressions are not evaluated yet and fall back to unlimited. +func parseMaxParallel(jobID, maxParallelString string) int { + if maxParallelString == "" { + return 0 + } + maxParallel, err := strconv.ParseFloat(maxParallelString, 64) + if err != nil || math.IsNaN(maxParallel) { + log.Debug("job %s: unsupported max-parallel value %q, treating as unlimited", jobID, maxParallelString) + return 0 + } + // a run can never hold more jobs than MaxJobNumPerRun, so clamping there keeps the cast total + return int(min(max(maxParallel, 0), actions_model.MaxJobNumPerRun)) +} + +// maxParallelSlots counts the jobs holding a max-parallel slot. Slots are scoped by ParentJobID as +// well as JobID so the same job name in two reusable workflow calls does not cross-link, matching +// how jobStatusResolver scopes needs. +type maxParallelSlots map[maxParallelKey]int + +type maxParallelKey struct { + parentJobID int64 + jobID string +} + +// hold seeds a job's slot from a status it already holds, bypassing the limit check. +// The status is passed separately because the job emitter tracks statuses outside the job model. +func (s maxParallelSlots) hold(job *actions_model.ActionRunJob, status actions_model.Status) { + if job.MaxParallel <= 0 { + return // an unlimited job's count is never read by take, don't accumulate it + } + // A Cancelling job still owns its runner, so it keeps its slot until cleanup finishes. + // An expanded reusable caller is underway even while its children aggregate it back to Blocked, so it holds a slot too. + if status.In(actions_model.StatusRunning, actions_model.StatusWaiting, actions_model.StatusCancelling) || + (job.IsReusableCaller && job.IsExpanded && !status.IsDone()) { + s[maxParallelKey{job.ParentJobID, job.JobID}]++ + } +} + +// available reports whether a slot is free without reserving it. +// Peek it before running the concurrency check: DO NOT cancel the group's other pending jobs if no available slots. +func (s maxParallelSlots) available(job *actions_model.ActionRunJob) bool { + return job.MaxParallel <= 0 || s[maxParallelKey{job.ParentJobID, job.JobID}] < job.MaxParallel +} + +// take reserves a slot and reports whether one was free. A job without a limit always succeeds. +func (s maxParallelSlots) take(job *actions_model.ActionRunJob) bool { + if !s.available(job) { + return false + } + if job.MaxParallel > 0 { + s[maxParallelKey{job.ParentJobID, job.JobID}]++ + } + return true +} + +// applyMaxParallel demotes a ready job to Blocked when its slots are full, leaving the job emitter to +// promote it once one frees. Call it after concurrency evaluation so a concurrency-blocked job does +// not consume a slot, and before the insert so a held-back reusable caller is not expanded. +func applyMaxParallel(job *actions_model.ActionRunJob, slots maxParallelSlots) { + if job.Status == actions_model.StatusWaiting && !slots.take(job) { + job.Status = actions_model.StatusBlocked + } +} diff --git a/services/actions/max_parallel_test.go b/services/actions/max_parallel_test.go new file mode 100644 index 0000000000..d63e40da86 --- /dev/null +++ b/services/actions/max_parallel_test.go @@ -0,0 +1,251 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package actions + +import ( + "testing" + + actions_model "gitea.dev/models/actions" + "gitea.dev/models/db" + repo_model "gitea.dev/models/repo" + "gitea.dev/models/unittest" + user_model "gitea.dev/models/user" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseMaxParallel(t *testing.T) { + tests := []struct { + input string + want int + }{ + {"", 0}, + {"3", 3}, + {"0", 0}, + {"-1", 0}, + {"1.5", 1}, // GitHub casts the YAML number to int + {"-1.5", 0}, // truncates to -1, which means unlimited + {"1e3", 256}, // clamped to MaxJobNumPerRun + {"nan", 0}, // must not reach the int cast + {"${{ vars.n }}", 0}, // expressions are not evaluated yet + {"abc", 0}, + } + for _, tt := range tests { + assert.Equal(t, tt.want, parseMaxParallel("job", tt.input), "input %q", tt.input) + } +} + +// a 5-job matrix capped at 2, so 2 jobs start and 3 stay blocked +const maxParallelWorkflow = `name: max-parallel +on: push +jobs: + build: + runs-on: ubuntu-latest + strategy: + max-parallel: 2 + matrix: + version: [1, 2, 3, 4, 5] + steps: + - run: echo hi +` + +func insertMaxParallelRun(t *testing.T, workflow string, needApproval bool) *actions_model.ActionRun { + t.Helper() + run := &actions_model.ActionRun{ + Title: "max-parallel", + RepoID: 4, + OwnerID: 1, + WorkflowID: "max-parallel.yaml", + TriggerUserID: 1, + Ref: "refs/heads/master", + CommitSHA: "c2d72f548424103f01ee1dc02889c1e2bff816b0", + Event: "push", + TriggerEvent: "push", + EventPayload: "{}", + NeedApproval: needApproval, + WorkflowRepoID: 4, + WorkflowCommitSHA: "c2d72f548424103f01ee1dc02889c1e2bff816b0", + } + require.NoError(t, PrepareRunAndInsert(t.Context(), []byte(workflow), run, nil)) + return run +} + +func runJobs(t *testing.T, runID, attemptID int64) actions_model.ActionJobList { + t.Helper() + jobs, err := actions_model.GetRunJobsByRunAndAttemptID(t.Context(), runID, attemptID) + require.NoError(t, err) + return jobs +} + +func statusCounts(jobs actions_model.ActionJobList) map[actions_model.Status]int { + counts := map[actions_model.Status]int{} + for _, job := range jobs { + counts[job.Status]++ + } + return counts +} + +func TestPrepareRunAndInsert_MaxParallel(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + + run := insertMaxParallelRun(t, maxParallelWorkflow, false) + jobs := runJobs(t, run.ID, run.LatestAttemptID) + assert.Equal(t, map[actions_model.Status]int{ + actions_model.StatusWaiting: 2, + actions_model.StatusBlocked: 3, + }, statusCounts(jobs)) + + for _, job := range jobs { + assert.Equal(t, 2, job.MaxParallel, "every matrix sibling carries the limit") + } +} + +// A reusable workflow declares its own strategy, so the limit must reach the child jobs. +func TestInsertCallerChildren_MaxParallel(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + ctx := t.Context() + + run := &actions_model.ActionRun{ + RepoID: 4, OwnerID: 1, Index: 9701, TriggerUserID: 1, TriggerEvent: "push", EventPayload: "{}", + } + require.NoError(t, db.Insert(ctx, run)) + require.NoError(t, run.LoadAttributes(ctx)) + + attempt := &actions_model.ActionRunAttempt{RepoID: run.RepoID, RunID: run.ID, Attempt: 1} + require.NoError(t, db.Insert(ctx, attempt)) + + caller := &actions_model.ActionRunJob{ + RunID: run.ID, RunAttemptID: attempt.ID, RepoID: run.RepoID, OwnerID: run.OwnerID, + JobID: "caller", AttemptJobID: 1, IsReusableCaller: true, + } + require.NoError(t, db.Insert(ctx, caller)) + + called := []byte(`name: called +on: workflow_call +jobs: + build: + runs-on: ubuntu-latest + strategy: + max-parallel: 2 + matrix: + version: [1, 2, 3] + steps: + - run: echo hi +`) + require.NoError(t, insertCallerChildren(ctx, run, attempt, caller, called, run.RepoID, run.CommitSHA, nil, nil)) + + children := 0 + for _, job := range runJobs(t, run.ID, attempt.ID) { + if job.ParentJobID == caller.ID { + children++ + assert.Equal(t, 2, job.MaxParallel) + } + } + assert.Equal(t, 3, children) +} + +// An approval-gated run inserts every job as Blocked, so the cap has to be applied on approval. +func TestApproveRuns_MaxParallel(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + + run := insertMaxParallelRun(t, maxParallelWorkflow, true) + assert.Equal(t, map[actions_model.Status]int{actions_model.StatusBlocked: 5}, statusCounts(runJobs(t, run.ID, run.LatestAttemptID))) + + repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: run.RepoID}) + doer := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 1}) + require.NoError(t, ApproveRuns(t.Context(), repo, doer, []int64{run.ID})) + + assert.Equal(t, map[actions_model.Status]int{ + actions_model.StatusWaiting: 2, + actions_model.StatusBlocked: 3, + }, statusCounts(runJobs(t, run.ID, run.LatestAttemptID))) +} + +func TestPrepareRunAndInsert_MaxParallelStarvedSkipsConcurrency(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + + holder := insertConcurrencyHolder(t, 9702, "cluster-b") + run := insertMaxParallelRun(t, maxParallelConcurrencyWorkflow, false) + + assert.Equal(t, map[actions_model.Status]int{ + actions_model.StatusWaiting: 1, + actions_model.StatusBlocked: 1, + }, statusCounts(runJobs(t, run.ID, run.LatestAttemptID))) + + holder = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: holder.ID}) + assert.Equal(t, actions_model.StatusRunning, holder.Status, "the starved job must not cancel the group holder") +} + +func Test_jobStatusResolver_MaxParallelStarvedSkipsConcurrency(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + + holder := insertConcurrencyHolder(t, 9703, "cluster-b") + run := insertMaxParallelRun(t, maxParallelConcurrencyWorkflow, false) + + jobs := runJobs(t, run.ID, run.LatestAttemptID) + for _, job := range jobs { + job.Run = run + } + assert.Empty(t, newJobStatusResolver(jobs, nil).Resolve(t.Context()), "the starved job must stay blocked") + + holder = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: holder.ID}) + assert.Equal(t, actions_model.StatusRunning, holder.Status) +} + +func TestApproveRuns_MaxParallelStarvedSkipsConcurrency(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + + holder := insertConcurrencyHolder(t, 9704, "cluster-b") + run := insertMaxParallelRun(t, maxParallelConcurrencyWorkflow, true) + repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: run.RepoID}) + doer := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 1}) + require.NoError(t, ApproveRuns(t.Context(), repo, doer, []int64{run.ID})) + + assert.Equal(t, map[actions_model.Status]int{ + actions_model.StatusWaiting: 1, + actions_model.StatusBlocked: 1, + }, statusCounts(runJobs(t, run.ID, run.LatestAttemptID))) + + holder = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: holder.ID}) + assert.Equal(t, actions_model.StatusRunning, holder.Status) +} + +// a 2-job matrix capped at 1 where each matrix job has its own concurrency group that cancels in progress +const maxParallelConcurrencyWorkflow = `name: max-parallel-concurrency +on: push +jobs: + deploy: + runs-on: ubuntu-latest + strategy: + max-parallel: 1 + matrix: + cluster: [a, b] + concurrency: + group: cluster-${{ matrix.cluster }} + cancel-in-progress: true + steps: + - run: echo hi +` + +// insertConcurrencyHolder inserts a running job occupying the given concurrency group +func insertConcurrencyHolder(t *testing.T, index int64, group string) *actions_model.ActionRunJob { + t.Helper() + ctx := t.Context() + + run := &actions_model.ActionRun{ + RepoID: 4, OwnerID: 1, Index: index, TriggerUserID: 1, TriggerEvent: "push", EventPayload: "{}", + Status: actions_model.StatusRunning, + } + require.NoError(t, db.Insert(ctx, run)) + attempt := &actions_model.ActionRunAttempt{RepoID: run.RepoID, RunID: run.ID, Attempt: 1, Status: actions_model.StatusRunning} + require.NoError(t, db.Insert(ctx, attempt)) + holder := &actions_model.ActionRunJob{ + RunID: run.ID, RunAttemptID: attempt.ID, RepoID: run.RepoID, OwnerID: run.OwnerID, + JobID: "holder", AttemptJobID: 1, Status: actions_model.StatusRunning, + ConcurrencyGroup: group, IsConcurrencyEvaluated: true, + } + require.NoError(t, db.Insert(ctx, holder)) + return holder +} diff --git a/services/actions/rerun.go b/services/actions/rerun.go index 6ec2463438..665d63a8b0 100644 --- a/services/actions/rerun.go +++ b/services/actions/rerun.go @@ -240,6 +240,7 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR // templateIDToNewID maps each template-attempt job's DB ID to its newly-inserted clone's DB ID templateIDToNewID := make(map[int64]int64, len(plan.templateJobs)) + slots := maxParallelSlots{} for _, templateJob := range plan.templateJobs { // descendants of a reset reusable caller are not cloned at all, the caller will re-insert them @@ -276,7 +277,8 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR newJob.CallPayload = "" } - if newJob.RawConcurrency != "" && !shouldBlockJob { + // A slot-starved job must not cancel its group peers. + if newJob.RawConcurrency != "" && !shouldBlockJob && slots.available(newJob) { if err := EvaluateJobConcurrencyFillModel(ctx, plan.run, newAttempt, newJob, vars, nil); err != nil { return fmt.Errorf("evaluate job concurrency: %w", err) } @@ -287,6 +289,7 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR cancelledConcurrencyJobs = append(cancelledConcurrencyJobs, jobsToCancel...) } + applyMaxParallel(newJob, slots) newJobsToRerun = append(newJobsToRerun, newJob) } else { newJob.TaskID = 0 @@ -522,6 +525,7 @@ func cloneRunJobForAttempt(templateJob *actions_model.ActionRunJob, attempt *act ConcurrencyGroup: templateJob.ConcurrencyGroup, ConcurrencyCancel: templateJob.ConcurrencyCancel, TokenPermissions: templateJob.TokenPermissions, + MaxParallel: templateJob.MaxParallel, // reusable workflow fields IsReusableCaller: templateJob.IsReusableCaller, diff --git a/services/actions/rerun_test.go b/services/actions/rerun_test.go index 9c0f63e165..b4cb4ad734 100644 --- a/services/actions/rerun_test.go +++ b/services/actions/rerun_test.go @@ -94,6 +94,13 @@ func TestCloneRunJobForAttempt(t *testing.T) { clone := cloneRunJobForAttempt(template, attempt) assert.False(t, clone.ContinueOnError) }) + + // without this the rerun of a capped matrix would run unlimited + t.Run("preserves max-parallel", func(t *testing.T) { + template := &actions_model.ActionRunJob{MaxParallel: 3} + clone := cloneRunJobForAttempt(template, attempt) + assert.Equal(t, 3, clone.MaxParallel) + }) } func TestRerunValidation(t *testing.T) { diff --git a/services/actions/reusable_workflow.go b/services/actions/reusable_workflow.go index 05056fcb4b..c178fb2189 100644 --- a/services/actions/reusable_workflow.go +++ b/services/actions/reusable_workflow.go @@ -354,6 +354,7 @@ func insertCallerChildren(ctx context.Context, run *actions_model.ActionRun, att Needs: needs, RunsOn: parsedChild.RunsOn(), ContinueOnError: parsedChild.GetContinueOnError(), + MaxParallel: parseMaxParallel(jobID, parsedChild.Strategy.MaxParallelString), Status: actions_model.StatusBlocked, ParentJobID: caller.ID, WorkflowSourceRepoID: sourceRepoID, diff --git a/services/actions/run.go b/services/actions/run.go index c8e476fa32..68c2e7fdb5 100644 --- a/services/actions/run.go +++ b/services/actions/run.go @@ -133,13 +133,16 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte runJobs := make([]*actions_model.ActionRunJob, 0, len(jobs)) var hasWaitingJobs bool + slots := maxParallelSlots{} + for _, v := range jobs { - runJob, jobsToCancel, jobNeedsPostCommitEmit, err := insertRunJob(ctx, run, runAttempt, v, vars, inputs) + runJob, jobsToCancel, jobNeedsPostCommitEmit, err := insertRunJob(ctx, run, runAttempt, v, vars, inputs, slots) if err != nil { return err } cancelledConcurrencyJobs = append(cancelledConcurrencyJobs, jobsToCancel...) needPostCommitEmit = needPostCommitEmit || jobNeedsPostCommitEmit + // A reusable caller is never dispatched to a runner, so it must not drive the task-version bump. hasWaitingJobs = hasWaitingJobs || (runJob.Status == actions_model.StatusWaiting && !runJob.IsReusableCaller) runJobs = append(runJobs, runJob) @@ -180,7 +183,7 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte // inline-expands (or skips) it. It returns the inserted job, any jobs cancelled by // job concurrency, and whether a post-commit emitter pass is needed to resolve the // caller's dependents. -func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt *actions_model.ActionRunAttempt, workflowJob *jobparser.SingleWorkflow, vars map[string]string, inputs map[string]any) (*actions_model.ActionRunJob, []*actions_model.ActionRunJob, bool, error) { +func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt *actions_model.ActionRunAttempt, workflowJob *jobparser.SingleWorkflow, vars map[string]string, inputs map[string]any, slots maxParallelSlots) (*actions_model.ActionRunJob, []*actions_model.ActionRunJob, bool, error) { id, job := workflowJob.Job() needs := job.Needs() if err := workflowJob.SetJob(id, job.EraseNeeds()); err != nil { @@ -215,6 +218,7 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt WorkflowSourceRepoID: run.WorkflowRepoID, WorkflowSourceCommitSHA: run.WorkflowCommitSHA, ContinueOnError: job.GetContinueOnError(), + MaxParallel: parseMaxParallel(id, job.Strategy.MaxParallelString), } // Parse workflow/job permissions (no clamping here) if perms := ExtractJobPermissionsFromWorkflow(workflowJob, job); perms != nil { @@ -244,7 +248,8 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt // If a job needs other jobs ("needs" is not empty), its status is set to StatusBlocked at the entry of the loop // No need to check job concurrency for a blocked job (it will be checked by job emitter later) - if runJob.Status == actions_model.StatusWaiting { + // A slot-starved job skips the check too: it will not start, so it must not cancel its group peers. + if runJob.Status == actions_model.StatusWaiting && slots.available(runJob) { var jobsToCancel []*actions_model.ActionRunJob runJob.Status, jobsToCancel, err = PrepareToStartJobWithConcurrency(ctx, runJob) if err != nil { @@ -254,6 +259,8 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt } } + applyMaxParallel(runJob, slots) + if err := db.Insert(ctx, runJob); err != nil { return nil, nil, false, err }