feat: support evaluation of concurrency clauses in runner (#827)

Pre-req for support `concurrency` clauses for Gitea Actions, later added to Gitea in PR: https://github.com/go-gitea/gitea/pull/32751. Unit tests added in this PR relative to upstream.

Squashes PRs https://gitea.com/gitea/act/pulls/124 & https://gitea.com/gitea/act/pulls/139, as noted in https://code.forgejo.org/forgejo/runner/issues/678

Reviewed-on: https://gitea.com/gitea/act/pulls/124
Reviewed-by: Lunny Xiao <lunny@noreply.gitea.com>
Co-authored-by: Zettat123 <zettat123@gmail.com>
Co-committed-by: Zettat123 <zettat123@gmail.com>
Reviewed-on: https://gitea.com/gitea/act/pulls/139
Reviewed-by: Zettat123 <zettat123@noreply.gitea.com>
Reviewed-by: Lunny Xiao <xiaolunwen@gmail.com>
Co-authored-by: ChristopherHX <christopher.homberger@web.de>
Co-committed-by: ChristopherHX <christopher.homberger@web.de>

<!--start release-notes-assistant-->
<!--URL:https://code.forgejo.org/forgejo/runner-->
- features
  - [PR](https://code.forgejo.org/forgejo/runner/pulls/827): <!--number 827 --><!--line 0 --><!--description ZmVhdDogc3VwcG9ydCBldmFsdWF0aW9uIG9mIGNvbmN1cnJlbmN5IGNsYXVzZXMgaW4gcnVubmVy-->feat: support evaluation of concurrency clauses in runner<!--description-->
<!--end release-notes-assistant-->

Reviewed-on: https://code.forgejo.org/forgejo/runner/pulls/827
Reviewed-by: earl-warren <earl-warren@noreply.code.forgejo.org>
Co-authored-by: Mathieu Fenniak <mathieu@fenniak.net>
Co-committed-by: Mathieu Fenniak <mathieu@fenniak.net>

act/jobparser/interpeter.go

@@ -16,6 +16,7 @@ func NewInterpeter(
 	gitCtx *model.GithubContext,
 	results map[string]*JobResult,
 	vars map[string]string,
+	inputs map[string]interface{},
 ) exprparser.Interpreter {
 	strategy := make(map[string]interface{})
 	if job.Strategy != nil {
@@ -62,7 +63,7 @@ func NewInterpeter(
 		Strategy: strategy,
 		Matrix:   matrix,
 		Needs:    using,
-		Inputs:   nil, // not supported yet
+		Inputs:   inputs,
 		Vars:     vars,
 	}
 

act/jobparser/jobparser.go

@@ -48,7 +48,7 @@ func Parse(content []byte, validate bool, options ...ParseOption) ([]*SingleWork
 		}
 		for _, matrix := range matricxes {
 			job := job.Clone()
-			evaluator := NewExpressionEvaluator(NewInterpeter(id, origin.GetJob(id), matrix, pc.gitContext, results, pc.vars))
+			evaluator := NewExpressionEvaluator(NewInterpeter(id, origin.GetJob(id), matrix, pc.gitContext, results, pc.vars, nil))
 			if job.Name == "" {
 				job.Name = nameWithMatrix(id, matrix)
 			} else {

act/jobparser/jobparser_test.go

@@ -42,6 +42,16 @@ func TestParse(t *testing.T) {
 			options: nil,
 			wantErr: false,
 		},
+		{
+			name:    "job_concurrency",
+			options: nil,
+			wantErr: false,
+		},
+		{
+			name:    "job_concurrency_eval",
+			options: nil,
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

act/jobparser/model.go

@@ -1,6 +1,7 @@
 package jobparser
 
 import (
+	"bytes"
 	"fmt"
 
 	"code.forgejo.org/forgejo/runner/v9/act/model"
@@ -82,6 +83,7 @@ type Job struct {
 	Uses           string                    `yaml:"uses,omitempty"`
 	With           map[string]interface{}    `yaml:"with,omitempty"`
 	RawSecrets     yaml.Node                 `yaml:"secrets,omitempty"`
+	RawConcurrency *model.RawConcurrency     `yaml:"concurrency,omitempty"`
 }
 
 func (j *Job) Clone() *Job {
@@ -104,6 +106,7 @@ func (j *Job) Clone() *Job {
 		Uses:           j.Uses,
 		With:           j.With,
 		RawSecrets:     j.RawSecrets,
+		RawConcurrency: j.RawConcurrency,
 	}
 }
 
@@ -190,6 +193,85 @@ func (evt *Event) Schedules() []map[string]string {
 	return evt.schedules
 }
 
+func ReadWorkflowRawConcurrency(content []byte) (*model.RawConcurrency, error) {
+	w := new(model.Workflow)
+	err := yaml.NewDecoder(bytes.NewReader(content)).Decode(w)
+	return w.RawConcurrency, err
+}
+
+func EvaluateConcurrency(rc *model.RawConcurrency, jobID string, job *Job, gitCtx map[string]any, results map[string]*JobResult, vars map[string]string, inputs map[string]any) (string, bool, error) {
+	actJob := &model.Job{}
+	if job != nil {
+		actJob.Strategy = &model.Strategy{
+			FailFastString:    job.Strategy.FailFastString,
+			MaxParallelString: job.Strategy.MaxParallelString,
+			RawMatrix:         job.Strategy.RawMatrix,
+		}
+		actJob.Strategy.FailFast = actJob.Strategy.GetFailFast()
+		actJob.Strategy.MaxParallel = actJob.Strategy.GetMaxParallel()
+	}
+
+	matrix := make(map[string]any)
+	matrixes, err := actJob.GetMatrixes()
+	if err != nil {
+		return "", false, err
+	}
+	if len(matrixes) > 0 {
+		matrix = matrixes[0]
+	}
+
+	evaluator := NewExpressionEvaluator(NewInterpeter(jobID, actJob, matrix, toGitContext(gitCtx), results, vars, inputs))
+	var node yaml.Node
+	if err := node.Encode(rc); err != nil {
+		return "", false, fmt.Errorf("failed to encode concurrency: %w", err)
+	}
+	if err := evaluator.EvaluateYamlNode(&node); err != nil {
+		return "", false, fmt.Errorf("failed to evaluate concurrency: %w", err)
+	}
+	var evaluated model.RawConcurrency
+	if err := node.Decode(&evaluated); err != nil {
+		return "", false, fmt.Errorf("failed to unmarshal evaluated concurrency: %w", err)
+	}
+	if evaluated.RawExpression != "" {
+		return evaluated.RawExpression, false, nil
+	}
+	return evaluated.Group, evaluated.CancelInProgress == "true", nil
+}
+
+func toGitContext(input map[string]any) *model.GithubContext {
+	gitContext := &model.GithubContext{
+		EventPath:        asString(input["event_path"]),
+		Workflow:         asString(input["workflow"]),
+		RunID:            asString(input["run_id"]),
+		RunNumber:        asString(input["run_number"]),
+		Actor:            asString(input["actor"]),
+		Repository:       asString(input["repository"]),
+		EventName:        asString(input["event_name"]),
+		Sha:              asString(input["sha"]),
+		Ref:              asString(input["ref"]),
+		RefName:          asString(input["ref_name"]),
+		RefType:          asString(input["ref_type"]),
+		HeadRef:          asString(input["head_ref"]),
+		BaseRef:          asString(input["base_ref"]),
+		Token:            asString(input["token"]),
+		Workspace:        asString(input["workspace"]),
+		Action:           asString(input["action"]),
+		ActionPath:       asString(input["action_path"]),
+		ActionRef:        asString(input["action_ref"]),
+		ActionRepository: asString(input["action_repository"]),
+		Job:              asString(input["job"]),
+		RepositoryOwner:  asString(input["repository_owner"]),
+		RetentionDays:    asString(input["retention_days"]),
+	}
+
+	event, ok := input["event"].(map[string]any)
+	if ok {
+		gitContext.Event = event
+	}
+
+	return gitContext
+}
+
 func ParseRawOn(rawOn *yaml.Node) ([]*Event, error) {
 	switch rawOn.Kind {
 	case yaml.ScalarNode:
@@ -348,3 +430,12 @@ func parseMappingNode[T any](node *yaml.Node) ([]string, []T, error) {
 
 	return scalars, datas, nil
 }
+
+func asString(v interface{}) string {
+	if v == nil {
+		return ""
+	} else if s, ok := v.(string); ok {
+		return s
+	}
+	return ""
+}

act/jobparser/model_test.go

@@ -339,3 +339,118 @@ func TestParseMappingNode(t *testing.T) {
 		})
 	}
 }
+
+func TestEvaluateConcurrency(t *testing.T) {
+	tests := []struct {
+		name             string
+		input            model.RawConcurrency
+		group            string
+		cancelInProgress bool
+	}{
+		{
+			name: "basic",
+			input: model.RawConcurrency{
+				Group:            "group-name",
+				CancelInProgress: "true",
+			},
+			group:            "group-name",
+			cancelInProgress: true,
+		},
+		{
+			name:             "undefined",
+			input:            model.RawConcurrency{},
+			group:            "",
+			cancelInProgress: false,
+		},
+		{
+			name: "group-evaluation",
+			input: model.RawConcurrency{
+				Group: "${{ github.workflow }}-${{ github.ref }}",
+			},
+			group:            "test_workflow-main",
+			cancelInProgress: false,
+		},
+		{
+			name: "cancel-evaluation-true",
+			input: model.RawConcurrency{
+				Group:            "group-name",
+				CancelInProgress: "${{ !contains(github.ref, 'release/')}}",
+			},
+			group:            "group-name",
+			cancelInProgress: true,
+		},
+		{
+			name: "cancel-evaluation-false",
+			input: model.RawConcurrency{
+				Group:            "group-name",
+				CancelInProgress: "${{ contains(github.ref, 'release/')}}",
+			},
+			group:            "group-name",
+			cancelInProgress: false,
+		},
+		{
+			name: "event-evaluation",
+			input: model.RawConcurrency{
+				Group: "user-${{ github.event.commits[0].author.username }}",
+			},
+			group:            "user-someone",
+			cancelInProgress: false,
+		},
+		{
+			name: "arbitrary-var",
+			input: model.RawConcurrency{
+				Group: "${{ vars.eval_arbitrary_var }}",
+			},
+			group:            "123",
+			cancelInProgress: false,
+		},
+		{
+			name: "arbitrary-input",
+			input: model.RawConcurrency{
+				Group: "${{ inputs.eval_arbitrary_input }}",
+			},
+			group:            "456",
+			cancelInProgress: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			group, cancelInProgress, err := EvaluateConcurrency(
+				&test.input,
+				"job-id",
+				nil, // job
+				map[string]any{
+					"workflow": "test_workflow",
+					"ref":      "main",
+					"event": map[string]interface{}{
+						"commits": []interface{}{
+							map[string]interface{}{
+								"author": map[string]interface{}{
+									"username": "someone",
+								},
+							},
+							map[string]interface{}{
+								"author": map[string]interface{}{
+									"username": "someone-else",
+								},
+							},
+						},
+					},
+				}, // gitCtx
+				map[string]*JobResult{
+					"job-id": {},
+				}, // results
+				map[string]string{
+					"eval_arbitrary_var": "123",
+				}, // vars
+				map[string]any{
+					"eval_arbitrary_input": "456",
+				}, // inputs
+			)
+			assert.NoError(t, err)
+			assert.EqualValues(t, test.group, group)
+			assert.EqualValues(t, test.cancelInProgress, cancelInProgress)
+		})
+	}
+}

act/jobparser/testdata/job_concurrency.in.yaml

@@ -0,0 +1,9 @@
+name: test
+jobs:
+  job1:
+    runs-on: linux
+    concurrency:
+      group: major-tests
+      cancel-in-progress: true
+    steps:
+      - run: uname -a

act/jobparser/testdata/job_concurrency.out.yaml

@@ -0,0 +1,10 @@
+name: test
+jobs:
+  job1:
+    name: job1
+    runs-on: linux
+    steps:
+      - run: uname -a
+    concurrency:
+      group: major-tests
+      cancel-in-progress: "true"

act/jobparser/testdata/job_concurrency_eval.in.yaml

@@ -0,0 +1,9 @@
+name: test
+jobs:
+  job1:
+    runs-on: linux
+    concurrency:
+      group: ${{ github.workflow }}
+      cancel-in-progress: ${{ !contains(github.ref, 'release/')}}
+    steps:
+      - run: uname -a

act/jobparser/testdata/job_concurrency_eval.out.yaml

@@ -0,0 +1,10 @@
+name: test
+jobs:
+  job1:
+    name: job1
+    runs-on: linux
+    steps:
+      - run: uname -a
+    concurrency:
+      group: ${{ github.workflow }}
+      cancel-in-progress: ${{ !contains(github.ref, 'release/')}}

act/model/workflow.go

@@ -26,6 +26,7 @@ type Workflow struct {
 	Defaults Defaults          `yaml:"defaults"`
 
 	RawNotifications yaml.Node       `yaml:"enable-email-notifications"`
+	RawConcurrency   *RawConcurrency `yaml:"concurrency"` // For Gitea
 }
 
 // On events for the workflow
@@ -806,3 +807,28 @@ func (w *Workflow) Notifications() (bool, error) {
 		return false, fmt.Errorf("enable-email-notifications: unknown type: %v", w.RawNotifications.Kind)
 	}
 }
+
+// For Gitea
+// RawConcurrency represents a workflow concurrency or a job concurrency with uninterpolated options
+type RawConcurrency struct {
+	Group            string `yaml:"group,omitempty"`
+	CancelInProgress string `yaml:"cancel-in-progress,omitempty"`
+	RawExpression    string `yaml:"-,omitempty"`
+}
+
+type objectConcurrency RawConcurrency
+
+func (r *RawConcurrency) UnmarshalYAML(n *yaml.Node) error {
+	if err := n.Decode(&r.RawExpression); err == nil {
+		return nil
+	}
+	return n.Decode((*objectConcurrency)(r))
+}
+
+func (r *RawConcurrency) MarshalYAML() (interface{}, error) {
+	if r.RawExpression != "" {
+		return r.RawExpression, nil
+	}
+
+	return (*objectConcurrency)(r), nil
+}

act/model/workflow_test.go

@@ -702,3 +702,48 @@ func TestStepUsesHash(t *testing.T) {
 		})
 	}
 }
+
+func TestReadWorkflow_Concurrency(t *testing.T) {
+	for _, testCase := range []struct {
+		expected *RawConcurrency
+		err      string
+		snippet  string
+	}{
+		{
+			expected: nil,
+			snippet:  "# nothing",
+		},
+		{
+			expected: &RawConcurrency{Group: "${{ github.workflow }}-${{ github.ref }}", CancelInProgress: ""},
+			snippet:  "concurrency: { group: \"${{ github.workflow }}-${{ github.ref }}\" }",
+		},
+		{
+			expected: &RawConcurrency{Group: "example-group", CancelInProgress: "true"},
+			snippet:  "concurrency: { group: example-group, cancel-in-progress: true }",
+		},
+	} {
+		t.Run(testCase.snippet, func(t *testing.T) {
+			yaml := fmt.Sprintf(`
+name: name-455
+on: push
+%s
+jobs:
+  valid-JOB-Name-455:
+    runs-on: docker
+    steps:
+      - run: echo hi
+`, testCase.snippet)
+
+			workflow, err := ReadWorkflow(strings.NewReader(yaml), true)
+			if testCase.err != "" {
+				assert.ErrorContains(t, err, testCase.err)
+			} else {
+				assert.NoError(t, err, "read workflow should succeed")
+
+				concurrency := workflow.RawConcurrency
+				// assert.NoError(t, err)
+				assert.Equal(t, testCase.expected, concurrency)
+			}
+		})
+	}
+}

act/schema/workflow_schema.json

@@ -1689,7 +1689,7 @@
             "description": "When a concurrent job or workflow is queued, if another job or workflow using the same concurrency group in the repository is in progress, the queued job or workflow will be `pending`. Any previously pending job or workflow in the concurrency group will be canceled. To also cancel any currently running job or workflow in the same concurrency group, specify `cancel-in-progress: true`."
           },
           "cancel-in-progress": {
-            "type": "boolean",
+            "type": "non-empty-string",
             "description": "To cancel any currently running job or workflow in the same concurrency group, specify cancel-in-progress: true."
           }
         }