From 5672b1c4cf69db180d17f25cb1f7873a4ac6347b Mon Sep 17 00:00:00 2001 From: Pascal Zimmermann Date: Tue, 28 Jul 2026 17:59:00 +0200 Subject: [PATCH] feat: Add support for dynamic matrix evaluation in Gitea Actions workflows (#36564) Adds dynamic matrix evaluation to Gitea Actions: a job's `strategy.matrix` can be built from the outputs of the jobs it needs. ```yaml jobs: generate: runs-on: ubuntu-latest outputs: matrix: ${{ steps.set.outputs.result }} steps: - id: set run: echo "result=[1,2,3]" >> $GITHUB_OUTPUT build: needs: [generate] runs-on: ubuntu-latest strategy: matrix: version: ${{ fromJson(needs.generate.outputs.matrix) }} steps: - run: echo "building ${{ matrix.version }}" ``` Such a matrix cannot be expanded at planning time, so the job is planned as a single placeholder and expanded by the job emitter once its needs finish. Each combination is then gated by `if:` and concurrency as usual. - A matrix that resolves to no combination fails the job, as on GitHub. - Expansion is capped at `MaxJobNumPerRun`. - Workflows without a needs-dependent matrix are unaffected. Fixes https://github.com/go-gitea/gitea/issues/25179 --------- Signed-off-by: Pascal Zimmermann Signed-off-by: ZPascal Co-authored-by: Claude Co-authored-by: silverwind Co-authored-by: Claude (Opus 4.8) Co-authored-by: bircni Co-authored-by: Zettat123 --- go.mod | 2 +- modelmigration/migrations.go | 1 + modelmigration/v1_28/v344.go | 25 ++ models/actions/run_job.go | 60 +++++ modules/actions/jobparser/jobparser.go | 216 ++++++++++++++--- modules/actions/jobparser/jobparser_test.go | 149 ++++++++++++ modules/actions/jobparser/model.go | 4 +- modules/actions/jobparser/workflow_call.go | 2 +- routers/web/repo/actions/actions.go | 4 +- services/actions/commit_status.go | 7 + services/actions/helper.go | 8 + services/actions/job_emitter.go | 102 ++++++-- services/actions/job_emitter_test.go | 12 +- services/actions/matrix.go | 226 ++++++++++++++++++ services/actions/matrix_test.go | 220 +++++++++++++++++ services/actions/max_parallel_test.go | 4 +- services/actions/rerun.go | 77 +++++- services/actions/reusable_workflow.go | 31 ++- services/actions/run.go | 6 + .../actions_dynamic_matrix_test.go | 165 +++++++++++++ 20 files changed, 1252 insertions(+), 69 deletions(-) create mode 100644 modelmigration/v1_28/v344.go create mode 100644 services/actions/matrix.go create mode 100644 services/actions/matrix_test.go create mode 100644 tests/integration/actions_dynamic_matrix_test.go diff --git a/go.mod b/go.mod index c13f3f702a..e905f84838 100644 --- a/go.mod +++ b/go.mod @@ -89,6 +89,7 @@ require ( github.com/prometheus/client_golang v1.24.0 github.com/quasoft/websspi v1.1.2 github.com/redis/go-redis/v9 v9.21.0 + github.com/rhysd/actionlint v1.7.12 github.com/robfig/cron/v3 v3.0.1 github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 github.com/sassoftware/go-rpmutils v0.4.0 @@ -244,7 +245,6 @@ require ( github.com/prometheus/common v0.70.0 // indirect github.com/prometheus/procfs v0.21.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect - github.com/rhysd/actionlint v1.7.12 // indirect github.com/rs/xid v1.6.0 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/shopspring/decimal v1.4.0 // indirect diff --git a/modelmigration/migrations.go b/modelmigration/migrations.go index 987ceec270..32ab0f9586 100644 --- a/modelmigration/migrations.go +++ b/modelmigration/migrations.go @@ -424,6 +424,7 @@ func prepareMigrationTasks() []*migration { // 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), + newMigration(344, "Add deferred-matrix columns to ActionRunJob", v1_28.AddDeferredMatrixColumnsToActionRunJob), } return preparedMigrations } diff --git a/modelmigration/v1_28/v344.go b/modelmigration/v1_28/v344.go new file mode 100644 index 0000000000..1820aec57a --- /dev/null +++ b/modelmigration/v1_28/v344.go @@ -0,0 +1,25 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package v1_28 + +import ( + "gitea.dev/modelmigration/base" + + "xorm.io/xorm" +) + +// AddDeferredMatrixColumnsToActionRunJob adds the columns backing deferred (dynamic) matrix expansion +func AddDeferredMatrixColumnsToActionRunJob(x base.EngineMigration) error { + type ActionRunJob struct { + // IsMatrixDeferred marks jobs whose matrix depends on other jobs' outputs and is therefore expanded only once those jobs finish; + IsMatrixDeferred bool `xorm:"NOT NULL DEFAULT FALSE"` + // DeferredMatrixPayload preserves the raw, unevaluated payload across expansion so a rerun can re-derive the matrix + DeferredMatrixPayload []byte `xorm:"LONGBLOB"` + } + _, err := x.SyncWithOptions(xorm.SyncOptions{ + IgnoreDropIndices: true, + IgnoreConstrains: true, + }, new(ActionRunJob)) + return err +} diff --git a/models/actions/run_job.go b/models/actions/run_job.go index 83dabff535..b4cf1caa94 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -13,6 +13,7 @@ import ( "gitea.dev/models/db" repo_model "gitea.dev/models/repo" "gitea.dev/modules/actions/jobparser" + "gitea.dev/modules/container" "gitea.dev/modules/log" "gitea.dev/modules/timeutil" "gitea.dev/modules/util" @@ -72,6 +73,18 @@ type ActionRunJob struct { // MaxParallel is strategy.max-parallel, shared by all matrix jobs with the same JobID (0 = unlimited). MaxParallel int `xorm:"NOT NULL DEFAULT 0"` + // IsMatrixDeferred marks a placeholder for a job whose matrix references `needs.*.outputs.*` and so + // could not be expanded at planning time. Its WorkflowPayload still carries the raw, unevaluated + // matrix; the job emitter expands it once the needs finish. Only a successful expansion clears the flag: + // it survives a terminal status (skipped, cancelled, failed expansion) so a rerun can recognize + // the row as unexpanded and re-derive the matrix instead of dispatching the raw payload. + IsMatrixDeferred bool `xorm:"NOT NULL DEFAULT FALSE"` + + // DeferredMatrixPayload preserves a deferred-matrix placeholder's original WorkflowPayload (the raw, unevaluated matrix). + // A rerun whose needs re-run collapses the combinations back into a single placeholder built from this payload, + // so the matrix is re-derived from the fresh outputs. + DeferredMatrixPayload []byte `xorm:"LONGBLOB"` + // RunAttemptID identifies the ActionRunAttempt this job belongs to. // A value of 0 indicates a legacy job created before ActionRunAttempt existed. RunAttemptID int64 `xorm:"index NOT NULL DEFAULT 0"` @@ -323,6 +336,53 @@ func GetPriorAttemptChildrenByParent(ctx context.Context, runID, currentAttemptI return nil, nil //nolint:nilnil // every prior attempt skipped this caller } +// GetPriorAttemptMatrixCombos returns the most recent prior attempt's combination rows of the given +// dynamic-matrix job, indexed by Name, so re-expansion keeps AttemptJobIDs stable across attempts. +func GetPriorAttemptMatrixCombos(ctx context.Context, runID, currentAttemptID, parentAttemptJobID int64, jobID string) (map[string]*ActionRunJob, error) { + // An unexpanded placeholder is not a combination, so it is skipped and the search looks further + // back past it. Only the columns the scope check and the result need are read: the rows carry + // two payload blobs, and every prior attempt of the job is a candidate. + var candidates []*ActionRunJob + if err := db.GetEngine(ctx). + Where("run_id = ? AND job_id = ? AND run_attempt_id < ? AND is_matrix_deferred = ?", runID, jobID, currentAttemptID, false). + Cols("id", "name", "attempt_job_id", "run_attempt_id", "parent_job_id"). + Desc("run_attempt_id"). + Find(&candidates); err != nil { + return nil, fmt.Errorf("find prior matrix combos: %w", err) + } + + // Every combination of one attempt shares a parent, so dedupe before the lookup. + parentIDs := container.FilterSlice(candidates, func(c *ActionRunJob) (int64, bool) { + return c.ParentJobID, c.ParentJobID > 0 + }) + parentAttemptIDByRowID := make(map[int64]int64, len(parentIDs)) + if len(parentIDs) > 0 { + var parents []*ActionRunJob + if err := db.GetEngine(ctx).In("id", parentIDs).Cols("id", "attempt_job_id").Find(&parents); err != nil { + return nil, fmt.Errorf("find prior matrix combo parents: %w", err) + } + for _, p := range parents { + parentAttemptIDByRowID[p.ID] = p.AttemptJobID + } + } + + // Rows arrive newest-attempt-first, so the first in-scope row fixes the attempt to take. + combos := map[string]*ActionRunJob{} + newestAttemptID := int64(0) + for _, c := range candidates { + if parentAttemptIDByRowID[c.ParentJobID] != parentAttemptJobID { + continue + } + if newestAttemptID == 0 { + newestAttemptID = c.RunAttemptID + } else if c.RunAttemptID != newestAttemptID { + break + } + combos[c.Name] = c + } + return combos, nil +} + // GetDirectChildJobsByParent returns the direct child jobs of a parent job (e.g. a reusable workflow caller). func GetDirectChildJobsByParent(ctx context.Context, parentJob *ActionRunJob) (ActionJobList, error) { var jobs []*ActionRunJob diff --git a/modules/actions/jobparser/jobparser.go b/modules/actions/jobparser/jobparser.go index 79c7b7b433..5981d27d0c 100644 --- a/modules/actions/jobparser/jobparser.go +++ b/modules/actions/jobparser/jobparser.go @@ -5,15 +5,61 @@ package jobparser import ( "bytes" + "errors" "fmt" + "slices" "sort" "strings" "gitea.com/gitea/runner/act/exprparser" "gitea.com/gitea/runner/act/model" + "github.com/rhysd/actionlint" "go.yaml.in/yaml/v4" ) +// HasDeferredMatrix reports whether the job's matrix can only be expanded once its needs finish: +// it reads the needs context and the job has needs to resolve that context against. +// Parse emits such a job as a single placeholder rather than one job per combination, so every +// caller that persists a job must agree with Parse on this condition. +func HasDeferredMatrix(job *Job) bool { + return len(job.Needs()) > 0 && rawMatrixReadsNeeds(&job.Strategy.RawMatrix) +} + +func rawMatrixReadsNeeds(node *yaml.Node) bool { + if node.Kind == yaml.ScalarNode { + return expressionReadsNeeds(node.Value) + } + return slices.ContainsFunc(node.Content, rawMatrixReadsNeeds) +} + +// expressionReadsNeeds reports whether value holds a ${{ }} expression reading the needs context. +// Every other context (github, vars, inputs, ...) is already available while planning, so deferring +// those too would replace their combinations with one placeholder and change the commit status +// contexts the run publishes, which a repository's required checks are configured against. +func expressionReadsNeeds(value string) bool { + for rest := value; ; { + _, after, found := strings.Cut(rest, "${{") + if !found { + return false + } + rest = after + // The lexer ends the expression at its closing `}}`, so it can be handed the whole remainder. + expr, err := actionlint.NewExprParser().Parse(actionlint.NewExprLexer(rest)) + if err != nil { + return true // unparseable here, let the expansion report it against the real values + } + readsNeeds := false + actionlint.VisitExprNode(expr, func(node, _ actionlint.ExprNode, entering bool) { + if variable, ok := node.(*actionlint.VariableNode); entering && ok && strings.EqualFold(variable.Name, "needs") { + readsNeeds = true + } + }) + if readsNeeds { + return true + } + } +} + func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) { origin, err := model.ReadWorkflow(bytes.NewReader(content)) if err != nil { @@ -37,7 +83,7 @@ func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) { results[id] = &JobResult{ Needs: job.Needs(), Result: pc.jobResults[id], - Outputs: nil, // not supported yet + Outputs: nil, // resolved at expansion time, not at plan time } } @@ -52,35 +98,34 @@ func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) { for i, id := range ids { job := jobs[i] - matricxes, err := getMatrixes(origin.GetJob(id)) - if err != nil { - return nil, fmt.Errorf("getMatrixes: %w", err) + originJob := origin.GetJob(id) + + if originJob == nil { + return nil, fmt.Errorf("job %s not found in origin workflow", id) } - for _, matrix := range matricxes { - job := job.Clone() - if job.Name == "" { - job.Name = id + + var combos []*Job + if HasDeferredMatrix(job) { + // The matrix reads values that do not exist yet (a needs output), so emit a single + // placeholder keeping it raw. Re-parsing that placeholder's payload yields it again, + // and the server expands it once the needs finish. + placeholder := job.Clone() + if placeholder.Name == "" { + placeholder.Name = id } - job.Strategy.RawMatrix = encodeMatrix(matrix) - evaluator := NewExpressionEvaluator(NewInterpeter(id, origin.GetJob(id), matrix, pc.gitContext, results, pc.vars, pc.inputs)) - job.Name = nameWithMatrix(job.Name, matrix, evaluator) - runsOn := origin.GetJob(id).RunsOn() - for i, v := range runsOn { - runsOn[i] = evaluator.Interpolate(v) + combos = []*Job{placeholder} + } else { + matricxes, err := getMatrixes(originJob) + if err != nil { + return nil, fmt.Errorf("getMatrixes: %w", err) } - job.RawRunsOn = encodeRunsOn(runsOn) - if err := evaluator.EvaluateYamlNode(&job.RawContinueOnError); err != nil { - return nil, fmt.Errorf("evaluate continue-on-error for job %q: %w", id, err) + if combos, err = buildMatrixCombos(id, job, matricxes, originJob, pc.gitContext, results, pc.vars, pc.inputs); err != nil { + return nil, err } - swf := &SingleWorkflow{ - Name: workflow.Name, - RawOn: workflow.RawOn, - Env: workflow.Env, - Defaults: workflow.Defaults, - RawPermissions: workflow.RawPermissions, - RunName: workflow.RunName, - } - if err := swf.SetJob(id, job); err != nil { + } + for _, combo := range combos { + swf := workflow.cloneHeader() + if err := swf.SetJob(id, combo); err != nil { return nil, fmt.Errorf("SetJob: %w", err) } ret = append(ret, swf) @@ -89,6 +134,121 @@ func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) { return ret, nil } +// cloneHeader returns a copy of w with its workflow-global fields but no jobs. +func (w *SingleWorkflow) cloneHeader() *SingleWorkflow { + return &SingleWorkflow{ + Name: w.Name, + RawOn: w.RawOn, + Env: w.Env, + Defaults: w.Defaults, + RawPermissions: w.RawPermissions, + RunName: w.RunName, + } +} + +// ExpandMatrixWithNeeds returns one Job per combination of job's matrix, resolved against the +// completed needs in results. As for EvaluateConcurrency, results must also describe jobID itself, +// which is where NewInterpeter reads the job's own needs from. +// maxCombinations caps how many combinations may be built: the values come from a needs output at +// runtime, so the cap has to be enforced before one Job is materialized per combination. +func ExpandMatrixWithNeeds(jobID string, job *Job, gitCtx *model.GithubContext, results map[string]*JobResult, vars map[string]string, inputs map[string]any, maxCombinations int) ([]*Job, error) { + actJob := &model.Job{Strategy: &model.Strategy{ + FailFastString: job.Strategy.FailFastString, + MaxParallelString: job.Strategy.MaxParallelString, + RawMatrix: job.Strategy.RawMatrix, + }} + + // Resolve fromJson(needs.*.outputs.*) and friends into concrete matrix values. + if err := NewExpressionEvaluator(NewInterpeter(jobID, actJob, nil, gitCtx, results, vars, inputs)). + EvaluateYamlNode(&actJob.Strategy.RawMatrix); err != nil { + return nil, fmt.Errorf("evaluate matrix: %w", err) + } + matrixes, err := getMatrixes(actJob) + if err != nil { + return nil, fmt.Errorf("getMatrixes: %w", err) + } + // act collapses a matrix that yields no combination (an empty vector or include, everything + // excluded, a whole-matrix expression that is not a mapping) into one empty combination, which + // would run the job once unparameterized. GitHub rejects such a matrix, so reject it too. + if len(matrixes) == 1 && len(matrixes[0]) == 0 { + return nil, errors.New("matrix must define at least one vector") + } + if len(matrixes) > maxCombinations { + return nil, fmt.Errorf("matrix expands to %d combinations, exceeding the limit of %d", len(matrixes), maxCombinations) + } + return buildMatrixCombos(jobID, job, matrixes, actJob, gitCtx, results, vars, inputs) +} + +// matrixesOf is this package's only entry to act's GetMatrixes, so that every caller is covered by +// the filter check below. A deferred placeholder is the first thing carrying a raw matrix this far, +// and the emitter reads its `if:` before expanding it. +// TODO: drop the check once gitea.com/gitea/runner validates the shape itself. +func matrixesOf(job *model.Job) ([]map[string]any, error) { + if err := validateMatrixFilters(job); err != nil { + return nil, err + } + matrixes, err := job.GetMatrixes() + if err != nil { + return nil, fmt.Errorf("GetMatrixes: %w", err) + } + return matrixes, nil +} + +// validateMatrixFilters rejects an `include`/`exclude` that is not a list of mappings. act asserts +// that shape without checking, so anything else panics there; an unevaluated ${{ }} expression, which +// is still a scalar, is the usual way to reach it. +func validateMatrixFilters(job *model.Job) error { + if job.Strategy == nil || job.Strategy.RawMatrix.Kind != yaml.MappingNode { + return nil + } + content := job.Strategy.RawMatrix.Content + for i := 0; i+1 < len(content); i += 2 { + name, value := content[i].Value, content[i+1] + if name != "include" && name != "exclude" { + continue + } + entries := []*yaml.Node{value} + if value.Kind == yaml.SequenceNode { + entries = value.Content + } + for _, entry := range entries { + if entry.Kind == yaml.AliasNode { + entry = entry.Alias + } + if entry.Kind != yaml.MappingNode { + return fmt.Errorf("matrix %s must be a list of mappings", name) + } + } + } + return nil +} + +// buildMatrixCombos builds one Job per matrix combination from src, baking the combination into the +// strategy and interpolating the name, runs-on and continue-on-error with it. +func buildMatrixCombos(jobID string, src *Job, matrixes []map[string]any, actJob *model.Job, gitCtx *model.GithubContext, results map[string]*JobResult, vars map[string]string, inputs map[string]any) ([]*Job, error) { + srcRunsOn := src.RunsOn() + combos := make([]*Job, 0, len(matrixes)) + for _, matrix := range matrixes { + combo := src.Clone() + if combo.Name == "" { + combo.Name = jobID + } + combo.Strategy.RawMatrix = encodeMatrix(matrix) + evaluator := NewExpressionEvaluator(NewInterpeter(jobID, actJob, matrix, gitCtx, results, vars, inputs)) + combo.Name = nameWithMatrix(combo.Name, matrix, evaluator) + runsOn := slices.Clone(srcRunsOn) + for i := range runsOn { + runsOn[i] = evaluator.Interpolate(runsOn[i]) + } + combo.RawRunsOn = encodeRunsOn(runsOn) + if err := evaluator.EvaluateYamlNode(&combo.RawContinueOnError); err != nil { + return nil, fmt.Errorf("evaluate continue-on-error for job %q: %w", jobID, err) + } + combos = append(combos, combo) + } + return combos, nil +} + func WithGitContext(context *model.GithubContext) ParseOption { return func(c *parseContext) { c.gitContext = context @@ -117,9 +277,9 @@ type parseContext struct { type ParseOption func(c *parseContext) func getMatrixes(job *model.Job) ([]map[string]any, error) { - ret, err := job.GetMatrixes() + ret, err := matrixesOf(job) if err != nil { - return nil, fmt.Errorf("GetMatrixes: %w", err) + return nil, err } sort.Slice(ret, func(i, j int) bool { return matrixName(ret[i]) < matrixName(ret[j]) diff --git a/modules/actions/jobparser/jobparser_test.go b/modules/actions/jobparser/jobparser_test.go index 05bb3151b7..96749f9c28 100644 --- a/modules/actions/jobparser/jobparser_test.go +++ b/modules/actions/jobparser/jobparser_test.go @@ -4,9 +4,11 @@ package jobparser import ( + "fmt" "strings" "testing" + "gitea.com/gitea/runner/act/model" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.yaml.in/yaml/v4" @@ -107,3 +109,150 @@ func TestParse(t *testing.T) { }) } } + +func TestParseDefersDynamicMatrix(t *testing.T) { + // A matrix referencing needs outputs yields one placeholder keeping the raw expression, rather + // than one job per resolvable static value. Any other matrix expands at plan time as usual. + const workflow = ` +on: push +jobs: + setup: + steps: [{run: echo}] + build: + %s + strategy: + matrix: + os: [a, b] + version: %s + steps: [{run: echo}] +` + for _, tt := range []struct { + name string + needs string + version string + deferred bool + want int + }{ + {"needs outputs", "needs: setup", "${{ fromJson(needs.setup.outputs.v) }}", true, 1}, + {"static", "needs: setup", "[1, 2]", false, 4}, + // Without needs there is nothing to resolve the expression from later, so deferring would + // strand the job as a single combination that never expands. + {"expression without needs", "", `["${{ github.sha }}"]`, false, 2}, + // A context that is already available while planning must keep expanding there, otherwise + // such a workflow would silently lose the per-combination commit statuses it used to create. + {"expression over another context", "needs: setup", `["${{ github.sha }}"]`, false, 2}, + // The needs context is looked up in the parsed expression, not in the raw text. + {"needs inside a string literal", "needs: setup", `["${{ format('needs.setup.outputs.v {0}', github.sha) }}"]`, false, 2}, + } { + t.Run(tt.name, func(t *testing.T) { + result, err := Parse(fmt.Appendf(nil, workflow, tt.needs, tt.version)) + require.NoError(t, err) + + var builds []*Job + for _, w := range result { + if id, job := w.Job(); id == "build" { + builds = append(builds, job) + } + } + require.Len(t, builds, tt.want) + assert.Equal(t, tt.deferred, HasDeferredMatrix(builds[0])) + }) + } +} + +func TestExpandMatrixWithNeeds(t *testing.T) { + // matrixYAML is the YAML value of the `matrix:` key, so a case can replace the whole node. + expandMax := func(t *testing.T, matrixYAML string, maxCombinations int) ([]*Job, error) { + t.Helper() + var strategy Strategy + require.NoError(t, yaml.Unmarshal([]byte("matrix:"+matrixYAML), &strategy)) + job := &Job{Name: "build", Strategy: strategy} + require.NoError(t, job.RawRunsOn.Encode("${{ matrix.os || 'ubuntu-latest' }}")) + require.NoError(t, job.RawNeeds.Encode([]string{"setup"})) + // The results map must describe the job itself too, as findJobNeedsAndFillJobResults does. + return ExpandMatrixWithNeeds("build", job, &model.GithubContext{}, map[string]*JobResult{ + "build": {Needs: []string{"setup"}}, + "setup": {Result: "success", Outputs: map[string]string{ + "versions": `["1.20", "1.21"]`, + "os": `["linux", "darwin"]`, + "include": `[{"os":"linux","fast":true},{"os":"windows","fast":false}]`, + "empty": "[]", + }}, + }, nil, nil, maxCombinations) + } + expand := func(t *testing.T, matrixYAML string) ([]*Job, error) { + t.Helper() + return expandMax(t, matrixYAML, 256) + } + + t.Run("expands the product and interpolates runs-on", func(t *testing.T) { + got, err := expand(t, "\n os: ${{ fromJson(needs.setup.outputs.os) }}\n version: ${{ fromJson(needs.setup.outputs.versions) }}\n") + require.NoError(t, err) + names := make([]string, 0, len(got)) + for _, combo := range got { + names = append(names, combo.Name) + assert.Contains(t, []string{"linux", "darwin"}, combo.RunsOn()[0]) + } + // Dimensions are appended in key order, as GitHub names multi-dimension combinations. + assert.ElementsMatch(t, []string{ + "build (linux, 1.20)", "build (linux, 1.21)", "build (darwin, 1.20)", "build (darwin, 1.21)", + }, names) + }) + + t.Run("static and dynamic dimensions expand together, once", func(t *testing.T) { + got, err := expand(t, "\n os: [linux, darwin]\n version: ${{ fromJson(needs.setup.outputs.versions) }}\n") + require.NoError(t, err) + assert.Len(t, got, 4) + }) + + t.Run("include-only matrix expands", func(t *testing.T) { + got, err := expand(t, "\n include: ${{ fromJson(needs.setup.outputs.include) }}\n") + require.NoError(t, err) + assert.Len(t, got, 2) + }) + + // GitHub rejects a matrix that yields no combinations instead of running the job unparameterized. + for _, tt := range []struct{ name, matrix string }{ + {"empty vector", "\n version: ${{ fromJson(needs.setup.outputs.empty) }}\n"}, + {"empty include", "\n include: ${{ fromJson(needs.setup.outputs.empty) }}\n"}, + {"whole matrix not a mapping", " ${{ fromJson(needs.setup.outputs.empty) }}\n"}, + } { + t.Run(tt.name+" errors", func(t *testing.T) { + _, err := expand(t, tt.matrix) + require.ErrorContains(t, err, "matrix must define at least one vector") + }) + } + + t.Run("unresolved need errors", func(t *testing.T) { + _, err := expand(t, "\n v: ${{ fromJson(needs.missing.outputs.v) }}\n") + require.ErrorContains(t, err, "evaluate matrix") + }) + + // The combination count comes from a runtime output, so it must be rejected before one Job per + // combination is built rather than after. + t.Run("too many combinations errors", func(t *testing.T) { + _, err := expandMax(t, "\n version: ${{ fromJson(needs.setup.outputs.versions) }}\n", 1) + require.ErrorContains(t, err, "exceeding the limit of 1") + }) +} + +func TestRejectsUnevaluatedMatrixFilters(t *testing.T) { + // act dereferences include/exclude entries as mappings without checking, so an unevaluated + // expression panics there. Every entry point into act's matrix expansion must reject it: Parse + // sees one in a workflow file and in a deferred placeholder's payload, and the emitter reads a + // placeholder's `if:` while its matrix is still raw. + for _, filter := range []string{"include", "exclude"} { + t.Run(filter, func(t *testing.T) { + _, err := Parse(fmt.Appendf(nil, + "name: t\non: push\njobs:\n build:\n runs-on: ubuntu-latest\n strategy:\n matrix:\n os: [a]\n %s: ${{ fromJson(vars.MATRIX) }}\n steps: [{run: echo}]\n", filter)) + require.ErrorContains(t, err, "must be a list of mappings") + + var strategy Strategy + require.NoError(t, yaml.Unmarshal(fmt.Appendf(nil, "matrix:\n os: [a]\n %s: ${{ fromJson(vars.MATRIX) }}\n", filter), &strategy)) + job := &Job{Name: "build", Strategy: strategy} + require.NoError(t, job.If.Encode("${{ true }}")) + _, err = EvaluateJobIfExpression("build", job, map[string]any{}, map[string]*JobResult{"build": {}}, nil, nil) + require.ErrorContains(t, err, "must be a list of mappings") + }) + } +} diff --git a/modules/actions/jobparser/model.go b/modules/actions/jobparser/model.go index 1df8d89350..6b423c038e 100644 --- a/modules/actions/jobparser/model.go +++ b/modules/actions/jobparser/model.go @@ -269,7 +269,7 @@ func EvaluateConcurrency(rc *model.RawConcurrency, jobID string, job *Job, gitCt } matrix := make(map[string]any) - matrixes, err := actJob.GetMatrixes() + matrixes, err := matrixesOf(actJob) if err != nil { return "", false, err } @@ -510,7 +510,7 @@ func EvaluateJobIfExpression(jobID string, job *Job, gitCtx map[string]any, resu // 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() + matrixes, err := matrixesOf(actJob) if err != nil { return false, err } diff --git a/modules/actions/jobparser/workflow_call.go b/modules/actions/jobparser/workflow_call.go index 7c534dca3c..1b01192af1 100644 --- a/modules/actions/jobparser/workflow_call.go +++ b/modules/actions/jobparser/workflow_call.go @@ -177,7 +177,7 @@ func EvaluateCallerWith( }} var matrix map[string]any - matrixes, err := actJob.GetMatrixes() + matrixes, err := matrixesOf(actJob) if err != nil { return nil, fmt.Errorf("get caller %q matrix: %w", jobID, err) } diff --git a/routers/web/repo/actions/actions.go b/routers/web/repo/actions/actions.go index 04a6142ed5..a84c4abffc 100644 --- a/routers/web/repo/actions/actions.go +++ b/routers/web/repo/actions/actions.go @@ -554,7 +554,9 @@ func (data *actionRunListData) processActionRuns(ctx *context.Context) bool { return false } for _, job := range jobs { - if !job.Status.In(actions_model.StatusWaiting, actions_model.StatusBlocked) { + // A deferred matrix is unresolvable until its needs finish, so the whole per-job block + // is skipped: parsing the payload would report a valid workflow as invalid. + if job.IsMatrixDeferred || !job.Status.In(actions_model.StatusWaiting, actions_model.StatusBlocked) { continue } if err := actions.ValidateWorkflowContent(job.WorkflowPayload); err != nil { diff --git a/services/actions/commit_status.go b/services/actions/commit_status.go index 39c7ff77aa..debacde117 100644 --- a/services/actions/commit_status.go +++ b/services/actions/commit_status.go @@ -54,6 +54,13 @@ func CreateCommitStatusForRunJobs(ctx context.Context, run *actions_model.Action } for _, job := range jobs { + // A deferred-matrix placeholder's name changes when it expands, so a status created while it + // waits would be orphaned. The emitter reloads the jobs after expanding and creates them + // then. A placeholder that reached a final status (skipped by its `if:`, cancelled with the + // run, or failed to expand) never expands, so it keeps its own name and still deserves a status. + if job.IsMatrixDeferred && !job.Status.IsDone() { + continue + } if err = createCommitStatus(ctx, run.Repo, event, commitID, scopedPrefix, run, job); err != nil { log.Error("Failed to create commit status for job %d: %v", job.ID, err) } diff --git a/services/actions/helper.go b/services/actions/helper.go index 37dbb326ed..5c85823fac 100644 --- a/services/actions/helper.go +++ b/services/actions/helper.go @@ -68,6 +68,14 @@ func evaluateJobIf(ctx context.Context, run *actions_model.ActionRun, attempt *a if err != nil { return false, err } + // GenerateGiteaContext dereferences the run's repo and trigger user, so load them here instead of + // relying on whatever the caller happened to load before. + if err := run.LoadRepo(ctx); err != nil { + return false, err + } + if err := run.LoadTriggerUser(ctx); err != nil { + return false, err + } gitCtx := GenerateGiteaContext(ctx, run, attempt, job) return jobparser.EvaluateJobIfExpression(job.JobID, parsedJob, gitCtx, jobResults, vars, inputs) } diff --git a/services/actions/job_emitter.go b/services/actions/job_emitter.go index 434ba9eac1..6e5f30a915 100644 --- a/services/actions/job_emitter.go +++ b/services/actions/job_emitter.go @@ -17,7 +17,6 @@ import ( "gitea.dev/modules/queue" "gitea.dev/modules/setting" "gitea.dev/modules/timeutil" - "gitea.dev/modules/util" "xorm.io/builder" ) @@ -270,6 +269,7 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action return result, nil } } + vars, err := actions_model.GetVariablesOfRun(ctx, run) if err != nil { return nil, err @@ -282,7 +282,10 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action job.Run = run } - updates := resolver.Resolve(ctx) + updates, err := resolver.Resolve(ctx) + if err != nil { + return err + } for _, job := range jobs { status, ok := updates[job.ID] if !ok { @@ -336,7 +339,10 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action return nil, err } - if expandedAnyCaller { + result.UpdatedJobs = append(result.UpdatedJobs, resolver.matrixUpdatedJobs...) + // Caller and matrix expansion both insert Blocked jobs, which only a follow-up pass resolves. + // Like the caller's children, matrix siblings are left out of result.Jobs and picked up there. + if expandedAnyCaller || resolver.matrixChanged { result.RunIDsToReEmit = append(result.RunIDsToReEmit, run.ID) } result.CancelledJobs = resolver.cancelledJobs @@ -352,6 +358,12 @@ type jobStatusResolver struct { jobMap map[int64]*actions_model.ActionRunJob vars map[string]string cancelledJobs []*actions_model.ActionRunJob + // matrixChanged is set when matrix expansion inserted siblings or failed a placeholder, both of + // which need a follow-up pass to resolve the dependents. + matrixChanged bool + // matrixUpdatedJobs holds jobs whose status matrix expansion persisted itself, so they are + // notified like the ones the caller updates from the resolved status map. + matrixUpdatedJobs []*actions_model.ActionRunJob } func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]string) *jobStatusResolver { @@ -392,19 +404,22 @@ func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]stri } } -func (r *jobStatusResolver) Resolve(ctx context.Context) map[int64]actions_model.Status { +func (r *jobStatusResolver) Resolve(ctx context.Context) (map[int64]actions_model.Status, error) { ret := map[int64]actions_model.Status{} for i := 0; i < len(r.statuses); i++ { - updated := r.resolve(ctx) + updated, err := r.resolve(ctx) + if err != nil { + return nil, err + } if len(updated) == 0 { - return ret + return ret, nil } for k, v := range updated { ret[k] = v r.statuses[k] = v } } - return ret + return ret, nil } func (r *jobStatusResolver) resolveCheckNeeds(id int64) (allDone, allSucceed bool) { @@ -426,7 +441,7 @@ func (r *jobStatusResolver) resolveCheckNeeds(id int64) (allDone, allSucceed boo return allDone, allSucceed } -func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model.Status { +func (r *jobStatusResolver) resolve(ctx context.Context) (map[int64]actions_model.Status, error) { ret := map[int64]actions_model.Status{} slots := maxParallelSlots{} @@ -455,13 +470,61 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model continue } + // Decide whether the job runs at all before expanding a deferred matrix: a job whose needs + // failed or were skipped has to be skipped too, not failed for a matrix those needs never + // produced the outputs for. A job-level `if:` cannot read `matrix.*`, so it does not need + // the combination, unlike the concurrency expression evaluated below. + shouldStartJob, err := evaluateJobIf(ctx, actionRunJob.Run, nil, actionRunJob, r.vars, allSucceed) + if err != nil { + // TODO: surface deterministic expression errors to users by failing the job with a message. + log.Error("evaluateJobIf failed, job will stay blocked: job: %d, err: %v", id, err) + continue + } + if !shouldStartJob { + ret[id] = actions_model.StatusSkipped + continue + } + + // Expand a needs-dependent matrix now that its needs are done and the job is going to run. + wasDeferred := actionRunJob.IsMatrixDeferred + siblings, err := expandDeferredMatrix(ctx, actionRunJob, r.vars) + if err != nil { + // Aborting the pass is required: the placeholder is already claimed as the first + // combination, so committing here would drop the remaining ones for good. + return nil, fmt.Errorf("expand matrix of job %d: %w", id, err) + } + if actionRunJob.Status != actions_model.StatusBlocked { + // expandDeferredMatrix already persisted the failure, so it bypasses `ret`. + r.statuses[id] = actionRunJob.Status + r.matrixUpdatedJobs = append(r.matrixUpdatedJobs, actionRunJob) + r.matrixChanged = true + continue + } + if actionRunJob.IsMatrixDeferred { + continue // could not be expanded yet, it stays blocked and is retried on the next pass + } + r.matrixChanged = r.matrixChanged || len(siblings) > 0 + if wasDeferred { + // The `if:` above was decided against the raw matrix, so this row still has to be gated + // by its own combination like the siblings are on the next pass. + shouldStartJob, err := evaluateJobIf(ctx, actionRunJob.Run, nil, actionRunJob, r.vars, allSucceed) + if err != nil { + log.Error("evaluateJobIf failed after matrix expansion, job will stay blocked: job: %d, err: %v", id, err) + continue + } + if !shouldStartJob { + ret[id] = actions_model.StatusSkipped + 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) + err = updateConcurrencyEvaluationForJobWithNeeds(ctx, actionRunJob, r.vars) if err != nil { // The err can be caused by different cases: database error, or syntax error, or the needed jobs haven't completed // At the moment there is no way to distinguish them. @@ -470,22 +533,11 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model continue } - shouldStartJob, err := evaluateJobIf(ctx, actionRunJob.Run, nil, actionRunJob, r.vars, allSucceed) + newStatus, cancelledJobs, err := PrepareToStartJobWithConcurrency(ctx, actionRunJob) if err != nil { - // TODO: surface deterministic expression errors to users by failing the job with a message. - log.Error("evaluateJobIf failed, job will stay blocked: job: %d, err: %v", id, err) - continue - } - - newStatus := util.Iif(shouldStartJob, actions_model.StatusWaiting, actions_model.StatusSkipped) - if newStatus == actions_model.StatusWaiting { - var cancelledJobs []*actions_model.ActionRunJob - newStatus, cancelledJobs, err = PrepareToStartJobWithConcurrency(ctx, actionRunJob) - if err != nil { - log.Error("ShouldBlockJobByConcurrency failed, this job will stay blocked: job: %d, err: %v", id, err) - } else { - r.cancelledJobs = append(r.cancelledJobs, cancelledJobs...) - } + log.Error("ShouldBlockJobByConcurrency failed, this job will stay blocked: job: %d, err: %v", id, err) + } else { + r.cancelledJobs = append(r.cancelledJobs, cancelledJobs...) } if newStatus == actions_model.StatusWaiting && !slots.take(actionRunJob) { @@ -496,7 +548,7 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model ret[id] = newStatus } } - return ret + return ret, nil } func updateConcurrencyEvaluationForJobWithNeeds(ctx context.Context, actionRunJob *actions_model.ActionRunJob, vars map[string]string) error { diff --git a/services/actions/job_emitter_test.go b/services/actions/job_emitter_test.go index 4f3c65855e..f9930b6485 100644 --- a/services/actions/job_emitter_test.go +++ b/services/actions/job_emitter_test.go @@ -14,6 +14,7 @@ import ( user_model "gitea.dev/models/user" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func minimalWorkflowPayload(jobID string) []byte { @@ -257,7 +258,9 @@ jobs: } r := newJobStatusResolver(tt.jobs, nil) - assert.Equal(t, want, r.Resolve(ctx)) + got, err := r.Resolve(ctx) + require.NoError(t, err) + assert.Equal(t, want, got) }) } } @@ -277,7 +280,9 @@ func Test_maxParallelConverges(t *testing.T) { } for cycle := range totalJobs + 1 { - for id, status := range newJobStatusResolver(jobs, nil).Resolve(ctx) { + updates, err := newJobStatusResolver(jobs, nil).Resolve(ctx) + require.NoError(t, err) + for id, status := range updates { jobs[id-1].Status = status } counts := statusCounts(jobs) @@ -562,7 +567,8 @@ func Test_maxParallelReusableCallerLifecycle(t *testing.T) { } for cycle := range 2 * len(callers) { - promoted := newJobStatusResolver(callers, nil).Resolve(ctx) + promoted, err := newJobStatusResolver(callers, nil).Resolve(ctx) + require.NoError(t, err) for id, status := range promoted { caller := idToCaller[id] assert.False(t, caller.IsExpanded, "cycle %d: resolver re-promoted already-expanded caller %d", cycle, id) diff --git a/services/actions/matrix.go b/services/actions/matrix.go new file mode 100644 index 0000000000..86225a8b39 --- /dev/null +++ b/services/actions/matrix.go @@ -0,0 +1,226 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package actions + +import ( + "context" + "errors" + "fmt" + "slices" + + actions_model "gitea.dev/models/actions" + "gitea.dev/models/db" + "gitea.dev/modules/actions/jobparser" + "gitea.dev/modules/container" + "gitea.dev/modules/log" + "gitea.dev/modules/timeutil" + "gitea.dev/modules/util" + + "go.yaml.in/yaml/v4" + "xorm.io/builder" +) + +// expandDeferredMatrix expands a deferred-matrix placeholder once its needs are done and the job is +// known to run, using the needs' outputs: the placeholder becomes the first combination and the rest +// are returned as inserted siblings, all left Blocked so the caller's resolver still applies the +// concurrency gate. +// +// It runs inside the caller's transaction (job_emitter's resolver) and must not open a nested +// db.WithTx, which would reuse the ambient session and roll the whole emitter pass back on error. +// The three outcomes are reported through the job itself: +// - expanded: IsMatrixDeferred is cleared and the job stays StatusBlocked. +// - the workflow's fault (a matrix that cannot resolve, or one too large): a terminal status is +// persisted here, reported by the job leaving StatusBlocked. IsMatrixDeferred stays set, marking +// the payload as still unexpanded for a later rerun to re-derive. +// - not now: the job is left deferred and blocked for the next emitter pass to retry. This covers +// a transient failure before anything was written, and losing the claim to a concurrent pass. +// +// A returned error is reserved for a failure after the placeholder was claimed and must roll the +// caller's transaction back: committing it would drop the remaining combinations for good. +func expandDeferredMatrix(ctx context.Context, job *actions_model.ActionRunJob, vars map[string]string) ([]*actions_model.ActionRunJob, error) { + if !job.IsMatrixDeferred { + return nil, nil + } + + // failTerminal fails the job here rather than through the resolver's status map, which a + // reusable caller (a job may be both) would drop: its branch only handles waiting and skipped. + // The commit status is unaffected by the surviving flag: suppression only applies while the job + // is not done. + failTerminal := func(cause error) ([]*actions_model.ActionRunJob, error) { + log.Warn("Matrix expansion failed for job %d (JobID: %s): %v", job.ID, job.JobID, cause) + prevStatus, prevStopped := job.Status, job.Stopped + job.Status = actions_model.StatusFailure + job.Stopped = timeutil.TimeStampNow() + // The flag survives, so it alone no longer tells a fresh placeholder from one a concurrent + // pass already failed: the status has to be part of the condition. + affected, err := actions_model.UpdateRunJob(ctx, job, + builder.Eq{"is_matrix_deferred": true, "status": actions_model.StatusBlocked}, + "status", "stopped") + if err != nil { + return nil, fmt.Errorf("fail deferred matrix job %d: %w", job.ID, err) + } + if affected != 1 { + // A concurrent pass already advanced the row. Restore the in-memory state so this pass + // does not report a failure that was never persisted. + job.Status, job.Stopped = prevStatus, prevStopped + } + return nil, nil + } + + // retryLater leaves the placeholder untouched. It is only used before anything is written, so a + // transient failure neither fails the job nor aborts the emitter pass for the whole run. + retryLater := func(cause error) ([]*actions_model.ActionRunJob, error) { + log.Error("Matrix expansion of job %d (JobID: %s) postponed to the next pass: %v", job.ID, job.JobID, cause) + return nil, nil + } + + // The resolver only calls this once every need is done, as it does for job concurrency. + results, err := findJobNeedsAndFillJobResults(ctx, job) + if err != nil { + return retryLater(fmt.Errorf("find needs: %w", err)) + } + + if err := job.LoadAttributes(ctx); err != nil { + return retryLater(fmt.Errorf("load attributes: %w", err)) + } + + // The payload still carries the raw, unevaluated matrix: planning only erases the needs. + var baseSWF jobparser.SingleWorkflow + if err := yaml.Unmarshal(job.WorkflowPayload, &baseSWF); err != nil { + return failTerminal(fmt.Errorf("unmarshal payload: %w", err)) + } + _, parsedJob := baseSWF.Job() + if parsedJob == nil { + return failTerminal(errors.New("payload contains no job")) + } + + // `strategy` may reference the inputs context as well as needs, so resolve it like `if:` does. + inputs, err := getInputsForJob(ctx, job.Run, job) + if err != nil { + return retryLater(fmt.Errorf("get inputs: %w", err)) + } + + existingJobs, err := actions_model.CountRunJobsByRunAndAttemptID(ctx, job.RunID, job.RunAttemptID) + if err != nil { + return retryLater(fmt.Errorf("count jobs of attempt %d: %w", job.RunAttemptID, err)) + } + // The placeholder is reused as the first combination, so the attempt only grows by len-1. + maxCombinations := int(actions_model.MaxJobNumPerRun - existingJobs + 1) + + giteaCtx := GenerateGiteaContext(ctx, job.Run, nil, job) + expandedJobs, err := jobparser.ExpandMatrixWithNeeds(job.JobID, parsedJob, giteaCtx.ToGitHubContext(), results, vars, inputs, maxCombinations) + if err != nil { + return failTerminal(fmt.Errorf("expand matrix: %w", err)) + } + // Combinations differ only in what the matrix feeds: the name, the payload, and a + // runs-on/continue-on-error that may interpolate matrix.*. + applyCombo := func(dst *actions_model.ActionRunJob, combo *jobparser.Job) error { + swf := baseSWF + if err := swf.SetJob(job.JobID, combo.EraseNeeds()); err != nil { + return fmt.Errorf("set expanded job: %w", err) + } + payload, err := swf.Marshal() + if err != nil { + return fmt.Errorf("marshal expanded job: %w", err) + } + dst.Name = util.EllipsisDisplayString(combo.Name, 255) + dst.WorkflowPayload, dst.RunsOn = payload, combo.RunsOn() + dst.ContinueOnError = combo.GetContinueOnError() + return nil + } + + siblings := make([]*actions_model.ActionRunJob, 0, len(expandedJobs)-1) + for _, combo := range expandedJobs[1:] { + // Inherit from the placeholder rather than listing fields, so a sibling cannot silently lose + // one (scope, permissions, `uses:`) as the job model grows. + sibling := *job + sibling.ID, sibling.TaskID, sibling.SourceTaskID, sibling.AttemptJobID = 0, 0, 0, 0 + sibling.Started, sibling.Stopped, sibling.IsMatrixDeferred = 0, 0, false + sibling.Status = actions_model.StatusBlocked + sibling.Needs = slices.Clone(job.Needs) + // Only the placeholder keeps the raw payload, which is what identifies it as the group's anchor. + sibling.DeferredMatrixPayload = nil + if err := applyCombo(&sibling, combo); err != nil { + return failTerminal(err) + } + siblings = append(siblings, &sibling) + } + + // Keep AttemptJobIDs stable across attempts (best-effort). A single combination reuses the + // placeholder's row, so there is nothing to look up. + if len(siblings) > 0 { + parentAttemptJobID := int64(0) + if job.ParentJobID > 0 { + parent, err := actions_model.GetRunJobByRunAndID(ctx, job.RunID, job.ParentJobID) + if err != nil { + return retryLater(fmt.Errorf("load parent of job %d: %w", job.ID, err)) + } + parentAttemptJobID = parent.AttemptJobID + } + priorCombos, err := actions_model.GetPriorAttemptMatrixCombos(ctx, job.RunID, job.RunAttemptID, parentAttemptJobID, job.JobID) + if err != nil { + return retryLater(fmt.Errorf("lookup prior attempt combos of job %d: %w", job.ID, err)) + } + usedIDs := container.SetOf(job.AttemptJobID) + for _, sibling := range siblings { + if prior, ok := priorCombos[sibling.Name]; ok && usedIDs.Add(prior.AttemptJobID) { + sibling.AttemptJobID = prior.AttemptJobID + } + } + } + + // Reusing the placeholder leaves no phantom skipped job behind to poison downstream needs. The + // conditional update is an atomic claim: only the caller that flips IsMatrixDeferred inserts. + beforeClaim := *job + if err := applyCombo(job, expandedJobs[0]); err != nil { + return failTerminal(err) + } + job.IsMatrixDeferred = false + affected, err := actions_model.UpdateRunJob(ctx, job, + builder.Eq{"is_matrix_deferred": true, "status": actions_model.StatusBlocked}, + "name", "workflow_payload", "runs_on", "continue_on_error", "is_matrix_deferred") + if err != nil { + return nil, fmt.Errorf("claim placeholder of job %d: %w", job.ID, err) + } + if affected != 1 { + // A concurrent pass won the claim and owns the siblings. Restore the in-memory state so this + // pass leaves the job alone and picks the winner's rows up on the next one. + *job = beforeClaim + return nil, nil + } + + if len(siblings) == 0 { + return nil, nil + } + for _, sibling := range siblings { + if sibling.AttemptJobID != 0 { + continue // matched a prior attempt above + } + if sibling.AttemptJobID, err = actions_model.GetNextAttemptJobID(ctx, job.RunID); err != nil { + return nil, fmt.Errorf("alloc attempt_job_id for job %d: %w", job.ID, err) + } + } + if err := db.Insert(ctx, siblings); err != nil { + return nil, fmt.Errorf("insert matrix siblings of job %d: %w", job.ID, err) + } + return siblings, nil +} + +// restoreDeferredMatrixPlaceholder rewinds a rerun clone of a dynamic-matrix combination into the unexpanded placeholder it grew from +func restoreDeferredMatrixPlaceholder(clone *actions_model.ActionRunJob) error { + var swf jobparser.SingleWorkflow + if err := yaml.Unmarshal(clone.DeferredMatrixPayload, &swf); err != nil { + return fmt.Errorf("unmarshal deferred matrix payload: %w", err) + } + _, parsed := swf.Job() + if parsed == nil { + return errors.New("deferred matrix payload contains no job") + } + clone.Name = util.EllipsisDisplayString(parsed.Name, 255) + clone.WorkflowPayload = slices.Clone(clone.DeferredMatrixPayload) + clone.RunsOn = parsed.RunsOn() + clone.ContinueOnError = parsed.GetContinueOnError() + clone.IsMatrixDeferred = true + return nil +} diff --git a/services/actions/matrix_test.go b/services/actions/matrix_test.go new file mode 100644 index 0000000000..853a886a55 --- /dev/null +++ b/services/actions/matrix_test.go @@ -0,0 +1,220 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package actions + +import ( + "fmt" + "strings" + "testing" + + 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" +) + +// testRunIndex hands out the per-repo run index, which is unique per action_run row. +var testRunIndex int64 = 9100 + +// setupDeferredMatrixJob plants a completed `generate` job exposing outputs and the blocked `build` +// placeholder that depends on them, and returns the placeholder. Both are children of a reusable +// workflow caller, the case where a sibling losing ParentJobID would break needs resolution. +// jobIf is the `build` job's `if:` expression, omitted entirely when empty. +func setupDeferredMatrixJob(t *testing.T, matrixValue, jobIf string, outputs map[string]string) *actions_model.ActionRunJob { + t.Helper() + ctx := t.Context() + + ifLine := "" + if jobIf != "" { + ifLine = " if: " + jobIf + "\n" + } + // The `build` job takes its matrix from `generate`'s outputs, so Parse defers it. + workflows, err := jobparser.Parse(fmt.Appendf(nil, ` +on: push +jobs: + generate: + steps: [{run: echo}] + build: + needs: generate +%s strategy: + matrix: + value: %s + steps: [{run: echo}] +`, ifLine, matrixValue)) + require.NoError(t, err) + var placeholder *jobparser.SingleWorkflow + for _, workflow := range workflows { + if id, _ := workflow.Job(); id == "build" { + placeholder = workflow + } + } + require.NotNil(t, placeholder, "the build job must be planned as a single deferred placeholder") + id, job := placeholder.Job() + require.NoError(t, placeholder.SetJob(id, job.EraseNeeds())) + payload, err := placeholder.Marshal() + require.NoError(t, err) + + testRunIndex++ + run := &actions_model.ActionRun{ + RepoID: 4, OwnerID: 1, TriggerUserID: 1, Index: testRunIndex, + WorkflowID: "dynamic.yml", Ref: "refs/heads/main", Status: actions_model.StatusRunning, + } + require.NoError(t, db.Insert(ctx, run)) + attempt := &actions_model.ActionRunAttempt{RepoID: 4, RunID: run.ID, Attempt: 1, Status: actions_model.StatusRunning} + require.NoError(t, db.Insert(ctx, attempt)) + run.LatestAttemptID = attempt.ID + require.NoError(t, actions_model.UpdateRun(ctx, run, "latest_attempt_id")) + + // Needs outputs are read straight from action_task_output, so no action_task row is required; + // the run's id doubles as a task id that is unique across subtests. + taskID := run.ID + for key, value := range outputs { + require.NoError(t, db.Insert(ctx, &actions_model.ActionTaskOutput{TaskID: taskID, OutputKey: key, OutputValue: value})) + } + + newJob := func(jobID string) *actions_model.ActionRunJob { + attemptJobID, err := actions_model.GetNextAttemptJobID(ctx, run.ID) + require.NoError(t, err) + return &actions_model.ActionRunJob{ + RunID: run.ID, RunAttemptID: attempt.ID, AttemptJobID: attemptJobID, RepoID: 4, OwnerID: 1, + JobID: jobID, Name: jobID, Status: actions_model.StatusRunning, + } + } + caller := newJob("call") + caller.IsReusableCaller, caller.IsExpanded = true, true + require.NoError(t, db.Insert(ctx, caller)) + + generate := newJob("generate") + generate.ParentJobID, generate.Status, generate.TaskID = caller.ID, actions_model.StatusSuccess, taskID + require.NoError(t, db.Insert(ctx, generate)) + + build := newJob("build") + build.ParentJobID, build.Status = caller.ID, actions_model.StatusBlocked + build.Needs, build.WorkflowPayload, build.IsMatrixDeferred = []string{"generate"}, payload, true + build.DeferredMatrixPayload = payload + // Values a sibling must inherit rather than silently reset. + build.WorkflowSourceRepoID, build.WorkflowSourceCommitSHA = 42, "abc123" + require.NoError(t, db.Insert(ctx, build)) + return build +} + +func TestExpandDeferredMatrix(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + + t.Run("expands into siblings", func(t *testing.T) { + job := setupDeferredMatrixJob(t, "${{ fromJson(needs.generate.outputs.values) }}", "", map[string]string{"values": `["a","b","c"]`}) + + siblings, err := expandDeferredMatrix(t.Context(), job, nil) + require.NoError(t, err) + require.Len(t, siblings, 2) + + // The placeholder is reused as the first combination and stays blocked for the `if:` gate. + assert.Equal(t, "build (a)", job.Name) + assert.False(t, job.IsMatrixDeferred) + assert.Equal(t, actions_model.StatusBlocked, job.Status) + + names := []string{job.Name} + for _, sibling := range siblings { + names = append(names, sibling.Name) + assert.Equal(t, actions_model.StatusBlocked, sibling.Status) + assert.False(t, sibling.IsMatrixDeferred, "a sibling must not be expanded again") + assert.Equal(t, []string{"generate"}, sibling.Needs) + assert.Equal(t, job.ParentJobID, sibling.ParentJobID, "needs resolution is scoped by ParentJobID") + assert.Equal(t, int64(42), sibling.WorkflowSourceRepoID) + assert.Equal(t, "abc123", sibling.WorkflowSourceCommitSHA) + assert.Greater(t, sibling.AttemptJobID, job.AttemptJobID, "siblings take fresh ids from the run-wide counter") + assert.Empty(t, sibling.DeferredMatrixPayload, "only the placeholder anchors the group with the raw payload") + } + assert.ElementsMatch(t, []string{"build (a)", "build (b)", "build (c)"}, names) + + reloaded := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: job.ID}) + assert.Equal(t, "build (a)", reloaded.Name) + assert.False(t, reloaded.IsMatrixDeferred) + assert.NotEmpty(t, reloaded.DeferredMatrixPayload, "the claim must not erase the raw payload") + }) + + // A matrix that can never produce runnable combinations fails the job instead of rolling the + // emitter pass back, so it is not retried forever. + for _, tt := range []struct { + name, matrixValue string + outputs map[string]string + }{ + {name: "unresolvable matrix", matrixValue: "${{ fromJson(needs.generate.outputs.missing) }}"}, + { + name: "expansion exceeding the per-attempt job limit", + matrixValue: "${{ fromJson(needs.generate.outputs.many) }}", + // One combination more than the attempt has room for: the cap is MaxJobNumPerRun less + // the rows the setup plants, plus the placeholder the first combination reuses. + outputs: map[string]string{"many": "[" + strings.Repeat("0,", actions_model.MaxJobNumPerRun-2) + "0]"}, + }, + } { + t.Run(tt.name+" fails the job", func(t *testing.T) { + outputs := tt.outputs + if outputs == nil { + outputs = map[string]string{"values": `["a","b","c"]`} + } + job := setupDeferredMatrixJob(t, tt.matrixValue, "", outputs) + + siblings, err := expandDeferredMatrix(t.Context(), job, nil) + require.NoError(t, err) + assert.Empty(t, siblings) + assert.Equal(t, actions_model.StatusFailure, job.Status) + assert.True(t, job.IsMatrixDeferred, "the flag survives, marking the payload as still unexpanded for reruns") + assert.NotZero(t, job.Stopped) + // A rejected expansion must not leave the first combination's name behind. + assert.Equal(t, "build", unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: job.ID}).Name) + }) + } +} + +// TestDeferredMatrixResolverGating covers the resolver deciding a placeholder's fate around the +// expansion. A need that did not succeed leaves no outputs to build the matrix from, so the job is +// skipped like any other job with such a need rather than failed over a matrix it never had to +// evaluate, and no combination is inserted. A job that does run is then gated by its own +// combination: the placeholder reused as the first one is judged by `matrix.*`, not by the raw +// expression the `if:` would have seen before expansion. +func TestDeferredMatrixResolverGating(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + + for _, tt := range []struct { + name string + needStatus actions_model.Status + jobIf string + outputs map[string]string + wantBuilds []string + }{ + {name: "failed need", needStatus: actions_model.StatusFailure, wantBuilds: []string{"build"}}, + {name: "skipped need", needStatus: actions_model.StatusSkipped, wantBuilds: []string{"build"}}, + { + name: "`if:` gated per combination", needStatus: actions_model.StatusSuccess, + jobIf: "${{ matrix.value != 'a' }}", outputs: map[string]string{"values": `["a","b"]`}, + wantBuilds: []string{"build (a)", "build (b)"}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + ctx := t.Context() + job := setupDeferredMatrixJob(t, "${{ fromJson(needs.generate.outputs.values) }}", tt.jobIf, tt.outputs) + _, err := db.Exec(ctx, "UPDATE `action_run_job` SET status = ? WHERE run_id = ? AND job_id = ?", int(tt.needStatus), job.RunID, "generate") + require.NoError(t, err) + + jobs := runJobs(t, job.RunID, job.RunAttemptID) + require.NoError(t, jobs.LoadRuns(ctx, false)) + + updates, err := newJobStatusResolver(jobs, nil).Resolve(ctx) + require.NoError(t, err) + assert.Equal(t, actions_model.StatusSkipped, updates[job.ID]) + + var names []string + for _, runJob := range runJobs(t, job.RunID, job.RunAttemptID) { + if runJob.JobID == "build" { + names = append(names, runJob.Name) + } + } + assert.ElementsMatch(t, tt.wantBuilds, names) + }) + } +} diff --git a/services/actions/max_parallel_test.go b/services/actions/max_parallel_test.go index d63e40da86..c7c45e4508 100644 --- a/services/actions/max_parallel_test.go +++ b/services/actions/max_parallel_test.go @@ -188,7 +188,9 @@ func Test_jobStatusResolver_MaxParallelStarvedSkipsConcurrency(t *testing.T) { for _, job := range jobs { job.Run = run } - assert.Empty(t, newJobStatusResolver(jobs, nil).Resolve(t.Context()), "the starved job must stay blocked") + updates, err := newJobStatusResolver(jobs, nil).Resolve(t.Context()) + assert.NoError(t, err) + assert.Empty(t, updates, "the starved job must stay blocked") holder = unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: holder.ID}) assert.Equal(t, actions_model.StatusRunning, holder.Status) diff --git a/services/actions/rerun.go b/services/actions/rerun.go index 665d63a8b0..7aee11ba89 100644 --- a/services/actions/rerun.go +++ b/services/actions/rerun.go @@ -136,6 +136,12 @@ type rerunPlan struct { // skipCloneTemplateJobIDs holds the template-attempt DB row IDs of descendants of any reusable caller in rerunAttemptJobIDs. // These jobs should not be cloned, since the caller's lazy expansion will re-insert them fresh. skipCloneTemplateJobIDs container.Set[int64] + + // matrixPlaceholderTemplateIDs holds, per dynamic-matrix job whose matrix must be re-derived in the new attempt, + // the template DB row ID to clone as a restored unexpanded placeholder. + // The group's remaining combination rows are in matrixSiblingSkipTemplateIDs and are not cloned: they'll be re-expanded. + matrixPlaceholderTemplateIDs container.Set[int64] + matrixSiblingSkipTemplateIDs container.Set[int64] } // buildRerunPlan constructs a rerunPlan for the given workflow run without writing to the database. @@ -174,6 +180,7 @@ func buildRerunPlan(ctx context.Context, run *actions_model.ActionRun, triggerUs return nil, err } plan.skipCloneTemplateJobIDs = plan.collectResetCallerDescendants() + plan.collectMatrixCollapse() return plan, nil } @@ -247,9 +254,19 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR if plan.skipCloneTemplateJobIDs.Contains(templateJob.ID) { continue } + // siblings of a collapsed dynamic-matrix job are not cloned either: re-expansion re-inserts them + if plan.matrixSiblingSkipTemplateIDs.Contains(templateJob.ID) { + continue + } newJob := cloneRunJobForAttempt(templateJob, newAttempt) + if plan.matrixPlaceholderTemplateIDs.Contains(templateJob.ID) { + if err := restoreDeferredMatrixPlaceholder(newJob); err != nil { + return fmt.Errorf("restore matrix placeholder from job %d: %w", templateJob.ID, err) + } + } + // Remap ParentJobID from template attempts's DB ID -> new attempt's DB ID. if templateJob.ParentJobID != 0 { newParentID, ok := templateIDToNewID[templateJob.ParentJobID] @@ -261,7 +278,9 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR } if plan.rerunAttemptJobIDs.Contains(templateJob.AttemptJobID) { - shouldBlockJob := shouldBlock || plan.hasRerunDependency(templateJob) + // 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 newJob.Status = util.Iif(shouldBlockJob, actions_model.StatusBlocked, actions_model.StatusWaiting) newJob.TaskID = 0 @@ -360,8 +379,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: let job_emitter resolve its child jobs - if hasWaitingCallerJobs { + // 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) } @@ -519,6 +539,8 @@ func cloneRunJobForAttempt(templateJob *actions_model.ActionRunJob, attempt *act Needs: slices.Clone(templateJob.Needs), RunsOn: slices.Clone(templateJob.RunsOn), ContinueOnError: templateJob.ContinueOnError, + IsMatrixDeferred: templateJob.IsMatrixDeferred, + DeferredMatrixPayload: slices.Clone(templateJob.DeferredMatrixPayload), Status: templateJob.Status, RawConcurrency: templateJob.RawConcurrency, IsConcurrencyEvaluated: templateJob.IsConcurrencyEvaluated, @@ -600,3 +622,52 @@ func createOriginalAttemptForLegacyRun(ctx context.Context, run *actions_model.A return actions_model.UpdateRun(ctx, run, "latest_attempt_id") }) } + +// collectMatrixCollapse decides, per dynamic-matrix job in the rerun set, whether the new +// attempt must re-derive the matrix instead of reusing the previous attempt's combinations, +// and fills matrixPlaceholderTemplateIDs / matrixSiblingSkipTemplateIDs accordingly. +func (p *rerunPlan) collectMatrixCollapse() { + p.matrixPlaceholderTemplateIDs = make(container.Set[int64]) + p.matrixSiblingSkipTemplateIDs = make(container.Set[int64]) + + // Group every template row by the key the rows of one matrix job share. + type groupKey struct { + parentJobID int64 + jobID string + } + groups := make(map[groupKey][]*actions_model.ActionRunJob) + for _, tj := range p.templateJobs { + key := groupKey{tj.ParentJobID, tj.JobID} + groups[key] = append(groups[key], tj) + } + + for _, rows := range groups { + anchorIdx := slices.IndexFunc(rows, func(j *actions_model.ActionRunJob) bool { + return len(j.DeferredMatrixPayload) > 0 + }) + if anchorIdx < 0 { + continue // not a dynamic-matrix job: a plain job, or a matrix expanded at plan time + } + anchor := rows[anchorIdx] + // Descendants of a reset caller are not cloned at all; the caller's expansion re-inserts the job. + if p.skipCloneTemplateJobIDs.Contains(anchor.ID) { + continue + } + inRerunSet := slices.ContainsFunc(rows, func(j *actions_model.ActionRunJob) bool { + return p.rerunAttemptJobIDs.Contains(j.AttemptJobID) + }) + if !inRerunSet { + continue // pass-through group, cloned as-is + } + unexpanded := len(rows) == 1 && anchor.IsMatrixDeferred + if !unexpanded && !p.hasRerunDependency(anchor) { + continue // needs keep their outputs, reuse the combinations + } + p.matrixPlaceholderTemplateIDs.Add(anchor.ID) + for _, row := range rows { + if row.ID != anchor.ID { + p.matrixSiblingSkipTemplateIDs.Add(row.ID) + } + } + } +} diff --git a/services/actions/reusable_workflow.go b/services/actions/reusable_workflow.go index c178fb2189..15e101d9e5 100644 --- a/services/actions/reusable_workflow.go +++ b/services/actions/reusable_workflow.go @@ -318,6 +318,7 @@ func insertCallerChildren(ctx context.Context, run *actions_model.ActionRun, att continue } needs := parsedChild.Needs() + isMatrixDeferred := jobparser.HasDeferredMatrix(parsedChild) if err := sw.SetJob(jobID, parsedChild.EraseNeeds()); err != nil { return err } @@ -328,11 +329,10 @@ func insertCallerChildren(ctx context.Context, run *actions_model.ActionRun, att parsedChild.Name = util.EllipsisDisplayString(parsedChild.Name, 255) - // AttemptJobID: prefer a prior-attempt match by (JobID, Name) and fall back to a fresh allocator value for newly-appearing logical jobs. - // The two-level key disambiguates matrix instances (same JobID, different Names) and distinct jobs that legally share the same Name (different JobIDs). + // AttemptJobID: prefer a prior-attempt match and fall back to a fresh allocator value for newly-appearing logical jobs. var attemptJobID int64 - if priorChild, ok := priorChildren[jobID][parsedChild.Name]; ok { - attemptJobID = priorChild.AttemptJobID + if priorID, ok := priorAttemptJobID(priorChildren[jobID], parsedChild.Name, isMatrixDeferred); ok { + attemptJobID = priorID } else { attemptJobID, err = actions_model.GetNextAttemptJobID(ctx, run.ID) if err != nil { @@ -359,6 +359,11 @@ func insertCallerChildren(ctx context.Context, run *actions_model.ActionRun, att ParentJobID: caller.ID, WorkflowSourceRepoID: sourceRepoID, WorkflowSourceCommitSHA: sourceCommitSHA, + IsMatrixDeferred: isMatrixDeferred, + } + if isMatrixDeferred { + // Expansion overwrites WorkflowPayload; keep the raw payload so a rerun can re-derive the matrix. + child.DeferredMatrixPayload = payload } if perms := ExtractJobPermissionsFromWorkflow(sw, parsedChild); perms != nil { child.TokenPermissions = perms @@ -411,3 +416,21 @@ func undoExpansion(ctx context.Context, caller *actions_model.ActionRunJob) erro } return nil } + +// priorAttemptJobID returns the AttemptJobID a re-inserted child should reuse, +// given the prior attempt's rows of the same JobID indexed by Name. +func priorAttemptJobID(priorSameJobID map[string]*actions_model.ActionRunJob, name string, isMatrixDeferred bool) (int64, bool) { + if isMatrixDeferred { + for _, prior := range priorSameJobID { + if len(prior.DeferredMatrixPayload) > 0 { + return prior.AttemptJobID, true + } + } + return 0, false + } + prior, ok := priorSameJobID[name] + if !ok { + return 0, false + } + return prior.AttemptJobID, true +} diff --git a/services/actions/run.go b/services/actions/run.go index 68c2e7fdb5..9055391364 100644 --- a/services/actions/run.go +++ b/services/actions/run.go @@ -186,6 +186,7 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte 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() + isMatrixDeferred := jobparser.HasDeferredMatrix(job) if err := workflowJob.SetJob(id, job.EraseNeeds()); err != nil { return nil, nil, false, err } @@ -218,8 +219,13 @@ func insertRunJob(ctx context.Context, run *actions_model.ActionRun, runAttempt WorkflowSourceRepoID: run.WorkflowRepoID, WorkflowSourceCommitSHA: run.WorkflowCommitSHA, ContinueOnError: job.GetContinueOnError(), + IsMatrixDeferred: isMatrixDeferred, MaxParallel: parseMaxParallel(id, job.Strategy.MaxParallelString), } + if isMatrixDeferred { + // Expansion overwrites WorkflowPayload; keep the raw payload so a rerun can re-derive the matrix. + runJob.DeferredMatrixPayload = payload + } // Parse workflow/job permissions (no clamping here) if perms := ExtractJobPermissionsFromWorkflow(workflowJob, job); perms != nil { runJob.TokenPermissions = perms diff --git a/tests/integration/actions_dynamic_matrix_test.go b/tests/integration/actions_dynamic_matrix_test.go new file mode 100644 index 0000000000..7c3b6341c4 --- /dev/null +++ b/tests/integration/actions_dynamic_matrix_test.go @@ -0,0 +1,165 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package integration + +import ( + "fmt" + "maps" + "net/http" + "net/url" + "strings" + "testing" + "time" + + runnerv1 "gitea.dev/actions-proto-go/runner/v1" + actions_model "gitea.dev/models/actions" + auth_model "gitea.dev/models/auth" + "gitea.dev/models/unittest" + user_model "gitea.dev/models/user" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestDynamicMatrixEvaluation covers a job whose matrix references ${{ needs.*.outputs.* }}: it is +// planned as a single placeholder and expanded once its dependency completes. `build` exercises the +// expansion, `report` that a downstream job sees the combinations' outputs, `gated` and `partial` the +// `if:` gates on either side of it, and `static` that a matrix expanded at plan time is left alone. +// A full rerun then re-derives the matrix from the new attempt's outputs instead of reusing the +// previous combinations, keeping the AttemptJobID of every combination that recurs. +func TestDynamicMatrixEvaluation(t *testing.T) { + onGiteaRun(t, func(t *testing.T, u *url.URL) { + user2 := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2}) + session := loginUser(t, user2.Name) + token := getTokenForLoggedInUser(t, session, auth_model.AccessTokenScopeWriteRepository, auth_model.AccessTokenScopeWriteUser) + + apiRepo := createActionsTestRepo(t, token, "actions-dynamic-matrix-eval", false) + runner := newMockRunner() + runner.registerAsRepoRunner(t, user2.Name, apiRepo.Name, "mock-runner", []string{"ubuntu-latest"}, false) + + const workflow = `name: dynamic-matrix +on: + push: + paths: ['.gitea/workflows/dynamic-matrix.yml'] +jobs: + generate: + runs-on: ubuntu-latest + outputs: + matrix: ${{ steps.set.outputs.matrix }} + steps: + - id: set + run: echo "matrix=[1,2]" >> "$GITHUB_OUTPUT" + build: + needs: [generate] + runs-on: ubuntu-latest + outputs: + result: ${{ steps.out.outputs.result }} + strategy: + matrix: + value: ${{ fromJson(needs.generate.outputs.matrix) }} + steps: + - id: out + run: echo "result=built-${{ matrix.value }}" >> "$GITHUB_OUTPUT" + static: + needs: [generate] + runs-on: ubuntu-latest + strategy: + matrix: + os: [a, b] + steps: + - run: echo "${{ matrix.os }}" + gated: + needs: [generate] + if: false + runs-on: ubuntu-latest + strategy: + matrix: + value: ${{ fromJson(needs.generate.outputs.matrix) }} + steps: + - run: echo "${{ matrix.value }}" + partial: + needs: [generate] + if: ${{ matrix.value != 1 }} + runs-on: ubuntu-latest + strategy: + matrix: + value: ${{ fromJson(needs.generate.outputs.matrix) }} + steps: + - run: echo "${{ matrix.value }}" + report: + needs: [build] + runs-on: ubuntu-latest + steps: + - run: echo '${{ toJSON(needs.build.outputs) }}' +` + opts := getWorkflowCreateFileOptions(user2, apiRepo.DefaultBranch, "create dynamic-matrix.yml", workflow) + createWorkflowFile(t, token, user2.Name, apiRepo.Name, ".gitea/workflows/dynamic-matrix.yml", opts) + + generateTask := runner.fetchTask(t, 10*time.Second) + require.Equal(t, "generate", getTaskJobNameByTaskID(t, token, user2.Name, apiRepo.Name, generateTask.Id)) + _, _, run := getTaskAndJobAndRunByTaskID(t, generateTask.Id) + runner.execTask(t, generateTask, &mockTaskOutcome{ + result: runnerv1.Result_RESULT_SUCCESS, + outputs: map[string]string{"matrix": "[1,2]"}, + }) + + // execAttempt runs the attempt's remaining tasks, recording each job's name and AttemptJobID(order is not fixed). + attemptJobIDs := map[string]int64{} + var reportTask *runnerv1.Task + execAttempt := func(t *testing.T, count int) []string { + names := make([]string, 0, count) + for range count { + task := runner.fetchTask(t, 10*time.Second) + _, taskJob, _ := getTaskAndJobAndRunByTaskID(t, task.Id) + names = append(names, taskJob.Name) + attemptJobIDs[taskJob.Name] = taskJob.AttemptJobID + outcome := &mockTaskOutcome{result: runnerv1.Result_RESULT_SUCCESS} + if value, ok := strings.CutPrefix(taskJob.Name, "build ("); ok { + outcome.outputs = map[string]string{"result": "built-" + strings.TrimSuffix(value, ")")} + } + if taskJob.JobID == "report" { + reportTask = task + } + runner.execTask(t, task, outcome) + } + runner.fetchNoTask(t, 300*time.Millisecond) + return names + } + + seen := execAttempt(t, 6) + firstAttemptIDs := maps.Clone(attemptJobIDs) + // `gated` is decided before the matrix is touched, so it never expands and is skipped as one job； + // `partial (1)` is decided afterwards, against its own combination. + assert.ElementsMatch(t, []string{"build (1)", "build (2)", "static (a)", "static (b)", "partial (2)", "report"}, seen) + skipped := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: run.ID, Name: "partial (1)"}) + assert.Equal(t, actions_model.StatusSkipped, skipped.Status) + + buildNeed, ok := reportTask.Needs["build"] + require.True(t, ok, "report must see the expanded combinations as its 'build' need") + assert.Equal(t, runnerv1.Result_RESULT_SUCCESS, buildNeed.Result) + assert.Contains(t, []string{"built-1", "built-2"}, buildNeed.Outputs["result"]) + + // Re-running the whole run collapses the combinations back into a placeholder and re-derives + // the matrix from the new attempt's outputs instead of reusing the previous combinations. + req := NewRequest(t, "POST", fmt.Sprintf("/%s/%s/actions/runs/%d/rerun", user2.Name, apiRepo.Name, run.ID)) + session.MakeRequest(t, req, http.StatusOK) + + generateTask2 := runner.fetchTask(t, 10*time.Second) + require.Equal(t, "generate", getTaskJobNameByTaskID(t, token, user2.Name, apiRepo.Name, generateTask2.Id)) + assert.Equal(t, "2", generateTask2.Context.GetFields()["run_attempt"].GetStringValue()) + runner.execTask(t, generateTask2, &mockTaskOutcome{ + result: runnerv1.Result_RESULT_SUCCESS, + outputs: map[string]string{"matrix": "[1,2,3]"}, + }) + + // The matrix now yields a third value, while `static` is cloned as-is rather than collapsed. + seenRerun := execAttempt(t, 8) + assert.ElementsMatch(t, []string{ + "build (1)", "build (2)", "build (3)", "static (a)", "static (b)", "partial (2)", "partial (3)", "report", + }, seenRerun) + for name, firstID := range firstAttemptIDs { + assert.Equal(t, firstID, attemptJobIDs[name], "%s keeps its AttemptJobID across attempts", name) + } + }) +}