mirror of
https://github.com/go-gitea/gitea.git
synced 2026-08-01 02:30:22 +00:00
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 <pascal.zimmermann@theiotstudio.com>
Signed-off-by: ZPascal <pascal.zimmermann@theiotstudio.com>
Co-authored-by: Claude <claude-sonnet-4-5@anthropic.com>
Co-authored-by: silverwind <me@silverwind.io>
Co-authored-by: Claude (Opus 4.8) <noreply@anthropic.com>
Co-authored-by: bircni <bircni@icloud.com>
Co-authored-by: Zettat123 <zettat123@gmail.com>
This commit is contained in:
Pascal Zimmermann
committed by
GitHub
GitHub
parent
717db275d5
commit
5672b1c4cf
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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])
|
||||
|
||||
@@ -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")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user