From 9f8dda632538350c9667489b4dc045c3a6c12d14 Mon Sep 17 00:00:00 2001 From: Mathieu Fenniak Date: Sat, 17 Jan 2026 23:09:49 +0100 Subject: [PATCH] fix: retry ActionRun updates when optimistic-concurrency-control indicates record changed (#10893) When concurrent updates occur to the `action_run` table, fetching a task via `FetchTask` can result in an error: ``` time="2026-01-16T16:02:30Z" level=error msg="failed to fetch task" error="internal: pick task: CreateTaskForRunner: update run 2358339: run has changed" ``` This is an error with no known harm. However, it is more efficient to recover from doing 90% of the work to fetch a task, and complete that work, rather than error out and rollback. Discovered while investigating https://code.forgejo.org/forgejo/runner/issues/1302#issuecomment-73859, although conclusions do not indicate that it is a source of this bug. ## Checklist The [contributor guide](https://forgejo.org/docs/next/contributor/) contains information that will be helpful to first time contributors. 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 - I added test coverage for Go changes... - [ ] in their respective `*_test.go` for unit tests. - [ ] in the `tests/integration` directory if it involves interactions with a live Forgejo server. - 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. - [ ] I did not document these changes and I do not expect someone else to do it. ### Release notes - [ ] 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/.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/10893 Reviewed-by: Michael Kriese Reviewed-by: Andreas Ahlenstorf Co-authored-by: Mathieu Fenniak Co-committed-by: Mathieu Fenniak --- models/actions/run.go | 9 ++++-- models/actions/run_job.go | 30 ++++++++++++++---- models/actions/run_job_test.go | 58 ++++++++++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 8 deletions(-) diff --git a/models/actions/run.go b/models/actions/run.go index 8924e04d37..126a4e7e38 100644 --- a/models/actions/run.go +++ b/models/actions/run.go @@ -479,6 +479,10 @@ func GetRunByIndex(ctx context.Context, repoID, index int64) (*ActionRun, error) return run, nil } +// Error returned when ActionRun's optimistic concurrency control has indicated that the record has been updated in the +// database by another session since it was loaded in-memory in this session. +var ErrActionRunOutOfDate = errors.New("run has changed") + // UpdateRun updates a run. // It requires the inputted run has Version set. // It will return error if the version is not matched (it means the run has been changed after loaded). @@ -495,8 +499,9 @@ func UpdateRunWithoutNotification(ctx context.Context, run *ActionRun, cols ...s return err } if affected == 0 { - return errors.New("run has changed") - // It's impossible that the run is not found, since Gitea never deletes runs. + // UPDATE has no conditions on it, and we never delete runs, so the only possible cause of this is + // `xorm:"version"` tagged field indicated that the version has changed since the record was loaded. + return ErrActionRunOutOfDate } if run.Status != 0 || slices.Contains(cols, "status") { diff --git a/models/actions/run_job.go b/models/actions/run_job.go index aee7cd9aa1..9ddae1e301 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -5,12 +5,14 @@ package actions import ( "context" + "errors" "fmt" "slices" "time" "forgejo.org/models/db" "forgejo.org/modules/container" + "forgejo.org/modules/log" "forgejo.org/modules/timeutil" "forgejo.org/modules/util" @@ -174,7 +176,7 @@ func UpdateRunJobWithoutNotification(ctx context.Context, job *ActionRunJob, con } } - { + for { // Other goroutines may aggregate the status of the run and update it too. // So we need load the run and its jobs before updating the run. run, err := GetRunByID(ctx, job.RunID) @@ -185,23 +187,39 @@ func UpdateRunJobWithoutNotification(ctx context.Context, job *ActionRunJob, con if err != nil { return 0, err } - run.Status = AggregateJobStatus(jobs) + + updateRequired := false + newStatus := AggregateJobStatus(jobs) + if run.Status != newStatus { + run.Status = newStatus + updateRequired = true + } if run.Started.IsZero() && run.Status.IsRunning() { run.Started = timeutil.TimeStampNow() + updateRequired = true } if run.Stopped.IsZero() && run.Status.IsDone() { run.Stopped = timeutil.TimeStampNow() + updateRequired = true } - // As the caller has to ensure the ActionRunNowDone notification is sent we can ignore doing so here. - if err := UpdateRunWithoutNotification(ctx, run, "status", "started", "stopped"); err != nil { - return 0, fmt.Errorf("update run %d: %w", run.ID, err) + if updateRequired { + // As the caller has to ensure the ActionRunNowDone notification is sent we can ignore doing so here. + if err := UpdateRunWithoutNotification(ctx, run, "status", "started", "stopped"); err != nil && errors.Is(err, ErrActionRunOutOfDate) { + // Retry update; another session affected `run` simultaneously. It wasn't necessarily another update + // from this same loop -- there are other codepaths that update `ActionRun`. + log.Debug("UpdateRunWithoutNotification failed with %v; looping for retry", err) + continue + } else if err != nil { + return 0, fmt.Errorf("update run %d: %w", run.ID, err) + } } + break // exit retry loop } return affected, nil } -func AggregateJobStatus(jobs []*ActionRunJob) Status { +var AggregateJobStatus = func(jobs []*ActionRunJob) Status { allSuccessOrSkipped := len(jobs) != 0 allSkipped := len(jobs) != 0 var hasFailure, hasCancelled, hasWaiting, hasRunning, hasBlocked bool diff --git a/models/actions/run_job_test.go b/models/actions/run_job_test.go index 7c204d4eb9..caaaae2cf8 100644 --- a/models/actions/run_job_test.go +++ b/models/actions/run_job_test.go @@ -8,6 +8,7 @@ import ( "forgejo.org/models/db" "forgejo.org/models/unittest" + "forgejo.org/modules/test" "code.forgejo.org/forgejo/runner/v12/act/jobparser" "github.com/stretchr/testify/assert" @@ -281,3 +282,60 @@ func TestActionRunJob_HasIncompleteWith(t *testing.T) { }) } } + +func TestUpdateRunJobWithoutNotificationConcurrency(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + + testJob := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: 192}) + testRun := unittest.AssertExistsAndLoadBean(t, &ActionRun{ID: testJob.RunID}) + + // UpdateRunJobWithoutNotification is intended to update the related `ActionRun`, setting its `Started`, `Stopped`, + // and `Status` field to an appropriate state considering the job update. It has a retry loop to perform this work + // even if `ActionRun` is updated concurrently. To test that loop, we're going to intercept the invocation of + // AggregateJobStatus and freeze that update process, perform a different modification to the run, and then release + // the frozen test. The retry loop should trigger and a second pass updating the `ActionRun` should succeed. + + syncBeginPoint := make(chan any) + syncMidPoint := make(chan any) + syncEndPoint := make(chan any) + firstPass := true + + defer test.MockVariableValue(&AggregateJobStatus, func(jobs []*ActionRunJob) Status { + // Synchronization here needs to handle the faact that `AggregateJobStatus` will be invoked twice -- pause + // correctly on the first run, but continue with no concerns on the second run. + if firstPass { + firstPass = false + // Signal that we're in AggregateJobStatus()... + close(syncBeginPoint) + // Wait until signalled to continue + <-syncMidPoint + } + return StatusCancelled + })() + + go func() { + testJob.Status = StatusCancelled + updated, err := UpdateRunJobWithoutNotification(t.Context(), testJob, nil, "status") + close(syncEndPoint) // close before asserts, so that the test doesn't hang if it fails + require.NoError(t, err) + assert.EqualValues(t, 1, updated) + }() + + // Wait until UpdateRunJobWithoutNotification reaches AggregateJobStatus()... + <-syncBeginPoint + + // Perform a concurrent modification to `ActionRun` + testRun.Status = StatusSkipped + err := UpdateRunWithoutNotification(t.Context(), testRun, "status") + require.NoError(t, err) + + // Signal for AggregateJobStatus to continue + close(syncMidPoint) + + // Wait for goroutine to complete + <-syncEndPoint + + // Reload the `ActionRun` + testRun = unittest.AssertExistsAndLoadBean(t, &ActionRun{ID: testJob.RunID}) + assert.Equal(t, StatusCancelled, testRun.Status) +}