From eab225f095ef988ba5673450dd21b5d8582f0644 Mon Sep 17 00:00:00 2001 From: Zettat123 Date: Thu, 6 Aug 2026 22:54:08 -0600 Subject: [PATCH] fix(actions): allow cancelling runs without running jobs (#35842) (#38812) --- models/actions/run_job.go | 122 +++++++++++++++++++------------ models/actions/run_job_test.go | 89 ++++++++++++++++++++++ routers/web/repo/actions/view.go | 12 ++- 3 files changed, 173 insertions(+), 50 deletions(-) diff --git a/models/actions/run_job.go b/models/actions/run_job.go index f00c5da51b0..9d396323928 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -448,58 +448,74 @@ func UpdateRunJob(ctx context.Context, job *ActionRunJob, cond builder.Cond, col return affected, RefreshReusableCallerStatus(ctx, parent) } - { - // Other goroutines may aggregate the status of the attempt/run and update it too. - // So we need to load the current jobs before updating the aggregate state. - if job.RunAttemptID > 0 { - attempt, err := GetRunAttemptByRepoAndID(ctx, job.RepoID, job.RunAttemptID) - if err != nil { - return 0, err - } - jobs, err := GetRunJobsByRunAndAttemptID(ctx, job.RunID, job.RunAttemptID) - if err != nil { - return 0, err - } - attempt.Status = AggregateJobStatus(jobs) - if attempt.Started.IsZero() && attempt.Status.IsRunning() { - attempt.Started = timeutil.TimeStampNow() - } - if attempt.Stopped.IsZero() && attempt.Status.IsDone() { - attempt.Stopped = timeutil.TimeStampNow() - } - if err := UpdateRunAttempt(ctx, attempt, "status", "started", "stopped"); err != nil { - return 0, fmt.Errorf("update run attempt %d: %w", attempt.ID, err) - } - } else { - // TODO: Remove this fallback in the future. - // Legacy fallback: jobs created before migration v331 have RunAttemptID=0 and are NOT backfilled. - // This path keeps those runs' status consistent when their jobs finish, including: - // - jobs created before migration v331 and complete on the new version starts - // - zombie/abandoned cleanup cron tasks that call UpdateRunJob on legacy jobs - run, err := GetRunByRepoAndID(ctx, job.RepoID, job.RunID) - if err != nil { - return 0, err - } - jobs, err := GetLatestAttemptJobsByRepoAndRunID(ctx, job.RepoID, job.RunID) - if err != nil { - return 0, err - } - run.Status = AggregateJobStatus(jobs) - if run.Started.IsZero() && run.Status.IsRunning() { - run.Started = timeutil.TimeStampNow() - } - if run.Stopped.IsZero() && run.Status.IsDone() { - run.Stopped = timeutil.TimeStampNow() - } - if err := UpdateRun(ctx, run, "status", "started", "stopped"); err != nil { - return 0, fmt.Errorf("update run %d: %w", run.ID, err) - } - } + if err := refreshRunStatus(ctx, job.RepoID, job.RunID, job.RunAttemptID, StatusUnknown); err != nil { + return 0, err } return affected, nil } +// refreshRunStatus recomputes the status of an attempt from the jobs currently stored and persists it. +// The latest attempt propagates its status to its run, an older one only updates itself. +// noJobsStatus settles an attempt without any job, which AggregateJobStatus cannot conclude on its own. +func refreshRunStatus(ctx context.Context, repoID, runID, runAttemptID int64, noJobsStatus Status) error { + // Other goroutines may aggregate the status of the attempt/run and update it too. + // So we need to load the current jobs before updating the aggregate state. + if runAttemptID > 0 { + attempt, err := GetRunAttemptByRepoAndID(ctx, repoID, runAttemptID) + if err != nil { + return err + } + jobs, err := GetRunJobsByRunAndAttemptID(ctx, runID, runAttemptID) + if err != nil { + return err + } + attempt.Status = AggregateJobStatus(jobs) + if len(jobs) == 0 { + attempt.Status = noJobsStatus + } + if attempt.Started.IsZero() && attempt.Status.IsRunning() { + attempt.Started = timeutil.TimeStampNow() + } + if attempt.Stopped.IsZero() && attempt.Status.IsDone() { + attempt.Stopped = timeutil.TimeStampNow() + } + if err := UpdateRunAttempt(ctx, attempt, "status", "started", "stopped"); err != nil { + return fmt.Errorf("update run attempt %d: %w", attempt.ID, err) + } + return nil + } + + // TODO: Remove this fallback in the future. + // Legacy fallback: jobs created before migration v331 have RunAttemptID=0 and are NOT backfilled. + // This path keeps those runs' status consistent when their jobs finish, including: + // - jobs created before migration v331 and complete on the new version starts + // - zombie/abandoned cleanup cron tasks that call UpdateRunJob on legacy jobs + // - cancelling a legacy run whose jobs are all already done + run, err := GetRunByRepoAndID(ctx, repoID, runID) + if err != nil { + return err + } + jobs, err := GetLatestAttemptJobsByRepoAndRunID(ctx, repoID, runID) + if err != nil { + return err + } + run.Status = AggregateJobStatus(jobs) + if len(jobs) == 0 { + run.Status = noJobsStatus + } + if run.Started.IsZero() && run.Status.IsRunning() { + run.Started = timeutil.TimeStampNow() + } + if run.Stopped.IsZero() && run.Status.IsDone() { + run.Stopped = timeutil.TimeStampNow() + } + if err := UpdateRun(ctx, run, "status", "started", "stopped"); err != nil { + return fmt.Errorf("update run %d: %w", run.ID, err) + } + return nil +} + // RefreshReusableCallerStatus recomputes a reusable workflow caller's Status, Started and Stopped from its current direct children and persists the change. // No-op if caller is not a reusable caller. // @@ -660,6 +676,8 @@ func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) return CancelJobs(ctx, jobsToCancel) } +// CancelJobs cancels every cancellable job it is given. It leaves the status of a run it +// cancelled nothing in untouched, SettleRunAfterCancel is what gives such a run a final one. func CancelJobs(ctx context.Context, jobs []*ActionRunJob) ([]*ActionRunJob, error) { cancelledJobs := make([]*ActionRunJob, 0, len(jobs)) @@ -684,6 +702,16 @@ func CancelJobs(ctx context.Context, jobs []*ActionRunJob) ([]*ActionRunJob, err return cancelledJobs, nil } +// SettleRunAfterCancel gives a run a final status when cancelling it updated no job at all. +// A run's status is otherwise only ever written as a side effect of a job update, so a run whose +// jobs are all done already, or that has no job at all, would stay unfinished forever. +func SettleRunAfterCancel(ctx context.Context, run *ActionRun) error { + if run.Status.IsDone() { + return nil + } + return refreshRunStatus(ctx, run.RepoID, run.ID, run.LatestAttemptID, StatusCancelled) +} + // cancelOneJob cancels a single job and returns the post-cancel row func cancelOneJob(ctx context.Context, job *ActionRunJob) (*ActionRunJob, error) { if job.Status.IsDone() { diff --git a/models/actions/run_job_test.go b/models/actions/run_job_test.go index 4437b5906df..34e32b5ecd0 100644 --- a/models/actions/run_job_test.go +++ b/models/actions/run_job_test.go @@ -8,6 +8,7 @@ import ( "gitea.dev/models/db" "gitea.dev/models/unittest" + "gitea.dev/modules/timeutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -197,3 +198,91 @@ func TestCancelJobs_NestedBlockedReusableCaller(t *testing.T) { gotRun := unittest.AssertExistsAndLoadBean(t, &ActionRun{ID: run.ID}) assert.Equal(t, StatusCancelled, gotRun.Status, "run must aggregate to Cancelled, not stay Blocked") } + +func TestSettleRunAfterCancel(t *testing.T) { + // A run that cancelling updates no job in, because its jobs all reached a final status already + // or because it has none at all. Its own row has to be settled explicitly, or the run can never + // finish and can never be deleted either. + + newStuckRun := func(t *testing.T, withAttempt, withJob bool) (*ActionRun, []*ActionRunJob) { + t.Helper() + ctx := t.Context() + + run := &ActionRun{ + Title: "stuck-waiting", + RepoID: 4, + Index: 9801, + OwnerID: 1, + WorkflowID: "test.yaml", + TriggerUserID: 1, + Ref: "refs/heads/master", + CommitSHA: "c2d72f548424103f01ee1dc02889c1e2bff816b0", + Event: "push", + TriggerEvent: "push", + EventPayload: "{}", + Status: StatusWaiting, + } + require.NoError(t, db.Insert(ctx, run)) + + var runAttemptID int64 + if withAttempt { + attempt := &ActionRunAttempt{RepoID: run.RepoID, RunID: run.ID, Attempt: 1, TriggerUserID: 1, Status: StatusWaiting} + require.NoError(t, db.Insert(ctx, attempt)) + run.LatestAttemptID = attempt.ID + require.NoError(t, UpdateRun(ctx, run, "latest_attempt_id")) + runAttemptID = attempt.ID + } + + if !withJob { + return run, nil + } + job := &ActionRunJob{ + RunID: run.ID, + RunAttemptID: runAttemptID, + RepoID: run.RepoID, + OwnerID: run.OwnerID, + CommitSHA: run.CommitSHA, + Name: "job1", + JobID: "job1", + Attempt: 1, + Status: StatusSuccess, + Stopped: timeutil.TimeStampNow(), + } + require.NoError(t, db.Insert(ctx, job)) + return run, []*ActionRunJob{job} + } + + cases := []struct { + name string + withAttempt bool + withJob bool + want Status + }{ + {"done job", true, true, StatusSuccess}, + // Runs created before migration v331 have no attempt, their status lives on the run row itself. + {"done job on a legacy run without attempt", false, true, StatusSuccess}, + // Aggregation cannot reach a final status without any job, so cancelling has to end the run itself. + {"no job at all", true, false, StatusCancelled}, + {"no job at all on a legacy run without attempt", false, false, StatusCancelled}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + run, jobs := newStuckRun(t, tc.withAttempt, tc.withJob) + + // mirrors what the CancelRun service does + cancelled, err := CancelJobs(t.Context(), jobs) + require.NoError(t, err) + assert.Empty(t, cancelled, "nothing is cancellable, so the run row has to be settled explicitly") + require.NoError(t, SettleRunAfterCancel(t.Context(), run)) + + if tc.withAttempt { + gotAttempt := unittest.AssertExistsAndLoadBean(t, &ActionRunAttempt{ID: run.LatestAttemptID}) + assert.Equal(t, tc.want, gotAttempt.Status) + } + gotRun := unittest.AssertExistsAndLoadBean(t, &ActionRun{ID: run.ID}) + assert.Equal(t, tc.want, gotRun.Status) + assert.NotZero(t, gotRun.Stopped) + }) + } +} diff --git a/routers/web/repo/actions/view.go b/routers/web/repo/actions/view.go index 61101967529..7426a45c8b3 100644 --- a/routers/web/repo/actions/view.go +++ b/routers/web/repo/actions/view.go @@ -1063,7 +1063,10 @@ func Cancel(ctx *context_module.Context) { return fmt.Errorf("cancel jobs: %w", err) } updatedJobs = append(updatedJobs, cancelledJobs...) - return nil + if len(updatedJobs) > 0 { + return nil // a job update already refreshed the run + } + return actions_model.SettleRunAfterCancel(ctx, run) }); err != nil { ctx.ServerError("StopTask", err) return @@ -1073,8 +1076,11 @@ func Cancel(ctx *context_module.Context) { actions_service.EmitJobsIfReadyByJobs(updatedJobs) actions_service.NotifyWorkflowJobsStatusUpdate(ctx, updatedJobs...) - if len(updatedJobs) > 0 { - actions_service.NotifyWorkflowRunStatusUpdateWithReload(ctx, run.RepoID, run.ID) + // SettleRunAfterCancel finishes a run without updating any job, so compare the run itself. + if reloaded, err := actions_model.GetRunByRepoAndID(ctx, run.RepoID, run.ID); err != nil { + log.Error("GetRunByRepoAndID: %v", err) + } else if len(updatedJobs) > 0 || reloaded.Status != run.Status { + actions_service.NotifyWorkflowRunStatusUpdate(ctx, reloaded) } ctx.JSONOK() }