fix

Author: Zettat123

routers/api/actions/runner/utils.go

@@ -77,7 +77,10 @@ func generateTaskContext(t *actions_model.ActionTask) *structpb.Struct {
 }
 
 func findTaskNeeds(ctx context.Context, task *actions_model.ActionTask) (map[string]*runnerv1.TaskNeed, error) {
-	taskNeeds, err := actions.FindTaskNeeds(ctx, task)
+	if err := task.LoadAttributes(ctx); err != nil {
+		return nil, fmt.Errorf("task LoadAttributes: %w", err)
+	}
+	taskNeeds, err := actions.FindTaskNeeds(ctx, task.Job)
 	if err != nil {
 		return nil, err
 	}

services/actions/concurrency.go

@@ -21,15 +21,31 @@ func evaluateJobConcurrency(run *actions_model.ActionRun, actionRunJob *actions_
 
 	gitCtx := jobparser.ToGitContext(GenerateGitContext(run, actionRunJob))
 
+	singleWorkflows, err := jobparser.Parse([]byte(actionRunJob.WorkflowPayload))
+	if err != nil {
+		return "", false, fmt.Errorf("parse single workflow: %w", err)
+	} else if len(singleWorkflows) != 1 {
+		return "", false, fmt.Errorf("not single workflow")
+	}
+
+	// FIXME: cannot get job needs
+	singleWorkflow := singleWorkflows[0]
+	singleWorkflow.Job()
+	actJobMap := map[string]*act_model.Job{}
+	if err := singleWorkflow.RawJobs.Decode(&actJobMap); err != nil {
+		return "", false, fmt.Errorf("decode act job: %w", err)
+	}
+
 	actWorkflow, err := act_model.ReadWorkflow(bytes.NewReader(actionRunJob.WorkflowPayload))
 	if err != nil {
 		return "", false, fmt.Errorf("read workflow: %w", err)
 	}
 	actJob := actWorkflow.GetJob(actionRunJob.JobID)
 
-	if len(jobResults) == 0 {
-		jobResults = map[string]*jobparser.JobResult{actionRunJob.JobID: {}}
+	if jobResults == nil {
+		jobResults = map[string]*jobparser.JobResult{}
 	}
+	jobResults[actionRunJob.JobID] = &jobparser.JobResult{}
 
 	concurrencyGroup, concurrencyCancel := jobparser.EvaluateJobConcurrency(rawConcurrency, actionRunJob.JobID, actJob, gitCtx, vars, jobResults)
 

services/actions/job_emitter.go

@@ -235,7 +235,7 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
 			// check concurrency
 			blockedByJobConcurrency, err := checkConcurrencyForJobWithNeeds(ctx, r.jobMap[id], r.vars)
 			if err != nil {
-				log.Error("Check run %d job %d concurrency: %v. This job will stay blocked.")
+				log.Error("Check job %d concurrency: %v. This job will stay blocked.", id, err)
 				continue
 			}
 
@@ -271,17 +271,13 @@ func checkConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_
 	if len(actionRunJob.RawConcurrencyGroup) == 0 {
 		return false, nil
 	}
-	if err := actionRunJob.LoadRun(ctx); err != nil {
+	if err := actionRunJob.LoadAttributes(ctx); err != nil {
 		return false, err
 	}
 
 	if !actionRunJob.IsConcurrencyEvaluated {
 		// empty concurrency group means the raw concurrency has not been evaluated
-		task, err := actions_model.GetTaskByID(ctx, actionRunJob.TaskID)
-		if err != nil {
-			return false, fmt.Errorf("get task by id: %w", err)
-		}
-		taskNeeds, err := FindTaskNeeds(ctx, task)
+		taskNeeds, err := FindTaskNeeds(ctx, actionRunJob)
 		if err != nil {
 			return false, fmt.Errorf("find task needs: %w", err)
 		}
@@ -298,13 +294,9 @@ func checkConcurrencyForJobWithNeeds(ctx context.Context, actionRunJob *actions_
 		if err != nil {
 			return false, fmt.Errorf("evaluate job concurrency: %w", err)
 		}
+		actionRunJob.IsConcurrencyEvaluated = true
 
-		if _, err := actions_model.UpdateRunJob(ctx, &actions_model.ActionRunJob{
-			ID:                     actionRunJob.ID,
-			IsConcurrencyEvaluated: true,
-			ConcurrencyGroup:       actionRunJob.ConcurrencyGroup,
-			ConcurrencyCancel:      actionRunJob.ConcurrencyCancel,
-		}, nil); err != nil {
+		if _, err := actions_model.UpdateRunJob(ctx, actionRunJob, nil, "concurrency_group", "concurrency_cancel", "is_concurrency_evaluated"); err != nil {
 			return false, fmt.Errorf("update run job: %w", err)
 		}
 	}

services/actions/utils.go

@@ -103,16 +103,13 @@ type TaskNeed struct {
 	Outputs map[string]string
 }
 
-func FindTaskNeeds(ctx context.Context, task *actions_model.ActionTask) (map[string]*TaskNeed, error) {
-	if err := task.LoadAttributes(ctx); err != nil {
-		return nil, fmt.Errorf("LoadAttributes: %w", err)
-	}
-	if len(task.Job.Needs) == 0 {
+func FindTaskNeeds(ctx context.Context, job *actions_model.ActionRunJob) (map[string]*TaskNeed, error) {
+	if len(job.Needs) == 0 {
 		return nil, nil
 	}
-	needs := container.SetOf(task.Job.Needs...)
+	needs := container.SetOf(job.Needs...)
 
-	jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: task.Job.RunID})
+	jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: job.RunID})
 	if err != nil {
 		return nil, fmt.Errorf("FindRunJobs: %w", err)
 	}

tests/integration/actions_concurrency_test.go

@@ -21,7 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestWorkflowConcurrency_NoCancellation(t *testing.T) {
+func TestWorkflowConcurrency(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)
@@ -34,7 +34,7 @@ func TestWorkflowConcurrency_NoCancellation(t *testing.T) {
 
 		// add a variable for test
 		req := NewRequestWithJSON(t, "POST",
-			fmt.Sprintf("/api/v1/repos/%s/%s/actions/variables/qwe", user2.Name, repo.Name), &api.CreateVariableOption{
+			fmt.Sprintf("/api/v1/repos/%s/%s/actions/variables/myvar", user2.Name, repo.Name), &api.CreateVariableOption{
 				Value: "abc123",
 			}).
 			AddTokenAuth(token)
@@ -61,7 +61,7 @@ on:
     paths:
       - '.gitea/workflows/concurrent-workflow-2.yml'
 concurrency:
-  group: workflow-${{ github.ref_name }}-${{ vars.qwe }}
+  group: workflow-${{ github.ref_name }}-${{ vars.myvar }}
 jobs:
   wf2-job:
     runs-on: ubuntu-latest
@@ -126,7 +126,7 @@ jobs:
 	})
 }
 
-func TestWorkflowConcurrency_WithCancellation(t *testing.T) {
+func TestWorkflowConcurrency_WithPullRequest(t *testing.T) {
 	onGiteaRun(t, func(t *testing.T, u *url.URL) {
 		// user2 is the owner of the base repo
 		user2 := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2})
@@ -276,6 +276,101 @@ jobs:
 	})
 }
 
+func TestJobConcurrency(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-concurrency", false)
+		repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: apiRepo.ID})
+		runner1 := newMockRunner()
+		runner1.registerAsRepoRunner(t, user2.Name, repo.Name, "mock-runner-1", []string{"runner1"})
+		runner2 := newMockRunner()
+		runner2.registerAsRepoRunner(t, user2.Name, repo.Name, "mock-runner-2", []string{"runner2"})
+
+		// add a variable for test
+		req := NewRequestWithJSON(t, "POST",
+			fmt.Sprintf("/api/v1/repos/%s/%s/actions/variables/version_var", user2.Name, repo.Name), &api.CreateVariableOption{
+				Value: "v1.23.0",
+			}).
+			AddTokenAuth(token)
+		MakeRequest(t, req, http.StatusNoContent)
+
+		wf1TreePath := ".gitea/workflows/concurrent-workflow-1.yml"
+		wf1FileContent := `name: concurrent-workflow-1
+on: 
+  push:
+    paths:
+      - '.gitea/workflows/concurrent-workflow-1.yml'
+jobs:
+  wf1-job1:
+    runs-on: runner1
+    concurrency:
+      group: job-main-${{ vars.version_var }}
+    steps:
+      - run: echo 'wf1-job1'
+`
+		wf2TreePath := ".gitea/workflows/concurrent-workflow-2.yml"
+		wf2FileContent := `name: concurrent-workflow-2
+on: 
+  push:
+    paths:
+      - '.gitea/workflows/concurrent-workflow-2.yml'
+jobs:
+  wf2-job1:
+    runs-on: runner2
+    outputs:
+      version: ${{ steps.version_step.outputs.app_version }} 
+    steps:
+      - id: version_step
+        run: echo "app_version=v1.23.0" >> "$GITHUB_OUTPUT"
+  wf2-job2:
+    runs-on: runner1
+    needs: [wf2-job1]
+    concurrency:
+      group: job-main-${{ needs.wf2-job1.outputs.version }}
+    steps:
+      - run: echo 'wf2-job2'
+`
+
+		opts1 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, fmt.Sprintf("create %s", wf1TreePath), wf1FileContent)
+		createWorkflowFile(t, token, user2.Name, repo.Name, wf1TreePath, opts1)
+		opts2 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, fmt.Sprintf("create %s", wf2TreePath), wf2FileContent)
+		createWorkflowFile(t, token, user2.Name, repo.Name, wf2TreePath, opts2)
+		// opts3 := getWorkflowCreateFileOptions(user2, repo.DefaultBranch, fmt.Sprintf("create %s", wf3TreePath), wf3FileContent)
+		// createWorkflowFile(t, token, user2.Name, repo.Name, wf3TreePath, opts3)
+
+		// fetch wf1-job1
+		wf1Job1Task := runner1.fetchTask(t)
+		_, wf1Job1ActionJob, _ := getTaskAndJobAndRunByTaskID(t, wf1Job1Task.Id)
+		assert.Equal(t, "job-main-v1.23.0", wf1Job1ActionJob.ConcurrencyGroup)
+		assert.True(t, wf1Job1ActionJob.Status.IsRunning())
+		// fetch and exec wf2-job1
+		wf2Job1Task := runner2.fetchTask(t)
+		runner2.execTask(t, wf2Job1Task, &mockTaskOutcome{
+			result: runnerv1.Result_RESULT_SUCCESS,
+			outputs: map[string]string{
+				"version": "v1.23.0",
+			},
+		})
+		// cannot fetch wf2-job2 because wf1-job1 is running
+		runner1.fetchNoTask(t)
+		// exec wf1-job1
+		runner1.execTask(t, wf1Job1Task, &mockTaskOutcome{
+			result: runnerv1.Result_RESULT_SUCCESS,
+		})
+		// fetch wf2-job2
+		wf2Job2Task := runner2.fetchTask(t)
+		_, wf2Job2ActionJob, _ := getTaskAndJobAndRunByTaskID(t, wf2Job2Task.Id)
+		assert.Equal(t, "job-main-v1.23.0", wf2Job2ActionJob.ConcurrencyGroup)
+		assert.True(t, wf1Job1ActionJob.Status.IsRunning())
+
+		httpContext := NewAPITestContext(t, user2.Name, repo.Name, auth_model.AccessTokenScopeWriteRepository)
+		doAPIDeleteRepository(httpContext)(t)
+	})
+}
+
 func getTaskAndJobAndRunByTaskID(t *testing.T, taskID int64) (*actions_model.ActionTask, *actions_model.ActionRunJob, *actions_model.ActionRun) {
 	actionTask := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionTask{ID: taskID})
 	actionRunJob := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: actionTask.JobID})