fix: update statuses in phases to prevent out of order updates from stalling workflows (#12685)

During workflow expansion, jobs are replaced or added. That means that their execution order does not necessarily match the order of their numeric IDs. For example, job 129 might depend on job 130. Unfortunately, Forgejo doesn't take this possibility into account and always examines and updates jobs by ascending ID. That means that while examining job 129, job 130 has still its old status, and, as a result, Forgejo won't schedule job 129 for execution because it's still waiting for job 130 to complete. This can lead to workflows getting stuck:

```
...tions/job_emitter.go:48:jobEmitterQueueHandler() [E] checkJobsOfRun failed for RunID = 49: error in tryHandleIncompleteMatrix: jobStatusResolver attempted to tryHandleIncompleteMatrix for a job (id=129) with an incomplete 'needs' job (id=130)
```

This is caused by calculating all status changes recursively in memory before writing them to the database. For example, job A that was completed would unblock job B that depended on it. In that case, Forgejo would simultaneously mark job A as completed and B as waiting. However, that would not work if B had a lower ID than A. That is remedied by writing updates to the database before entering the next recursion. So, continuing the example, job A is marked as completed in the database before the next iteration detects that B is no longer blocked and writes that to the database.

Resolves https://codeberg.org/forgejo/forgejo/issues/12641.

## Checklist

The [contributor guide](https://forgejo.org/docs/next/contributor/) contains information that will be helpful to first time contributors. All work and communication must conform to Forgejo's [AI Agreement](https://codeberg.org/forgejo/governance/src/branch/main/AIAgreement.md). There also are a few [conditions for merging Pull Requests in Forgejo repositories](https://codeberg.org/forgejo/governance/src/branch/main/PullRequestsAgreement.md). You are also welcome to join the [Forgejo development chatroom](https://matrix.to/#/#forgejo-development:matrix.org).

### Tests for Go changes

(can be removed for JavaScript changes)

- I added test coverage for Go changes...
  - [x] in their respective `*_test.go` for unit tests.
  - [ ] in the `tests/integration` directory if it involves interactions with a live Forgejo server.
- I ran...
  - [x] `make pr-go` before pushing

### Tests for JavaScript changes

(can be removed for Go changes)

- I added test coverage for JavaScript changes...
  - [ ] in `web_src/js/*.test.js` if it can be unit tested.
  - [ ] in `tests/e2e/*.test.e2e.js` if it requires interactions with a live Forgejo server (see also the [developer guide for JavaScript testing](https://codeberg.org/forgejo/forgejo/src/branch/forgejo/tests/e2e/README.md#end-to-end-tests)).

### Documentation

- [ ] I created a pull request [to the documentation](https://codeberg.org/forgejo/docs) to explain to Forgejo users how to use this change.
- [x] I did not document these changes and I do not expect someone else to do it.

### Release notes

- [x] This change will be noticed by a Forgejo user or admin (feature, bug fix, performance, etc.). I suggest to include a release note for this change.
- [ ] This change is not visible to a Forgejo user or admin (refactor, dependency upgrade, etc.). I think there is no need to add a release note for this change.

*The decision if the pull request will be shown in the release notes is up to the mergers / release team.*

The content of the `release-notes/<pull request number>.md` file will serve as the basis for the release notes. If the file does not exist, the title of the pull request will be used instead.

Reviewed-on: https://codeberg.org/forgejo/forgejo/pulls/12685
Reviewed-by: Mathieu Fenniak <mfenniak@noreply.codeberg.org>

services/actions/Test_checkJobsOfRun/action_run.yml

@@ -16,3 +16,23 @@
   updated: 1683636626
   need_approval: 0
   approved_by: 0
+
+# Test case with needs, only
+- id: 901
+  title: ".forgejo/workflows/test.yaml"
+  repo_id: 63
+  owner_id: 2
+  workflow_id: "test.yaml"
+  workflow_directory: ".forgejo/workflows"
+  index: 5
+  trigger_user_id: 2
+  ref: "refs/heads/main"
+  commit_sha: "c29b3a4b23e21ea3261291d82bfb8e7af7797e26"
+  trigger_event: "workflow_dispatch"
+  is_fork_pull_request: 0
+  status: 7 # running
+  started: 1779368408
+  created: 1779368408
+  updated: 1779368421
+  need_approval: 0
+  approved_by: 0

services/actions/Test_checkJobsOfRun/action_run_job.yml

@@ -38,3 +38,107 @@
   task_id: 100
   status: 1 # success
   runs_on: '["fedora"]'
+
+# Jobs for run 901. For the test to be effective, it is important that the job with `job_id: c` has a lower `id` than
+# `job_id: b` because that reflects real-word ordering issues that need to be handled.
+- id: 674
+  run_id: 901
+  repo_id: 63
+  owner_id: 2
+  commit_sha: c29b3a4b23e21ea3261291d82bfb8e7af7797e26
+  attempt: 1
+  job_id: a
+  needs: null
+  runs_on: '["fedora"]'
+  status: 1 # success
+  task_id: 101
+  workflow_payload: |
+    "on":
+      workflow_dispatch:
+    jobs:
+      a:
+        name: a
+        runs-on: ubuntu-latest
+        steps:
+          - id: set
+            run: echo "argument=something" >> "$FORGEJO_OUTPUT"
+        outputs:
+          argument: ${{ steps.set.outputs.argument }}
+
+- id: 676
+  run_id: 901
+  repo_id: 63
+  owner_id: 2
+  commit_sha: c29b3a4b23e21ea3261291d82bfb8e7af7797e26
+  attempt: 1
+  job_id: c
+  needs: '["a","b"]'
+  runs_on: '["fedora"]'
+  status: 7 # blocked
+  workflow_payload: |
+    "on":
+      workflow_dispatch:
+    jobs:
+      c:
+        name: c
+        runs-on: []
+        uses: ./.forgejo/workflows/reusable.yaml
+        with:
+          argument: ${{ needs.a.outputs.argument }}
+    incomplete_with: true
+    incomplete_with_needs:
+      job: a
+      output: argument
+    __metadata:
+      workflow_call_id: e0c1155c1286a2f6e213ec7986500f7961418d1e11664de1409a2d8072d6845e
+
+- id: 677
+  run_id: 901
+  repo_id: 63
+  owner_id: 2
+  commit_sha: c29b3a4b23e21ea3261291d82bfb8e7af7797e26
+  attempt: 1
+  job_id: b
+  needs: '["b.reusable","a"]'
+  runs_on: '["fedora"]'
+  status: 7 # blocked
+  workflow_payload: |
+    "on":
+      workflow_dispatch:
+    jobs:
+      b:
+        name: b
+        runs-on: []
+        if: false
+    __metadata:
+      workflow_call_inputs:
+        argument: something
+      workflow_call_id: ec2c61f51535f715c1b7b3089c6b2d46578698a4c5d10f6dd4a4466b2f44f81c
+
+- id: 678
+  run_id: 901
+  repo_id: 63
+  owner_id: 2
+  commit_sha: c29b3a4b23e21ea3261291d82bfb8e7af7797e26
+  attempt: 1
+  job_id: b.reusable
+  needs: '["a"]'
+  runs_on: '["fedora"]'
+  status: 1 # success
+  workflow_payload: |
+    "on":
+      workflow_call:
+        inputs:
+          argument:
+            default: something
+            type: string
+    jobs:
+      b.reusable:
+        name: reusable
+        runs-on: ubuntu-latest
+        steps:
+          - run: |
+              echo "Argument: ${{ inputs.argument }}"
+    __metadata:
+      workflow_call_parent: ec2c61f51535f715c1b7b3089c6b2d46578698a4c5d10f6dd4a4466b2f44f81c
+

services/actions/Test_checkJobsOfRun/action_task_output.yml

@@ -3,3 +3,10 @@
   task_id: 100
   output_key: colors
   output_value: '["red", "blue", "green"]'
+
+-
+  id: 101
+  task_id: 101
+  output_key: argument
+  output_value: "something"
+

services/actions/job_emitter.go

@@ -61,17 +61,15 @@ func checkJobsOfRun(ctx context.Context, runID int64, recursionCount int) error
 		return fmt.Errorf("checkJobsOfRun for runID %d hit recursion limit %d", runID, recursionCount)
 	}
 
+	var jobs actions_model.ActionJobList
 	jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: runID})
 	if err != nil {
 		return err
 	}
-	if err := db.WithTx(ctx, func(ctx context.Context) error {
-		idToJobs := make(map[string][]*actions_model.ActionRunJob, len(jobs))
-		for _, job := range jobs {
-			idToJobs[job.JobID] = append(idToJobs[job.JobID], job)
-		}
 
-		updates := newJobStatusResolver(jobs).Resolve()
+	var updates map[int64]actions_model.Status
+	if err := db.WithTx(ctx, func(ctx context.Context) error {
+		updates = newJobStatusResolver(jobs).Resolve()
 		for _, job := range jobs {
 			if status, ok := updates[job.ID]; ok {
 				job.Status = status
@@ -115,15 +113,16 @@ func checkJobsOfRun(ctx context.Context, runID int64, recursionCount int) error
 	}); err != nil {
 		return err
 	}
+
 	CreateCommitStatus(ctx, jobs...)
 
 	// tryHandleIncompleteMatrix can create new jobs in this run which may initially be persisted in the DB as blocked
 	// because they have non-empty `needs`. In that case, we need to recursively run the job emitter so that new jobs
 	// are recognized as having their `needs` completed and be set as unblocked. Check if any new jobs were created and
-	// rerun the job emitter if so.
+	// rerun the job emitter if so. The same is necessary if updates completed jobs that unblocked other jobs.
 	if hasNewJobs, err := actions_model.RunHasOtherJobs(ctx, runID, jobs); err != nil {
 		return fmt.Errorf("RunHasOtherJobs error: %w", err)
-	} else if hasNewJobs {
+	} else if hasNewJobs || len(updates) > 0 {
 		return checkJobsOfRun(ctx, runID, recursionCount+1)
 	}
 
@@ -172,21 +171,6 @@ func newJobStatusResolver(jobs actions_model.ActionJobList) *jobStatusResolver {
 }
 
 func (r *jobStatusResolver) Resolve() map[int64]actions_model.Status {
-	ret := map[int64]actions_model.Status{}
-	for i := 0; i < len(r.statuses); i++ {
-		updated := r.resolve()
-		if len(updated) == 0 {
-			return ret
-		}
-		for k, v := range updated {
-			ret[k] = v
-			r.statuses[k] = v
-		}
-	}
-	return ret
-}
-
-func (r *jobStatusResolver) resolve() map[int64]actions_model.Status {
 	ret := map[int64]actions_model.Status{}
 	for id, status := range r.statuses {
 		if status != actions_model.StatusBlocked {

services/actions/job_emitter_test.go

@@ -68,8 +68,9 @@ func Test_jobStatusResolver_Resolve(t *testing.T) {
 				{ID: 3, JobID: "3", Status: actions_model.StatusBlocked, Needs: []string{"2"}},
 			},
 			want: map[int64]actions_model.Status{
+				// Resolve() does only one update pass and does not update jobs recursively. Therefore, job 3, which
+				// depends on 2, is not marked as skipped. It would only be marked as skipped if it depended on job 1.
 				2: actions_model.StatusSkipped,
-				3: actions_model.StatusSkipped,
 			},
 		},
 		{
@@ -876,6 +877,61 @@ func Test_tryHandleWorkflowCallOuterJob(t *testing.T) {
 	}
 }
 
+func Test_checkJobsOfRun(t *testing.T) {
+	defer unittest.OverrideFixtures("services/actions/Test_checkJobsOfRun")()
+	require.NoError(t, unittest.PrepareTestDatabase())
+
+	reusableWorkflow := `
+on:
+  workflow_call:
+    inputs:
+      argument:
+        type: string
+        
+jobs:
+  reusable:
+    runs-on: ubuntu-latest
+    steps:
+      - run: |
+          echo "Argument: ${{ inputs.argument }}"
+`
+
+	defer test.MockVariableValue(&lazyRepoExpandLocalReusableWorkflow,
+		func(ctx context.Context, repoID int64, commitSHA string) (jobparser.LocalWorkflowFetcher, CleanupFunc) {
+			fetcher := func(job *jobparser.Job, path string) ([]byte, error) {
+				return []byte(reusableWorkflow), nil
+			}
+			cleanup := func() {
+			}
+			return fetcher, cleanup
+		})()
+
+	jobs, err := actions_model.GetRunJobsByRunID(t.Context(), 901)
+	require.NoError(t, err)
+	require.Len(t, jobs, 4)
+
+	require.NoError(t, checkJobsOfRun(t.Context(), 901, 0))
+
+	jobs, err = actions_model.GetRunJobsByRunID(t.Context(), 901)
+	require.NoError(t, err)
+	assert.Len(t, jobs, 5)
+
+	assert.Equal(t, "a", jobs[0].JobID)
+	assert.Equal(t, actions_model.StatusSuccess, jobs[0].Status)
+
+	assert.Equal(t, "b", jobs[1].JobID)
+	assert.Equal(t, actions_model.StatusSuccess, jobs[1].Status)
+
+	assert.Equal(t, "b.reusable", jobs[2].JobID)
+	assert.Equal(t, actions_model.StatusSuccess, jobs[2].Status)
+
+	assert.Equal(t, "c", jobs[3].JobID)
+	assert.Equal(t, actions_model.StatusBlocked, jobs[3].Status)
+
+	assert.Equal(t, "c.reusable", jobs[4].JobID)
+	assert.Equal(t, actions_model.StatusWaiting, jobs[4].Status)
+}
+
 func Test_checkJobsOfRun_ExpandsMatrixWithCorrectOutputJobStatuses(t *testing.T) {
 	defer unittest.OverrideFixtures("services/actions/Test_checkJobsOfRun")()
 	require.NoError(t, unittest.PrepareTestDatabase())