Files
Gitea/services/actions/approve.go
T
Pascal Zimmermann 10c678a0a0 feat: Add max-parallel Support for Gitea Actions (#36357)
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 <pascal.zimmermann@theiotstudio.com>
Signed-off-by: ZPascal <pascal.zimmermann@theiotstudio.com>
Signed-off-by: wxiaoguang <wxiaoguang@gmail.com>
Co-authored-by: silverwind <me@silverwind.io>
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
Co-authored-by: Zettat123 <zettat123@gmail.com>
2026-07-27 16:18:43 +00:00

127 lines
3.9 KiB
Go

// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package actions
import (
"context"
"errors"
"fmt"
actions_model "gitea.dev/models/actions"
"gitea.dev/models/db"
repo_model "gitea.dev/models/repo"
user_model "gitea.dev/models/user"
"gitea.dev/modules/container"
"gitea.dev/modules/log"
)
func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_model.User, runIDs []int64) error {
updatedJobs := make([]*actions_model.ActionRunJob, 0)
cancelledConcurrencyJobs := make([]*actions_model.ActionRunJob, 0)
// Track runs whose reusable callers were just expanded so we can re-emit after the tx commits.
expandedCallerRunIDs := make(container.Set[int64])
err := db.WithTx(ctx, func(ctx context.Context) (err error) {
for _, runID := range runIDs {
run, err := actions_model.GetRunByRepoAndID(ctx, repo.ID, runID)
if err != nil {
return err
}
if !run.NeedApproval {
continue
}
run.NeedApproval = false
run.ApprovedBy = doer.ID
if err := actions_model.UpdateRun(ctx, run, "need_approval", "approved_by"); err != nil {
return err
}
jobs, err := actions_model.GetLatestAttemptJobsByRepoAndRunID(ctx, repo.ID, run.ID)
if err != nil {
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
}
n, err := actions_model.UpdateRunJob(ctx, job, nil, "status")
if err != nil {
return err
}
if n == 0 {
continue
}
updatedJobs = append(updatedJobs, job)
// A top-level reusable caller was just unblocked by approval, expand it
if job.IsReusableCaller && !job.IsExpanded {
attempt, has, err := run.GetLatestAttempt(ctx)
if err != nil {
return fmt.Errorf("get latest attempt of run %d: %w", run.ID, err)
}
if !has {
return errors.New("run has no attempt")
}
vars, err := actions_model.GetVariablesOfRun(ctx, run)
if err != nil {
return err
}
if err := expandReusableWorkflowCaller(ctx, run, attempt, job, vars); err != nil {
return fmt.Errorf("expand caller %d on approval: %w", job.ID, err)
}
if err := actions_model.RefreshReusableCallerStatus(ctx, job); err != nil {
return fmt.Errorf("refresh caller %d status after approval-time expansion: %w", job.ID, err)
}
expandedCallerRunIDs.Add(run.ID)
}
}
}
return nil
})
if err != nil {
return err
}
// Re-emit AFTER the tx commits so the newly inserted callee rows transition Blocked -> Waiting.
for runID := range expandedCallerRunIDs {
if err := EmitJobsIfReadyByRun(runID); err != nil {
log.Error("emit run %d after approval-time caller expansion: %v", runID, err)
}
}
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, updatedJobs)
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, cancelledConcurrencyJobs)
EmitJobsIfReadyByJobs(cancelledConcurrencyJobs)
return nil
}