Support Simplified Concurrency (#139)

- update RawConcurrency struct to parse and serialize string-based concurrency notation
- update EvaluateConcurrency to handle new RawConcurrency format

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>

pkg/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,
 	}
 

pkg/jobparser/jobparser.go

@@ -8,6 +8,7 @@ import (
 
 	"gopkg.in/yaml.v3"
 
+	"github.com/nektos/act/pkg/exprparser"
 	"github.com/nektos/act/pkg/model"
 )
 
@@ -40,6 +41,10 @@ func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) {
 	if err != nil {
 		return nil, fmt.Errorf("invalid jobs: %w", err)
 	}
+
+	evaluator := NewExpressionEvaluator(exprparser.NewInterpeter(&exprparser.EvaluationEnvironment{Github: pc.gitContext, Vars: pc.vars}, exprparser.Config{}))
+	workflow.RunName = evaluator.Interpolate(workflow.RunName)
+
 	for i, id := range ids {
 		job := jobs[i]
 		matricxes, err := getMatrixes(origin.GetJob(id))
@@ -52,7 +57,7 @@ func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) {
 				job.Name = id
 			}
 			job.Strategy.RawMatrix = encodeMatrix(matrix)
-			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))
 			job.Name = nameWithMatrix(job.Name, matrix, evaluator)
 			runsOn := origin.GetJob(id).RunsOn()
 			for i, v := range runsOn {
@@ -60,10 +65,12 @@ func Parse(content []byte, options ...ParseOption) ([]*SingleWorkflow, error) {
 			}
 			job.RawRunsOn = encodeRunsOn(runsOn)
 			swf := &SingleWorkflow{
-				Name:     workflow.Name,
-				RawOn:    workflow.RawOn,
-				Env:      workflow.Env,
-				Defaults: workflow.Defaults,
+				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 {
 				return nil, fmt.Errorf("SetJob: %w", err)

pkg/jobparser/model.go

@@ -1,6 +1,7 @@
 package jobparser
 
 import (
+	"bytes"
 	"fmt"
 
 	"github.com/nektos/act/pkg/model"
@@ -9,11 +10,13 @@ import (
 
 // SingleWorkflow is a workflow with single job and single matrix
 type SingleWorkflow struct {
-	Name     string            `yaml:"name,omitempty"`
-	RawOn    yaml.Node         `yaml:"on,omitempty"`
-	Env      map[string]string `yaml:"env,omitempty"`
-	RawJobs  yaml.Node         `yaml:"jobs,omitempty"`
-	Defaults Defaults          `yaml:"defaults,omitempty"`
+	Name           string            `yaml:"name,omitempty"`
+	RawOn          yaml.Node         `yaml:"on,omitempty"`
+	Env            map[string]string `yaml:"env,omitempty"`
+	RawJobs        yaml.Node         `yaml:"jobs,omitempty"`
+	Defaults       Defaults          `yaml:"defaults,omitempty"`
+	RawPermissions yaml.Node         `yaml:"permissions,omitempty"`
+	RunName        string            `yaml:"run-name,omitempty"`
 }
 
 func (w *SingleWorkflow) Job() (string, *Job) {
@@ -82,6 +85,8 @@ 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"`
+	RawPermissions yaml.Node                 `yaml:"permissions,omitempty"`
 }
 
 func (j *Job) Clone() *Job {
@@ -104,6 +109,8 @@ func (j *Job) Clone() *Job {
 		Uses:           j.Uses,
 		With:           j.With,
 		RawSecrets:     j.RawSecrets,
+		RawConcurrency: j.RawConcurrency,
+		RawPermissions: j.RawPermissions,
 	}
 }
 
@@ -241,6 +248,85 @@ func parseWorkflowDispatchInputs(inputs map[string]interface{}) ([]WorkflowDispa
 	return results, nil
 }
 
+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:
@@ -335,6 +421,13 @@ func ParseRawOn(rawOn *yaml.Node) ([]*Event, error) {
 								return nil, err
 							}
 							acts[act] = t
+						case yaml.ScalarNode:
+							var t string
+							err := content.Decode(&t)
+							if err != nil {
+								return nil, err
+							}
+							acts[act] = []string{t}
 						case yaml.MappingNode:
 							if k != "workflow_dispatch" || act != "inputs" {
 								return nil, fmt.Errorf("map should only for workflow_dispatch but %s: %#v", act, content)
@@ -422,3 +515,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 ""
+}

pkg/jobparser/model_test.go

@@ -58,6 +58,19 @@ func TestParseRawOn(t *testing.T) {
 				},
 			},
 		},
+		{
+			input: "on:\n  push:\n    branches: main",
+			result: []*Event{
+				{
+					Name: "push",
+					acts: map[string][]string{
+						"branches": {
+							"main",
+						},
+					},
+				},
+			},
+		},
 		{
 			input: "on:\n  branch_protection_rule:\n    types: [created, deleted]",
 			result: []*Event{

pkg/model/workflow.go

@@ -17,12 +17,14 @@ import (
 
 // Workflow is the structure of the files in .github/workflows
 type Workflow struct {
-	File     string
-	Name     string            `yaml:"name"`
-	RawOn    yaml.Node         `yaml:"on"`
-	Env      map[string]string `yaml:"env"`
-	Jobs     map[string]*Job   `yaml:"jobs"`
-	Defaults Defaults          `yaml:"defaults"`
+	File           string
+	Name           string            `yaml:"name"`
+	RawOn          yaml.Node         `yaml:"on"`
+	Env            map[string]string `yaml:"env"`
+	Jobs           map[string]*Job   `yaml:"jobs"`
+	Defaults       Defaults          `yaml:"defaults"`
+	RawConcurrency *RawConcurrency   `yaml:"concurrency"`
+	RawPermissions yaml.Node         `yaml:"permissions"`
 }
 
 // On events for the workflow
@@ -199,6 +201,7 @@ type Job struct {
 	Uses           string                    `yaml:"uses"`
 	With           map[string]interface{}    `yaml:"with"`
 	RawSecrets     yaml.Node                 `yaml:"secrets"`
+	RawPermissions yaml.Node                 `yaml:"permissions"`
 	Result         string
 }
 
@@ -769,3 +772,28 @@ func decodeNode(node yaml.Node, out interface{}) bool {
 	}
 	return true
 }
+
+// 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
+}

pkg/runner/job_executor.go

@@ -73,10 +73,6 @@ func newJobExecutor(info jobInfo, sf stepFactory, rc *RunContext) common.Executo
 
 		preExec := step.pre()
 		preSteps = append(preSteps, useStepLogger(rc, stepModel, stepStagePre, func(ctx context.Context) error {
-			if rc.caller != nil { // For Gitea
-				rc.caller.reusedWorkflowJobResults[rc.JobName] = "pending"
-			}
-
 			logger := common.Logger(ctx)
 			preErr := preExec(ctx)
 			if preErr != nil {
@@ -189,34 +185,7 @@ func setJobResult(ctx context.Context, info jobInfo, rc *RunContext, success boo
 	info.result(jobResult)
 	if rc.caller != nil {
 		// set reusable workflow job result
-
-		rc.caller.updateResultLock.Lock()
-		rc.caller.reusedWorkflowJobResults[rc.JobName] = jobResult
-
-		allJobDone := true
-		hasFailure := false
-		for _, result := range rc.caller.reusedWorkflowJobResults {
-			if result == "pending" {
-				allJobDone = false
-				break
-			}
-			if result == "failure" {
-				hasFailure = true
-			}
-		}
-
-		if allJobDone {
-			reusedWorkflowJobResult := "success"
-			reusedWorkflowJobResultMessage := "succeeded"
-			if hasFailure {
-				reusedWorkflowJobResult = "failure"
-				reusedWorkflowJobResultMessage = "failed"
-			}
-			rc.caller.runContext.result(reusedWorkflowJobResult)
-			logger.WithField("jobResult", reusedWorkflowJobResult).Infof("\U0001F3C1  Job %s", reusedWorkflowJobResultMessage)
-		}
-
-		rc.caller.updateResultLock.Unlock()
+		rc.caller.setReusedWorkflowJobResult(rc.JobName, jobResult) // For Gitea
 		return
 	}
 

pkg/runner/reusable_workflow.go

@@ -175,7 +175,11 @@ func newReusableWorkflowExecutor(rc *RunContext, directory string, workflow stri
 			return err
 		}
 
-		return runner.NewPlanExecutor(plan)(ctx)
+		// return runner.NewPlanExecutor(plan)(ctx)
+		return common.NewPipelineExecutor( // For Gitea
+			runner.NewPlanExecutor(plan),
+			setReusedWorkflowCallerResult(rc, runner),
+		)(ctx)
 	}
 }
 
@@ -271,3 +275,47 @@ func newRemoteReusableWorkflow(uses string) *remoteReusableWorkflow {
 		URL:      "https://github.com",
 	}
 }
+
+// For Gitea
+func setReusedWorkflowCallerResult(rc *RunContext, runner Runner) common.Executor {
+	return func(ctx context.Context) error {
+		logger := common.Logger(ctx)
+
+		runnerImpl, ok := runner.(*runnerImpl)
+		if !ok {
+			logger.Warn("Failed to get caller from runner")
+			return nil
+		}
+		caller := runnerImpl.caller
+
+		allJobDone := true
+		hasFailure := false
+		for _, result := range caller.reusedWorkflowJobResults {
+			if result == "pending" {
+				allJobDone = false
+				break
+			}
+			if result == "failure" {
+				hasFailure = true
+			}
+		}
+
+		if allJobDone {
+			reusedWorkflowJobResult := "success"
+			reusedWorkflowJobResultMessage := "succeeded"
+			if hasFailure {
+				reusedWorkflowJobResult = "failure"
+				reusedWorkflowJobResultMessage = "failed"
+			}
+
+			if rc.caller != nil {
+				rc.caller.setReusedWorkflowJobResult(rc.JobName, reusedWorkflowJobResult)
+			} else {
+				rc.result(reusedWorkflowJobResult)
+				logger.WithField("jobResult", reusedWorkflowJobResult).Infof("\U0001F3C1  Job %s", reusedWorkflowJobResultMessage)
+			}
+		}
+
+		return nil
+	}
+}

pkg/runner/run_context.go

@@ -92,7 +92,12 @@ func (rc *RunContext) GetEnv() map[string]string {
 }
 
 func (rc *RunContext) jobContainerName() string {
-	return createSimpleContainerName(rc.Config.ContainerNamePrefix, "WORKFLOW-"+rc.Run.Workflow.Name, "JOB-"+rc.Name)
+	nameParts := []string{rc.Config.ContainerNamePrefix, "WORKFLOW-" + rc.Run.Workflow.Name, "JOB-" + rc.Name}
+	if rc.caller != nil {
+		nameParts = append(nameParts, "CALLED-BY-"+rc.caller.runContext.JobName)
+	}
+	// return createSimpleContainerName(rc.Config.ContainerNamePrefix, "WORKFLOW-"+rc.Run.Workflow.Name, "JOB-"+rc.Name)
+	return createSimpleContainerName(nameParts...) // For Gitea
 }
 
 // Deprecated: use `networkNameForGitea`
@@ -653,6 +658,7 @@ func (rc *RunContext) Executor() (common.Executor, error) {
 	return func(ctx context.Context) error {
 		res, err := rc.isEnabled(ctx)
 		if err != nil {
+			rc.caller.setReusedWorkflowJobResult(rc.JobName, "failure") // For Gitea
 			return err
 		}
 		if res {
@@ -748,6 +754,10 @@ func (rc *RunContext) isEnabled(ctx context.Context) (bool, error) {
 	}
 
 	if !runJob {
+		if rc.caller != nil { // For Gitea
+			rc.caller.setReusedWorkflowJobResult(rc.JobName, "skipped")
+			return false, nil
+		}
 		l.WithField("jobResult", "skipped").Debugf("Skipping job '%s' due to '%s'", job.Name, job.If.Value)
 		return false, nil
 	}

pkg/runner/runner.go

@@ -210,6 +210,9 @@ func (runner *runnerImpl) NewPlanExecutor(plan *model.Plan) common.Executor {
 					if len(rc.String()) > maxJobNameLen {
 						maxJobNameLen = len(rc.String())
 					}
+					if rc.caller != nil { // For Gitea
+						rc.caller.setReusedWorkflowJobResult(rc.JobName, "pending")
+					}
 					stageExecutor = append(stageExecutor, func(ctx context.Context) error {
 						jobName := fmt.Sprintf("%-*s", maxJobNameLen, rc.String())
 						executor, err := rc.Executor()
@@ -281,3 +284,10 @@ func (runner *runnerImpl) newRunContext(ctx context.Context, run *model.Run, mat
 
 	return rc
 }
+
+// For Gitea
+func (c *caller) setReusedWorkflowJobResult(jobName string, result string) {
+	c.updateResultLock.Lock()
+	defer c.updateResultLock.Unlock()
+	c.reusedWorkflowJobResults[jobName] = result
+}