From a15f0320260970056173dbbfbaebbb3262f059ec Mon Sep 17 00:00:00 2001 From: Zettat123 Date: Sat, 26 Sep 2026 01:01:21 -0600 Subject: [PATCH] fix(actions): evaluate job-level `if:` before concurrency check (#39437) Gitea doesn't evaluate a job's `if:` before checking the job's concurrency group, which causes a job that should have been skipped to incorrectly cancel other jobs in the same concurrency group. This PR makes Gitea decide `if:` for every job before it becomes waiting, including jobs without `needs` at insertion, on approval and on rerun. A skipped job therefore no longer takes part in job concurrency or holds a max-parallel slot, and a reusable caller whose `if:` is false is no longer expanded on approval or rerun. An invalid `if:` skips the job with an error summary. After this PR, Gitea decides all jobs' `if:` expressions and sends `if: always()` to the runner, so the runner no longer needs to evaluate a job's `if:` again ([gitea/runner `run_context.go`](https://gitea.com/gitea/runner/src/commit/81add274599355ec1838b6ebe45804890d40bab9/act/runner/run_context.go#L1195)). --------- Co-authored-by: silverwind --- models/actions/run_attempt.go | 4 ++ models/actions/run_job.go | 8 +-- services/actions/approve.go | 44 ++++++++++------ services/actions/approve_test.go | 16 ++++++ services/actions/context_test.go | 42 ++++++++++++++++ services/actions/helper.go | 31 ++++++++++-- services/actions/rerun.go | 43 ++++++++-------- services/actions/rerun_test.go | 52 +++++++++++++++++++ services/actions/run.go | 61 +++++++++-------------- services/actions/task.go | 22 +++++++- services/actions/task_test.go | 15 ++++++ tests/integration/api_actions_run_test.go | 8 +++ 12 files changed, 263 insertions(+), 83 deletions(-) diff --git a/models/actions/run_attempt.go b/models/actions/run_attempt.go index 73298cfe33e..f592be34cfd 100644 --- a/models/actions/run_attempt.go +++ b/models/actions/run_attempt.go @@ -163,6 +163,10 @@ func UpdateRunAttempt(ctx context.Context, attempt *ActionRunAttempt, cols ...st attempt.Started = timeutil.TimeStampNow() cols = append(cols, "started") } + if slices.Contains(cols, "status") && attempt.Stopped.IsZero() && attempt.Status.IsDone() { + attempt.Stopped = timeutil.TimeStampNow() + cols = append(cols, "stopped") + } if slices.Contains(cols, "status") && !attempt.Stopped.IsZero() && !attempt.Status.IsDone() { attempt.Stopped = 0 cols = append(cols, "stopped") diff --git a/models/actions/run_job.go b/models/actions/run_job.go index 003859725fd..c32ce538f8a 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -545,13 +545,7 @@ func refreshRunStatus(ctx context.Context, repoID, runID, runAttemptID int64, no if len(jobs) == 0 { attempt.Status = noJobsStatus } - if attempt.Started.IsZero() && attempt.Status.IsRunning() { - attempt.Started = timeutil.TimeStampNow() - } - if attempt.Stopped.IsZero() && attempt.Status.IsDone() { - attempt.Stopped = timeutil.TimeStampNow() - } - if err := UpdateRunAttempt(ctx, attempt, "status", "started", "stopped"); err != nil { + if err := UpdateRunAttempt(ctx, attempt, "status"); err != nil { return fmt.Errorf("update run attempt %d: %w", attempt.ID, err) } return nil diff --git a/services/actions/approve.go b/services/actions/approve.go index cbee1a4e15c..1c58b908fa1 100644 --- a/services/actions/approve.go +++ b/services/actions/approve.go @@ -21,8 +21,7 @@ import ( func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_model.User, runIDs []int64) ([]*actions_model.ActionRun, 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]) + runIDsToEmit := make(container.Set[int64]) err := db.WithTx(ctx, func(ctx context.Context) (err error) { for _, runID := range runIDs { @@ -43,6 +42,11 @@ func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_mo return err } + vars, err := actions_model.GetVariablesOfRun(ctx, run) + 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 { @@ -58,6 +62,25 @@ func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_mo // 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 skipped job must neither cancel its group peers nor take a slot + if isUnblocking { + shouldStart, err := evaluateJobIf(ctx, run, nil, job, vars, true) + if err != nil { + return fmt.Errorf("evaluate job %d if on approval: %w", job.ID, err) + } + if !shouldStart { + job.Status = actions_model.StatusSkipped + n, err := actions_model.UpdateRunJob(ctx, job, nil, "status") + if err != nil { + return err + } + if n > 0 { + updatedJobs = append(updatedJobs, job) + runIDsToEmit.Add(run.ID) + } + continue + } + } // A slot-starved job cannot start, skip the following checks. if isUnblocking && !slots.available(job) { continue @@ -92,17 +115,10 @@ func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_mo if !has { return errors.New("run has no attempt") } - vars, err := actions_model.GetVariablesOfRun(ctx, run) - if err != nil { + if err := expandInlineReusableCaller(ctx, run, attempt, job, vars); 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) + runIDsToEmit.Add(run.ID) } } } @@ -112,10 +128,10 @@ func ApproveRuns(ctx context.Context, repo *repo_model.Repository, doer *user_mo return nil, err } - // Re-emit AFTER the tx commits so the newly inserted callee rows transition Blocked -> Waiting. - for runID := range expandedCallerRunIDs { + // Re-emit AFTER the tx commits so callee rows and dependents of skipped jobs get resolved. + for runID := range runIDsToEmit { if err := EmitJobsIfReadyByRun(runID); err != nil { - log.Error("emit run %d after approval-time caller expansion: %v", runID, err) + log.Error("emit run %d after approval: %v", runID, err) } } diff --git a/services/actions/approve_test.go b/services/actions/approve_test.go index d9d90ef8599..6f611559a17 100644 --- a/services/actions/approve_test.go +++ b/services/actions/approve_test.go @@ -11,6 +11,7 @@ import ( repo_model "gitea.dev/models/repo" "gitea.dev/models/unittest" user_model "gitea.dev/models/user" + "gitea.dev/modules/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -37,6 +38,7 @@ func TestApproveRuns(t *testing.T) { RunID: run.ID, RepoID: run.RepoID, OwnerID: run.OwnerID, CommitSHA: run.CommitSHA, Name: "job1", Attempt: 1, JobID: "job1", Status: status, RunsOn: []string{"ubuntu-latest"}, Needs: needs, + WorkflowPayload: minimalWorkflowPayload("job1"), } require.NoError(t, db.Insert(t.Context(), job)) return job @@ -55,6 +57,20 @@ func TestApproveRuns(t *testing.T) { assert.Equal(t, actions_model.StatusWaiting, unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: job.ID}).Status) }) + t.Run("approve skips a job whose if is false", func(t *testing.T) { + defer test.MockVariableValue(&EmitJobsIfReadyByRun, func(int64) error { return nil })() + run := insertRun(1006, actions_model.StatusBlocked, true, 0) + job := insertJob(run, actions_model.StatusBlocked) + job.WorkflowPayload = []byte("jobs:\n job1:\n if: false\n") + _, err := actions_model.UpdateRunJob(t.Context(), job, nil, "workflow_payload") + require.NoError(t, err) + + _, err = ApproveRuns(t.Context(), repo, doer, []int64{run.ID}) + require.NoError(t, err) + + assert.Equal(t, actions_model.StatusSkipped, unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: job.ID}).Status) + }) + t.Run("a job with unmet dependencies stays blocked", func(t *testing.T) { run := insertRun(1002, actions_model.StatusBlocked, true, 0) job := insertJob(run, actions_model.StatusBlocked, "some-other-job") diff --git a/services/actions/context_test.go b/services/actions/context_test.go index 9049de2c523..494e88a9b24 100644 --- a/services/actions/context_test.go +++ b/services/actions/context_test.go @@ -17,6 +17,7 @@ import ( actions_module "gitea.dev/modules/actions" "gitea.dev/modules/json" api "gitea.dev/modules/structs" + "gitea.dev/modules/test" webhook_module "gitea.dev/modules/webhook" "github.com/stretchr/testify/assert" @@ -97,6 +98,47 @@ jobs: assert.NotEmpty(t, persisted.RawConcurrency) } +func TestPrepareRunAndInsert_JobIf(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + defer test.MockVariableValue(&EmitJobsIfReadyByRun, func(int64) error { return nil })() + + run := insertMaxParallelRun(t, `on: push +jobs: + start: + if: github.event_name == 'push' + runs-on: ubuntu-latest + steps: + - run: echo + skip: + if: github.event_name != 'push' + runs-on: ubuntu-latest + concurrency: skip + steps: + - run: echo + skip-caller: + if: false + uses: ./.gitea/workflows/callee.yml + invalid: + if: fromJSON('{') + runs-on: ubuntu-latest + steps: + - run: echo +`, false) + + jobs := map[string]*actions_model.ActionRunJob{} + for _, job := range runJobs(t, run.ID, run.LatestAttemptID) { + jobs[job.JobID] = job + } + assert.Equal(t, actions_model.StatusWaiting, jobs["start"].Status) + assert.Equal(t, actions_model.StatusSkipped, jobs["skip"].Status) + assert.False(t, jobs["skip"].IsConcurrencyEvaluated) + assert.Equal(t, actions_model.StatusSkipped, jobs["skip-caller"].Status) + assert.Equal(t, actions_model.StatusSkipped, jobs["invalid"].Status) + summary, err := actions_model.GetActionRunJobSummary(t.Context(), run.RepoID, run.ID, run.LatestAttemptID, jobs["invalid"].ID, 0) + require.NoError(t, err) + assert.Contains(t, summary.Content, "Error when evaluating `if` for job `invalid`") +} + func TestComputeReusableCallerOutputs(t *testing.T) { require.NoError(t, unittest.PrepareTestDatabase()) ctx := t.Context() diff --git a/services/actions/helper.go b/services/actions/helper.go index 2e3a9d504c6..6ec6b88984a 100644 --- a/services/actions/helper.go +++ b/services/actions/helper.go @@ -5,6 +5,7 @@ package actions import ( "context" + "errors" "fmt" actions_model "gitea.dev/models/actions" @@ -92,11 +93,35 @@ func pullRequestTargetBaseSHA(run *actions_model.ActionRun) (string, bool) { return payload.PullRequest.Base.Sha, true } -// evaluateJobIf evaluates a job's `if:` +// evaluateJobIf evaluates a job's `if:`. An invalid `if:` skips the job and is reported in its summary. func evaluateJobIf(ctx context.Context, run *actions_model.ActionRun, attempt *actions_model.ActionRunAttempt, job *actions_model.ActionRunJob, vars map[string]string, allNeedsSucceed bool) (bool, error) { + shouldStart, err := resolveJobIf(ctx, run, attempt, job, vars, allNeedsSucceed) + if errors.Is(err, util.ErrInvalidArgument) { + return false, upsertJobErrorSummary(ctx, job, "if", err) + } + return shouldStart, err +} + +// decideJobIf skips a waiting job whose `if:` is false before its insertion, an invalid `if:` is returned for the job's summary. +func decideJobIf(ctx context.Context, run *actions_model.ActionRun, attempt *actions_model.ActionRunAttempt, job *actions_model.ActionRunJob, vars map[string]string) (invalidIf, err error) { + if job.Status != actions_model.StatusWaiting { + return nil, nil + } + shouldStart, err := resolveJobIf(ctx, run, attempt, job, vars, true) + if errors.Is(err, util.ErrInvalidArgument) { + invalidIf, err = err, nil + } + if !shouldStart { + job.Status = actions_model.StatusSkipped + } + return invalidIf, err +} + +// resolveJobIf evaluates a job's `if:` and returns an invalid `if:` as util.ErrInvalidArgument. +func resolveJobIf(ctx context.Context, run *actions_model.ActionRun, attempt *actions_model.ActionRunAttempt, job *actions_model.ActionRunJob, vars map[string]string, allNeedsSucceed bool) (bool, error) { parsedJob, err := job.ParseJob() if err != nil { - return false, upsertJobErrorSummary(ctx, job, "if", err) + return false, util.NewInvalidArgumentErrorf("%v", err) } // Empty `if:` reduces to implicit `success()` - true iff every need finished as Success. if len(parsedJob.If.Value) == 0 { @@ -122,7 +147,7 @@ func evaluateJobIf(ctx context.Context, run *actions_model.ActionRun, attempt *a gitCtx["job"] = "" // github.com decides a job's `if:` before the job exists shouldStart, err := jobparser.EvaluateJobIfExpression(job.JobID, parsedJob, gitCtx, jobResults, vars, inputs) if err != nil { - return false, upsertJobErrorSummary(ctx, job, "if", err) + return false, util.NewInvalidArgumentErrorf("%v", err) } return shouldStart, nil } diff --git a/services/actions/rerun.go b/services/actions/rerun.go index 9ec81f2edb7..0efe7d09355 100644 --- a/services/actions/rerun.go +++ b/services/actions/rerun.go @@ -190,7 +190,7 @@ func buildRerunPlan(ctx context.Context, run *actions_model.ActionRun, triggerUs // Inside a single database transaction it then inserts the new attempt, clones all template jobs, evaluates job-level concurrency for rerun jobs, // and updates the run's latest_attempt_id. // Jobs not in the rerun set are cloned as pass-through: their status is preserved and SourceTaskID points to the original task so the UI can still display their results. -// The attempt's final status is derived only from the rerun jobs, not the pass-through jobs. +// The attempt's status aggregates all its jobs, pass-through ones included, as rerun jobs skipped by `if:` get no later update. // Notifications and commit statuses are sent after the transaction commits. func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionRunAttempt, error) { vars, err := actions_model.GetVariablesOfRun(ctx, plan.run) @@ -222,7 +222,6 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR var newJobs, newJobsToRerun actions_model.ActionJobList var cancelledConcurrencyJobs []*actions_model.ActionRunJob - var hasWaitingCallerJobs bool err = db.WithTx(ctx, func(ctx context.Context) error { newAttemptStatus, jobsToCancel, err := PrepareToStartRunWithConcurrency(ctx, newAttempt) @@ -281,10 +280,10 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR newJob.ParentJobID = newParentID } + var invalidIf error if plan.rerunAttemptJobIDs.Contains(templateJob.AttemptJobID) { - // A deferred-matrix placeholder must go through the emitter, which is the only place - // that expands it: dispatching it directly would hand the runner the raw payload. - shouldBlockJob := shouldBlock || plan.hasRerunDependency(templateJob) || newJob.IsMatrixDeferred + // the emitter decides `if:` once all needs have results, and is the only place expanding a deferred matrix + shouldBlockJob := shouldBlock || len(newJob.Needs) > 0 || newJob.IsMatrixDeferred newJob.Status = util.Iif(shouldBlockJob, actions_model.StatusBlocked, actions_model.StatusWaiting) newJob.TaskID = 0 @@ -300,8 +299,13 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR newJob.CallPayload = "" } + invalidIf, err = decideJobIf(ctx, plan.run, newAttempt, newJob, vars) + if err != nil { + return fmt.Errorf("evaluate job if: %w", err) + } + // A slot-starved job must not cancel its group peers. - if newJob.RawConcurrency != "" && !shouldBlockJob && slots.available(newJob) { + if newJob.RawConcurrency != "" && newJob.Status == actions_model.StatusWaiting && slots.available(newJob) { if err := EvaluateJobConcurrencyFillModel(ctx, plan.run, newAttempt, newJob, vars, nil); err != nil { return fmt.Errorf("evaluate job concurrency: %w", err) } @@ -327,17 +331,17 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR return err } templateIDToNewID[templateJob.ID] = newJob.ID + if invalidIf != nil { + if err := upsertJobErrorSummary(ctx, newJob, "if", invalidIf); err != nil { + return err + } + } // expand reusable caller if newJob.IsReusableCaller && newJob.Status == actions_model.StatusWaiting && !newJob.IsExpanded { - if err := expandReusableWorkflowCaller(ctx, plan.run, newAttempt, newJob, vars); err != nil { - return fmt.Errorf("inline trigger caller %d ready: %w", newJob.ID, err) + if err := expandInlineReusableCaller(ctx, plan.run, newAttempt, newJob, vars); err != nil { + return err } - // refresh the caller status - if err := actions_model.RefreshReusableCallerStatus(ctx, newJob); err != nil { - return fmt.Errorf("refresh caller %d status: %w", newJob.ID, err) - } - hasWaitingCallerJobs = true } // A reusable caller is never dispatched to a runner, so it must not drive the task-version bump. @@ -356,7 +360,7 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR } } - newAttempt.Status = actions_model.AggregateJobStatus(newJobsToRerun) + newAttempt.Status = actions_model.AggregateJobStatus(newJobs) if err := actions_model.UpdateRunAttempt(ctx, newAttempt, "status"); err != nil { return err } @@ -383,12 +387,9 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR CreateCommitStatusForRunJobs(ctx, plan.run, newJobs...) NotifyWorkflowJobsAndRunsStatusUpdate(ctx, newJobsToRerun) - // Post-commit kick for expanded callers and restored matrix placeholders: let job_emitter - // resolve child jobs, and re-expand a placeholder whose needs may all be pass-through and done. - if hasWaitingCallerJobs || len(plan.matrixPlaceholderTemplateIDs) > 0 { - if err := EmitJobsIfReadyByRun(plan.run.ID); err != nil { - log.Error("emit run %d after rerun: %v", plan.run.ID, err) - } + // Post-commit kick: resolve rerun jobs with needs, children of expanded callers and dependents of skipped jobs. + if err := EmitJobsIfReadyByRun(plan.run.ID); err != nil { + log.Error("emit run %d after rerun: %v", plan.run.ID, err) } return newAttempt, nil @@ -487,7 +488,7 @@ func (p *rerunPlan) expandRerunJobIDs(jobsToRerun []*actions_model.ActionRunJob) // hasRerunDependency reports whether `job` has a needs-reference that points to a job which is itself being rerun (in rerunAttemptJobIDs) // or is an ancestor caller whose subtree is being rerun (in ancestorAttemptJobIDs). -// Either case means `job` should start in Blocked status. +// Either case means the needs of `job` may produce different results or outputs in the new attempt. func (p *rerunPlan) hasRerunDependency(job *actions_model.ActionRunJob) bool { if len(job.Needs) == 0 { return false diff --git a/services/actions/rerun_test.go b/services/actions/rerun_test.go index 5cfeff0bb09..4c7383ba1fb 100644 --- a/services/actions/rerun_test.go +++ b/services/actions/rerun_test.go @@ -7,8 +7,11 @@ import ( "testing" actions_model "gitea.dev/models/actions" + repo_model "gitea.dev/models/repo" + "gitea.dev/models/unittest" user_model "gitea.dev/models/user" "gitea.dev/modules/container" + "gitea.dev/modules/test" "gitea.dev/modules/util" "github.com/stretchr/testify/assert" @@ -453,3 +456,52 @@ func TestCollectMatrixCollapse(t *testing.T) { assert.Empty(t, plan.matrixSiblingSkipTemplateIDs) }) } + +func TestRerunDecidesJobIf(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + defer test.MockVariableValue(&EmitJobsIfReadyByRun, func(int64) error { return nil })() + ctx := t.Context() + + repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 4}) + variable, err := actions_model.InsertVariable(ctx, 0, repo.ID, "DEPLOY", "yes", "") + require.NoError(t, err) + + run := insertMaxParallelRun(t, `on: push +jobs: + build: + runs-on: ubuntu-latest + steps: + - run: echo + deploy: + if: vars.DEPLOY == 'yes' + runs-on: ubuntu-latest + steps: + - run: echo +`, false) + jobs := map[string]*actions_model.ActionRunJob{} + for _, job := range runJobs(t, run.ID, run.LatestAttemptID) { + require.Equal(t, actions_model.StatusWaiting, job.Status) + job.Status = actions_model.StatusSuccess + _, err = actions_model.UpdateRunJob(ctx, job, nil, "status") + require.NoError(t, err) + jobs[job.JobID] = job + } + run = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run.ID}) + + variable.Data = "no" + _, err = actions_model.UpdateVariableCols(ctx, variable, "data") + require.NoError(t, err) + attempt, err := RerunWorkflowRunJobs(ctx, repo, run, &user_model.User{ID: 1}, []*actions_model.ActionRunJob{jobs["deploy"]}) + require.NoError(t, err) + rerunJobs := map[string]*actions_model.ActionRunJob{} + for _, job := range runJobs(t, run.ID, attempt.ID) { + rerunJobs[job.JobID] = job + } + assert.Equal(t, actions_model.StatusSuccess, rerunJobs["build"].Status) + assert.Equal(t, actions_model.StatusSkipped, rerunJobs["deploy"].Status) + + attempt = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunAttempt{ID: attempt.ID}) + assert.Equal(t, actions_model.StatusSuccess, attempt.Status) + assert.NotZero(t, attempt.Stopped) + assert.Equal(t, actions_model.StatusSuccess, unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run.ID}).Status) +} diff --git a/services/actions/run.go b/services/actions/run.go index 2379b3d8c97..4fc70788880 100644 --- a/services/actions/run.go +++ b/services/actions/run.go @@ -178,11 +178,7 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte return nil } -// insertRunJob builds a single run job from a parsed workflow job, evaluates its -// job-level concurrency, inserts it, and — for a ready no-needs reusable caller — -// 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. +// insertRunJob returns the inserted job, the jobs its concurrency cancelled, and whether a post-commit emitter pass is needed. 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() @@ -236,6 +232,12 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt runJob.CallUses = job.Uses } + // a skipped job must neither cancel its group peers nor take a slot + invalidIf, err := decideJobIf(ctx, run, runAttempt, runJob, vars) + if err != nil { + return nil, nil, false, fmt.Errorf("evaluate job if: %w", err) + } + var cancelledConcurrencyJobs []*actions_model.ActionRunJob // check job concurrency if job.RawConcurrency != nil { @@ -245,8 +247,8 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt } runJob.RawConcurrency = string(rawConcurrency) - // do not evaluate job concurrency when it requires `needs`, the jobs with `needs` will be evaluated later by job emitter - if len(needs) == 0 { + // the job emitter evaluates it for jobs with `needs`, a skipped job never takes part + if len(needs) == 0 && runJob.Status != actions_model.StatusSkipped { if err := EvaluateJobConcurrencyFillModel(ctx, run, runAttempt, runJob, vars, inputs); err != nil { return nil, nil, false, fmt.Errorf("evaluate job concurrency: %w", err) } @@ -270,43 +272,28 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt if err := db.Insert(ctx, runJob); err != nil { return nil, nil, false, err } - - // expand reusable caller - var needPostCommitEmit bool - if isReusableWorkflowCaller && runJob.Status == actions_model.StatusWaiting { - if err := processInlineReusableCaller(ctx, run, runAttempt, runJob, vars); err != nil { + if invalidIf != nil { + if err := upsertJobErrorSummary(ctx, runJob, "if", invalidIf); err != nil { return nil, nil, false, err } - // A processed caller always needs a resolver pass: - // - if the caller is expanded, resolve its children jobs; - // - if the caller is skipped, propagate its state to its dependents - needPostCommitEmit = true } - return runJob, cancelledConcurrencyJobs, needPostCommitEmit, nil + if isReusableWorkflowCaller && runJob.Status == actions_model.StatusWaiting { + if err := expandInlineReusableCaller(ctx, run, runAttempt, runJob, vars); err != nil { + return nil, nil, false, err + } + } + + // the emitter resolves an expanded caller's children and a skipped job's dependents + return runJob, cancelledConcurrencyJobs, runJob.IsExpanded || runJob.Status == actions_model.StatusSkipped, nil } -// processInlineReusableCaller evaluates a no-needs reusable caller's own `if:` and -// either inline-expands it into child jobs or marks it skipped. -// (A caller with needs is Blocked and gets its `if:` evaluated by the job emitter instead.) -func processInlineReusableCaller(ctx context.Context, run *actions_model.ActionRun, runAttempt *actions_model.ActionRunAttempt, caller *actions_model.ActionRunJob, vars map[string]string) error { - shouldStart, err := evaluateJobIf(ctx, run, runAttempt, caller, vars, true) - if err != nil { - return fmt.Errorf("evaluate caller %d if: %w", caller.ID, err) +func expandInlineReusableCaller(ctx context.Context, run *actions_model.ActionRun, runAttempt *actions_model.ActionRunAttempt, caller *actions_model.ActionRunJob, vars map[string]string) error { + if err := expandReusableWorkflowCaller(ctx, run, runAttempt, caller, vars); err != nil { + return fmt.Errorf("inline trigger caller %d ready: %w", caller.ID, err) } - if shouldStart { - if err := expandReusableWorkflowCaller(ctx, run, runAttempt, caller, vars); err != nil { - return fmt.Errorf("inline trigger caller %d ready: %w", caller.ID, err) - } - // refresh the caller status - if err := actions_model.RefreshReusableCallerStatus(ctx, caller); err != nil { - return fmt.Errorf("refresh caller %d status: %w", caller.ID, err) - } - return nil - } - caller.Status = actions_model.StatusSkipped - if _, err := actions_model.UpdateRunJob(ctx, caller, nil, "status"); err != nil { - return fmt.Errorf("skip caller %d: %w", caller.ID, err) + if err := actions_model.RefreshReusableCallerStatus(ctx, caller); err != nil { + return fmt.Errorf("refresh caller %d status: %w", caller.ID, err) } return nil } diff --git a/services/actions/task.go b/services/actions/task.go index 35d31788b20..c744306ecb6 100644 --- a/services/actions/task.go +++ b/services/actions/task.go @@ -14,10 +14,12 @@ import ( actions_model "gitea.dev/models/actions" "gitea.dev/models/db" secret_model "gitea.dev/models/secret" + "gitea.dev/modules/actions/jobparser" "gitea.dev/modules/graceful" "gitea.dev/modules/log" "gitea.dev/modules/setting" + "go.yaml.in/yaml/v4" "google.golang.org/protobuf/types/known/structpb" ) @@ -156,9 +158,14 @@ func buildRunnerTask(ctx context.Context, t *actions_model.ActionTask) (*runnerv return nil, nil, fmt.Errorf("generateTaskContext: %w", err) } + payload, err := runnerWorkflowPayload(job) + if err != nil { + return nil, nil, fmt.Errorf("runnerWorkflowPayload: %w", err) + } + return &runnerv1.Task{ Id: t.ID, - WorkflowPayload: t.Job.WorkflowPayload, + WorkflowPayload: payload, Context: taskContext, Secrets: secrets, Vars: vars, @@ -166,6 +173,19 @@ func buildRunnerTask(ctx context.Context, t *actions_model.ActionTask) (*runnerv }, job, nil } +// runnerWorkflowPayload sets the job `if:` to `always()`, as Gitea has decided it and a runner must not re-evaluate it. +func runnerWorkflowPayload(job *actions_model.ActionRunJob) ([]byte, error) { + swf, parsedJob, err := jobparser.ParseRawSingleWorkflow(job.WorkflowPayload) + if err != nil { + return nil, err + } + parsedJob.If = yaml.Node{Kind: yaml.ScalarNode, Value: "always()"} + if err := swf.SetJob(job.JobID, parsedJob); err != nil { + return nil, err + } + return swf.Marshal() +} + func generateTaskContext(ctx context.Context, t *actions_model.ActionTask) (*structpb.Struct, error) { giteaRuntimeToken, err := CreateAuthorizationToken(t.ID, t.Job.RunID, t.JobID) if err != nil { diff --git a/services/actions/task_test.go b/services/actions/task_test.go index 21f8615623d..397266059f3 100644 --- a/services/actions/task_test.go +++ b/services/actions/task_test.go @@ -9,6 +9,7 @@ import ( actions_model "gitea.dev/models/actions" "gitea.dev/models/db" "gitea.dev/models/unittest" + "gitea.dev/modules/actions/jobparser" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -72,3 +73,17 @@ func TestReleaseTaskForRunnerCleanup(t *testing.T) { assert.Zero(t, released.TaskID) unittest.AssertNotExistsBean(t, &actions_model.ActionTask{ID: task.ID}) } + +func TestRunnerWorkflowPayload(t *testing.T) { + payload, err := runnerWorkflowPayload(&actions_model.ActionRunJob{ + JobID: "build", + WorkflowPayload: []byte("name: ci\njobs:\n build:\n if: false\n steps:\n - run: echo\n"), + }) + require.NoError(t, err) + + swf, parsedJob, err := jobparser.ParseRawSingleWorkflow(payload) + require.NoError(t, err) + assert.Equal(t, "ci", swf.Name) + assert.Equal(t, "always()", parsedJob.If.Value) + assert.Len(t, parsedJob.Steps, 1) +} diff --git a/tests/integration/api_actions_run_test.go b/tests/integration/api_actions_run_test.go index 81ea0a7fbb3..6433c2789c1 100644 --- a/tests/integration/api_actions_run_test.go +++ b/tests/integration/api_actions_run_test.go @@ -208,6 +208,13 @@ func TestAPIActionsRerunWorkflowRun(t *testing.T) { writeToken := getTokenForLoggedInUser(t, session, auth_model.AccessTokenScopeWriteRepository) readToken := getTokenForLoggedInUser(t, session, auth_model.AccessTokenScopeReadRepository) + for _, jobID := range []int64{198, 199} { + job := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: jobID}) + job.WorkflowPayload = minimalConcurrentWorkflowPayload(job.JobID) + _, err := actions_model.UpdateRunJob(t.Context(), job, nil, "workflow_payload") + require.NoError(t, err) + } + t.Run("RunLogs", func(t *testing.T) { // run 795 (workflow "test.yaml") has job 198 "job_1" on task 53 and job 199 "job_2" on task 54 seedTaskLogs(t, 53, "hello from job_1") @@ -522,6 +529,7 @@ func testAPIActionsApproveWorkflowRun(t *testing.T) { require.NoError(t, db.Insert(t.Context(), &actions_model.ActionRunJob{ RunID: run.ID, RepoID: run.RepoID, OwnerID: run.OwnerID, CommitSHA: run.CommitSHA, Name: "job1", Attempt: 1, JobID: "job1", Status: actions_model.StatusBlocked, RunsOn: []string{"ubuntu-latest"}, + WorkflowPayload: minimalConcurrentWorkflowPayload("job1"), })) return run }