From 159f74040c09d8f8f0090e8975fe16e61871ff70 Mon Sep 17 00:00:00 2001 From: Zettat123 Date: Thu, 26 Mar 2026 21:08:01 -0600 Subject: [PATCH] Fix missing `workflow_run` notifications when updating jobs from multiple runs (#36997) (#37003) Backport #36997 This PR fixes `notifyWorkflowJobStatusUpdate` to send `WorkflowRunStatusUpdate` for each affected workflow run instead of only the first run in the input job list. --- services/actions/clear_tasks.go | 27 ++++++--- tests/integration/repo_webhook_test.go | 83 ++++++++++++++++++++++++++ 2 files changed, 103 insertions(+), 7 deletions(-) diff --git a/services/actions/clear_tasks.go b/services/actions/clear_tasks.go index 3c7aa0b1a5..dd7111f9e2 100644 --- a/services/actions/clear_tasks.go +++ b/services/actions/clear_tasks.go @@ -36,14 +36,27 @@ func StopEndlessTasks(ctx context.Context) error { } func notifyWorkflowJobStatusUpdate(ctx context.Context, jobs []*actions_model.ActionRunJob) { - if len(jobs) > 0 { - CreateCommitStatus(ctx, jobs...) - for _, job := range jobs { - _ = job.LoadAttributes(ctx) - notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil) + if len(jobs) == 0 { + return + } + + CreateCommitStatus(ctx, jobs...) + + runs := make(map[int64]*actions_model.ActionRun, len(jobs)) + for _, job := range jobs { + if err := job.LoadAttributes(ctx); err != nil { + log.Error("Failed to load job attributes: %v", err) + continue } - job := jobs[0] - notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run) + + notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil) + if _, ok := runs[job.RunID]; !ok { + runs[job.RunID] = job.Run + } + } + + for _, run := range runs { + notify_service.WorkflowRunStatusUpdate(ctx, run.Repo, run.TriggerUser, run) } } diff --git a/tests/integration/repo_webhook_test.go b/tests/integration/repo_webhook_test.go index 3b02152adb..054dee4b7b 100644 --- a/tests/integration/repo_webhook_test.go +++ b/tests/integration/repo_webhook_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + actions_model "code.gitea.io/gitea/models/actions" auth_model "code.gitea.io/gitea/models/auth" "code.gitea.io/gitea/models/perm" "code.gitea.io/gitea/models/repo" @@ -1150,6 +1151,10 @@ func Test_WebhookWorkflowRun(t *testing.T) { testWorkflowRunEventsOnCancellingAbandonedRun(t, webhookData, false) }, }, + { + name: "WorkflowRunOnStoppingEndlessTasksForMultipleRuns", + testFunc: testWorkflowRunOnStoppingEndlessTasksForMultipleRuns, + }, } for _, obj := range testCases { t.Run(obj.name, func(t *testing.T) { @@ -1586,6 +1591,84 @@ jobs: assert.Equal(t, "user2/"+repoName, webhookData.payloads[1].Repo.FullName) } +func testWorkflowRunOnStoppingEndlessTasksForMultipleRuns(t *testing.T, webhookData *workflowRunWebhook) { + defer test.MockVariableValue(&setting.Actions.EndlessTaskTimeout, time.Second)() + + user2 := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2}) + session := loginUser(t, "user2") + token := getTokenForLoggedInUser(t, session, auth_model.AccessTokenScopeWriteRepository, auth_model.AccessTokenScopeWriteUser) + + repoName := "test-workflow-run-stop-endless-tasks" + testRepo := unittest.AssertExistsAndLoadBean(t, &repo.Repository{ID: createActionsTestRepo(t, token, repoName, false).ID}) + + testAPICreateWebhookForRepo(t, session, "user2", repoName, webhookData.URL, "workflow_run") + + runners := make([]*mockRunner, 2) + for i := range runners { + runners[i] = newMockRunner() + runners[i].registerAsRepoRunner(t, "user2", repoName, fmt.Sprintf("mock-runner-%d", i), []string{"ubuntu-latest"}, false) + } + + workflowPath1 := ".gitea/workflows/endless-1.yml" + workflowPath2 := ".gitea/workflows/endless-2.yml" + workflowContent1 := `name: endless-1 +on: + push: + paths: + - '.gitea/workflows/endless-1.yml' +jobs: + job-1: + runs-on: ubuntu-latest + steps: + - run: echo 'job-1' +` + workflowContent2 := `name: endless-2 +on: + push: + paths: + - '.gitea/workflows/endless-2.yml' +jobs: + job-2: + runs-on: ubuntu-latest + steps: + - run: echo 'job-2' +` + + opts1 := getWorkflowCreateFileOptions(user2, testRepo.DefaultBranch, "create "+workflowPath1, workflowContent1) + createWorkflowFile(t, token, "user2", repoName, workflowPath1, opts1) + opts2 := getWorkflowCreateFileOptions(user2, testRepo.DefaultBranch, "create "+workflowPath2, workflowContent2) + createWorkflowFile(t, token, "user2", repoName, workflowPath2, opts2) + + task1 := runners[0].fetchTask(t) + task2 := runners[1].fetchTask(t) + _, job1, _ := getTaskAndJobAndRunByTaskID(t, task1.Id) + _, job2, _ := getTaskAndJobAndRunByTaskID(t, task2.Id) + require.NotEqual(t, job1.RunID, job2.RunID) + + initialRunEventsLen := len(webhookData.payloads) + + time.Sleep(2 * time.Second) + + require.NoError(t, actions.StopEndlessTasks(t.Context())) + + require.Len(t, webhookData.payloads, initialRunEventsLen+2) + + var completedRunIDs []int64 + for _, payload := range webhookData.payloads[initialRunEventsLen:] { + assert.Equal(t, "completed", payload.Action) + assert.Equal(t, "completed", payload.WorkflowRun.Status) + completedRunIDs = append(completedRunIDs, payload.WorkflowRun.ID) + } + assert.Len(t, completedRunIDs, 2) + assert.Contains(t, completedRunIDs, job1.RunID) + assert.Contains(t, completedRunIDs, job2.RunID) + + run1 := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: job1.RunID}) + run2 := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: job2.RunID}) + assert.Equal(t, actions_model.StatusFailure, run1.Status) + assert.Equal(t, actions_model.StatusFailure, run2.Status) +} + func testWebhookWorkflowRun(t *testing.T, webhookData *workflowRunWebhook) { // 1. create a new webhook with special webhook for repo1 user2 := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2})