From 7cfd421aa82f45c0489c4d5a64bf52e4db98c159 Mon Sep 17 00:00:00 2001 From: Zettat123 Date: Tue, 21 Jul 2026 00:43:45 -0600 Subject: [PATCH] fix(actions): support `matrix` when evaluating workflow `if` expression (#38474) Partially fixes #38466 Added `matrix` to the evaluator for `if` expressions --------- Signed-off-by: Zettat123 Co-authored-by: bircni --- modules/actions/jobparser/model.go | 14 +- modules/actions/jobparser/model_test.go | 46 ++++ services/actions/run.go | 223 ++++++++++-------- .../actions_reusable_workflow_test.go | 68 ++++++ 4 files changed, 258 insertions(+), 93 deletions(-) diff --git a/modules/actions/jobparser/model.go b/modules/actions/jobparser/model.go index 97a36235d21..5950feb5c8f 100644 --- a/modules/actions/jobparser/model.go +++ b/modules/actions/jobparser/model.go @@ -490,7 +490,19 @@ func EvaluateJobIfExpression(jobID string, job *Job, gitCtx map[string]any, resu RawMatrix: job.Strategy.RawMatrix, }, } - evaluator := NewExpressionEvaluator(NewInterpeter(jobID, actJob, nil, toGitContext(gitCtx), results, vars, inputs)) + // Each per-matrix job carries its single matrix combination in RawMatrix so resolve it and pass it in; + // otherwise `matrix.*` references in `if:` evaluate to null. + // GetMatrixes always returns at least one element (an empty map for a job without a matrix), + // so only a non-empty combination should populate `matrix.*`, leaving it nil otherwise. + var matrix map[string]any + matrixes, err := actJob.GetMatrixes() + if err != nil { + return false, err + } + if len(matrixes) > 0 && len(matrixes[0]) > 0 { + matrix = matrixes[0] + } + evaluator := NewExpressionEvaluator(NewInterpeter(jobID, actJob, matrix, toGitContext(gitCtx), results, vars, inputs)) expr, err := rewriteSubExpression(job.If.Value, false) if err != nil { return false, err diff --git a/modules/actions/jobparser/model_test.go b/modules/actions/jobparser/model_test.go index e3b2d2b95f9..23d7489fedb 100644 --- a/modules/actions/jobparser/model_test.go +++ b/modules/actions/jobparser/model_test.go @@ -4,6 +4,7 @@ package jobparser import ( + "fmt" "strings" "testing" @@ -465,6 +466,51 @@ func TestParseMappingNode(t *testing.T) { } } +func TestEvaluateJobIfExpressionMatrix(t *testing.T) { + ifExprs := []string{ + `${{ contains(fromJSON('["linux","windows"]'), matrix.target) }}`, + `${{ contains('["linux","windows"]', matrix.target) }}`, + } + + want := map[string]bool{ + "build (linux)": true, + "build (windows)": true, + "build (macos)": false, + } + + for _, ifExpr := range ifExprs { + t.Run(ifExpr, func(t *testing.T) { + content := fmt.Sprintf(` +name: test +on: push +jobs: + build: + runs-on: ubuntu-latest + if: %s + strategy: + fail-fast: false + matrix: + target: [linux, windows, macos] + steps: + - run: echo ${{ matrix.target }} +`, ifExpr) + + swfs, err := Parse([]byte(content)) + require.NoError(t, err) + require.Len(t, swfs, 3) + + got := make(map[string]bool, len(swfs)) + for _, swf := range swfs { + id, job := swf.Job() + shouldRun, err := EvaluateJobIfExpression(id, job, map[string]any{}, map[string]*JobResult{id: {}}, nil, nil) + require.NoError(t, err) + got[job.Name] = shouldRun + } + assert.Equal(t, want, got) + }) + } +} + func TestEvaluateJobIfExpression(t *testing.T) { kases := []struct { name string diff --git a/services/actions/run.go b/services/actions/run.go index fc2631e19d7..c8e476fa329 100644 --- a/services/actions/run.go +++ b/services/actions/run.go @@ -60,7 +60,7 @@ func PrepareRunAndInsert(ctx context.Context, content []byte, run *actions_model // 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, content []byte, vars map[string]string, inputs map[string]any, wfRawConcurrency *act_model.RawConcurrency) error { var cancelledConcurrencyJobs []*actions_model.ActionRunJob - var hasWaitingCallerJobs bool + var needPostCommitEmit bool if err := db.WithTx(ctx, func(ctx context.Context) error { index, err := db.GetNextResourceIndex(ctx, "action_run_index", run.RepoID) if err != nil { @@ -133,99 +133,15 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte runJobs := make([]*actions_model.ActionRunJob, 0, len(jobs)) var hasWaitingJobs 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() - - isReusableWorkflowCaller := job.Uses != "" - shouldBlockJob := runAttempt.Status == actions_model.StatusBlocked || len(needs) > 0 || run.NeedApproval - - attemptJobID, err := actions_model.GetNextAttemptJobID(ctx, run.ID) + runJob, jobsToCancel, jobNeedsPostCommitEmit, err := insertRunJob(ctx, run, runAttempt, v, vars, inputs) if err != nil { - return fmt.Errorf("alloc attempt_job_id: %w", err) - } - - job.Name = util.EllipsisDisplayString(job.Name, 255) - runJob := &actions_model.ActionRunJob{ - RunID: run.ID, - RunAttemptID: runAttempt.ID, - RepoID: run.RepoID, - OwnerID: run.OwnerID, - CommitSHA: run.CommitSHA, - IsForkPullRequest: run.IsForkPullRequest, - Name: job.Name, - Attempt: runAttempt.Attempt, - WorkflowPayload: payload, - JobID: id, - AttemptJobID: attemptJobID, - Needs: needs, - RunsOn: job.RunsOn(), - Status: util.Iif(shouldBlockJob, actions_model.StatusBlocked, actions_model.StatusWaiting), - WorkflowSourceRepoID: run.WorkflowRepoID, - WorkflowSourceCommitSHA: run.WorkflowCommitSHA, - ContinueOnError: job.GetContinueOnError(), - } - // Parse workflow/job permissions (no clamping here) - if perms := ExtractJobPermissionsFromWorkflow(v, job); perms != nil { - runJob.TokenPermissions = perms - } - - if isReusableWorkflowCaller { - runJob.IsReusableCaller = true - runJob.CallUses = job.Uses - } - - // check job concurrency - if job.RawConcurrency != nil { - rawConcurrency, err := yaml.Marshal(job.RawConcurrency) - if err != nil { - return fmt.Errorf("marshal raw concurrency: %w", err) - } - runJob.RawConcurrency = string(rawConcurrency) - - // do not evaluate job concurrency when it requires `needs`, the jobs with `needs` will be evaluated later by job emitter - if len(needs) == 0 { - err = EvaluateJobConcurrencyFillModel(ctx, run, runAttempt, runJob, vars, inputs) - if err != nil { - return fmt.Errorf("evaluate job concurrency: %w", err) - } - } - - // 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 { - var jobsToCancel []*actions_model.ActionRunJob - runJob.Status, jobsToCancel, err = PrepareToStartJobWithConcurrency(ctx, runJob) - if err != nil { - return fmt.Errorf("prepare to start job with concurrency: %w", err) - } - cancelledConcurrencyJobs = append(cancelledConcurrencyJobs, jobsToCancel...) - } - } - - // 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 && !isReusableWorkflowCaller) - if err := db.Insert(ctx, runJob); err != nil { return err } - - // expand reusable caller - if isReusableWorkflowCaller && runJob.Status == actions_model.StatusWaiting { - if err := expandReusableWorkflowCaller(ctx, run, runAttempt, runJob, vars); err != nil { - return fmt.Errorf("inline trigger caller %d ready: %w", runJob.ID, err) - } - // refresh the caller status - if err := actions_model.RefreshReusableCallerStatus(ctx, runJob); err != nil { - return fmt.Errorf("refresh caller %d status: %w", runJob.ID, err) - } - hasWaitingCallerJobs = true - } - + 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) } @@ -249,8 +165,8 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte NotifyWorkflowJobsAndRunsStatusUpdate(ctx, cancelledConcurrencyJobs) EmitJobsIfReadyByJobs(cancelledConcurrencyJobs) - // Post-commit kick for expanded callers: let job_emitter resolve its child jobs - if hasWaitingCallerJobs { + // Post-commit kick: let the job emitter resolve jobs if needed + if needPostCommitEmit { if err := EmitJobsIfReadyByRun(run.ID); err != nil { log.Error("emit run %d after InsertRun: %v", run.ID, err) } @@ -258,3 +174,126 @@ 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. +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) { + id, job := workflowJob.Job() + needs := job.Needs() + if err := workflowJob.SetJob(id, job.EraseNeeds()); err != nil { + return nil, nil, false, err + } + payload, _ := workflowJob.Marshal() + + isReusableWorkflowCaller := job.Uses != "" + shouldBlockJob := runAttempt.Status == actions_model.StatusBlocked || len(needs) > 0 || run.NeedApproval + + attemptJobID, err := actions_model.GetNextAttemptJobID(ctx, run.ID) + if err != nil { + return nil, nil, false, fmt.Errorf("alloc attempt_job_id: %w", err) + } + + job.Name = util.EllipsisDisplayString(job.Name, 255) + runJob := &actions_model.ActionRunJob{ + RunID: run.ID, + RunAttemptID: runAttempt.ID, + RepoID: run.RepoID, + OwnerID: run.OwnerID, + CommitSHA: run.CommitSHA, + IsForkPullRequest: run.IsForkPullRequest, + Name: job.Name, + Attempt: runAttempt.Attempt, + WorkflowPayload: payload, + JobID: id, + AttemptJobID: attemptJobID, + Needs: needs, + RunsOn: job.RunsOn(), + Status: util.Iif(shouldBlockJob, actions_model.StatusBlocked, actions_model.StatusWaiting), + WorkflowSourceRepoID: run.WorkflowRepoID, + WorkflowSourceCommitSHA: run.WorkflowCommitSHA, + ContinueOnError: job.GetContinueOnError(), + } + // Parse workflow/job permissions (no clamping here) + if perms := ExtractJobPermissionsFromWorkflow(workflowJob, job); perms != nil { + runJob.TokenPermissions = perms + } + + if isReusableWorkflowCaller { + runJob.IsReusableCaller = true + runJob.CallUses = job.Uses + } + + var cancelledConcurrencyJobs []*actions_model.ActionRunJob + // check job concurrency + if job.RawConcurrency != nil { + rawConcurrency, err := yaml.Marshal(job.RawConcurrency) + if err != nil { + return nil, nil, false, fmt.Errorf("marshal raw concurrency: %w", err) + } + 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 { + if err := EvaluateJobConcurrencyFillModel(ctx, run, runAttempt, runJob, vars, inputs); err != nil { + return nil, nil, false, fmt.Errorf("evaluate job concurrency: %w", err) + } + } + + // 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 { + var jobsToCancel []*actions_model.ActionRunJob + runJob.Status, jobsToCancel, err = PrepareToStartJobWithConcurrency(ctx, runJob) + if err != nil { + return nil, nil, false, fmt.Errorf("prepare to start job with concurrency: %w", err) + } + cancelledConcurrencyJobs = append(cancelledConcurrencyJobs, jobsToCancel...) + } + } + + 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 { + 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 +} + +// 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) + } + 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) + } + return nil +} diff --git a/tests/integration/actions_reusable_workflow_test.go b/tests/integration/actions_reusable_workflow_test.go index 632b41f2b37..ab79fcfed67 100644 --- a/tests/integration/actions_reusable_workflow_test.go +++ b/tests/integration/actions_reusable_workflow_test.go @@ -19,6 +19,7 @@ import ( user_model "gitea.dev/models/user" "gitea.dev/modules/git" "gitea.dev/modules/json" + "gitea.dev/modules/queue" api "gitea.dev/modules/structs" "github.com/stretchr/testify/assert" @@ -763,6 +764,73 @@ jobs: run = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: runID}) assert.Equal(t, actions_model.StatusSuccess, run.Status) }) + + t.Run("No-needs caller if evaluated inline: false skips, true expands", func(t *testing.T) { + // A no-needs reusable-workflow caller is processed inline during InsertRun, where its own + // `if:` is now evaluated before expansion: + // - a false `if:` skips the caller without inserting any children, and the skip is + // propagated to a dependent job (via the post-commit emitter kick); + // - a true `if:` still expands the caller into its child jobs. + apiRepo := createActionsTestRepo(t, user2Token, "caller-inline-if-test", false) + repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: apiRepo.ID}) + + createRepoWorkflowFile(t, user2, user2Token, repo, ".gitea/workflows/lib.yaml", + `name: Lib +on: + workflow_call: + +jobs: + inner: + runs-on: ubuntu-latest + steps: + - run: echo inner +`) + createRepoWorkflowFile(t, user2, user2Token, repo, ".gitea/workflows/caller.yaml", + `name: Caller +on: + push: + paths: + - '.gitea/workflows/caller.yaml' +jobs: + will_skip: + if: ${{ false }} + uses: ./.gitea/workflows/lib.yaml + + will_run: + if: ${{ true }} + uses: ./.gitea/workflows/lib.yaml + + after_skip: + needs: [will_skip] + runs-on: ubuntu-latest + steps: + - run: echo after_skip +`) + + // drain the emitter queue so the skip has propagated to the dependent job + assert.NoError(t, queue.GetManager().FlushAll(t.Context(), 5*time.Second)) + + run := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{RepoID: repo.ID}) + runID := run.ID + + // will_skip: a caller with a false `if:` is skipped inline and never expands (no children inserted). + willSkip := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: runID, JobID: "will_skip"}) + assert.True(t, willSkip.IsReusableCaller) + assert.False(t, willSkip.IsExpanded) + assert.Equal(t, actions_model.StatusSkipped, willSkip.Status) + unittest.AssertNotExistsBean(t, &actions_model.ActionRunJob{RunID: runID, ParentJobID: willSkip.ID}) + + // will_run: a caller with a true `if:` still expands into its child job. + willRun := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: runID, JobID: "will_run"}) + assert.True(t, willRun.IsReusableCaller) + assert.True(t, willRun.IsExpanded) + innerChild := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: runID, JobID: "inner"}) + assert.Equal(t, willRun.ID, innerChild.ParentJobID) + + // after_skip: a dependent of the skipped caller resolves to Skipped instead of staying Blocked. + afterSkip := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: runID, JobID: "after_skip"}) + assert.Equal(t, actions_model.StatusSkipped, afterSkip.Status) + }) }) }